feat(pull): let scheduled work reach the durable queue (#2391)

[obasilakis]

Fixes #2391. Option 2, deferred by #2048. Parent: #1081.

What was broken

task_execution_service — the producer behind the scheduler, i.e. all cron, plus webhooks, reminders, loops and fan-out — dispatched with overflow_policy="reject" unconditionally. capacity_manager.acquire only offers a row to the durable queue when its producer passed "queue_persistent", so PULL_MODE_PILOT_AGENTS was structurally inert for the fleet's dominant traffic class no matter how it was set. Pull carried agent-to-agent traffic only, which is why exactly one agent on eu2 had the volume to serve as a soak pilot.

What changed

build_pull_queue_payload returns a PersistentTaskPayload when — and only when — pull_owns_dispatch says a pilot owns the trigger. That selects overflow_policy="queue_persistent"; the row is enqueued, execute_task returns QUEUED before any activity, agent call or dispatch marker, and the agent's own worker claims it via GET /api/internal/next-task.

PULL_REACHABLE_TRIGGERS: {agent, event} → {agent, event, schedule, webhook, reminder}.

AC 1 — the capacity-pressure decision

The gate is pull_owns_dispatch and nothing else.

• Flag OFF (every agent, by default): byte-for-byte unchanged. Policy stays reject; a scheduled fire arriving at capacity still fails fast with Agent at capacity (N/N parallel tasks running). Pinned by a test that reads the real acquire call's kwargs, not by assertion in prose.
• Flag ON: rejection stops existing for that agent's pullable triggers — acquire's pull_exclusive branch never offers a slot, so "at capacity" is not a state that can reject. Capacity becomes physical (feat: pull / work-stealing coordination (push→pull dispatch) #1081 Phase 5, pilot-scoped). Backpressure moves to agent_ownership.max_backlog_depth; a full backlog still raises CapacityFull, and the row's error now names the backlog rather than parallel slots.

The issue framed the risk as changing scheduled semantics "whether or not pull is enabled for that agent". That is true of an unconditional queue_persistent on this producer — which is what #2048 declined to bundle, and which this does not do. Gating the policy on the predicate that already decides pull ownership keeps the blast radius exactly equal to the pilot allowlist.

Operational note for the soak: cron fires on a fixed cadence regardless of whether the agent is keeping up, so a slow cron pilot accumulates queue depth that a reject policy used to discard. Watch max_backlog_depth (M8).

AC 4 — #1083 fire-and-forget

They do not stack, and pull wins by construction, not precedence: a pull-queued row is never dispatched, so there is no request for the agent to ACK with 202 and async_result is never sent. dispatch_async_eligible is still evaluated and its result is simply unreachable. On a non-pilot agent #1083 behaves exactly as before (pinned both ways). The two already share the machinery that matters — the eid-keyed slot lease, the lease reaper, and the claim_token-gated CAS terminal write — so the recovery story is single, not layered.

Scheduler: queued is not a terminal

_poll_execution_completion treated anything but running as an outcome. A queued row would have been published as schedule_execution_completed(status=queued), classified a failure (only success is not), and handed to _maybe_schedule_retry — duplicating work that was queued and about to run. ExecutionStatus.QUEUED + _NON_TERMINAL_POLL_STATES fix it.

⚠️ Deviation from the issue's scope, needs a call

The issue's AC lists loop and fan_out among the triggers to widen. They are not widened here, because their callers consume the result synchronously:

• loop_service renders the next iteration's template from result.response (feat: run_agent_loop MCP tool — sequential bounded task execution with configurable stop conditions #740)
• fan_out_service builds each FanOutTaskResult from the returned result

A queued row is claimed and run later and returns nothing for either to read. This is the same constraint ASYNC_DISPATCH_ELIGIBLE_TRIGGERS (#1083) already encodes in its own comment, and #1081's own phasing puts the async fan-out join at Phase 4. Widening them needs that join, not this change. a2a is stranded for the same reason (ent#157). operator_response is out by choice rather than structure — it dispatches through this same producer, but the respond endpoint records result.status as the ent#329 dispatch receipt, and queued is not the outcome that contract reports.

So AC 2 is delivered for schedule / webhook / reminder — which is all of the scheduler's traffic and the entire volume case the issue makes — and deliberately not for loop / fan_out. If Phase 4 should be pulled forward into this issue, say so and it becomes a separate change.

#2048's tripwire

Fired as designed, and was re-derived, not weakened. It now pins that this producer carries both policies — the shape a conditional widening has and an unconditional one does not — plus a new assertion that queue_persistent is reachable only through pull_owns_dispatch and the payload builder's two preconditions. That is the property #2048 could not assert and the one that actually bounds the change.

Tests

tests/unit/test_2391_scheduled_pull_reach.py (35 cases): flag-OFF unchanged (policy, payload, CapacityFull terminal, normal dispatch); flag-ON queued (policy, payload contents, QUEUED return, no agent call, no slot release, no dispatch marker, backlog-full wording); stranded triggers; interactive carve-out; builder preconditions; #1083 both directions; an end-to-end class driving the real CapacityManager + BacklogService + _build_claim_response so "it enqueued" is not mistaken for "a worker can run it" (the #2317 failure mode); scheduler poll-through-queued.

Full tests/unit green (13,613 passed). The 21 failures in test_736_a2a_outbound_*, test_ent14_registry_url_ssrf, test_ent399_ipv6_origin, test_mcp_validator reproduce identically on unmodified dev — local IPv6/DNS environment, unrelated.

Follow-up that becomes binding

#2392. This moves the fleet's dominant traffic class onto a mechanism built on re-running the same execution, and effect_guard still fails open when the agent omits the execution id. Do not run a side-effect-bearing cron pilot before it is resolved. Noted in the soak doc.

Docs

PULL_MIGRATION_TESTING.md §9 — M1 expectations, the reach table, and pilot selection (a cron-driven agent is viable now, reversing the advice #2048 added; the eu2 oracles that scored 2–5 pull_eligible score ~128 under the new query). Plus capacity-management.md, task-execution-service.md, architecture.md.

🤖 Generated with Claude Code

https://claude.ai/code/session_019Tm7UEkd4G9KQZD5oeLRSa

[vybe]

/validate-pr + structural review — APPROVE WITH NOTES

Validation: closing keyword present (Fixes #2391 → auto-promotes to status-in-dev), targets dev, 12 files, security battery clean (no keys/emails/IPs/env/credential files, no new os.getenv, no infra), docs updated (architecture + two feature flows + PULL_MIGRATION_TESTING), named regression suite test_2391_scheduled_pull_reach.py (35 cases) + registry entry. Full CI matrix green after re-running the one hung head-seed job (45-min cancel, same as two other PRs today).

Local verification on origin/dev + this branch merged (clean merge at 00b6f025): the 141 unit files touching task_execution_service / pull_pilot / pull_coordination / capacity_manager / backlog_service / scheduler → 3183 passed, 10 skipped, 0 failed.

Structural review: no critical findings. The gate is exactly pull_owns_dispatch with the two policies as literals, so non-pilot dispatch is byte-identical; the QUEUED handoff returns before mark_execution_dispatched, any activity row, and any agent POST, so #1083 cannot stack; no slot is ZADDed on the pull branch so the finally has nothing to leak; scheduler polls through queued instead of classifying it as a failed terminal + duplicate retry. AC1–AC7 covered (AC2's loop/fan_out deferral to #1081 Phase 4 is stated in the body).

Four pilot-scoped gaps, verified against the merged tree, filed as #2514 rather than blocking (all are unreachable unless the agent is in PULL_MODE_PILOT_AGENTS): (1) pulled cron turns carry no schedule_context into the platform prompt — the #2317 class; (2) RETRY-001 retries are triggered_by="retry", so they are pushed onto a pilot (breaks #1766 exclusivity on the failure path) and a poison-parked row is a retry candidate; (3) _dispatch_and_record_outcome still compares != RUNNING while its two siblings were widened; (4) no breaker-open test on the pull path.

Merging next in sequence; #2264 rebases on top of this (import-block conflict expected in task_execution_service.py, nothing in _write_terminal_and_gate/apply_result).