feat(pull): route interactive producers onto the durable queue on pull pilots (#3114)

docker/base-image/agent_server/services/pull_worker.py

@@ -389,6 +389,7 @@ async def _run_and_report(
     ``persist_session``), matching the SESSION_TAB ``--resume`` convention — the
     minimal reversible reading of the schema's "presence ⇒ persist/resume"."""
     from ..services.runtime_adapter import get_runtime
+    from .process_registry import get_process_registry
 
     execution_id = claim.get("execution_id")
     claim_token = claim.get("claim_token")
@@ -398,9 +399,25 @@ async def _run_and_report(
     session_id = payload.get("session_id")
     overrides = payload.get("task_overrides") or {}
     turn_timeout = _resolve_turn_timeout(overrides)
-
+    # #3114: a turn queued by the backend's `execute_task` carries the push
+    # payload's `persist_session` and `images`. Older rows carry neither, so
+    # persistence falls back to "a session id means resume and persist".
+    persist_session = payload.get("persist_session")
+    if persist_session is None:
+        persist_session = bool(session_id)
+
+    # #3114: a claimed turn is ACCEPTED-not-spawned until the runtime starts it.
+    # Registered as pending, `/api/executions/{id}/terminate` reaches it in that
+    # window (#2433): the cancel is recorded, the turn is skipped (or killed at
+    # spawn) and reported `cancelled` below.
+    registry = get_process_registry()
+    registry.register_pending(
+        execution_id, timeout_seconds=turn_timeout, metadata={"type": "pull"},
+    )
     agent_state.record_task_start()
     try:
+        if execution_id and registry.was_terminated(execution_id):
+            raise HTTPException(status_code=409, detail="Execution cancelled before it started")
         response_text, raw_messages, metadata, ran_session_id = await get_runtime().execute_headless(
             prompt=message,
             model=overrides.get("model"),
@@ -410,7 +427,8 @@ async def _run_and_report(
             max_turns=overrides.get("max_turns"),
             execution_id=execution_id,
             resume_session_id=session_id,
-            persist_session=bool(session_id),
+            persist_session=bool(persist_session),
+            images=payload.get("images") or None,
         )
         body = _success_result_body(claim_token, response_text, raw_messages, metadata, ran_session_id)
         finish_success: Optional[bool] = True
@@ -424,6 +442,25 @@ async def _run_and_report(
     except Exception as exc:  # noqa: BLE001 — any failure must still report a terminal
         finish_success = False
         body = _failed_result_body(claim_token, str(exc) or type(exc).__name__)
+    finally:
+        # #2433: promoted at spawn; dropped here if it never spawned.
+        registry.discard_pending(execution_id)
+
+    # #679 / #3114: a terminated turn reports `cancelled` (result_callback
+    # parity). Auth and rate-limit terminals keep their class so the breaker
+    # and SUB-003 still see them.
+    if (
+        execution_id
+        and registry.was_terminated(execution_id)
+        and body.get("error_code") not in ("auth", "billing")
+    ):
+        body = {
+            **body,
+            "status": "cancelled",
+            "error_code": None,
+            "error": "Execution terminated by user",
+        }
+        finish_success = None
 
     agent_state.record_task_finish(success=finish_success)
 

docs/memory/architecture/execution.md

@@ -154,8 +154,9 @@ The last blocking dispatch callers, moved onto execution terminals so they work
 
 - **Fan-out.** `FanOutService` rebuilds the sync aggregate from the batch's `schedule_executions` rows (`fan_out_id` + `fan_out_task_id`), not from `execute_task` return values. Each row is created **at slot grant** inside the `max_concurrency` semaphore — never up front: a pre-created `RUNNING` row waiting behind the semaphore matches the #106 no-session sweep, the watchdog orphan reconcile and the stale sweep (all admission-anchored, #2433/#2435), and a `QUEUED` one is claimable by `claim_next_queued` while the service also dispatches it. The sync caller waits for the (shielded) dispatch, then `sync_waiter.wait_for_fan_out_batch`; with no caller deadline the budget is `ceil(N / min(max_concurrency, max_parallel_tasks)) × execution_timeout + 120s`. `async_mode` returns a receipt; the not-yet-dispatched tail is in-process and does not survive a restart.
 - **The join** (`fan_out_service.join_fan_out_on_terminal`) and the #2523 loop advance hang off `event_dispatch_service.spawn_task_terminal_event` via `_terminal_side_effects`, each under its own guard — outside the subscriber-gated `emit_task_terminal_event`. One PK read per fleet terminal; a `(fan_out_id, status)`-indexed COUNT only for fan-out rows.
-- **Sync edge adapter** (`task_execution_service.dispatch_and_await_terminal` + `result_from_execution_row`): `execute_task`, and on a `QUEUED` return wait for that row's terminal via `sync_waiter.wait_for_sync_terminal` and rebuild the `TaskExecutionResult` from the row. Used by `routers/a2a._run_a2a_task` and `operator_resume_service`. A timed-out wait returns `FAILED/TIMEOUT` without writing a terminal. The pull sink does not signal that registry, so the wake is the 5s DB poll (≤5s added tail latency on a pilot agent).
+- **Sync edge adapter** (`task_execution_service.dispatch_and_await_terminal` + `result_from_execution_row`): `execute_task`, and on a `QUEUED` return wait for that row's terminal via `sync_waiter.wait_for_sync_terminal` and rebuild the `TaskExecutionResult` from the row. Used by `routers/a2a._run_a2a_task`, `operator_resume_service` and, since #3114, every sync interactive caller (below). A timed-out wait returns `FAILED/TIMEOUT` ("Execution timed out after …") without writing a terminal. The pull sink signals the in-process waiter when its CAS write wins; a caller on another uvicorn worker wakes on the 5s DB poll.
 - `PULL_REACHABLE_TRIGGERS` now equals `_AUTONOMOUS_TRIGGERS` but stays an enumerated allow-list (pinned by `test_2048_pull_pilot_reach.py`).
+- **Interactive turns on a pilot (#3114).** `pull_owns_dispatch` also returns True for `pull_pilot.PULL_REACHABLE_NON_AUTONOMOUS` = `INTERACTIVE_TRIGGERS - {"chat"} | {"validation"}`; the UI `/chat` path (`dispatch_admission_service` → `run_chat_turn`, trigger `chat`) still pushes. Sync callers — `session_turn_service.run_resumable_turn` (Session tab + Workspace), `public_chat_service`, `adapters/message_router`, `shared_sessions` rooms, `routers/paid`, `mcp_auth_service`, `validation_service`, internal sync run-now — call `dispatch_and_await_terminal` with a `conversation_key` (`session:`/`public:`/`channel:`/`room:`/`paid:`). For interactive and validation triggers the adapter first waits for a claim, bounded by `claim_budget` (default one agent execution timeout; rooms pass 300s); on expiry it stores the row FAILED (same queued-only CAS as `cancel_queued_execution`) and returns FAILED/`CAPACITY`, a caller that went away leaves it CANCELLED, and a lost CAS (just claimed) falls through to the terminal wait. The pull sink spawns `channel_completion_report` on its CAS-won branch like the push terminals; the agent's pull worker registers a claimed turn as pending so a terminate before spawn reports `cancelled`. Room wakes of one agent serialise on a per-(room, agent) `ResumeLock` (`room_wake_lock:`), re-entrant within one mention chain. Images ride `backlog_metadata`. `result_from_execution_row` rebuilds a push-shaped result (`[code]` error prefix → `error_code`, compact events in `raw_response`, dispatch sentinels never a session id). The Session `ResumeLock` and the Workspace in-flight markers stay and span the queued wait: their TTLs add `pull_pilot.pull_queue_allowance` (one agent timeout on a pilot, 0 elsewhere). The live-stream proxies hold while the row is queued (`sync_waiter.wait_while_queued`). Pinned by `test_3114_pull_route_interactive.py` and `test_3114_pull_route_workspace.py`.
 - **Pull claim order (#2842, #2843).** `claim_next_task` passes `pull_pilot.INTERACTIVE_TRIGGERS` into `claim_next_queued`: interactive rows first, oldest first within each group, no anti-starvation rule. The pull claim also skips a row whose `conversation_key` has a `running` row for the agent; the partial unique index `idx_executions_one_running_turn` (agent, conversation_key WHERE running) turns a concurrent second claim into an `IntegrityError` that the claim retries. The guard is the row's status, so terminal, lease expiry and requeue all release it. `backlog_service.enqueue` stamps the key (`chat_session_id`, else `resume_session_id`) on pilot agents only. The push drain calls `claim_next_queued(agent_name)` with neither, so its order is plain oldest-first. Pinned by `test_2842_2843_pull_claim_order.py`.
 
 ### Agent Self-Reminders (#1296)
@@ -177,7 +178,7 @@ Durable one-shot deferred self-trigger — the time-deferred sibling of loops (
 | `routers/sessions.py` | `agent_sessions` / `agent_session_messages` | platform user |
 | `client_portal/service.py` (Workspace) | `enterprise_portal_sessions` / `_messages` | verified client email |
 
-`cached_claude_session_id` is the load-bearing field on both tables. The engine: runtime gate (drop the cached id for a runtime without `--resume`, i.e. Codex) → resume lock → `execute_task(persist_session=True, resume_session_id=…)` → **one** cold retry when Claude reports the JSONL missing. Callers pass `on_resume_failure` (clear the cache, count the failure, inside the lock) and may pass a distinct `cold_message` for that retry. `ResumeLockBusy` subclasses `HTTPException(429)` so the Session router surfaces it unchanged while the Workspace catches the precise type and re-raises `ClientPortalError(429)`.
+`cached_claude_session_id` is the load-bearing field on both tables. The engine: runtime gate (drop the cached id for a runtime without `--resume`, i.e. Codex) → resume lock → `dispatch_and_await_terminal(persist_session=True, resume_session_id=…, conversation_key="session:<key>")` (plain `execute_task` off a pull pilot; on a pilot the lock is held across the queued wait, #3114) → **one** cold retry when Claude reports the JSONL missing. Callers pass `on_resume_failure` (clear the cache, count the failure, inside the lock) and may pass a distinct `cold_message` for that retry. `ResumeLockBusy` subclasses `HTTPException(429)` so the Session router surfaces it unchanged while the Workspace catches the precise type and re-raises `ClientPortalError(429)`.
 
 **Workspace absorbed the Session surface (ent#358).** Agent Detail no longer renders a Session surface: `SessionPanel.vue` and the Session-mode toggle are gone, the Chat tab is stateless-only (plus a "Continue in Workspace" link), and `?tab=session` **redirects** to `/workspace?agent=<name>` (query-preserving, `router.replace`, guarded first in both `onMounted` and `onActivated` since AgentDetail is KeepAlive-cached). The removal was gated on continuity parity, not on streaming — the Session surface never streamed (synchronous POST + reattach poller, #1376/#759), so Workspace streaming (ent#286) was **not** a prerequisite. What *was* a prerequisite: Workspace chat ran stateless with a history prefix, so it moved onto this engine first. `agent_sessions` rows, endpoints and store stay readable; only the entry point went away. Since ent#456 the "Continue in Workspace" link and the NavBar entry open a NEW tab; the redirect stays same-tab.
 

docs/memory/feature-flows/persistent-task-backlog.md

@@ -205,10 +205,25 @@ ON schedule_executions(agent_name, conversation_key)
 WHERE status = 'running' AND conversation_key IS NOT NULL;
 ```
 
-`backlog_service.enqueue` stamps `conversation_key` (`chat_session_id`, else
-`resume_session_id`) on pilot agents only. The ordering expression cannot use
+`backlog_service.enqueue` stamps `conversation_key` on pilot agents only: the
+explicit key `execute_task` was given (`session:<key>`, `public:<id>`,
+`channel:<id>`, `room:<id>`, `paid:<id>`, #3114), else `chat_session_id`, else
+`resume_session_id`. The ordering expression cannot use
 `idx_executions_queued`; per-agent queues are small.
 
+**Interactive turns on a pilot (#3114).** `pull_pilot.pull_owns_dispatch`
+covers every interactive trigger except `chat` (the UI `/chat` path, which
+still pushes), plus `validation`. Their sync callers go through
+`task_execution_service.dispatch_and_await_terminal`: phase 1 polls the row
+while it is `queued`, for at most one agent execution timeout, and on expiry
+cancels it (`cancel_queued_execution`) and returns FAILED/`CAPACITY`; a cancel
+that loses to a claim falls through to phase 2, the terminal wait (agent
+timeout + 120s from the claim). The pull sink signals the in-process waiter
+when its CAS write wins. The live-stream proxies (`routers/chat.py`,
+`routers/public.py`, `client_portal/router.py`) hold the SSE connection with
+`: queued` comments while the row is queued (`sync_waiter.wait_while_queued`),
+then retry an agent 404 briefly while the claimed row is `running`.
+
 ### backlog_metadata JSON Shape
 
 Captured at enqueue time, replayed on drain. No credential values, only
@@ -237,10 +252,20 @@ identity and request parameters:
   "triggered_by": "manual",
   "collaboration_activity_id": null,
   "is_self_task": false,
-  "self_task_activity_id": null
+  "self_task_activity_id": null,
+  "persist_session": false,
+  "schedule_context": null,
+  "attempt": null,
+  "images": null
 }
 ```
 
+`persist_session`, `schedule_context`, `attempt` and `images` (#3114) are read
+by the pull claim only: the envelope hands `persist_session` and `images`
+(base64 vision input) to the worker, and the claim-time prompt uses
+`schedule_context` and `attempt`. Canary G-04 skips `images` when scanning for
+credentials.
+
 ## Lifecycle
 
 ```

docs/planning/PULL_MIGRATION_STATUS.md

@@ -38,7 +38,8 @@ Dispatch topology, not policy. `pull_pilot.PULL_REACHABLE_TRIGGERS` is the sourc
 |---|---|---|
 | **On `dev` today** | `agent`, `event`, `schedule`, `webhook`, `reminder`, `loop` | 6 of 9 |
 | **Adds with Phase 4** | `fan_out`, `a2a`, `operator_response` | → 9 of 9 |
-| **Pending** | interactive chat / Session-tab turns | claim order (#2842) and conversation guard (#2843) are in; producers not yet routed (decided 2026-09-16, #1989) |
+| **Interactive (#3114)** | `manual`, `mcp`, `session`, `public`, `voice`, `voip`, `room`, `user`, `paid`, `slack`, `telegram`, `whatsapp`, plus `validation` (`pull_pilot.PULL_REACHABLE_NON_AUTONOMOUS`) | routed on pilots; sync callers wait through `dispatch_and_await_terminal` |
+| **Pending** | `chat` — the UI `/chat` path (`dispatch_admission_service` → `chat_execution_service.run_chat_turn`) | still pushes on pilots; routed in a later change |
 | **Unreached, unclassified** | `retry` | in neither trigger set — #2845. 4.7% of `eu2` traffic; looks like an oversight, not a decision |
 
 `schedule` / `webhook` / `reminder` landed with #2391; `loop` with #2523. Before #2391 the pilot flag was
@@ -62,10 +63,12 @@ The spec names the gates (`TARGET_ARCHITECTURE.md`, §Re-Delivery and Side-Effec
    and logged. Pilots must run a base image with #2392, or every effect call whose model omits the id is
    refused. *(Trace fidelity #548/#333 closed Aug/Jun; `prior_trace` injection #1401 closed 2026-07-08;
    #1402 closed 2026-07-26.)*
-3. **A soak on an agent that actually emits.** The current pilot (`cornelius-oracle` on eu2) emits no
-   messages, calls or shares — measured 2026-09-02: 219 `idempotency_keys` rows, all `agent:*`, zero
-   `effect:*`. It has therefore never entered the code path item 2 protects, so a clean window on it is not
-   evidence about that gate. Mechanics for a second, disposable emitting pilot are in the ops repo
+3. **A soak on an agent that actually emits.** eu2 runs three pilot arms: `cornelius-oracle` and
+   `oracle-6-science` emit no messages, calls or shares; `oracle-7-ai-semi` (flipped 2026-09-15) is the
+   emitting arm, with a 2-hourly schedule that writes a byte-stable file to its public folder and calls
+   `share_file`, so a re-delivered execution must emit once. What the gate still needs is the pulled-run rate on
+   that arm: effect rows per pulled execution, with no duplicate share on a re-delivery. All three arms share
+   one subscription seat, so a quota blackout darkens every arm at once. Mechanics are in the ops repo
    (`trinity-ops-agent:docs/pull-soak-eu2.md`). System of record for the soak is
    [#1766](https://github.com/abilityai/trinity/issues/1766)'s comment thread — read it before measuring.
 4. **Phase 5: flip default-ON and delete the legacy machinery** — the 9-path cleanup pyramid, the slot ZSET,
@@ -84,9 +87,30 @@ The spec names the gates (`TARGET_ARCHITECTURE.md`, §Re-Delivery and Side-Effec
    row whose `conversation_key` already has a `running` row is skipped, and the unique index
    `idx_executions_one_running_turn` stops two concurrent claimers. **No worker is reserved** for interactive
    arrivals: one of N held idle is 33% of a 3-worker agent. Multi-container affinity (`replica_count > 1`,
-   #927) is out of scope. The claim guard is the one mechanism for pulled turns: when session turns are routed
-   onto the queue, `session_turn_service`'s Redis `ResumeLock` is removed from that path, not kept beside it. Next: route the interactive producers onto the queue. Until then they still run on
-   the old path — a **migration state, not a design boundary**.
+   #927) is out of scope.
+
+   **Routed on pilots (#3114)** for every interactive trigger except `chat` (the UI `/chat` path, still
+   pushed). Sync callers (Session tab, Workspace, public links, channels, rooms, paid, MCP key auth,
+   validation, internal run-now) go through `task_execution_service.dispatch_and_await_terminal`: it waits up
+   to one agent execution timeout for a worker to claim the row (rooms: `ROOM_CLAIM_BUDGET_SECONDS`, 300s,
+   via `claim_budget`), stores it FAILED/`CAPACITY` if none does (a caller that went away leaves it
+   CANCELLED), then waits for the terminal. Each producer passes a `conversation_key` (`session:`, `public:`, `channel:`,
+   `room:`, `paid:`), which is what the claim guard serialises on. Images ride the queue in
+   `backlog_metadata`, which the #1449 retention sweep NULLs once the row is terminal. Live-stream proxies hold the SSE connection
+   while the row is queued.
+
+   **The Session-tab `ResumeLock` and the Workspace in-flight markers are retained** and span the queued
+   wait: their TTLs add `pull_pilot.pull_queue_allowance` (one agent timeout on a pilot). The lock also
+   keeps the cached resume id fresh across turns, which the claim guard does not do. Rooms take a
+   per-(room, agent) wake lock (`ResumeLock` under `room_wake_lock:<room>:<agent>`) from the cursor read
+   through the posted reply, so a second wake reads what the first left; a mention chain back to the same
+   agent re-enters it. The room working marker adds the queue allowance (capped at the room claim budget).
+   The claim guard is the only serialisation for public links and channels.
+
+   **Report-back and stop.** The pull sink (`apply_task_result`) spawns the channel / portal / room
+   completion report on its CAS-won branch, as the push terminals do. The agent's pull worker registers a
+   claimed turn as pending in the process registry, so `/api/executions/{id}/terminate` before spawn skips
+   the turn and reports it `cancelled`.
 
 ### The soak duration requirement is mis-cited — correct it when you touch it