diff --git a/docs/testing/PULL_MIGRATION_TESTING.md b/docs/testing/PULL_MIGRATION_TESTING.md index 6f776a869..2258ad88b 100644 --- a/docs/testing/PULL_MIGRATION_TESTING.md +++ b/docs/testing/PULL_MIGRATION_TESTING.md @@ -329,8 +329,208 @@ Ran on `gtm-synthesizer` (rebuilt image, `AGENT_AUTH_SECRET` fixed; chosen becau ## 8. Net remaining before a default-ON decision -Canary lease-awareness (E-05/E-01, §3 T3.6/T3.7) · Tier-6 `effect_guard` `execution_id` injection · ~~G3 -canary-on-PG (#1540)~~ ✅ closed · B6 runtime-verify on the rebuilt image · the ≥2-week soak (#856). +~~Canary lease-awareness E-05 (§3 T3.6)~~ ✅ **closed by #1766** — E-05 now excludes leased rows, mirroring +S-01 and the `mark_no_session_executions_failed` sweep it watches (which already carried +`lease_expires_at IS NULL`); without it E-05 fired on every pull turn past 60s · canary lease-awareness +E-01 (§3 T3.7) · Tier-6 `effect_guard` `execution_id` injection · ~~G3 canary-on-PG (#1540)~~ ✅ closed · +B6 runtime-verify on the rebuilt image · the ≥2-week soak (#856 / #1766, measurement set in §9). + +**Also closed by #1766:** the pilot flag was purely additive, so a pilot ran push AND pull concurrently — +the producer never force-queued (a free slot still meant a push, so rows only queued on overflow) and the +backend's own `drain_next` raced the agent's worker for whatever did queue. Two independent capacity +counters meant up to 2x `max_parallel_tasks`, invisible to S-02. The flag is now a true either/or for +autonomous triggers; interactive turns keep the synchronous path (Open Question 7 scope cut). + +## 9. Soak measurement set (#1766) + +The queries that turn "we ran it for a few days" into a verdict. **Nothing else +currently covers this**: `instance-monitor` has zero references to leases, +`claim_token`, `redelivery_count`, or queue depth — its probes report generic +health and will stay green whether pull carries the fleet or does nothing at all. +Run these instead (or fold M1/M3/M5 into the monitor's deep-probe). + +PostgreSQL flavour. All timestamp columns are `Text` holding ISO-Z strings +(Invariant #16), hence the explicit `::timestamptz` casts. Substitute the pilot +name for `''`. + +### Pre-flight — capture a baseline BEFORE the flip + +Run M4 + M6 + M7 on the pilot for the 7 days *preceding* the flip and keep the +output. Without a baseline, a 4% success-rate dip during the soak is +unattributable and the write-up degrades to "felt fine". + +### The set + +| ID | Answers | AC | +|----|---------|-----| +| M1 | Is pull actually carrying load? | coverage (the #1766 premise) | +| M2 | Push/pull split over time | coverage | +| M3 | Is the agent over its configured concurrency? | no slot/backlog drift | +| M4 | Lost or phantom executions | no lost/phantom | +| M5 | Re-delivery + poison-park behaviour | leases behave as designed | +| M6 | Outcome regression vs baseline | no regression | +| M7 | Canary state | canary green throughout | +| M8 | Queue starvation | no drift | + +**M1 — is the pull path carrying anything?** The single most important query: if +`pulled` is ~0 the soak is measuring nothing and everything downstream is void. + +```sql +SELECT COUNT(*) FILTER (WHERE claimed_by_worker IS NOT NULL) AS pulled, + COUNT(*) FILTER (WHERE claimed_by_worker IS NULL) AS pushed +FROM schedule_executions +WHERE agent_name = '' + AND started_at::timestamptz > now() - interval '24 hours'; +``` + +Expected after the #1766 gate: autonomous triggers ~100% `pulled`; anything still +`pushed` should be interactive (`manual`/`user`/`chat`) — confirm with: + +```sql +SELECT triggered_by, + COUNT(*) FILTER (WHERE claimed_by_worker IS NOT NULL) AS pulled, + COUNT(*) FILTER (WHERE claimed_by_worker IS NULL) AS pushed +FROM schedule_executions +WHERE agent_name = '' + AND started_at::timestamptz > now() - interval '24 hours' +GROUP BY 1 ORDER BY 1; +``` + +A pushed **autonomous** row means the producer gate did not engage — check the +backend actually has `PULL_MODE_PILOT_AGENTS` in its env. + +**M2 — split over time.** Confirms the flip took effect at the moment you think +it did, and shows drift. + +```sql +SELECT date_trunc('hour', started_at::timestamptz) AS hr, + COUNT(*) FILTER (WHERE claimed_by_worker IS NOT NULL) AS pulled, + COUNT(*) FILTER (WHERE claimed_by_worker IS NULL) AS pushed +FROM schedule_executions +WHERE agent_name = '' + AND started_at::timestamptz > now() - interval '7 days' +GROUP BY 1 ORDER BY 1; +``` + +**M3 — overbooking.** Concurrent leased rows must never exceed the agent's +`max_parallel_tasks`. This is the 2x-concurrency failure the #1766 gate closes; +canary S-02 cannot see it (it counts `ZCARD` only), so it needs its own check. +Sample on a short interval during peak, not once. + +```sql +SELECT o.agent_name, o.max_parallel_tasks AS cap, + COUNT(*) FILTER (WHERE e.lease_expires_at IS NOT NULL) AS leased_running, + COUNT(*) FILTER (WHERE e.lease_expires_at IS NULL) AS pushed_running +FROM agent_ownership o +LEFT JOIN schedule_executions e + ON e.agent_name = o.agent_name AND e.status = 'running' +WHERE o.agent_name = '' +GROUP BY 1, 2; +``` + +Verdict: `leased_running <= cap` always. `leased_running > 0 AND pushed_running > 0` +for an autonomous workload means coexistence is still live. + +**M4 — lost / phantom executions.** Any non-terminal row older than the lease +window is the headline AC failing. + +```sql +SELECT id, status, triggered_by, started_at, lease_expires_at, + claimed_by_worker, redelivery_count, + EXTRACT(epoch FROM (now() - started_at::timestamptz))::int AS age_s +FROM schedule_executions +WHERE agent_name = '' + AND status IN ('queued', 'running') + AND started_at::timestamptz < now() - interval '2 hours' +ORDER BY started_at; +``` + +Expect **zero rows**. A `running` row past its `lease_expires_at` that the reaper +has not touched is a reaper failure; a `queued` row aging with idle workers is a +claim failure. + +**M5 — re-delivery + poison-park.** Proves the lease machinery behaves under real +load rather than in the 2026-07-08 synthetic pilot. + +```sql +SELECT redelivery_count, COUNT(*) +FROM schedule_executions +WHERE agent_name = '' + AND started_at::timestamptz > now() - interval '7 days' +GROUP BY 1 ORDER BY 1; +``` + +Healthy: overwhelmingly `0`, a thin tail at 1–2, nothing at `MAX_REDELIVERY` +without a matching operator alert. Cross-check the parks: + +```sql +SELECT id, agent_name, title, created_at, status +FROM operator_queue +WHERE id LIKE 'poison-%' AND created_at::timestamptz > now() - interval '7 days' +ORDER BY created_at DESC; +``` + +Every row at the cap in the first query must have a `poison-` item here. A park +with no alert is a silent drop — treat as a blocker. + +**M6 — outcome regression.** Compare against the pre-flight baseline. + +```sql +SELECT status, COUNT(*), + ROUND(AVG(duration_ms)::numeric, 0) AS avg_ms, + ROUND(SUM(cost)::numeric, 4) AS cost +FROM schedule_executions +WHERE agent_name = '' + AND started_at::timestamptz > now() - interval '7 days' +GROUP BY 1 ORDER BY 2 DESC; +``` + +Watch for a `failed` share above baseline and for `duration_ms` inflation (pull +adds up to one poll interval of latency by design — the worker idles up to 15s +between claims, so a small p50 rise is expected, a p95 blow-up is not). + +**M7 — canary.** With E-05 lease-excluded (this branch), violations should be +genuinely rare. + +```sql +SELECT invariant_id, severity, COUNT(*), MAX(snapshot_time) AS last_seen +FROM canary_violations +WHERE snapshot_time::timestamptz > now() - interval '7 days' +GROUP BY 1, 2 ORDER BY 3 DESC; +``` + +Any **S-01 / S-02 / E-01 / E-02 / L-03** hit on the pilot is an abort signal. +E-05 hits mean the lease exclusion is not deployed — verify the build. + +**M8 — starvation.** Queue depth should oscillate, not climb monotonically. + +```sql +SELECT COUNT(*) AS queued, + EXTRACT(epoch FROM (now() - MIN(queued_at)::timestamptz))::int AS oldest_s +FROM schedule_executions +WHERE agent_name = '' AND status = 'queued'; +``` + +A steadily rising `oldest_s` with healthy workers means claims are failing; +with busy workers it just means the agent is undersized — check M3 before +concluding. + +### Abort criteria + +Stop the soak and revert if any of these hold: a non-empty **M4**; **M3** +`leased_running > cap`; a critical canary violation on the pilot (**M7**); a +park with no operator alert (**M5**); or a `failed` share materially above the +pre-flight baseline (**M6**). + +### Rollback + +Remove the agent from `PULL_MODE_PILOT_AGENTS` → restart backend → **recreate the +agent** (the flag only clears at create/recreate; `PULL_MODE_ENV_KEYS` is popped +on recreate precisely so de-piloting takes effect). Queued rows are picked up by +the backend drain again as soon as the pilot guard stops matching, so in-flight +work is not stranded. + +--- ## Appendix — key file references diff --git a/src/backend/canary/invariants/e05_dispatched_rows_have_session.py b/src/backend/canary/invariants/e05_dispatched_rows_have_session.py index d4b66ccb2..3c51b0117 100644 --- a/src/backend/canary/invariants/e05_dispatched_rows_have_session.py +++ b/src/backend/canary/invariants/e05_dispatched_rows_have_session.py @@ -31,6 +31,17 @@ and the observability link to the JSONL file in the container. Real operational problem, but not the lights-out kind S-01/S-02/E-01 flag. +## Pull-claimed rows are excluded (#1766) + +A `#1081` pull-CLAIMED row (`lease_expires_at IS NOT NULL`) is `running` +with a NULL `claude_session_id` by design — the claim is a pure SQL +UPDATE and the worker reports its session id back only with the terminal. +`mark_no_session_executions_failed`, the very sweep this invariant +watches, already carries `lease_expires_at IS NULL` for that reason, so +without the same exclusion here E-05 fires on every pull turn older than +60s: flagging rows the sweep is deliberately leaving alone, and burying a +real #106 regression under noise for a whole soak window. + Tier B, severity major. """ @@ -62,6 +73,18 @@ def check(snapshot: Snapshot) -> List[ViolationReport]: for agent in snapshot.agents: for eid in sorted(agent.running_exec_ids): + # #1766: exclude pull-CLAIMED rows (mirrors S-01's exclusion and, + # decisively, the very sweep this invariant watches — + # `mark_no_session_executions_failed` already carries + # `lease_expires_at IS NULL` for exactly this reason). A leased row + # is `running` with a NULL claude_session_id BY DESIGN: the claim is + # a pure SQL UPDATE and the worker reports its session id back only + # with the terminal. Without this, E-05 fires on every pull turn + # older than 60s — flagging the rows the sweep is deliberately + # leaving alone, and drowning a real #106 regression in noise for + # the whole soak window (#1081 Phase 3 / T3.6). + if agent.running_lease_expires_at.get(eid) is not None: + continue session_id = agent.running_claude_session_ids.get(eid) if session_id: continue diff --git a/src/backend/canary/snapshot.py b/src/backend/canary/snapshot.py index e6f8e6fc7..016af6720 100644 --- a/src/backend/canary/snapshot.py +++ b/src/backend/canary/snapshot.py @@ -222,8 +222,11 @@ class AgentSnapshot: # owned EXCLUSIVELY by the lease-reaper and NEVER enters the slot ZSET (a # claim is a pure SQL UPDATE with no ZADD). S-01 uses this to exclude leased # rows from the SQL side of its slot–row bijection, so a legitimately- - # unslotted pull row is not flagged `in_sql_only`. Only S-01 reads this; - # E-01/E-02/E-05 keep seeing the full `running_exec_ids` unchanged. + # unslotted pull row is not flagged `in_sql_only`. Read by S-01 and, since + # #1766, by E-05 — a leased row is `running` with a NULL claude_session_id + # by design, and `mark_no_session_executions_failed` (the sweep E-05 watches) + # already excludes leased rows for that reason. E-01/E-02 keep seeing the + # full `running_exec_ids` unchanged. running_lease_expires_at: Dict[str, Optional[str]] = field(default_factory=dict) # `claude_session_id` per running id (str or None); used by E-05 to detect # dispatched rows that never acquired a backing session. diff --git a/src/backend/routers/internal.py b/src/backend/routers/internal.py index 3a752b14e..f08866602 100644 --- a/src/backend/routers/internal.py +++ b/src/backend/routers/internal.py @@ -113,7 +113,7 @@ def _pull_authorized(request: Request, agent_name: str) -> bool: # with a valid scoped key, so rollback takes effect on backend restart rather # than only after a container recreate. The trusted-backend (internal-secret) # path is unchanged. - from services.agent_service.pull_mode import is_pull_pilot_agent + from services.pull_pilot import is_pull_pilot_agent if not is_pull_pilot_agent(agent_name): return False return heartbeat_service.authorize_heartbeat(_validated_agent_key(request), agent_name) diff --git a/src/backend/services/agent_service/pull_mode.py b/src/backend/services/agent_service/pull_mode.py index c33579233..af462cd7e 100644 --- a/src/backend/services/agent_service/pull_mode.py +++ b/src/backend/services/agent_service/pull_mode.py @@ -27,17 +27,23 @@ from __future__ import annotations import os -from typing import Dict, Set - - -def _pilot_allowlist() -> Set[str]: - raw = os.getenv("PULL_MODE_PILOT_AGENTS", "") - return {name.strip() for name in raw.split(",") if name.strip()} - - -def is_pull_pilot_agent(agent_name: str) -> bool: - """True when ``agent_name`` is in the ``PULL_MODE_PILOT_AGENTS`` allowlist.""" - return agent_name in _pilot_allowlist() +from typing import Dict + +# The predicates live in the LEAF module `services/pull_pilot.py` and are +# re-exported here so every existing importer keeps working. They CANNOT live in +# this file: `services/agent_service/__init__.py` eagerly imports the whole +# agent-lifecycle stack (helpers/lifecycle/crud/deploy/terminal → models), so a +# dispatch-path module doing `from services.agent_service.pull_mode import ...` +# drags all of it in. See the docstring in `services/pull_pilot.py`. +# +# Dispatch-path callers (`capacity_manager`, `backlog_service`, `routers/internal`) +# import from `services.pull_pilot` DIRECTLY — importing them from here would +# reintroduce exactly the dependency this split removes. +from services.pull_pilot import ( # noqa: F401 — re-exported for back-compat + _pilot_allowlist, + is_pull_pilot_agent, + pull_owns_dispatch, +) # The container env keys this module manages. Recreate (``lifecycle.py``) pops diff --git a/src/backend/services/backlog_service.py b/src/backend/services/backlog_service.py index 645097444..22b96e2d4 100644 --- a/src/backend/services/backlog_service.py +++ b/src/backend/services/backlog_service.py @@ -152,6 +152,26 @@ async def drain_next(self, agent_name: str) -> bool: Returns True if a row was drained, False otherwise. """ + # ---- #1766: a pull pilot's queue has exactly one consumer ---------- + # The backend and the agent's worker pool both claim through + # `db.claim_next_queued`, so before this guard they raced for every + # queued row — the atomic UPDATE meant no double-run, but the winner was + # whoever polled first, and the backend was structurally favoured (it + # drains on slot release plus the 60s `drain_orphans_all` sweep, while a + # worker idles up to 15s between polls). That made pull coverage + # nondeterministic and drain-biased: exactly the thing a soak is supposed + # to measure. One guard here covers every drain path, since the release + # callback, the orphan sweep, and `drain_on_release` all funnel through + # this method. + from services.pull_pilot import is_pull_pilot_agent + + if is_pull_pilot_agent(agent_name): + logger.debug( + f"[Backlog] Drain skipped for pull pilot '{agent_name}': " + "its worker pool is the sole consumer (#1766)" + ) + return False + from database import db if db.get_queued_count(agent_name) == 0: diff --git a/src/backend/services/capacity_manager.py b/src/backend/services/capacity_manager.py index 7c1769a45..982fca1e8 100644 --- a/src/backend/services/capacity_manager.py +++ b/src/backend/services/capacity_manager.py @@ -345,15 +345,38 @@ async def acquire( raise CircuitOpen(agent_name, breaker.retry_after_seconds()) return AcquireResult(state="admitted", execution_id=execution_id) - # First try to acquire a slot directly. SlotService already does the - # atomic ZADD + count check. - admitted = await self._slots.acquire_slot( - agent_name=agent_name, - execution_id=execution_id, - max_parallel_tasks=max_concurrent, - message_preview=message_preview, - timeout_seconds=timeout_seconds, + # ---- #1766: a pull pilot's autonomous work is queue-ONLY ----------- + # The pilot flag used to be purely additive: the agent started pulling, + # but the backend kept admitting-and-pushing whenever a slot was free, so + # the two paths ran in parallel over one queue with two independent + # capacity counters. Skipping admission here makes the durable queue the + # single entry point for this agent, so its worker pool IS its capacity + # (#1081 Phase 5, pilot-scoped). `pull_owns_dispatch` excludes + # interactive triggers (Open Question 7 scope cut) and fails safe to + # push, so a non-pilot's path is byte-for-byte unchanged. + from services.pull_pilot import pull_owns_dispatch + + pull_exclusive = ( + overflow_policy == "queue_persistent" + and overflow_payload is not None + and pull_owns_dispatch(agent_name, overflow_payload.triggered_by) ) + + if pull_exclusive: + # No ZADD: the row falls through to the persistent enqueue below and + # is claimed by the agent's own worker. Capacity is enforced + # physically by the pool, not by this counter. + admitted = False + else: + # First try to acquire a slot directly. SlotService already does the + # atomic ZADD + count check. + admitted = await self._slots.acquire_slot( + agent_name=agent_name, + execution_id=execution_id, + max_parallel_tasks=max_concurrent, + message_preview=message_preview, + timeout_seconds=timeout_seconds, + ) if admitted: return AcquireResult(state="admitted", execution_id=execution_id) @@ -517,7 +540,7 @@ async def get_all_states( # Gated on the EXISTING pilot allowlist; empty allowlist (the default) # short-circuits to the unchanged ZSET-only path — inert, zero added # cost/delta. Metering ONLY: acquire/release are not touched. - from services.agent_service.pull_mode import _pilot_allowlist + from services.pull_pilot import _pilot_allowlist pilots = _pilot_allowlist() & set(clamped) if pilots: @@ -556,7 +579,7 @@ async def get_slot_state(self, agent_name: str, max_concurrent: int): # short-circuits per-agent, so a non-pilot pays nothing and its output # is byte-for-byte identical whether the allowlist is set or not. # Metering ONLY: acquire/release are not touched. - from services.agent_service.pull_mode import is_pull_pilot_agent + from services.pull_pilot import is_pull_pilot_agent if is_pull_pilot_agent(agent_name): from database import db diff --git a/src/backend/services/pull_pilot.py b/src/backend/services/pull_pilot.py new file mode 100644 index 000000000..e0f33b23d --- /dev/null +++ b/src/backend/services/pull_pilot.py @@ -0,0 +1,85 @@ +"""Pull-pilot predicates — the LEAF half of ``agent_service/pull_mode.py``. + +Answers two questions and nothing else: *is this agent a pull pilot*, and *must +this ``(agent, trigger)`` pair reach it only via the durable queue*. + +**Why this module exists separately (#1766).** These predicates are consulted +from the dispatch hot path — ``capacity_manager.acquire`` and +``backlog_service.drain_next`` — but their original home, +``services/agent_service/pull_mode.py``, sits inside a package whose +``__init__.py`` eagerly imports ``helpers``, ``lifecycle``, ``crud``, ``deploy`` +and ``terminal``. So ``from services.agent_service.pull_mode import ...`` drags +the entire agent-lifecycle stack (and transitively ``models``) in behind it. In +production that import is already warm and the cost is invisible; under a unit +test that stubs ``models``/``database`` it explodes with a bare +``ImportError: cannot import name 'AgentGitConfig'`` that names nothing to do +with the actual dependency. A capacity/backlog module must not need the agent +CRUD stack to answer "is this name in an env var". + +Stdlib-only by construction — keep it that way. ``pull_mode`` re-exports these +names, so every existing importer is unaffected. +""" +from __future__ import annotations + +import logging +import os +from typing import Optional, Set + +logger = logging.getLogger(__name__) + + +def _pilot_allowlist() -> Set[str]: + raw = os.getenv("PULL_MODE_PILOT_AGENTS", "") + return {name.strip() for name in raw.split(",") if name.strip()} + + +def is_pull_pilot_agent(agent_name: str) -> bool: + """True when ``agent_name`` is in the ``PULL_MODE_PILOT_AGENTS`` allowlist.""" + return agent_name in _pilot_allowlist() + + +def pull_owns_dispatch(agent_name: str, triggered_by: Optional[str]) -> bool: + """True when this ``(agent, trigger)`` pair must reach the agent ONLY by the + agent claiming it from the durable queue — the backend neither pushes it nor + drains it (#1766, the pilot-scoped slice of #1081 Phase 5 "capacity becomes + physical"). + + Without this, a pilot agent runs BOTH systems at once: the backend still + admits-and-pushes whenever a slot is free (``acquire`` had no pilot branch), + so a row only ever queued on overflow, and the backend's own + ``backlog_service.drain_next`` then raced the agent's worker for it. Two + independent capacity counters (Redis ZSET vs the container's worker pool) + meant a pilot could run up to 2x ``max_parallel_tasks`` — invisible to canary + S-02, which counts ``ZCARD`` only. Making the pilot flag a true either/or + restores one capacity owner per agent. + + **Interactive turns are deliberately excluded.** Only the autonomous trigger + set queues; a human chat / Session-tab turn keeps today's synchronous push + path and today's Redis session lock. That is the scope cut in + ``TARGET_ARCHITECTURE.md`` Open Question 7 ("Does human-interactive chat + belong in the queue at all?" — *under consideration, not decided*), and it is + load-bearing here: one FIFO ordered by ``queued_at`` would park a human turn + behind N batch tasks until the held connection timed out, and N competing + workers could claim two turns of the same session concurrently — the exact + concurrent ``--resume`` on one JSONL the session lock exists to prevent. + + Fail-safe: any error resolving the trigger set returns ``False``, i.e. the + unchanged push behaviour. The dangerous direction would be silently claiming + dispatch for a trigger we could not classify. + """ + if not is_pull_pilot_agent(agent_name): + return False + try: + # Lazy: task_execution_service imports the capacity stack, and this is + # called from inside it. Single source of truth for the trigger set — + # never a second copy that can drift. + from services.task_execution_service import _AUTONOMOUS_TRIGGERS + + return triggered_by in _AUTONOMOUS_TRIGGERS + except Exception: # noqa: BLE001 — unresolvable trigger set ⇒ push, as today + logger.warning( + "[#1766] could not resolve the autonomous-trigger set for %s " + "(trigger=%r); falling back to push dispatch", + agent_name, triggered_by, + ) + return False diff --git a/tests/test_canary_invariants.py b/tests/test_canary_invariants.py index 934cdd3b3..6eb499f9f 100644 --- a/tests/test_canary_invariants.py +++ b/tests/test_canary_invariants.py @@ -1591,7 +1591,7 @@ def test_skips_row_with_malformed_started_at(self): class TestInvariantE05: @staticmethod - def _snap(*, started_at, session_id): + def _snap(*, started_at, session_id, lease_expires_at=None): from canary.snapshot import Snapshot, AgentSnapshot return Snapshot( snapshot_time="2026-05-18T12:00:00Z", @@ -1604,6 +1604,7 @@ def _snap(*, started_at, session_id): running_exec_ids={"e1"}, running_started_at={"e1": started_at}, running_claude_session_ids={"e1": session_id}, + running_lease_expires_at={"e1": lease_expires_at}, ) ], ) @@ -1640,6 +1641,31 @@ def test_fires_when_old_row_lacks_session(self): assert v.severity == "major" assert v.observed_state["age_seconds"] == 3600 + # -- #1766: pull-claimed rows are excluded ------------------------------ + + def test_leased_row_is_excluded(self): + """A pull-CLAIMED row is `running` with a NULL claude_session_id BY + DESIGN — `mark_no_session_executions_failed` already skips leased rows, + so E-05 must too or it fires on every pull turn for a whole soak.""" + snap = self._snap( + started_at="2026-05-18T11:00:00Z", # 1h old + session_id=None, # no session — would fire + lease_expires_at="2026-05-18T12:30:00Z", + ) + from canary.invariants import e05_dispatched_rows_have_session as e05 + assert e05.check(snap) == [] + + def test_non_leased_control_still_fires(self): + """The exclusion must be keyed on the lease, not blanket-silence E-05: + an identical row with a NULL lease is still the #106 bug class.""" + snap = self._snap( + started_at="2026-05-18T11:00:00Z", + session_id=None, + lease_expires_at=None, + ) + from canary.invariants import e05_dispatched_rows_have_session as e05 + assert len(e05.check(snap)) == 1 + # --------------------------------------------------------------------------- # B-01 — queue-status coherence diff --git a/tests/unit/test_1766_pull_pilot_exclusive.py b/tests/unit/test_1766_pull_pilot_exclusive.py new file mode 100644 index 000000000..edc1ccf38 --- /dev/null +++ b/tests/unit/test_1766_pull_pilot_exclusive.py @@ -0,0 +1,335 @@ +"""Pull pilots own their dispatch — #1766 (pilot-scoped slice of #1081 Phase 5). + +Before this change the ``PULL_MODE_PILOT_AGENTS`` flag was purely **additive**: +the agent's container started pulling, but the backend kept admitting-and-pushing +whenever a slot was free (``CapacityManager.acquire`` had no pilot branch), and +the backend's own ``backlog_service.drain_next`` kept claiming queued rows +through the SAME ``db.claim_next_queued`` the agent's worker uses. The result was +push and pull coexisting **inside one agent**: + + * a row only ever reached the queue on capacity overflow, so the pull path + carried almost no traffic on a healthy instance; + * when a row did queue, backend-drain and agent-worker raced for it — no + double-run (the claim is one atomic UPDATE), but the winner was whoever + polled first, and the backend was structurally favoured; + * two independent capacity counters (Redis ZSET vs the container's worker + pool, both sized ``max_parallel_tasks``) meant a pilot could run up to **2x** + its configured concurrency, invisible to canary S-02 (``ZCARD`` only) and to + S-01 (which excludes leased rows by design). + +The three properties proven here make the pilot flag a true **either/or**: + + 1. **Producer** — a pilot's autonomous work is never admitted; it goes straight + to the durable queue, and no slot is ZADDed. + 2. **Consumer** — the backend never drains a pilot's queue, so the agent's + worker pool is the sole claimant. One guard covers every drain path + (release callback, 60s orphan sweep, ``drain_on_release``). + 3. **Carve-out** — interactive triggers are excluded and still take the + synchronous push path (``TARGET_ARCHITECTURE.md`` Open Question 7's scope + cut), so human chat is not parked behind N batch tasks and per-session + ``--resume`` serialization is untouched. + +Plus the inertness property the whole dark-ship rests on: with an empty +allowlist (the default) every path is byte-for-byte unchanged. + +Pure unit test — mocked ``SlotService`` / ``BacklogService`` collaborators, +mirroring ``tests/unit/test_capacity_manager.py``. No Redis, no DB, no agent. +""" + +from __future__ import annotations + +import asyncio +import sys +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock + +import pytest + +# Bootstrap src/backend on sys.path. +# +# Deliberately WITHOUT the `sys.modules.pop("utils", ...)` preamble that +# test_capacity_manager.py carries: tests/unit/conftest.py already installs +# src/backend/utils as the canonical `utils` package via an importlib file +# loader, so evicting it leaves `utils` unbound and pytest's prepend import mode +# rebinds it to the tests/ helper package (the failure mode spelled out in +# test_1081_physical_meter.py). Every backend module here is imported lazily +# inside a test or fixture, so the path insert is all this file needs — and it +# keeps the file clean under tests/lint_sys_modules.py. +_THIS = Path(__file__).resolve() +_BACKEND = _THIS.parent.parent.parent / "src" / "backend" +_BACKEND_STR = str(_BACKEND) +if _BACKEND_STR not in sys.path: + sys.path.insert(0, _BACKEND_STR) + +pytestmark = pytest.mark.unit + + +# --------------------------------------------------------------------------- +# Fixtures (mirror tests/unit/test_capacity_manager.py) +# --------------------------------------------------------------------------- + + +@pytest.fixture +def fake_redis(): + class _FakeRedis: + def __init__(self): + self.lists: dict[str, list[str]] = {} + self.zsets: dict[str, dict[str, float]] = {} + + def lpush(self, key, *values): + self.lists.setdefault(key, []) + for v in values: + self.lists[key].insert(0, v) + return len(self.lists[key]) + + def rpop(self, key): + if not self.lists.get(key): + return None + return self.lists[key].pop() + + def llen(self, key): + return len(self.lists.get(key, [])) + + def lrange(self, key, start, end): + items = self.lists.get(key, []) + end = len(items) if end == -1 else end + 1 + return items[start:end] + + def delete(self, key): + self.lists.pop(key, None) + self.zsets.pop(key, None) + return 1 + + def exists(self, key): + return int(key in self.lists or key in self.zsets) + + def zscore(self, key, member): + return self.zsets.get(key, {}).get(member) + + def zadd(self, key, mapping): + self.zsets.setdefault(key, {}).update(mapping) + return len(mapping) + + def set(self, *_a, **_kw): + return True + + return _FakeRedis() + + +@pytest.fixture +def slot_service(): + s = AsyncMock() + s.slots_prefix = "agent:slots:" + s.acquire_slot = AsyncMock(return_value=True) + s.release_slot = AsyncMock() + s._registered_callbacks = [] + s.register_on_release = lambda cb: s._registered_callbacks.append(cb) + return s + + +@pytest.fixture +def backlog_service(): + b = AsyncMock() + b.enqueue = AsyncMock(return_value=True) + b.drain_next = AsyncMock(return_value=False) + return b + + +@pytest.fixture +def capacity(monkeypatch, fake_redis, slot_service, backlog_service): + from services import capacity_manager as cm_module + + monkeypatch.setattr(cm_module.redis, "from_url", lambda *_a, **_kw: fake_redis) + return cm_module.CapacityManager( + redis_url="redis://test", + slot_service=slot_service, + backlog_service=backlog_service, + ) + + +def _payload(triggered_by: str): + from services.capacity_manager import PersistentTaskPayload + + return PersistentTaskPayload( + request=MagicMock(), + effective_timeout=900, + user_id=1, + user_email="u@x", + subscription_id=None, + x_source_agent=None, + x_mcp_key_id=None, + x_mcp_key_name=None, + triggered_by=triggered_by, + collaboration_activity_id=None, + ) + + +@pytest.fixture +def pilot(monkeypatch): + """Put `alice` in the pilot allowlist. `bob` stays a normal push agent.""" + monkeypatch.setenv("PULL_MODE_PILOT_AGENTS", "alice") + + +# --------------------------------------------------------------------------- +# 1. pull_owns_dispatch — the predicate +# --------------------------------------------------------------------------- + + +class TestPullOwnsDispatch: + def test_non_pilot_never_owns_dispatch(self, pilot): + from services.agent_service.pull_mode import pull_owns_dispatch + + assert pull_owns_dispatch("bob", "schedule") is False + + @pytest.mark.parametrize( + "trigger", ["schedule", "webhook", "loop", "event", "fan_out", "agent", "reminder"] + ) + def test_pilot_owns_every_autonomous_trigger(self, pilot, trigger): + from services.agent_service.pull_mode import pull_owns_dispatch + + assert pull_owns_dispatch("alice", trigger) is True + + @pytest.mark.parametrize("trigger", ["manual", "user", "chat", "voip", "voice", None]) + def test_pilot_does_not_own_interactive_triggers(self, pilot, trigger): + """Open Question 7 scope cut: human turns keep the synchronous path.""" + from services.agent_service.pull_mode import pull_owns_dispatch + + assert pull_owns_dispatch("alice", trigger) is False + + def test_empty_allowlist_is_inert(self, monkeypatch): + monkeypatch.setenv("PULL_MODE_PILOT_AGENTS", "") + from services.agent_service.pull_mode import pull_owns_dispatch + + assert pull_owns_dispatch("alice", "schedule") is False + + def test_unresolvable_trigger_set_falls_back_to_push(self, pilot, monkeypatch): + """Fail-safe direction: if the trigger set can't be resolved we must + behave exactly as today (push), never silently claim dispatch.""" + import services.agent_service.pull_mode as pm + + monkeypatch.setitem(sys.modules, "services.task_execution_service", None) + assert pm.pull_owns_dispatch("alice", "schedule") is False + + +# --------------------------------------------------------------------------- +# 2. Producer — a pilot's autonomous work is queue-only +# --------------------------------------------------------------------------- + + +class TestProducerGate: + def test_pilot_autonomous_work_bypasses_admission( + self, capacity, slot_service, backlog_service, pilot + ): + """The core fix: a free slot no longer means a push for a pilot.""" + slot_service.acquire_slot = AsyncMock(return_value=True) # slot IS free + result = asyncio.run( + capacity.acquire( + agent_name="alice", + execution_id="exec-1", + max_concurrent=3, + overflow_policy="queue_persistent", + overflow_payload=_payload("schedule"), + ) + ) + assert result.state == "queued_persistent" + # No ZADD: capacity for this agent is its worker pool, not the ZSET. + slot_service.acquire_slot.assert_not_awaited() + backlog_service.enqueue.assert_awaited_once() + + def test_pilot_interactive_work_still_pushes( + self, capacity, slot_service, backlog_service, pilot + ): + """The carve-out, on the same agent: a human turn is admitted.""" + result = asyncio.run( + capacity.acquire( + agent_name="alice", + execution_id="exec-2", + max_concurrent=3, + overflow_policy="queue_persistent", + overflow_payload=_payload("manual"), + ) + ) + assert result.state == "admitted" + slot_service.acquire_slot.assert_awaited_once() + backlog_service.enqueue.assert_not_awaited() + + def test_non_pilot_autonomous_work_unchanged( + self, capacity, slot_service, backlog_service, pilot + ): + result = asyncio.run( + capacity.acquire( + agent_name="bob", + execution_id="exec-3", + max_concurrent=3, + overflow_policy="queue_persistent", + overflow_payload=_payload("schedule"), + ) + ) + assert result.state == "admitted" + slot_service.acquire_slot.assert_awaited_once() + + def test_in_memory_policy_never_force_queued( + self, capacity, slot_service, pilot + ): + """`/chat` uses queue_in_memory; the gate must not touch it even for a + pilot, or the interactive path changes shape.""" + result = asyncio.run( + capacity.acquire( + agent_name="alice", + execution_id="exec-4", + max_concurrent=3, + overflow_policy="queue_in_memory", + ) + ) + assert result.state == "admitted" + slot_service.acquire_slot.assert_awaited_once() + + +# --------------------------------------------------------------------------- +# 3. Consumer — the backend never drains a pilot's queue +# --------------------------------------------------------------------------- + + +class TestConsumerGate: + def test_drain_skipped_for_pilot(self, monkeypatch, pilot): + """Guard sits ahead of the queued COUNT, so the DB is never touched.""" + from services.backlog_service import BacklogService + + svc = BacklogService() + called = {"count": 0} + + import database + + monkeypatch.setattr( + database.db, + "get_queued_count", + lambda *_a, **_kw: called.__setitem__("count", called["count"] + 1) or 5, + ) + assert asyncio.run(svc.drain_next("alice")) is False + assert called["count"] == 0, "pilot drain must short-circuit before the COUNT" + + def test_drain_proceeds_for_non_pilot(self, monkeypatch, pilot): + from services.backlog_service import BacklogService + + svc = BacklogService() + import database + + monkeypatch.setattr(database.db, "get_queued_count", lambda *_a, **_kw: 0) + # Reaches the COUNT (0 ⇒ False) rather than short-circuiting on identity. + assert asyncio.run(svc.drain_next("bob")) is False + + def test_drain_unchanged_when_allowlist_empty(self, monkeypatch): + from services.backlog_service import BacklogService + + monkeypatch.setenv("PULL_MODE_PILOT_AGENTS", "") + svc = BacklogService() + import database + + seen = {} + monkeypatch.setattr( + database.db, + "get_queued_count", + lambda name, *_a, **_kw: seen.setdefault("name", name) and 0 or 0, + ) + assert asyncio.run(svc.drain_next("alice")) is False + assert seen.get("name") == "alice", "default path must still reach the COUNT"