fix(pull): the pull sink applies post-turn delivery settings (#2329)

[obasilakis]

Summary

• On a pull pilot, a /task turn with trigger agent, mcp or manual is claimed by the agent's worker and its terminal lands in apply_task_result. That sink wrote the row and stopped. The Chat tab's /task (save_to_session: true, trigger manual) succeeded and never reached the chat session, and the collaboration activity stayed started.
• The three post-turn steps now live in one helper, run_post_turn_delivery, which both the push drain and the pull sink call. The queued request is rebuilt by one function, request_from_metadata.
• A token-gated SUCCESS write now refuses a row already SUCCESS or SKIPPED. The claim token survives the terminal write, so a worker's retried result POST won the CAS twice. With delivery in the sink, that would have saved the chat turn twice.
• The collaboration / self-task activity a queued row names is closed by the single close owner (close_execution_activity) and by the bulk close, so every terminal writer closes it: the pull sink, lease-reaper park and the watchdog. Backlog expiry (expire_stale_queued) calls no closer; the 120-minute sweep covers it.

Changes

• services/chat_execution_service.py: run_post_turn_delivery (moved out of run_async_task); sync /task keeps the chat_session_id the sink signals.
• services/backlog_service.py: request_from_metadata, used by _spawn_drain and the sink.
• services/pull_coordination_service.py: delivery spawned on the CAS-won branch when the metadata asks for it; waiter signalled after delivery in finally, or at once when there is nothing to deliver.
• db/schedules/executions.py: SUCCESS CAS under a claim token excludes SUCCESS / SKIPPED rows. A late SUCCESS still corrects FAILED.
• services/activity_service.py, db/activities.py: queued-row activities closed by exact id (queued_activity_ids), isolated from the dispatch close.
• Docs: architecture/execution.md, feature-flows/persistent-task-backlog.md, the scrub docstring, a learnings fragment, and the /cso --diff report (0 findings).

Behaviour notes

• A lease-reaper re-queue leaves the row's collaboration / self-task activity open for the re-delivered attempt. Only a row that is terminal has those activities closed.
• On push, the close owner now runs before run_async_task's own collaboration close, so its details win: execution_time_ms comes from the row's duration_ms (measured from drain/claim), and on FAILED error is the terminal's error text. The detail keys are unchanged.
• A sync /task caller woken by the DB poll (another worker, or a poll tick that beats delivery) gets no chat_session_id in its response. The turn is still saved. No current producer sends sync /task with save_to_session.

Test Plan

• cd tests && pytest unit/test_2329_pull_delivery_parity.py -v: 21 pass. They run the real sink, drain and close paths against the harness DB.
• Mutations: every fix reverted turns tests red (sink delivery 5, SUCCESS CAS 2, owner close 1, bulk close 1, sync session id 1, re-queue guard 1, applied-status guard 1).
• Full unit suite: no failure introduced. Every failure on the branch is pre-existing (it fails on origin/dev in isolation too) or depends on test order (test_ent477_definitions_endpoint.py passes 25/25 alone and alongside the new file).
• Manual, on a pull-pilot agent: send from the Chat tab and check that the reply appears in the session after the worker reports.

Fixes #2329

🤖 Generated with Claude Code

[vybe]

merge-train: not on today's train. This is not a defect. #3222 is on the train and edits the same region of src/backend/services/pull_coordination_service.py. A member-vs-member conflict in a .py file means the later PR rides the next train, off a dev that already contains the first.

Once #3222 lands, merge dev and resolve against it. Validation found nothing else to fix on this PR.

[github-actions]

⚠️ Nightly unit-suite check skipped — merge conflict against dev.

Resolve by running git merge dev locally and pushing the result. The next nightly run will re-test once the conflict is gone.

[github-actions]

⚠️ Live-instance suite skipped — merge conflict against dev.

Resolve by merging dev locally and pushing the result; the next nightly re-tests.

[AndriiPasternak31]

Reviewed #3227 (structural /review + /validate-pr). Looks right. Ready to merge once it's rebased.

What I checked:

• Mutations: I reverted each piece of the fix from a scratch copy, and each one turns its test red (7/7). The counts match your Test Plan exactly (sink 6, CAS 2, owner close 1, bulk close 1, applied-status 1, re-queue guard 1, sync session id 1). Every test runs the real code; none of them asserts on source text.
• I traced post-turn delivery across push drain, pull sink, duplicate SUCCESS, FAILED→late SUCCESS, lease-reaper park/re-queue and the dispatch_and_await_terminal callers. It runs once on each path, and only on the CAS-won branch.

Before merge:

1. Rebase. It conflicts with dev's #2514 (520c100af) in pull_coordination_service.py, and the conflict is mechanical. Keep _spawn_breaker_verdict(...), then _spawn_post_turn_delivery(execution, row_status), and drop dev's bare signal_sync_waiter(execution_id, None, None). If both signals stay, a same-worker sync /task wakes before delivery and loses its chat_session_id. With that resolution, the pull/1804/2514/3114/backlog suites pass locally on the merged tree (254 passed).
2. PR body, behaviour note 2. On push, the close owner runs first only for _write_terminal_and_gate terminals (task_execution_service.py:1135). The push SUCCESS and failed-envelope branches close by activity_id through complete_activity (:3017, :3153), so on a push SUCCESS run_async_task still closes the collaboration activity, with its own execution_time_ms.
3. Learnings fragment wording. A sequential retried POST was already answered replayed at apply_task_result's authoritative-terminal check (pull_coordination_service.py:607). The CAS clause closes the concurrent case, where both POSTs read RUNNING. Your test models exactly that; the ledger line should say so too.

Follow-ups (not this PR):

• execution.md now says every terminal writer closes the collaboration / self-task activity. expire_stale_queued, cancel_queued_for_agent, fail_queued_for_agent (db/schedules/queue.py:664/558/584) and the drain corrupt-metadata / spawn-failed writes (backlog_service.py:301/322) don't. That's pre-existing #1804-class residue; I'd narrow the sentence or file it.
• Delivery is spawned after the committed terminal, and a retry can no longer re-drive it, so a restart or exception in that window loses the chat save for good. Push has the same property, so this isn't a regression. Logging only type(e).__name__ at pull_coordination_service.py:548 makes a lost delivery hard to see, though.
• #3145 overlap. No textual overlap beyond the same #2514 hunk, and no double save: pilot /chat rows carry no save_to_session or collaboration id, so this sink only signals. But the two PRs use different delivery models. Here the sink owns delivery. In #3145 the waiting caller does, so a /chat turn whose caller hits the queued_timeout 504 is never saved even if the turn later succeeds. That's the #2329 class on the /chat leg. Worth choosing one model on purpose.

Not run: /verify-local and the manual Chat-tab check on a live pull pilot (still unchecked in the Test Plan).

[vybe]

merge-train (2026-10-06): pushed two mechanical commits to feature/2329-pull-delivery-settings so this can ride today's train:

• f9f9e8a99 merges origin/dev. The only conflict was pull_coordination_service.py against fix(pull): close the scheduler and breaker gaps on pull pilots (#2514) #3222 (fix(pull): close the pilot-scoped gaps left by #2391 — schedule context, RETRY-001 stacking, _dispatch_and_record_outcome, breaker test #2514). I kept both sides: _spawn_breaker_verdict(...) runs first, then _spawn_post_turn_delivery(execution, row_status) replaces the inline signal_sync_waiter. The 2329/2514/3114/1804/1449/2643/1081 unit files pass: 312 passed, 1 skipped.
• 0d2732d37 removes "expire" from the close-owner docstrings (db/activities.py, activity_service.py, the test docstrings). expire_stale_queued writes FAILED and calls no activity closer, so the claim was wrong. The 24h expiry is far past the 120-minute sweep, so nothing changes at runtime. Wiring expire into the closer would be your call, so I left it out.

I also changed the PR body: Fixes abilityai/trinity#2329 is now Fixes #2329, because the cross-repo form closes the issue but doesn't promote it. And the expire claim is corrected there too.

[vybe]

merge-train: batch validated on train/20261006-0833 (#3266)