diff --git a/docs/KNOWN_ISSUES.md b/docs/KNOWN_ISSUES.md index 25fa53f0fb..ad1172280c 100644 --- a/docs/KNOWN_ISSUES.md +++ b/docs/KNOWN_ISSUES.md @@ -60,7 +60,7 @@ **Symptoms:** - Agent A calls `mcp__trinity__chat_with_agent(agent_name="B", message="...", timeout_seconds=900)` and B works longer than the client's ceiling -- Before the mitigation, the call failed with a timeout error and the result was lost. Now the MCP server stops waiting at `MCP_CHAT_TIMEOUT_MS` (25 seconds by default) and answers with `{status: "queued_timeout", execution_id, …}` while B keeps working; read the outcome with `get_execution_result` and never re-send +- Before the mitigation, the call failed with a timeout error and the result was lost. Now the MCP server stops waiting at `MCP_CHAT_TIMEOUT_MS` (25 seconds by default) and answers with `{status: "queued_timeout", execution_id, …}` while B keeps working; read the outcome with `get_execution_result` and never re-send — unless that outcome is a confirmed `failed` or `cancelled`, which a word-for-word re-send retries (#3245) - The `timeout_seconds` parameter has no effect on either limit **Cause:** diff --git a/docs/MULTI_AGENT_SYSTEM_GUIDE.md b/docs/MULTI_AGENT_SYSTEM_GUIDE.md index bf442304c7..57224b182e 100644 --- a/docs/MULTI_AGENT_SYSTEM_GUIDE.md +++ b/docs/MULTI_AGENT_SYSTEM_GUIDE.md @@ -819,7 +819,7 @@ mcp__trinity__chat_with_agent( **What this means for agent design:** - A long synchronous call no longer just fails: when the server can match it to its execution, it answers with a receipt and the target keeps running it; when it cannot, the error says so and names `list_recent_executions` -- The caller reads the outcome with `get_execution_result(agent_name, execution_id)` and never re-sends — a reworded re-send runs the work twice. Every agent is taught this as the delegation contract in its platform prompt (§Agent Collaboration); the `chat_with_agent` description carries the same text +- The caller reads the outcome with `get_execution_result(agent_name, execution_id)` and never re-sends — unless the outcome is a confirmed `failed` or `cancelled`, which a word-for-word re-send retries (if the same `execution_id` comes back, set a reminder and end the turn). A reworded re-send runs the work twice. Every agent is taught this as the delegation contract in its platform prompt (§Agent Collaboration); the `chat_with_agent` description carries the same text - The calling turn still decides what to do while it waits: read later, arm `set_reminder`, or — for a `parallel=true` run — subscribe to the target's `agent.task.completed` / `agent.task.failed` **Design patterns that work within this constraint:** diff --git a/docs/TRINITY_COMPATIBLE_AGENT_GUIDE.md b/docs/TRINITY_COMPATIBLE_AGENT_GUIDE.md index 380eb513b1..ed7e964fea 100644 --- a/docs/TRINITY_COMPATIBLE_AGENT_GUIDE.md +++ b/docs/TRINITY_COMPATIBLE_AGENT_GUIDE.md @@ -1069,7 +1069,7 @@ Permission checks before any agent-to-agent communication: > **Design Limitation**: an MCP client gives up on a tool call at its own ceiling (30–60 seconds is typical). Trinity's MCP server stops waiting first — at `MCP_CHAT_TIMEOUT_MS`, 25 seconds by default — and answers a synchronous `chat_with_agent` with a receipt (`status: "queued_timeout"` plus the `execution_id`) while the work keeps running. `timeout_seconds` does not extend that wait. When designing agents that collaborate with other agents, ensure that: -- A receipt is treated as "running", never as a failure: the outcome is read with `get_execution_result`, and the call is never re-sent. Every agent is taught this as the delegation contract in its platform prompt (§Agent Collaboration); the `chat_with_agent` tool description carries the same text +- A receipt is treated as "running", never as a failure: the outcome is read with `get_execution_result`, and the call is never re-sent unless the outcome is a confirmed `failed` or `cancelled` — then it is re-sent word for word to retry. Every agent is taught this as the delegation contract in its platform prompt (§Agent Collaboration); the `chat_with_agent` tool description carries the same text - Complex tasks use the async pattern (`parallel=true, async=true`) from the start - Large data exchanges use shared folders instead of MCP return values diff --git a/docs/memory/architecture/execution.md b/docs/memory/architecture/execution.md index c454b299f9..76e9b7d448 100644 --- a/docs/memory/architecture/execution.md +++ b/docs/memory/architecture/execution.md @@ -143,7 +143,9 @@ schema change (all state is Redis). Details in [redelivery-governor.md](../featu ### Idempotency (RELIABILITY-006, #525) -Trigger-boundary dedup — policy in Architectural Invariant #18, table DDL under `idempotency_keys`, details in [idempotency-keys.md](../feature-flows/idempotency-keys.md). `services/idempotency_service.py` (`begin`/`complete`/`fail`) over `db/idempotency.py`. The `(scope, key)` PRIMARY KEY is the atomic claim: `claim()` INSERTs an `in_flight` row; a concurrent loser catches `IntegrityError` and reads the surviving row (cross-process safe over the shared SQLite file). Lifecycle: `claim` → (`attach_execution`) → `complete` (stores `response_snapshot` for replay) or `release` (deletes the in_flight row so retry is possible; never deletes a `completed` row). Rows >24h expire and re-claim (cleanup purges via `idempotency_purge_expired`). Duplicates within 24h short-circuit with the original result + `X-Idempotent-Replay: true`; an in-flight duplicate returns 409. Fail-open. +Trigger-boundary dedup — policy in Architectural Invariant #18, table DDL under `idempotency_keys`, details in [idempotency-keys.md](../feature-flows/idempotency-keys.md). `services/idempotency_service.py` (`begin`/`complete`/`fail`) over `db/idempotency.py`. The `(scope, key)` PRIMARY KEY is the atomic claim: `claim()` INSERTs an `in_flight` row; a concurrent loser catches `IntegrityError` and reads the surviving row (cross-process safe over the shared SQLite file). Lifecycle: `claim` → (`attach_execution`) → `complete` (stores `response_snapshot` for replay) or `release` (deletes the in_flight row so retry is possible; never deletes a `completed` row — two exceptions, each deleting only a row whose replay would be a lie: #2040's `discard_completed` for an agent that no longer exists, and #3245's compare-and-delete below). Rows >24h expire and re-claim (cleanup purges via `idempotency_purge_expired`). Duplicates within 24h short-circuit with the original result + `X-Idempotent-Replay: true`; an in-flight duplicate returns 409. Fail-open. + +**A receipt replays only while its run can still succeed (#3245).** An async `/task`, a timed-out sync `/task` and a timed-out pulled `/chat` store a *receipt* (`async_mode: true`, status `accepted`/`queued`/`queued_timeout`) as the `completed` row, and nothing revisits it, so before #3245 an identical retry got the dead run's receipt for 24h. (a) **Where:** the check runs at the shared admission seam (`dispatch_admission_service._reclaim_ended_receipt`, called right after `begin` in `admit_chat_request` and `begin_task_idempotency`), at replay time — never on a terminal writer, so it adds no side effect to the #1804 writer set and heals rows written before the fix. Only receipt-shaped snapshots are checked (a sync success body replays with no read); `sched:` keys are skipped; the skill gate answers a gated request before `begin`. (b) **Verdicts:** `idempotency_service.execution_liveness(execution_id, agent_name)` returns `live` (`queued`/`running`/`pending_retry`), `succeeded`, `ended` (`failed`/`cancelled`/`skipped`), `gone` (no row), `maybe_alive` or `indeterminate` (lookup error, foreign owner, `manual`, unknown status) — reusable by any caller that needs "can this run still succeed?". Only `ended` and `gone` reclaim; everything else replays (fail toward not re-running). It reads `get_execution_gate_state` (two columns) and, only on `failed`, `get_execution_failure_stamp` (status, error, completed_at). (c) **The `lease_expired` window:** a FAILED whose error starts `lease_expired:` is `maybe_alive` until `completed_at + agent timeout + SLOT_TTL_BUFFER`, then `ended`. The anchor is `completed_at` (the moment the slot reaper wrote FAILED) because the reaper fires only after the slot TTL already ran out from acquire, so a `started_at` anchor would never hold. This is a **heuristic, not a bound**: the agent's restart sweep re-posts persisted results with no age limit and the callback route still accepts SUCCESS over an async FAILED (#378), so a restart after the window can still land SUCCESS beside a fresh run; the timeout is the agent's *current* setting; only the two `fail_stale_slot_execution` writers in `cleanup_service._process_stale_slot_reclaims` carry the tag (an AST guard pins that each such call carries it, not that no other writer ever does) — the Phase-1 stale sweep and the no-session sweep are untagged and go fresh at once; and the tag does not mean "async" (sync push rows are tagged too, whose late SUCCESS can only come from the backend's own dispatcher) — the hold is conservative protection for every tagged row. Nothing wakes the caller when the window ends; the delegation contract tells it to set a reminder. (d) **Race:** two identical retries that see the same dead run start one new run — `idempotency_discard_completed_if_execution` deletes only the row still naming the dead execution id, so a slow retry cannot erase a faster retry's fresh record, and the re-`begin` runs once (no loop). That CAS is sound only because nothing writes a claim after its receipt is stored (pinned by an AST guard over `run_async_task`, `backlog_service`, `pull_coordination_service` and both writers that store a receipt and then raise — the sync `/task` backlog timeout in `_dispatch_sync_backlog` and the pulled `/chat` timeout in `run_pulled_chat_turn`). A late SUCCESS landing in the milliseconds between the liveness read and the delete is part of the residual above (closing it needs cross-domain SQL in `db/idempotency.py`). A delete error replays the original (no second `begin`); a second `begin` that fails open after the compare-and-delete (whether it deleted the row or matched nothing) runs the request without dedupe (logged `outcome=no_dedup`), so a concurrent identical retry can run too — the usual dedupe-layer-failure exception. Each reclaim logs `[#3245] receipt_reclaim … cas=… outcome=…` and each held replay `[#3245] receipt_hold … until=…` (never the raw key). (e) **Leftover risks:** effects can repeat across attempts (effect dedupe is `effect:{execution_id}`); a retry storm is no longer braked by the stale replay (bounded by the caller and the dispatch breaker); a claim still `in_flight` when its run died keeps answering 409 for 24h (needs claim owner tokens; #3298). The MCP client marks a replay `idempotent_replay: true`. **Effect-scoped extension (#1084):** trigger-boundary dedup stops a re-POSTed `/chat`/webhook from creating a *second execution*; it does NOT reach an agent's individual outbound tool calls. So a re-delivered turn (the at-least-once semantics pull-mode / work-stealing will introduce, Epic #1045/#1081) re-emits the same side effect (re-sends a message, re-charges a payment). The same `idempotency_service` adds a per-sink guard — `effect_guard(effect_type, identifying_args, *, execution_id, agent_name, dedup_label, payment_request_id)` — enforced at the SINK, per resolved action identity. Scopes: `effect:{execution_id}` for messages/voip/share_file (after `resolve_and_validate_execution` confirms the execution belongs to the agent — generalizing MEM-001), `payment:{agent_request_id}` for Nevermined settles (a Nevermined observability id, **not** a provider exactly-once token — this local guard, not the provider, enforces at-most-once per id; residual at-least-once retry tracked by #1408). Key = `{effect_type}:sha256(execution_id ∥ effect_type ∥ resolved_identifying_args ∥ dedup_label)` on **resolved, immutable** identity only (recipient/channel/account) — **never the LLM-generated body** (non-deterministic across a re-run → would defeat dedup); `dedup_label` lets an agent intentionally repeat an effect to the same target in one turn. `in_flight ≠ completed`: a completed replay returns the stored sanitized snapshot (no re-emit); an in-flight replay raises `EffectInProgressError` (router → 409, never a silent skip-and-succeed). Reuses the 24h default TTL (already exceeds the lease window, so a completed row outlives a late re-delivery — no new TTL plumbing). Wired sinks: `proactive_message_service.send_message`, `voip_service.place_outbound_call`, `agent_shared_files_service.create_share`, `a2a_outbound_service.call_agent`, `voice_reply_service`, `channel_completion_report`, `nevermined_payment_service.settle_payment_once`; the platform supplies `execution_id` via the agent's MCP header `X-Trinity-Execution-Id` (the tool arg is the older-image fallback), and on a pull-mode agent an effect with no usable id is **refused** with an operator alarm, elsewhere sent and logged `effect_guard.degraded` (#2392; git push is idempotent-by-construction and needs no key). **Target-direction note:** `TARGET_ARCHITECTURE.md` v2 reframes this from a **per-agent** gate to **per-effect** — read/analysis-only + reversible + capability-confined-irreversible effects default on, only irreversible-**un-confineable** effects gate via the async operator queue (#1402); `effect_guard` (this section) is the reversible/backend-sink slice and retry-with-prior-trace (#1401) is the general recovery. See [effect-idempotency.md](../feature-flows/effect-idempotency.md). diff --git a/docs/memory/feature-flows.md b/docs/memory/feature-flows.md index cb1860991c..f4bfd89fa4 100644 --- a/docs/memory/feature-flows.md +++ b/docs/memory/feature-flows.md @@ -24,6 +24,7 @@ | Date | ID | Change | Flow | |------|-----|--------|------| | 2026-10-06 | #3123 | chore(skills): **fresh installs seed the bundled community skills catalog at `trinity-skills` v0.3.0** (was v0.2.0). The tag adds `update-dashboard`, the playbook the metrics tiles already name, and drops the four retired `add-*` installers (marketplace-only now): 38 skills. `config.DEFAULT_SKILL_SOURCE_REF`, `.env.example` and the user doc move together; `tests/unit/test_2545_skill_source_pin.py` raises its floor to v0.3.0 and adds the user doc to its parity checks. The user doc gains the steps for moving an existing instance's bundled row to a newer tag (existing installs keep their row; ent#529). No runtime code change beyond the default: the seed and the annotated-tag peel (#2550) already handle the tag. | [skills-library-sync.md](feature-flows/skills-library-sync.md) | +| 2026-10-06 | #3245 | fix(idempotency): **a failed async run stops replaying its receipt** — an async `/task`, a timed-out sync `/task` and a timed-out pulled `/chat` store a dispatch receipt as the `completed` idempotency row, and for 24 h an identical retry got the dead run back. The shared admission seam (`dispatch_admission_service._reclaim_ended_receipt`, right after `begin` in `/chat` and `/task`) now asks `idempotency_service.execution_liveness` whether that run can still succeed; an `ended` (failed/cancelled/skipped) or `gone` run is compare-and-deleted (`discard_completed_if_execution`, only while the row still names it, so two racing retries start one run) and the key claimed once more. A `lease_expired` FAILED is held until `completed_at + timeout + 300 s`. Read-time only — no terminal writer changes (#1804). MCP marks replays `idempotent_replay: true`; the delegation contract gains the word-for-word retry bullet (pin 1,650 → 1,730) | [idempotency-keys.md](feature-flows/idempotency-keys.md), [mcp-orchestration.md](feature-flows/mcp-orchestration.md), [agent-to-agent-collaboration.md](feature-flows/agent-to-agent-collaboration.md), [system-wide-trinity-prompt.md](feature-flows/system-wide-trinity-prompt.md) | | 2026-10-05 | ent#752 | feat(gate): **a gated skill is refused inside the agent unless its run was approved for it** (Claude Code only). A PreToolUse hook on `Skill` and on `Agent`/`Task` skill preloads (`/opt/trinity/hooks/skill-gate.py`, its own managed-settings drop-in, exec form) asks `POST /api/skill-gate/check` with the agent's own key; the run is cleared by #751's approved-run record or a new `self_approved` record written where a self-approval lets a run through (no migration). No answer → a root-owned marker decides (gated agents fail closed, others keep working); never exit 1. A self-approved `/chat` turn runs in its own session; `/health` reports `skill_gate_hook` | [skill-gate.md](feature-flows/skill-gate.md) | | 2026-10-05 | ent#568 | feat(delegation): **every caller is taught that a receipt means the work is running — never re-send on silence**. One text (`DELEGATION_CONTRACT`, byte-identical in `platform_prompt_service.py` and `src/mcp-server/src/delegation_contract.ts`) is spliced into the platform prompt's §Agent Collaboration and published verbatim in the `chat_with_agent` and every `chat_with_` description; `fan_out` / `send_message` repeat its rule sentence. Every `chat_with_*` `execution_id` receipt carries "Do not re-send: read the outcome with `get_execution_result(…)`", and the backend's REST "poll GET" receipts are rewritten with liveness-neutral leads. Each carrying description stays under Claude Code's 2,048-char cap — `chat_with_agent` had been cut there since #2958 | [mcp-orchestration.md](feature-flows/mcp-orchestration.md), [system-wide-trinity-prompt.md](feature-flows/system-wide-trinity-prompt.md), [mcp-agent-exposure.md](feature-flows/mcp-agent-exposure.md) | | 2026-10-03 | #3185 | feat(a2a): **the outbound client speaks 402 — a priced remote is an outcome, not an outage**. `_read_capped` collapsed every HTTP ≥ 400 into `rpc_http_error` without reading the body, so a remote answering 402 was unreachable: the caller saw neither the price nor a way to attach a token. 402/403 on the RPC hop are now classified BEFORE the encoding and length guards (a gzipped or oversized "pay me" read as an outage), both rails are read (the HTTP 402's `payment-required` header or body, and an in-band `x402.payment.status` task — checked before `_parse_task` on send AND poll, or a priced `input-required` reaches the agent as a prompt it polls forever), and the route answers **402** with an allowlisted `detail.payment` = `{summary, x402, truncated}`, scrubbed against the token, its base64 forms and its decoded leaves. `payment_rejected` / `rpc_forbidden` stay 502 with `remote_status`, so 402 ("buy") is always distinguishable from 403 ("top up"). The endpoint credential carries a `credential_kind` (`api_key` default — every pre-#3185 record — or `payment_token`, inferred from the value when omitted with the SAME predicate the client sends on), an x402 v3 token is flagged `credential_single_use` rather than refused, and the MCP tools return `payment_required` + `payment` + `task_id` + `do_not_retry` with a description that names who can act. Trinity never buys, never retries a priced call, and never snapshots a 402 — a stored one would replay "pay me" after the operator paid | [a2a-outbound-call.md](feature-flows/a2a-outbound-call.md), [architecture/backend.md](architecture/backend.md), [architecture/mcp-server.md](architecture/mcp-server.md) | diff --git a/docs/memory/feature-flows/agent-to-agent-collaboration.md b/docs/memory/feature-flows/agent-to-agent-collaboration.md index 2cffda7810..357082af6f 100644 --- a/docs/memory/feature-flows/agent-to-agent-collaboration.md +++ b/docs/memory/feature-flows/agent-to-agent-collaboration.md @@ -251,7 +251,7 @@ When `MCP_AGENT_CHAT_PULL_ENABLED` is ON, a **sequential** (`parallel=false`) ag authContext?.scope === "agent" && !isSelfTask; ``` -- **Pull branch** — chat.ts:391-402: calls `apiClient.task(agent_name, message, { async_mode: true }, sourceAgent, mcpKeyInfo, idempotencyKey)` and returns the receipt JSON. Only `async_mode` is forwarded; `model`/`allowed_tools`/`system_prompt` are parallel-only and were never applied in sequential mode, so omitting them preserves sequential semantics (agent defaults). +- **Pull branch** — chat.ts:391-402: calls `apiClient.task(agent_name, message, { async_mode: true }, sourceAgent, mcpKeyInfo, idempotencyKey)` and returns the receipt JSON (a replayed one carries `idempotent_replay: true`, #3245). If that run ends `failed` or `cancelled`, an identical re-send dispatches a new run under the same key (#3245) — for a `lease_expired` failure only after its hold window (agent timeout + 5 min after the failure); until then the original receipt comes back. Only `async_mode` is forwarded; `model`/`allowed_tools`/`system_prompt` are parallel-only and were never applied in sequential mode, so omitting them preserves sequential semantics (agent defaults). ### D8 — Route is part of the idempotency identity @@ -706,7 +706,7 @@ An MCP client gives up on a tool call at its own ceiling (30–60 seconds is typ **Impact**: The `timeout_seconds` parameter controls the backend execution timeout (how long Trinity waits for the target agent), not how long the call waits. A long synchronous call no longer just fails — when the server can match it to its execution it answers with a receipt (otherwise the error names `list_recent_executions`) — but the caller has to know a receipt means "running". Every agent is taught that as the delegation contract (ent#568) in its platform prompt's §Agent Collaboration, and the `chat_with_agent` description carries the same text. **Workarounds**: -1. Treat a receipt as running: read it with `get_execution_result`, never re-send +1. Treat a receipt as running: read it with `get_execution_result`, never re-send — unless the outcome is a confirmed `failed` or `cancelled`, which a word-for-word re-send retries (#3245) 2. Use `async=true` with `parallel=true` for long-running tasks (returns `execution_id` immediately) 3. Use shared folders for result exchange instead of synchronous return values diff --git a/docs/memory/feature-flows/idempotency-keys.md b/docs/memory/feature-flows/idempotency-keys.md index 7fe2dad2d5..583b4028ec 100644 --- a/docs/memory/feature-flows/idempotency-keys.md +++ b/docs/memory/feature-flows/idempotency-keys.md @@ -23,7 +23,7 @@ Enforcement lives at the **router** layer, not solely in `TaskExecutionService`, | `POST /api/webhooks/{token}` | `webhook:{token}` | header **or auto-derived** `(token, body_hash)` | `routers/webhooks.py:248` | | `POST /api/agents/{name}/fan-out` | `agent:{name}` | optional `Idempotency-Key` header | `routers/fan_out.py:134` | | Scheduler dispatch | `agent:{name}` | `sched:{execution_id}` (deterministic) | `src/scheduler/service.py:1042` | -| MCP `chat_with_agent` / `fan_out` | `agent:{name}` | `mcp:{sha256(args)}` (deterministic) | `src/mcp-server/src/tools/chat.ts:301,518` | +| MCP `chat_with_agent` / `fan_out` | `agent:{name}` | `mcp:{sha256(args)}` (deterministic) | `src/mcp-server/src/tools/chat.ts:284,746` | **Invariant #18 contract:** *Any new trigger type must accept an idempotency key before merge.* The dedup layer is fail-open, so the cost of wiring it is one `begin`/`complete`/`fail` triple. @@ -33,7 +33,7 @@ Enforcement lives at the **router** layer, not solely in `TaskExecutionService`, - **DB layer**: `db/idempotency.py` — `IdempotencyOperations` (no HTTP, no key-derivation logic). - **Service layer**: `services/idempotency_service.py` — key derivation + `begin`/`complete`/`fail` orchestration over the DB layer. - **Router layer**: each boundary calls `begin()` → dispatch → `complete()` or `fail()`. -- **Facade**: `database.py` exposes `idempotency_claim` / `idempotency_attach_execution` / `idempotency_complete` / `idempotency_release` / `idempotency_purge_expired` (`database.py:2064-2080`), backed by `IdempotencyOperations` constructed in `__init__` (`database.py:297`). +- **Facade**: `database.py` exposes `idempotency_claim` / `idempotency_attach_execution` / `idempotency_complete` / `idempotency_release` / `idempotency_purge_expired` (`database.py:4231-4257`), backed by `IdempotencyOperations` constructed in `__init__` (`database.py:1045`). #3245 adds `idempotency_discard_completed_if_execution` (the compare-and-delete below) and `get_execution_failure_stamp` (a `(status, error, completed_at)` read on `schedule_executions`, used by the replay-liveness check). ### DB layer — `db/idempotency.py` The table's `PRIMARY KEY (scope, idempotency_key)` **is** the atomic claim. Atomicity relies on SQLite database-level write locking, which holds across processes (multiple uvicorn workers + the standalone scheduler share one DB file). @@ -42,6 +42,7 @@ The table's `PRIMARY KEY (scope, idempotency_key)` **is** the atomic claim. Atom - `attach_execution(scope, key, execution_id)` — `db/idempotency.py:87`. Best-effort `UPDATE` recording the dispatched `execution_id` on an in-flight claim, so an in-flight 409 can hand back a pollable id. - `complete(scope, key, execution_id, snapshot)` — `db/idempotency.py:96`. Sets `status='completed'`, `COALESCE`s the execution_id, JSON-encodes and stores `response_snapshot` for replay. - `release(scope, key)` — `db/idempotency.py:126`. Deletes the row **only when still `in_flight`** so a failed first attempt can retry; never removes a `completed` row (which must stay to keep replaying the original result). +- `discard_completed_if_execution(scope, key, execution_id) -> bool` (#3245). Deletes a `completed` row **only while it still names `execution_id`**; returns whether a row went. `execution_id` is required. Unlike #2040's `discard_completed` (any completed row for the key), a slow retry cannot erase the record a faster retry re-completed with its new run. - `purge_expired(ttl_hours=24)` — `db/idempotency.py:139`. Deletes rows past the TTL; returns the row count. Time math uses `iso_cutoff(hours)` and `utc_now_iso()` from `utils/helpers.py` (Invariant #16 — ISO-Z TEXT columns). @@ -56,9 +57,13 @@ Time math uses `iso_cutoff(hours)` and `utc_now_iso()` from `utils/helpers.py` ( - `attach_execution(decision, execution_id)` — `idempotency_service.py:112`. No-op on replay / disabled / missing id. - `complete(decision, execution_id, snapshot)` — `idempotency_service.py:122`. Finalizes a fresh claim; no-op on replay / disabled. - `fail(decision)` — `idempotency_service.py:132`. Releases a fresh in-flight claim so a failed first attempt can retry; no-op on replay / disabled. +- **Replay liveness (#3245)**: + - `execution_liveness(execution_id, agent_name, *, now=None) -> Liveness(verdict, hold_until)`. Never raises. `live` (`queued`/`running`/`pending_retry`), `succeeded`, `ended` (`failed`/`cancelled`/`skipped`), `gone` (no row), `maybe_alive`, `indeterminate` (no id or `manual`, read error, another agent's row, unknown status). Reads `get_execution_gate_state` and, only on `failed`, `get_execution_failure_stamp`. A `failed` row whose `error` starts `lease_expired:` (the slot reaper's mark) is `maybe_alive` until `lease_expired_horizon(completed_at, agent)` = `completed_at` + the agent's current timeout + `SLOT_TTL_BUFFER` (300 s), then `ended`; the tag is checked before the stamp or the timeout, and a missing/unparseable stamp or an unreadable timeout is `indeterminate`. `_now()` is the clock hook. + - `discard_replay_of(scope, key, execution_id) -> True | False | None` wraps the compare-and-delete; `None` = the delete raised (logged, swallowed). ### Router flow (representative — `/chat`, `routers/chat.py`) 1. `begin(make_agent_scope(name), idempotency_key)` at `chat.py:148` (inside `_admit_chat_request`) — gated before consuming a capacity slot. +1a. **Stale-receipt reclaim (#3245)**: `dispatch_admission_service._reclaim_ended_receipt(idem, name=...)` runs right after `begin` in `admit_chat_request` (`/chat`) and `begin_task_idempotency` (`/task`). Only a `completed` replay whose snapshot is a dispatch receipt (`async_mode: true`, status `accepted`/`queued`/`queued_timeout`) is checked; `sched:` keys are skipped; the skill gate has already answered a gated request. On `ended`/`gone` it compare-and-deletes the row and calls `begin` once more (no loop): `new` dispatches a fresh run, a concurrent winner's `in_flight` → 409, its `completed` receipt → replay of the winner's run. A delete error replays the original. Logs `[#3245] receipt_reclaim … cas=deleted|no_match|error outcome=new|in_flight|completed|no_dedup|replay_original` and, for a held `lease_expired` failure, `[#3245] receipt_hold … until=` (never the raw key). 2. **Replay path** (`idem.replay`, `chat.py:151`): writes a `EXECUTION / idempotent_replay` platform-audit event; if `idem.in_flight`, raises `409 {error: "request_in_progress", execution_id}` (`chat.py:169`); otherwise returns `idem.snapshot` (or a minimal `{execution.task_execution_id}`) with header `X-Idempotent-Replay: true` (`chat.py:181`). 3. **Fresh path**: dispatch through `CapacityManager`. `attach_execution(idem, task_execution_id)` at `chat.py:361` (inside `_prepare_chat_execution`). On a final response, `complete(idem, task_execution_id, response_data)` at `chat.py:755` (inside `_run_chat_and_finalize`). On an upfront rejection where **nothing was dispatched** — capacity-full (`chat.py:264`) **or** dispatch-breaker-open (`chat.py:216`, #526, added by #1051) — `fail(idem)` releases the claim so the caller can retry with the same key once capacity frees or the breaker recovers. @@ -77,9 +82,10 @@ Idempotency covers the whole batch — a duplicate replays the original `FanOutR `src/scheduler/service.py:1042` sets `headers["Idempotency-Key"] = f"sched:{execution_id}"` when dispatching to the backend internal endpoint. The execution_id is created once per fire and reused across an HTTP-level resend of the same dispatch, so a transient backend 5xx + resend resolves to the same key and short-circuits the duplicate. Intentional #271 retries create a fresh execution_id → fresh key → not suppressed. ## MCP Layer (Invariant #13 — third surface in sync) -- `src/mcp-server/src/tools/chat.ts:20` — `deriveMcpIdempotencyKey(parts)` returns `mcp:{sha256(parts.join(" "))}`. -- `chat_with_agent` derives the key over `[caller, agent, "chat"|"task", model, "sync"|"async", message]` (`chat.ts:301`) and passes it into `apiClient.task(...)` / `apiClient.chat(...)`. -- `fan_out` derives the key over `[caller, agent, "fan_out", model, JSON.stringify(tasks)]` (`chat.ts:518`) and passes it into `apiClient.fanOut(...)`. +- `src/mcp-server/src/tools/chat.ts:23` — `deriveMcpIdempotencyKey(parts)` returns `mcp:{sha256(parts.join(" "))}`. +- `chat_with_agent` derives the key over `[caller, agent, "chat"|"task", model, "sync"|"async", message]` (`chat.ts:284`) and passes it into `apiClient.task(...)` / `apiClient.chat(...)`. +- `fan_out` derives the key over `[caller, agent, "fan_out", model, JSON.stringify(tasks)]` (`chat.ts:746`) and passes it into `apiClient.fanOut(...)`. +- **Replay marker (#3245)**: `client.ts` `markIdempotentReplay` copies `X-Idempotent-Replay: true` into the body as `idempotent_replay: true` on `task()` / `chat()` 200 replays, and the 409 in-flight receipt (`inFlightReplayReceipt`) carries it too. The client's own timeout-recovery receipts carry no field — absence is not proof of a fresh dispatch. `runAgentChat`'s receipt re-wording keeps the field. - `src/mcp-server/src/client.ts` forwards the key as the `Idempotency-Key` header on the `chat` / `task` / `fanOut` methods (`client.ts:485,498-501`, `client.ts:647,656-658`, `client.ts:762,787-789`). A transport-level retry of a byte-identical MCP call within 24h dedupes. ## Database @@ -115,6 +121,10 @@ CREATE TABLE idempotency_keys ( | First-seen key | Claim taken `in_flight`; request dispatches | | Duplicate, prior claim still running | `409` (`request_in_progress` for chat/task, `detail` string for webhook), with original `execution_id` where known | | Duplicate, prior claim completed | `200`/`202` replay of stored `response_snapshot` + `X-Idempotent-Replay: true` | +| Duplicate on `/chat` or `/task`, stored receipt's run ended `failed`/`cancelled`/`skipped` or its row is gone (#3245) | Row compare-and-deleted, key re-claimed: a **new** run, no replay header | +| Same, but the `failed` error starts `lease_expired:` and the hold window (`completed_at` + timeout + 300 s) is still open | Replay of the original receipt (logged `receipt_hold`); fresh once the window ends | +| Same, but the key is `sched:…` | Replay (the scheduler's own key is never reclaimed) | +| Same, but the request names a gated skill with a gate record | The gate's answer (pending / `409 request_`) — the receipt is not checked | | Upfront at-capacity rejection (chat/task) | Claim released (`fail`) so caller can retry once capacity frees | | Dispatch error (webhook → scheduler) | Claim released (`fail`) so a legitimate re-delivery can retry | | Dedup-layer DB error | **Fail-open** — `begin` logs a warning and returns a no-dedup decision; the real execution always proceeds | @@ -137,6 +147,10 @@ CREATE TABLE idempotency_keys ( **Expected**: normal dispatch; **no** `idempotency_keys` row created. 6. **TTL purge**: age a row past 24h (or wait), run a cleanup cycle. **Verify**: cleanup report shows `idempotency_keys_purged > 0`; row gone. +7. **Retry after a failed async run (#3245)**: `POST /api/agents/{name}/task` with `{"message": "...", "async_mode": true}` and `Idempotency-Key: t7` to a running agent, then end the run `failed`/`cancelled` (for example `POST /api/agents/{name}/executions/{execution_id}/terminate`; a request to a stopped agent is refused with 503 before any key is claimed, so it tests nothing); re-send the identical request. + **Expected**: a 200 receipt with a **new** `execution_id` and no `X-Idempotent-Replay` header. A third identical send while that new run is still running replays it (`X-Idempotent-Replay: true`, same new id). + +Unit coverage: `tests/unit/test_3245_async_replay_after_failure.py` (verdict matrix, `lease_expired` time axis, both seams incl. the pulled `/chat` receipt, AC2 over the real `/task` endpoint, the race, the log lines, structural pins); `src/mcp-server/src/task-receipt.test.ts` (the `idempotent_replay` marker). ## Related Flows - [Webhook Triggers](webhook-triggers.md) — the boundary that auto-derives `(token, body_hash)`. @@ -150,3 +164,4 @@ CREATE TABLE idempotency_keys ( | Date | ID | Change | |------|-----|--------| | 2026-06-02 | RELIABILITY-006 (#525), PR #1019 (commit 1bb8f271) | Initial implementation — `idempotency_keys` table + migration, `db/idempotency.py`, `services/idempotency_service.py`, wiring at `/chat`, `/task`, `/api/internal/execute-task`, `/api/webhooks/{token}`, `/api/agents/{name}/fan-out`, the scheduler, and MCP `chat_with_agent`/`fan_out`; 24h purge in the cleanup service. Established Architectural Invariant #18. | +| 2026-10-06 | #3245 | A dispatch receipt replays only while its run can still succeed: `execution_liveness` + `discard_replay_of` (service), `discard_completed_if_execution` + `get_execution_failure_stamp` (DB), `_reclaim_ended_receipt` at the `/chat` and `/task` admission seams, a `lease_expired` hold window, and the MCP `idempotent_replay` marker. | diff --git a/docs/memory/feature-flows/mcp-orchestration.md b/docs/memory/feature-flows/mcp-orchestration.md index 91c3995eb5..a075580739 100644 --- a/docs/memory/feature-flows/mcp-orchestration.md +++ b/docs/memory/feature-flows/mcp-orchestration.md @@ -497,7 +497,12 @@ chat_with_agent({ // right for a REST caller, impossible for an MCP one — so `runAgentChat` rewrites the // `message` of an `accepted`/`queued` receipt (ent#568); every other field is untouched. // The lead claims nothing about liveness: the backend stores this receipt as the -// idempotency snapshot at dispatch and replays it for 24 h (#3245). +// idempotency snapshot at dispatch and replays it within 24 h while the run is live, +// succeeded or may still report back (a `lease_expired` failure inside its hold +// window). Since #3245 a run that ended failed/cancelled is NOT replayed — an identical +// re-send starts a new run — and a replayed answer carries `"idempotent_replay": true` +// (the client copies the backend's `X-Idempotent-Replay` header into the body; absence +// is not proof of a fresh dispatch). ``` **When `async: true` (with `parallel: true`)**: @@ -648,7 +653,8 @@ silent wrong data, strictly worse than the loud error it replaced. | `/task` recovers on `AbortError` **only** | A `TypeError` may mean the request never landed, and on this route a concurrent peer row is the normal state | **Retry guidance the tool description now states:** an *identical* re-send is -deduplicated server-side and answers with the original `execution_id`; a +deduplicated server-side and answers with the original `execution_id` (since +#3245, not once that run ended `failed` or `cancelled`); a **reworded** one derives a different idempotency key and dispatches a second execution. That asymmetry is why the opaque error was dangerous — it invited the rewrite. @@ -671,13 +677,18 @@ The receipts above only help a caller who knows what they mean. The rule is now re-send (`agent_busy` included — it is not "nothing ran", #3244); `pending_approval` is not retried; `parallel=true, async=true` for long work, whose end fires `agent.task.*` for a subscriber (a sequential `/chat` turn fires none). + #3245 added: after a confirmed `failed` or `cancelled`, re-send word for word + to retry; if the same `execution_id` comes back (a `lease_expired` failure is + held until its window ends), `set_reminder` and end the turn. - **The 2,048-char cap.** Claude Code cuts every MCP tool description at `CLAUDE_CODE_MAX_MCP_DESCRIPTION_LENGTH` (default 2,048). #2958 (2026-09-24) took the `chat_with_agent` description from 2,035 to 2,424 chars, so from then on Claude Code cut its last paragraph — the `list_recent_executions` advice and "prefer `parallel=true, async=true`" — before the model read it (other runtimes' MCP clients were not measured). It is now a - lead, one line of modes and the contract (1,940 chars published); per-mode detail moved into the `parallel` / `async` / + lead, one line of modes and the contract (2,030 chars published since #3245, which + added the confirmed-failure bullet, raised the contract's own pin 1,650 → 1,730 + and dropped "(the primary way to use sub-agents)" from the lead); per-mode detail moved into the `parallel` / `async` / `timeout_seconds` parameter descriptions, which are not cut. `src/delegation-contract.test.ts` reads every carrying description back through a real `tools/list` and pins it under the cap. diff --git a/docs/memory/feature-flows/system-wide-trinity-prompt.md b/docs/memory/feature-flows/system-wide-trinity-prompt.md index 43afb05696..f2ba498c60 100644 --- a/docs/memory/feature-flows/system-wide-trinity-prompt.md +++ b/docs/memory/feature-flows/system-wide-trinity-prompt.md @@ -119,7 +119,7 @@ Platform instructions are injected at runtime on every Claude Code invocation ### Platform Prompt Service (Single Source of Truth) - `src/backend/services/platform_prompt_service.py` - `PLATFORM_INSTRUCTIONS` constant — static platform instructions (collaboration tools, operator queue ref, package persistence) - - `DELEGATION_CONTRACT` (ent#568) — spliced into §Agent Collaboration at the `__DELEGATION_CONTRACT__` marker, the `__REPORT_PAYLOAD_MAX__` pattern: what a dispatch receipt means and what to do instead of re-sending. It is byte-identical to `src/mcp-server/src/delegation_contract.ts`, which the `chat_with_agent` / `chat_with_` descriptions carry, so at `PromptTier.MINIMAL` (which drops §Agent Collaboration) the description is the copy. Bare tool names only, no `###` heading. `tests/unit/test_ent568_delegation_contract.py` pins parity, every tool / argument / status it names, and its budget (≤ 1,650 chars; §Agent Collaboration shorter than §Operator Communication) + - `DELEGATION_CONTRACT` (ent#568) — spliced into §Agent Collaboration at the `__DELEGATION_CONTRACT__` marker, the `__REPORT_PAYLOAD_MAX__` pattern: what a dispatch receipt means and what to do instead of re-sending. It is byte-identical to `src/mcp-server/src/delegation_contract.ts`, which the `chat_with_agent` / `chat_with_` descriptions carry, so at `PromptTier.MINIMAL` (which drops §Agent Collaboration) the description is the copy. Bare tool names only, no `###` heading. `tests/unit/test_ent568_delegation_contract.py` pins parity, every tool / argument / status it names, and its budget (≤ 1,730 chars since #3245 added the confirmed-failure retry bullet; §Agent Collaboration shorter than §Operator Communication) - `get_platform_system_prompt()` — combines static instructions with `trinity_prompt` DB setting ### Sections, tiers, and the two one-shot-runtime rules (#2454 → #2468) diff --git a/docs/memory/requirements/scheduling.md b/docs/memory/requirements/scheduling.md index 74b4cceead..04f3c3ade7 100644 --- a/docs/memory/requirements/scheduling.md +++ b/docs/memory/requirements/scheduling.md @@ -510,6 +510,7 @@ - MCP `chat_with_agent` / `fan_out` forward a deterministic key over the call args so a transport retry dedupes. - Header is OPTIONAL on chat/task/MCP (absent → no dedup, full back-compat); upfront at-capacity rejections release the claim so the caller can retry; in-flight duplicate → 409. - Audit event `idempotent_replay` on every replay (duplicate-storms observable); 24h TTL purge folded into the cleanup-service retention sweep. + - **A dispatch receipt is replayed only while its run can still succeed (#3245).** On `/chat` and `/task`, a stored receipt (`async_mode: true`, status `accepted` / `queued` / `queued_timeout`) whose execution ended `failed`, `cancelled` or `skipped`, or whose execution row is gone, is not replayed: an identical request dispatches a new run under the same key. **Exception:** a `failed` run whose error starts `lease_expired:` (the slot reaper's mark, written when no result arrived before the slot TTL) may still report success, so it keeps replaying until the agent's execution timeout + 5 min after the failure, then goes fresh. Carve-outs: a gated skill's approval record answers before the receipt is checked; `sched:` keys (the scheduler's own) are never reclaimed; non-receipt snapshots (sync results) replay as before. Concurrency: identical retries that observe the same dead run start one new run (compare-and-delete on the stored execution id), barring a dedupe-layer failure. Effects can repeat across attempts — effect dedupe is per execution, and a new attempt has a new execution id. The MCP client marks a replayed receipt `idempotent_replay: true`. - **Architectural Invariant**: #18 — every new trigger type must accept an `Idempotency-Key` before merge. #### 10.10.1 Effect-Scoped Idempotency for Outbound Side Effects @@ -1372,7 +1373,11 @@ schedules: Never re-send because a call timed out or its delivery could not be confirmed. Read with `get_execution_result` / `get_fan_out_result`; to finish later, `set_reminder` naming the `execution_id` and end the turn; - never report delegated work as done from a receipt. An error without an + never report delegated work as done from a receipt. After a confirmed + `failed` or `cancelled` outcome, re-send word for word to request a new + attempt; if the same `execution_id` comes back (a `lease_expired` failure + keeps replaying until its hold window ends, §10.10), `set_reminder` and end + the turn (#3245). An error without an `execution_id` — `agent_busy` included — is checked against `list_recent_executions` before a word-for-word re-send (`agent_busy` after `retry_after_seconds`). `agent_busy` is not "nothing ran": the MCP client @@ -1407,8 +1412,12 @@ schedules: `parallel=true, async=true`" before the model read them (other runtimes' MCP clients were not measured). Every description that carries the contract is pinned under the cap on the text the server - publishes (1,940 for `chat_with_agent`), and mode details moved into - parameter descriptions, which are not cut. + publishes (2,030 for `chat_with_agent`), and mode details moved into + parameter descriptions, which are not cut. #3245 added the + confirmed-failure bullet: the contract budget was raised deliberately from + 1,650 to 1,730 chars (the contract is 1,720), and `chat_with_agent`'s lead + dropped "(the primary way to use sub-agents)" to stay under the cap (it + would publish 2,066 otherwise). - **Receipt text.** Every `execution_id` receipt a `chat_with_*` caller gets carries the same "Do not re-send: read the outcome with `get_execution_result(agent_name=…, execution_id=…)`" line: the @@ -1418,7 +1427,9 @@ schedules: whose REST "poll GET …" line `runAgentChat` rewrites. Those rewritten leads claim nothing about liveness ("it may still be running or already done"): the backend stores the async receipt as the idempotency snapshot at dispatch - and replays it for 24 h, even after the run failed (#3245). `fan_out_timeout` + and replays it while the run is live, succeeded or may still report back — + since #3245 a run that ended `failed`/`cancelled` is not replayed, and a + replayed answer carries `idempotent_replay: true` (§10.10). `fan_out_timeout` keeps its own `get_fan_out_result` line (#2670). The "concurrent-duplicate guard will kill mid-execution" claim the receipts and the old description made is gone — no such guard exists in the backend or the agent server. @@ -1427,7 +1438,7 @@ schedules: typed outcome (ent#569), the unforwarded `chat_with_agent` `execution_id` (#3232), the system-scoped gate-outcome gap (#3233), the post-run 429 labelled `agent_busy` (#3244), the async receipt replayed after a failed run - (#3245), and the trinity-pm canon pointer (AC6, post-merge). + (#3245 — since fixed, §10.10), and the trinity-pm canon pointer (AC6, post-merge). ## 38. Sequential Agent Loops (#740) diff --git a/docs/user-docs/api-reference/chat-api.md b/docs/user-docs/api-reference/chat-api.md index 0c85539b96..9e5734fc2e 100644 --- a/docs/user-docs/api-reference/chat-api.md +++ b/docs/user-docs/api-reference/chat-api.md @@ -100,11 +100,47 @@ curl -X POST http://localhost:8000/api/agents/my-agent/task \ -d '{"message": "Summarize the latest reports"}' ``` -- The same key within 24 hours returns the original result with the header `X-Idempotent-Replay: true` — no second execution is created. +- The same key within 24 hours returns the original result with the header `X-Idempotent-Replay: true` — no second execution is created. The one exception is a **receipt** whose run has ended without success (see the table below): then the same key starts a new run. - A duplicate sent while the first request is still running returns **409** with the original `execution_id` to poll (for `/fan-out`, that field carries the batch's `fan_out_id`). - If the first attempt was rejected before dispatch (e.g., at capacity), the key is released so the retry goes through. - The header is optional and fail-open: omitting it preserves normal behavior, and a dedup-layer error never blocks a real request. +### Retrying after a failed run (`/chat` and `/task`) + +An async `/task` answers with a receipt — `{"status": "accepted" | "queued" | "queued_timeout", "execution_id": ..., "async_mode": true}`. A sync `/task`, or a `/chat` turn on a pull-mode agent, that outlives its wait answers `504` but stores the same kind of receipt (`queued_timeout`). That stored receipt is what an identical request replays, as a `200` with `X-Idempotent-Replay: true`. Whether it is replayed depends on how its run is doing: + +| The receipt's run is… | An identical request with the same key gets | +|---|---| +| `queued`, `running` or `pending_retry` | the original receipt (`X-Idempotent-Replay: true`) | +| `success` | the original receipt — read the result from the execution | +| `failed`, `cancelled` or `skipped`, or the execution no longer exists | a **new** run with a new `execution_id`, no replay header | +| `failed` with an error starting `lease_expired:` (no result arrived before the slot expired, so the run may still report back) | the original receipt until the agent's timeout + 5 minutes after the failure; after that, a new run | + +A non-receipt result (a completed sync reply) always replays. Scheduler-issued keys (`sched:…`) are never reclaimed, and a request for an approval-gated skill is answered by its approval record first. + +```bash +# 1. Dispatch async with a key +curl -s -X POST http://localhost:8000/api/agents/my-agent/task \ + -H "Authorization: Bearer " -H "Idempotency-Key: report-2026-10-06" \ + -H "Content-Type: application/json" \ + -d '{"message": "Summarize the latest reports", "async_mode": true}' +# -> 200 {"status": "accepted", "execution_id": "", "async_mode": true, ...} + +# 2. Check the outcome +curl -s -H "Authorization: Bearer " \ + http://localhost:8000/api/agents/my-agent/executions/ +# -> {"status": "failed", ...} + +# 3. Re-send the identical request: a new run, no X-Idempotent-Replay header +curl -s -X POST http://localhost:8000/api/agents/my-agent/task \ + -H "Authorization: Bearer " -H "Idempotency-Key: report-2026-10-06" \ + -H "Content-Type: application/json" \ + -d '{"message": "Summarize the latest reports", "async_mode": true}' +# -> 200 {"status": "accepted", "execution_id": "", ...} +``` + +Checking the status tells you **whether** to retry; it cannot make a retry safe. Anything the failed run already did (a message sent, a file written) may happen again in the new run — effect de-duplication is per execution, and the new run has a new `execution_id`. Two identical retries that arrive together start one new run. + Wired boundaries: `/api/agents/{name}/chat`, `/api/agents/{name}/task`, `/api/agents/{name}/fan-out`, `/api/agents/{name}/voip/call`, [webhook triggers](webhook-triggers.md) (key auto-derived from token + body when the header is absent), and the MCP `chat_with_agent` / `fan_out` tools (deterministic key derived from the call arguments). ## See Also diff --git a/docs/user-docs/faq/mcp-and-api.md b/docs/user-docs/faq/mcp-and-api.md index d6f434423b..75001a7fd4 100644 --- a/docs/user-docs/faq/mcp-and-api.md +++ b/docs/user-docs/faq/mcp-and-api.md @@ -81,11 +81,11 @@ JWT tokens are invalidated when the backend restarts, so every logged-in session ## What is the Idempotency-Key header for? -It makes retries safe on endpoints that trigger an execution (`/chat`, `/task`, `/fan-out`, VoIP calls, webhook triggers). Send any unique string per logical request; if you resend the same key within 24 hours, Trinity returns the original result with the header `X-Idempotent-Replay: true` instead of creating a second execution. A duplicate sent while the first request is still running returns 409 with the original `execution_id` to poll (for `/fan-out`, that field carries the batch's `fan_out_id`). A first attempt that was rejected before dispatch, or a sync `/task` that failed, timed out, or was cancelled, releases the key so a legitimate retry goes through. The header is optional and fail-open — omitting it preserves normal behavior. See [Chat API](../api-reference/chat-api.md#idempotency). +It makes retries safe on endpoints that trigger an execution (`/chat`, `/task`, `/fan-out`, VoIP calls, webhook triggers). Send any unique string per logical request; if you resend the same key within 24 hours, Trinity returns the original result with the header `X-Idempotent-Replay: true` instead of creating a second execution — except that a stored `/chat` or `/task` receipt whose run ended `failed`, `cancelled` or `skipped` is not replayed: the same key then starts a new run (a `lease_expired` failure first waits out the agent's timeout + 5 minutes). A duplicate sent while the first request is still running returns 409 with the original `execution_id` to poll (for `/fan-out`, that field carries the batch's `fan_out_id`). A first attempt that was rejected before dispatch, or a sync `/task` that failed, timed out, or was cancelled, releases the key so a legitimate retry goes through. The header is optional and fail-open — omitting it preserves normal behavior. See [Chat API](../api-reference/chat-api.md#idempotency). ## Why is `chat_with_agent` returning `queued_timeout` instead of a reply? -Synchronous `chat_with_agent` calls — sequential chat (`parallel=false`) and the sync task route (`parallel=true, async=false`) alike — give up at `MCP_CHAT_TIMEOUT_MS` (default 25 seconds), before the MCP gateway does. Instead of a transport error such as `fetch failed`, the tool returns a receipt `{status: "queued_timeout", agent, execution_id, message}`; the task keeps running on the agent, and you poll `get_execution_result` with that `execution_id`. The receipt is issued only when the running execution can be attributed to *your* call unambiguously — otherwise the error says so and points you at `list_recent_executions`. Calls carry a deterministic idempotency key, so an identical re-send answers with the same `execution_id`, but a **reworded** re-send is a new call and dispatches a second execution. For work you know will run long, use `parallel=true, async=true` from the start. See [MCP Server](../integrations/mcp-server.md#key-tools-worth-knowing). +Synchronous `chat_with_agent` calls — sequential chat (`parallel=false`) and the sync task route (`parallel=true, async=false`) alike — give up at `MCP_CHAT_TIMEOUT_MS` (default 25 seconds), before the MCP gateway does. Instead of a transport error such as `fetch failed`, the tool returns a receipt `{status: "queued_timeout", agent, execution_id, message}`; the task keeps running on the agent, and you poll `get_execution_result` with that `execution_id`. The receipt is issued only when the running execution can be attributed to *your* call unambiguously — otherwise the error says so and points you at `list_recent_executions`. Calls carry a deterministic idempotency key, so an identical re-send answers with the same `execution_id` — marked `"idempotent_replay": true` — while that run is live or has succeeded; a **reworded** re-send is a new call and dispatches a second execution. Once `get_execution_result` shows the run `failed` or `cancelled`, re-send the call word for word to retry it. A run that failed with `lease_expired` is the exception: it keeps answering with the same `execution_id` for the agent's timeout + 5 minutes after the failure, so if the same `execution_id` comes back, set a reminder and try again later (`idempotent_replay` is positive evidence of a replay; its absence does not prove a fresh run). For work you know will run long, use `parallel=true, async=true` from the start. See [MCP Server](../integrations/mcp-server.md#key-tools-worth-knowing). ## How do I poll a fan-out batch that outlived the call? diff --git a/docs/user-docs/integrations/mcp-server.md b/docs/user-docs/integrations/mcp-server.md index f1bc7801b3..c3f4505b1e 100644 --- a/docs/user-docs/integrations/mcp-server.md +++ b/docs/user-docs/integrations/mcp-server.md @@ -156,7 +156,7 @@ Every refusal is recorded in the audit log as a refusal, not as a successful cal | Tool | Why it exists | |------|---------------| -| `chat_with_agent` | Send a message to another agent. **Gateway-timeout safe in every sync mode** — sequential chat (`parallel=false`) and the sync task route (`parallel=true, async=false`) alike: if the call exceeds `MCP_CHAT_TIMEOUT_MS` (default 25s), it returns `{status: "queued_timeout", agent, execution_id, message}` so the caller polls `get_execution_result` instead of duplicate-queueing the request. The receipt is only issued when the running execution can be attributed to *your* call unambiguously; otherwise the error says so and names `list_recent_executions`. Calls carry a deterministic idempotency key, so an identical re-send dedupes server-side and answers with the original `execution_id` — a **reworded** re-send is a new call and dispatches a second execution. For work you know will outlive the gateway, use `parallel=true, async=true` from the start. See [Agent Network](../collaboration/agent-network.md) for the async pattern. | +| `chat_with_agent` | Send a message to another agent. **Gateway-timeout safe in every sync mode** — sequential chat (`parallel=false`) and the sync task route (`parallel=true, async=false`) alike: if the call exceeds `MCP_CHAT_TIMEOUT_MS` (default 25s), it returns `{status: "queued_timeout", agent, execution_id, message}` so the caller polls `get_execution_result` instead of duplicate-queueing the request. The receipt is only issued when the running execution can be attributed to *your* call unambiguously; otherwise the error says so and names `list_recent_executions`. Calls carry a deterministic idempotency key, so an identical re-send dedupes server-side and answers with the original `execution_id`, marked `"idempotent_replay": true`, while that run is live or has succeeded — a **reworded** re-send is a new call and dispatches a second execution. After a confirmed `failed` or `cancelled` outcome, the identical re-send starts a new run instead (a `lease_expired` failure first waits out the agent's timeout + 5 minutes). `idempotent_replay` is positive evidence only: its absence does not prove a fresh run. For work you know will outlive the gateway, use `parallel=true, async=true` from the start. See [Agent Network](../collaboration/agent-network.md) for the async pattern. | | `fan_out` | Dispatch N independent tasks to an agent in parallel and collect all the results. **Gateway-timeout safe**: a batch runs longer than any single task in it, so this is the tool most likely to outlive the 25s ceiling — when it does, it returns `{status: "fan_out_timeout", agent, fan_out_id, execution_ids, task_count, message}` and the batch keeps running. Poll `get_fan_out_result(agent_name, fan_out_id)`. Re-sending the *identical* call is deduplicated server-side and answers with the same batch; **rewording it dispatches all N tasks again**. For a batch you know will run long, pass `async_mode: true`: the tool returns `{fan_out_id, status: "accepted"}` at once and you poll. `timeout_seconds` bounds only the wait — tasks still open at the deadline report `running` and keep going. See [Fan-Out](../automation/fan-out.md). | | `get_fan_out_result` | Poll a fan-out batch: `running` while any task can still change, then `completed`, `partial` (some succeeded — normal for a best-effort batch) or `failed`, with per-task status and results. Each result carries the `task_id` you gave the task. | | `run_agent_loop` | Run the same task against an agent repeatedly (bounded, sequential), with templated messages and an optional stop signal. Poll with `get_loop_status`; stop gracefully with `stop_loop`. An agent-scoped key can loop only on itself or on agents it has permission to call — the same rule as `chat_with_agent` — and the refusal happens before any loop starts. `get_loop_status` and `stop_loop` apply the same rule to the loop's agent. See [Agent Loops](../automation/agent-loops.md). | diff --git a/src/backend/database.py b/src/backend/database.py index 87965245d7..ec29ae5540 100644 --- a/src/backend/database.py +++ b/src/backend/database.py @@ -1904,6 +1904,10 @@ def get_execution(self, execution_id: str): def get_execution_gate_state(self, execution_id: str): return self._schedule_ops.get_execution_gate_state(execution_id) + def get_execution_failure_stamp(self, execution_id: str): + """`(status, error, completed_at)` or None — the #3245 liveness read.""" + return self._schedule_ops.get_execution_failure_stamp(execution_id) + def get_fan_out_executions(self, agent_name: str, fan_out_id: str, limit: int = 200): """Every execution row of one fan-out batch (#2670).""" return self._schedule_ops.get_fan_out_executions(agent_name, fan_out_id, limit) @@ -4244,6 +4248,10 @@ def idempotency_discard_completed(self, scope: str, key: str) -> None: """Delete a completed replay row whose recorded resource is gone (#2040 F3).""" return self._idempotency_ops.discard_completed(scope, key) + def idempotency_discard_completed_if_execution(self, scope: str, key: str, execution_id: str) -> bool: + """Compare-and-delete a completed row still naming execution_id (#3245).""" + return self._idempotency_ops.discard_completed_if_execution(scope, key, execution_id) + def idempotency_purge_expired(self, ttl_hours: int = 24) -> int: """Purge idempotency rows older than ttl_hours. Returns rows removed.""" return self._idempotency_ops.purge_expired(ttl_hours=ttl_hours) diff --git a/src/backend/db/idempotency.py b/src/backend/db/idempotency.py index 8b32ceb561..9b32c3ff7d 100644 --- a/src/backend/db/idempotency.py +++ b/src/backend/db/idempotency.py @@ -236,6 +236,31 @@ def discard_completed(self, scope: str, key: str) -> None: ) ) + def discard_completed_if_execution(self, scope: str, key: str, execution_id: str) -> bool: + """Compare-and-delete a COMPLETED row only while it still names ``execution_id``. + + #3245: a stored dispatch receipt whose run ended is dropped so an + identical retry can dispatch fresh. Two retries that both saw the dead + run's row race here: the faster one deletes it, re-claims and + re-completes the key with its NEW execution id; the slower one's delete + must then match nothing, or it would erase the fresh record and start a + second run (``discard_completed`` deletes any completed row, so it is + not safe here). ``execution_id`` is required — there is no + unconditional form. Returns True iff a row was deleted. + """ + with get_engine().begin() as conn: + result = conn.execute( + delete(idempotency_keys).where( + and_( + idempotency_keys.c.scope == scope, + idempotency_keys.c.idempotency_key == key, + idempotency_keys.c.status == STATE_COMPLETED, + idempotency_keys.c.execution_id == execution_id, + ) + ) + ) + return (result.rowcount or 0) > 0 + def purge_expired(self, ttl_hours: int = 24) -> int: """Delete rows older than ttl_hours. Returns rows removed.""" cutoff = iso_cutoff(hours=ttl_hours) diff --git a/src/backend/db/schedules/executions.py b/src/backend/db/schedules/executions.py index 1d290600ae..dba7411e02 100644 --- a/src/backend/db/schedules/executions.py +++ b/src/backend/db/schedules/executions.py @@ -800,6 +800,22 @@ def get_execution_gate_state(self, execution_id: str) -> Optional[tuple]: row = conn.execute(stmt).first() return (row.agent_name, row.status) if row else None + def get_execution_failure_stamp(self, execution_id: str) -> Optional[tuple]: + """`(status, error, completed_at)` of one execution, or None (#3245). + + The replay-liveness check reads this only after the gate-state read + said ``failed``, to tell a slot-reaper ``lease_expired`` FAILED (which + may still report success) from an ordinary one, and when it was + written. Three columns — never the row's response or log.""" + stmt = select( + schedule_executions.c.status, + schedule_executions.c.error, + schedule_executions.c.completed_at, + ).where(schedule_executions.c.id == execution_id) + with get_engine().connect() as conn: + row = conn.execute(stmt).first() + return (row.status, row.error, row.completed_at) if row else None + # ent#525 — the in-flight rows a Workspace chat is waiting on. _CHAT_INFLIGHT_STATUSES = ("running", "queued") diff --git a/src/backend/services/dispatch_admission_service.py b/src/backend/services/dispatch_admission_service.py index db9c050eb4..797f37cc08 100644 --- a/src/backend/services/dispatch_admission_service.py +++ b/src/backend/services/dispatch_admission_service.py @@ -350,6 +350,7 @@ async def admit_chat_request( idem = idempotency_service.begin( idempotency_service.make_agent_scope(name), idempotency_key ) + idem = _reclaim_ended_receipt(idem, name=name) if idem.replay: await _audit_idempotent_replay( name=name, endpoint=f"/api/agents/{name}/chat", x_via_mcp=x_via_mcp, @@ -457,6 +458,83 @@ async def admit_chat_request( ) +# #3245: the receipt shapes a dispatch stores as its `completed` row — async +# `/task` (`accepted`/`queued`), and the timed-out sync `/task` and pulled +# `/chat` (`queued_timeout`). A sync success body is never one of these. +_RECEIPT_STATUSES = frozenset({"accepted", "queued", "queued_timeout"}) +# Liveness verdicts after which a receipt replay would be a lie. Everything +# else (live, succeeded, maybe_alive, indeterminate) replays. +_FRESH_AFTER = frozenset({ + idempotency_service.LIVENESS_ENDED, idempotency_service.LIVENESS_GONE, +}) + + +def _is_dispatch_receipt(snapshot) -> bool: + return ( + isinstance(snapshot, dict) + and snapshot.get("async_mode") is True + and snapshot.get("status") in _RECEIPT_STATUSES + ) + + +def _decision_outcome(decision) -> str: + if not decision.enabled: + return "no_dedup" + if not decision.replay: + return "new" + return "in_flight" if decision.in_flight else "completed" + + +def _reclaim_ended_receipt(idem, *, name: str): + """Let a retry dispatch fresh when the receipt it would replay names a run + that ended without success (#3245). Shared by ``/chat`` and ``/task``. + + Before #3245 the stored receipt replayed for 24h even after its run + failed, so a caller retrying word for word got the dead run back forever. + This runs at replay time, between ``begin`` and the replay branch: it adds + no side effect to any terminal writer (#1804) and heals rows written + before the fix. Only receipt-shaped ``completed`` snapshots are checked; + ``sched:`` keys are the scheduler's own and never reclaimed. On an ended + or gone run the row is compare-and-deleted (only while it still names the + dead execution, so two racing retries start one run) and the key is + claimed again exactly once — no loop. A delete error replays the original. + Returns the decision the caller should act on. + """ + if not (idem.replay and not idem.in_flight and idem.execution_id): + return idem + if str(idem.key or "").startswith("sched:"): + return idem + if not _is_dispatch_receipt(idem.snapshot): + return idem + liveness = idempotency_service.execution_liveness(idem.execution_id, name) + verdict = liveness.verdict + if verdict == idempotency_service.LIVENESS_MAYBE_ALIVE: + hold_until = liveness.hold_until.isoformat() if liveness.hold_until else None + logger.info( + "[#3245] receipt_hold agent=%s stale_exec=%s verdict=%s until=%s", + name, idem.execution_id, verdict, hold_until, + ) + return idem + if verdict not in _FRESH_AFTER: + return idem + cas = idempotency_service.discard_replay_of(idem.scope, idem.key, idem.execution_id) + if cas is None: + logger.info( + "[#3245] receipt_reclaim agent=%s stale_exec=%s verdict=%s cas=error " + "outcome=replay_original replay_exec=%s", + name, idem.execution_id, verdict, idem.execution_id, + ) + return idem + decision = idempotency_service.begin(idem.scope, idem.key) + logger.info( + "[#3245] receipt_reclaim agent=%s stale_exec=%s verdict=%s cas=%s outcome=%s " + "replay_exec=%s", + name, idem.execution_id, verdict, "deleted" if cas else "no_match", + _decision_outcome(decision), decision.execution_id if decision.replay else None, + ) + return decision + + def begin_task_idempotency( *, name: str, @@ -473,6 +551,7 @@ def begin_task_idempotency( idem = idempotency_service.begin( idempotency_service.make_agent_scope(name), idempotency_key ) + idem = _reclaim_ended_receipt(idem, name=name) if idem.replay: return idem, ChatAdmissionReplay( execution_id=idem.execution_id, in_flight=idem.in_flight, snapshot=idem.snapshot, diff --git a/src/backend/services/idempotency_service.py b/src/backend/services/idempotency_service.py index 4a1abacec7..5122ab8027 100644 --- a/src/backend/services/idempotency_service.py +++ b/src/backend/services/idempotency_service.py @@ -20,10 +20,15 @@ import logging from contextlib import asynccontextmanager from dataclasses import dataclass -from typing import Any, AsyncIterator, Optional, Union +from datetime import datetime, timedelta +from typing import Any, AsyncIterator, NamedTuple, Optional, Union from database import db from db.idempotency import STATE_COMPLETED, STATE_IN_FLIGHT, STATE_NEW +from models import TaskExecutionStatus +from services.execution_envelope import TaskExecutionErrorCode +from services.slot_service import SLOT_TTL_BUFFER +from utils.helpers import parse_iso_timestamp, utc_now logger = logging.getLogger(__name__) @@ -763,6 +768,143 @@ def discard_stale_replay(scope: Optional[str], key: Optional[str]) -> None: logger.warning("Idempotency discard_stale_replay failed (scope=%s): %s", scope, e) +# --------------------------------------------------------------------------- +# Replay liveness (#3245) +# +# A stored dispatch receipt names an execution. Replaying it is truthful only +# while that run can still succeed; once it ended without success (or its row +# is gone) an identical request should dispatch a new run. These two helpers +# are the mechanism; the policy (which snapshots, which keys, when to reclaim) +# lives at the admission seam, `dispatch_admission_service._reclaim_ended_receipt`. +# --------------------------------------------------------------------------- + +LIVENESS_LIVE = "live" # queued / running / pending_retry +LIVENESS_SUCCEEDED = "succeeded" +LIVENESS_ENDED = "ended" # failed / cancelled / skipped — will not succeed +LIVENESS_GONE = "gone" # no execution row +LIVENESS_MAYBE_ALIVE = "maybe_alive" # lease_expired FAILED inside its hold window +LIVENESS_INDETERMINATE = "indeterminate" # cannot tell — callers must not re-run + +_LIVE_STATUSES = frozenset({ + TaskExecutionStatus.QUEUED.value, + TaskExecutionStatus.RUNNING.value, + TaskExecutionStatus.PENDING_RETRY.value, +}) +# Decision D1/Q6 (#3245): `skipped` and `cancelled` count as ended. +_ENDED_STATUSES = frozenset({ + TaskExecutionStatus.CANCELLED.value, + TaskExecutionStatus.SKIPPED.value, +}) +# The slot reaper's mark (`cleanup_service._LEASE_EXPIRED_TAG`, parity-tested). +# `.value`, never the member: `TaskExecutionErrorCode` is a fieldless +# `@dataclass` str-Enum whose members all compare equal (#1085). +_LEASE_EXPIRED_PREFIX = f"{TaskExecutionErrorCode.LEASE_EXPIRED.value}:" + + +class Liveness(NamedTuple): + """`execution_liveness` result. `hold_until` is set only for `maybe_alive`.""" + verdict: str + hold_until: Optional[datetime] = None + + +def _now() -> datetime: + """Aware-UTC clock; a module hook so tests can pin time without sleeping.""" + return utc_now() + + +def _status_value(status: Any) -> str: + return str(getattr(status, "value", status)) + + +def lease_expired_horizon(completed_at: Optional[str], agent_name: str) -> datetime: + """When a `lease_expired` FAILED stops being "maybe alive" (#3245, decision 1). + + ``completed_at`` (the moment the reaper wrote FAILED) + the agent's CURRENT + execution timeout + ``SLOT_TTL_BUFFER``. Anchored at the failure, not the + start: the reaper fires only after the slot TTL (timeout + buffer from + acquire) already ran out, so a start anchor would never hold. A heuristic, + not a bound — the agent's restart sweep re-posts results with no age limit. + Raises on a missing/unparseable stamp or an unreadable timeout. + """ + if not completed_at: + raise ValueError("no completed_at on a lease_expired FAILED row") + failed_at = parse_iso_timestamp(completed_at) # naive legacy → UTC, always aware + timeout = int(db.get_execution_timeout(agent_name)) + return failed_at + timedelta(seconds=timeout + SLOT_TTL_BUFFER) + + +def execution_liveness( + execution_id: Optional[str], agent_name: str, *, now: Optional[datetime] = None, +) -> Liveness: + """Can the run a stored receipt names still succeed? Never raises. + + Reads `get_execution_gate_state` (owner + status) and, only when that says + ``failed``, `get_execution_failure_stamp` (status, error, completed_at). + Every doubt resolves to ``indeterminate`` — which callers treat as "do not + re-run" — except a missing row, which is ``gone`` (decision 3: a vanished + row must never wedge a key). + """ + if not execution_id or execution_id == MANUAL_EXECUTION_ID: + return Liveness(LIVENESS_INDETERMINATE) + try: + state = db.get_execution_gate_state(execution_id) + except Exception as e: + logger.warning("execution_liveness: gate-state read failed for %s: %s", execution_id, e) + return Liveness(LIVENESS_INDETERMINATE) + if state is None: + # The effect resolver calls this `unknown` (no such execution); for + # replay purposes it is `gone`. + return Liveness(LIVENESS_GONE) + owner, status = state + if owner != agent_name: + return Liveness(LIVENESS_INDETERMINATE) + status = _status_value(status) + if status in _LIVE_STATUSES: + return Liveness(LIVENESS_LIVE) + if status == TaskExecutionStatus.SUCCESS.value: + return Liveness(LIVENESS_SUCCEEDED) + if status in _ENDED_STATUSES: + return Liveness(LIVENESS_ENDED) + if status != TaskExecutionStatus.FAILED.value: + return Liveness(LIVENESS_INDETERMINATE) + try: + stamp = db.get_execution_failure_stamp(execution_id) + except Exception as e: + logger.warning("execution_liveness: failure-stamp read failed for %s: %s", execution_id, e) + return Liveness(LIVENESS_INDETERMINATE) + if stamp is None or _status_value(stamp[0]) != TaskExecutionStatus.FAILED.value: + return Liveness(LIVENESS_INDETERMINATE) # changed between the two reads + _, error, completed_at = stamp + # Tag first: an ordinary failure is ended without needing a timestamp or + # the timeout, so neither can hold it. + if not (isinstance(error, str) and error.startswith(_LEASE_EXPIRED_PREFIX)): + return Liveness(LIVENESS_ENDED) + try: + hold_until = lease_expired_horizon(completed_at, agent_name) + except Exception as e: + logger.warning("execution_liveness: no lease_expired horizon for %s: %s", execution_id, e) + return Liveness(LIVENESS_INDETERMINATE) + current = now if now is not None else _now() + if current < hold_until: + return Liveness(LIVENESS_MAYBE_ALIVE, hold_until) + return Liveness(LIVENESS_ENDED) + + +def discard_replay_of( + scope: Optional[str], key: Optional[str], execution_id: Optional[str], +) -> Optional[bool]: + """Compare-and-delete the completed row for (scope, key) while it still + names ``execution_id`` (#3245). True deleted, False no match (another retry + already moved the key on), None the delete raised (logged, swallowed).""" + if not scope or not key or not execution_id: + return False + try: + return bool(db.idempotency_discard_completed_if_execution(scope, key, execution_id)) + except Exception as e: + logger.warning("Idempotency discard_replay_of failed (scope=%s): %s", scope, e) + return None + + def fail(decision: IdempotencyDecision) -> None: """Release a fresh in-flight claim so a failed first attempt can be retried.""" if not decision.enabled or decision.replay: diff --git a/src/backend/services/platform_prompt_service.py b/src/backend/services/platform_prompt_service.py index e47454c1eb..ad5968eff7 100644 --- a/src/backend/services/platform_prompt_service.py +++ b/src/backend/services/platform_prompt_service.py @@ -287,6 +287,7 @@ - A `chat_with_*` or `fan_out` call answers with the reply or with a receipt: an `execution_id` (a `fan_out_id` for a batch) in place of the reply, whatever its status (`accepted`, `queued`, `queued_timeout`, `fan_out_timeout`). The work arrived and is queued, running or done, even if your call timed out. - Never re-send because a call timed out or its delivery could not be confirmed. An exact repeat is normally answered with the original; a reworded one can run the work twice. - Read the result with `get_execution_result(agent_name, execution_id)` (`get_fan_out_result` for a batch); `running` is not stuck. To finish later, call `set_reminder` with a message naming the `execution_id`, then end your turn. A receipt is not a result: never report the work as done. +- Confirmed `failed` or `cancelled`: re-send word for word to retry; same `execution_id` back: `set_reminder`, end your turn. - An error without an `execution_id`, `agent_busy` included, is not proof that nothing ran. Look for your exact message in `list_recent_executions(agent_name)`: one match is your receipt; otherwise re-send it word for word, same options (for `agent_busy`, after `retry_after_seconds`). - `pending_approval`: nothing ran. Do not retry or route it through another agent — the outcome will be sent to you. On `retryable: false`, do what its `message` says. - For long work use `parallel=true, async=true`: the receipt comes back at once, and the run's end fires the target's `agent.task.completed` / `agent.task.failed`. A `subscribe_to_event` subscription to those wakes you for every run of that agent, so match the `execution_id`.""" diff --git a/src/mcp-server/src/client.ts b/src/mcp-server/src/client.ts index feade4c131..dbb84a13d6 100644 --- a/src/mcp-server/src/client.ts +++ b/src/mcp-server/src/client.ts @@ -291,6 +291,27 @@ export function extractIdempotencyExecutionId(body: string): string | undefined } } +/** + * #3245: a body the backend REPLAYED from an idempotency snapshot, marked. + * + * The backend says so in `X-Idempotent-Replay: true`, a header the model never + * sees. A replayed receipt names a run that is live, succeeded or may still + * report back (a run that ended failed/cancelled is no longer replayed, so an + * identical re-send starts a new one) — the caller can tell "this is the run I + * already started" from "this is a new run" only by this field. Present only + * as positive evidence: its absence does not prove a fresh dispatch (the + * client's own timeout-recovery receipts carry none). + */ +export type Replayable = T & { idempotent_replay?: true }; + +export function markIdempotentReplay(response: Response, body: T): Replayable { + const replayed = response.headers.get("X-Idempotent-Replay") === "true"; + if (!replayed || body === null || typeof body !== "object" || Array.isArray(body)) { + return body as Replayable; + } + return { ...body, idempotent_replay: true } as Replayable; +} + /** * #2806: the backend refused an agent-to-agent hop past the chain-depth limit * (403, `detail.error === "inter_agent_depth_exceeded"`). Returned as a RESULT @@ -1046,9 +1067,9 @@ export class TrinityClient { idempotencyKey?: string, turn?: string ): Promise< - | ChatResponse + | Replayable | { error: string; queue_status: "busy" | "queue_full"; retry_after: number; agent: string; details?: Record } - | { status: "queued_timeout"; agent: string; execution_id: string; message: string } + | Replayable<{ status: "queued_timeout"; agent: string; execution_id: string; message: string }> | DepthRefusal | GateResult > { @@ -1199,7 +1220,8 @@ export class TrinityClient { throw new Error(`API error (${response.status}): ${error}`); } - return (await response.json()) as ChatResponse; + // #3245: a 200 replay of a stored snapshot says so. + return markIdempotentReplay(response, (await response.json()) as ChatResponse); } /** @@ -1322,11 +1344,11 @@ export class TrinityClient { turn?: string ): Promise< | GateResult - | ChatResponse - | { status: "accepted"; execution_id: string; agent_name: string; message: string; async_mode: true } + | Replayable + | Replayable<{ status: "accepted"; execution_id: string; agent_name: string; message: string; async_mode: true }> // #2661: sync mode may answer with the gateway-timeout receipt (same shape // chat() returns for #914) instead of a completed ChatResponse. - | { status: "queued_timeout"; agent: string; execution_id: string; message: string } + | Replayable<{ status: "queued_timeout"; agent: string; execution_id: string; message: string }> | DepthRefusal > { // Prepare headers @@ -1483,7 +1505,8 @@ export class TrinityClient { throw new Error(`API error (${response.status}): ${error}`); } - return (await response.json()) as ChatResponse; + // #3245: a 200 replay of a stored snapshot says so. + return markIdempotentReplay(response, (await response.json()) as ChatResponse); } /** @@ -1499,7 +1522,7 @@ export class TrinityClient { private inFlightReplayReceipt( name: string, executionId: string, - ): { status: "queued_timeout"; agent: string; execution_id: string; message: string } { + ): { status: "queued_timeout"; agent: string; execution_id: string; message: string; idempotent_replay: true } { return { status: "queued_timeout", agent: name, @@ -1507,6 +1530,8 @@ export class TrinityClient { message: `This exact call was already dispatched to '${name}' and has no result to replay yet. ` + `${readNotResend(name, executionId)} A reworded re-send would dispatch a SECOND execution (#2661).`, + // #3245: a 409 is a replay of the in-flight claim, not a new dispatch. + idempotent_replay: true, }; } diff --git a/src/mcp-server/src/delegation-contract.test.ts b/src/mcp-server/src/delegation-contract.test.ts index 4ca605051a..c6aa1357c6 100644 --- a/src/mcp-server/src/delegation-contract.test.ts +++ b/src/mcp-server/src/delegation-contract.test.ts @@ -397,6 +397,13 @@ describe("ent#568 the async receipt chat_with_agent answers with", () => { assert.equal(out.message, `${LEAD.accepted} ${readNotResendFor("ex_9")}`); }); + it("#3245: a replayed receipt keeps its idempotent_replay marker through the re-wording", async () => { + const out = JSON.parse(await toolAnswering({ ...BACKEND.accepted, idempotent_replay: true }, false, []) + .execute({ agent_name: "agent-a", message: "m", parallel: true, async: true }, agentSession)); + assert.equal(out.idempotent_replay, true); + assert.equal(out.message, `${LEAD.accepted} ${readNotResendFor("ex_9")}`); + }); + it("a completed reply is passed through untouched", async () => { const seen: Seen[] = []; const out = JSON.parse(await toolAnswering(BACKEND.accepted, false, seen).execute( diff --git a/src/mcp-server/src/delegation_contract.ts b/src/mcp-server/src/delegation_contract.ts index 6121752c4b..d80a271672 100644 --- a/src/mcp-server/src/delegation_contract.ts +++ b/src/mcp-server/src/delegation_contract.ts @@ -27,6 +27,7 @@ export const DELEGATION_CONTRACT_LINES: readonly string[] = [ "- A `chat_with_*` or `fan_out` call answers with the reply or with a receipt: an `execution_id` (a `fan_out_id` for a batch) in place of the reply, whatever its status (`accepted`, `queued`, `queued_timeout`, `fan_out_timeout`). The work arrived and is queued, running or done, even if your call timed out.", "- Never re-send because a call timed out or its delivery could not be confirmed. An exact repeat is normally answered with the original; a reworded one can run the work twice.", "- Read the result with `get_execution_result(agent_name, execution_id)` (`get_fan_out_result` for a batch); `running` is not stuck. To finish later, call `set_reminder` with a message naming the `execution_id`, then end your turn. A receipt is not a result: never report the work as done.", + "- Confirmed `failed` or `cancelled`: re-send word for word to retry; same `execution_id` back: `set_reminder`, end your turn.", "- An error without an `execution_id`, `agent_busy` included, is not proof that nothing ran. Look for your exact message in `list_recent_executions(agent_name)`: one match is your receipt; otherwise re-send it word for word, same options (for `agent_busy`, after `retry_after_seconds`).", "- `pending_approval`: nothing ran. Do not retry or route it through another agent — the outcome will be sent to you. On `retryable: false`, do what its `message` says.", "- For long work use `parallel=true, async=true`: the receipt comes back at once, and the run's end fires the target's `agent.task.completed` / `agent.task.failed`. A `subscribe_to_event` subscription to those wakes you for every run of that agent, so match the `execution_id`.", diff --git a/src/mcp-server/src/task-receipt.test.ts b/src/mcp-server/src/task-receipt.test.ts index 5d7f3a2923..9f28525b7c 100644 --- a/src/mcp-server/src/task-receipt.test.ts +++ b/src/mcp-server/src/task-receipt.test.ts @@ -31,10 +31,10 @@ function stubFetch(routes: Array<[string, Handler]>) { return calls; } -function json(body: unknown, status = 200) { +function json(body: unknown, status = 200, headers: Record = {}) { return new Response(JSON.stringify(body), { status, - headers: { "content-type": "application/json" }, + headers: { "content-type": "application/json", ...headers }, }); } @@ -194,3 +194,59 @@ describe("#2661 chat(): the sequential route honours the same 409 rule", () => { await assert.rejects(() => client.chat("agent-a", "m"), /API error \(409\)/); }); }); + +describe("#3245 a replayed response says it is a replay (idempotent_replay)", () => { + beforeEach(() => { globalThis.fetch = realFetch; }); + afterEach(() => { + globalThis.fetch = realFetch; + if (realTimeout === undefined) delete process.env.MCP_CHAT_TIMEOUT_MS; + else process.env.MCP_CHAT_TIMEOUT_MS = realTimeout; + }); + + const RECEIPT = { status: "accepted", execution_id: "ex-replayed-3245", agent_name: "agent-a", + async_mode: true, message: "Task accepted." }; + const REPLAY = { "X-Idempotent-Replay": "true" }; + + for (const route of ["task", "chat"] as const) { + it(`${route}(): a 200 with X-Idempotent-Replay: true carries idempotent_replay: true`, async () => { + stubFetch([[`/${route}`, () => json(RECEIPT, 200, REPLAY)]]); + const client = new TrinityClient("http://backend:8000", "tok"); + const out = await client[route]("agent-a", "m") as Record; + assert.equal(out.idempotent_replay, true); + assert.equal(out.execution_id, "ex-replayed-3245"); + }); + + it(`${route}(): a 200 without the header has no idempotent_replay field`, async () => { + stubFetch([[`/${route}`, () => json(RECEIPT)]]); + const client = new TrinityClient("http://backend:8000", "tok"); + const out = await client[route]("agent-a", "m") as Record; + assert.equal("idempotent_replay" in out, false); + }); + + it(`${route}(): the 409 in-flight receipt is marked as a replay`, async () => { + stubFetch([[`/${route}`, () => json({ detail: { execution_id: "ex-409-3245" } }, 409)]]); + const client = new TrinityClient("http://backend:8000", "tok"); + const out = await client[route]("agent-a", "m") as Record; + assert.equal(out.idempotent_replay, true); + assert.equal(out.execution_id, "ex-409-3245"); + }); + } + + it("a non-object replay body is returned unchanged", async () => { + stubFetch([["/task", () => json("plain", 200, REPLAY)]]); + const client = new TrinityClient("http://backend:8000", "tok"); + assert.equal(await client.task("agent-a", "m"), "plain"); + }); + + it("the client's own timeout-recovery receipt carries no marker (absence is not 'fresh')", async () => { + process.env.MCP_CHAT_TIMEOUT_MS = "20"; + stubFetch([ + ["/task", hangUntilAbort], + ["/executions?limit=50", () => json([row({ id: "ex-live" })])], + ]); + const client = new TrinityClient("http://backend:8000", "tok"); + const out = await client.task("agent-a", "do the thing", undefined, undefined, { keyId: "key-1" }) as Record; + assert.equal(out.execution_id, "ex-live"); + assert.equal("idempotent_replay" in out, false); + }); +}); diff --git a/src/mcp-server/src/tools/chat.ts b/src/mcp-server/src/tools/chat.ts index a214fd8ba7..790643cf25 100644 --- a/src/mcp-server/src/tools/chat.ts +++ b/src/mcp-server/src/tools/chat.ts @@ -146,10 +146,13 @@ function depthRefusalResult(refusal: DepthRefusal): string { * * The leads claim nothing about liveness: the backend stores the async receipt * as the idempotency snapshot at dispatch time and replays it for 24 h, after the - * run may have finished or failed. Three shapes: async `accepted` / `queued`, and - * the `queued_timeout` a sync `/task` stores when its long-poll gives up — told - * apart from the MCP server's own `queued_timeout` (client.ts), which already - * carries `readNotResend`, by that sentence. + * run may have finished (since #3245 a run that ended failed/cancelled is no + * longer replayed — an identical re-send starts a new run — but one that + * succeeded, or may still report back, is). Three shapes: async `accepted` / + * `queued`, and the `queued_timeout` a sync `/task` stores when its long-poll + * gives up — told apart from the MCP server's own `queued_timeout` (client.ts), + * which already carries `readNotResend`, by that sentence. Every other field is + * kept, including `idempotent_replay: true` on a replay (#3245). */ function asyncReceipt(agent: string, response: T): T { const r = response as { status?: unknown; execution_id?: unknown; message?: unknown }; @@ -424,7 +427,7 @@ export function createChatTools( // the parameter descriptions below, which are not cut. // delegation-contract.test.ts pins the published length. description: - "Delegate a task to another agent (the primary way to use sub-agents) and get back its reply or a receipt." + + "Delegate a task to another agent and get back its reply or a receipt." + "\n\n**Modes:** `parallel=false` (default) continues the agent's own chat session, one queue for every " + "caller. `parallel=true` runs a stateless task; several can run at once. `async=true` (with " + "`parallel=true`) answers with a receipt at once.\n\n" + diff --git a/tests/registry.json b/tests/registry.json index 20e86797cf..81a7521f17 100644 --- a/tests/registry.json +++ b/tests/registry.json @@ -1,5 +1,17 @@ { "test_files": [ + { + "file": "unit/test_3245_async_replay_after_failure.py", + "feature": "abilityai/trinity#3245", + "added": "2026-10-06", + "categories": [ + "backend", + "unit", + "reliability", + "database" + ], + "description": "A dispatch receipt replays only while its run can still succeed (#3245). An async `/task`, a timed-out sync `/task` and a timed-out pulled `/chat` store a receipt (`async_mode: true`, `accepted`/`queued`/`queued_timeout`) as the `completed` idempotency row; before the fix an identical retry got the dead run's receipt for 24 h. Real SQLite (`IdempotencyOperations`, temp DB) with non-ambient ids. Pins: the compare-and-delete `discard_completed_if_execution` deletes only the row still naming the dead execution, so a slow retry cannot erase a faster retry's fresh record (both interleavings, plus a teeth case where an unconditional delete starts two runs); `execution_liveness` maps queued/running/pending_retry → live, success → succeeded, failed/cancelled/skipped → ended, no row → gone, and every doubt (lookup error, foreign owner, `manual`, a status flip between the two reads) → indeterminate, which replays; the `lease_expired` hold on a fixed clock at the boundary (T+899 s replays and logs `receipt_hold until=`, T+900 s and T+901 s go fresh), with the tag checked before the stamp or the timeout, a naive stored `completed_at` read as UTC and the agent's current timeout used; both admission seams (`begin_task_idempotency`, `admit_chat_request`) incl. the receipt written by the real `run_pulled_chat_turn` timeout branch, `sched:` keys never reclaimed, a skill-gate record answering before the reclaim on `/chat` and `/task`, pre-fix rows healed, the reclaim log outcomes, and AC2 end to end over the real `execute_parallel_task` (dispatch, fail, identical re-send → new `execution_id` without `X-Idempotent-Replay`, a third send while running replays the new id). AST guards: both seams call the reclaim right after `begin`; every `fail_stale_slot_execution` error carries the `lease_expired:` tag (parity with `TaskExecutionErrorCode.LEASE_EXPIRED`); nothing writes a claim after its receipt is stored (the CAS precondition). Mutation-proven (reclaim calls removed: 24 red; CAS without the execution-id predicate: 5; no hold: 7; gate moved after the claim on `/task`: 2). SQLite only, no PostgreSQL proof. No Docker, no live backend." + }, { "file": "unit/test_2545_skill_source_pin.py", "feature": "#2545, #3123", diff --git a/tests/unit/test_3245_async_replay_after_failure.py b/tests/unit/test_3245_async_replay_after_failure.py new file mode 100644 index 0000000000..4a011c5314 --- /dev/null +++ b/tests/unit/test_3245_async_replay_after_failure.py @@ -0,0 +1,916 @@ +"""#3245: a failed async run stops replaying its receipt. + +An async ``/task`` (or a timed-out sync ``/task`` / pulled ``/chat``) stores a +*receipt* — ``{"status": "accepted"|"queued"|"queued_timeout", "execution_id", +"async_mode": True}`` — as the ``completed`` idempotency row. Before #3245 +nothing revisited it, so for 24h an identical request got the dead run's +receipt back. The fix checks, at replay time in the shared admission seam, +whether the run can still succeed; if it ended (or its row is gone) the row is +compare-and-deleted and the request dispatches fresh. + +Runs on a throwaway SQLite file (``DATABASE_URL`` + ``dispose_engines``) with +the real ``IdempotencyOperations`` / ``ScheduleOperations``. No ``sys.modules`` +surgery; the service under test is patched through the module object the code +actually reads (``dispatch_admission_service.idempotency_service``). +""" +from __future__ import annotations + +from types import SimpleNamespace + +import pytest +from sqlalchemy import insert, update + +STALE = "exec-stale-7f3" +FRESH = "exec-fresh-b21" +AGENT = "agent-3245" + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def store(monkeypatch, tmp_path): + """Real idempotency + execution tables on a temp SQLite file.""" + monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path / 't3245.db'}") + from db import engine as eng + eng.dispose_engines() + from db.tables import metadata, idempotency_keys, schedule_executions + metadata.create_all(eng.get_engine(), tables=[idempotency_keys, schedule_executions]) + from db.idempotency import IdempotencyOperations + from db.schedules import ScheduleOperations + yield SimpleNamespace( + idem=IdempotencyOperations(), + sched=ScheduleOperations(None, None), + engine=eng.get_engine(), + executions=schedule_executions, + keys=idempotency_keys, + ) + eng.dispose_engines() + + +def _insert_exec(store, execution_id, status, *, agent=AGENT, error=None, + completed_at=None, started_at="2026-10-06T11:00:00Z"): + with store.engine.begin() as conn: + conn.execute(insert(store.executions).values( + id=execution_id, agent_name=agent, status=status, error=error, + started_at=started_at, completed_at=completed_at, triggered_by="agent", + )) + + +def _set_exec(store, execution_id, **values): + with store.engine.begin() as conn: + conn.execute(update(store.executions) + .where(store.executions.c.id == execution_id).values(**values)) + + +def _receipt(execution_id, status="accepted"): + return {"status": status, "execution_id": execution_id, + "agent_name": AGENT, "async_mode": True} + + +# --------------------------------------------------------------------------- +# DB layer: compare-and-delete + the narrow failure-stamp read +# --------------------------------------------------------------------------- + +class TestDiscardCompletedIfExecution: + S, K = f"agent:{AGENT}", "k-cas" + + def test_deletes_only_the_row_still_naming_that_execution(self, store): + assert store.idem.claim(self.S, self.K)["state"] == "new" + store.idem.complete(self.S, self.K, STALE, _receipt(STALE)) + assert store.idem.discard_completed_if_execution(self.S, self.K, FRESH) is False + assert store.idem.claim(self.S, self.K)["execution_id"] == STALE + assert store.idem.discard_completed_if_execution(self.S, self.K, STALE) is True + assert store.idem.claim(self.S, self.K)["state"] == "new" + + def test_slow_retry_cannot_erase_the_fast_retrys_fresh_record(self, store): + """Dossier §8: A and B both saw the completed row naming STALE. A + discards and re-completes as FRESH; B's late discard must be a no-op.""" + store.idem.claim(self.S, self.K) + store.idem.complete(self.S, self.K, STALE, _receipt(STALE)) + assert store.idem.discard_completed_if_execution(self.S, self.K, STALE) is True # A + assert store.idem.claim(self.S, self.K)["state"] == "new" # A + store.idem.complete(self.S, self.K, FRESH, _receipt(FRESH)) # A + assert store.idem.discard_completed_if_execution(self.S, self.K, STALE) is False # B + after = store.idem.claim(self.S, self.K) + assert after["state"] == "completed" and after["execution_id"] == FRESH + + def test_never_deletes_an_in_flight_claim(self, store): + store.idem.claim(self.S, self.K) + store.idem.attach_execution(self.S, self.K, STALE) + assert store.idem.discard_completed_if_execution(self.S, self.K, STALE) is False + assert store.idem.claim(self.S, self.K)["state"] == "in_flight" + + def test_facade_delegates(self, store): + from database import DatabaseManager + fake_self = SimpleNamespace(_idempotency_ops=store.idem) + store.idem.claim(self.S, self.K) + store.idem.complete(self.S, self.K, STALE, _receipt(STALE)) + assert DatabaseManager.idempotency_discard_completed_if_execution( + fake_self, self.S, self.K, STALE) is True + + +class TestExecutionFailureStamp: + def test_returns_status_error_and_completed_at(self, store): + _insert_exec(store, STALE, "failed", error="lease_expired: Stale execution", + completed_at="2026-10-06T12:00:00Z") + assert store.sched.get_execution_failure_stamp(STALE) == ( + "failed", "lease_expired: Stale execution", "2026-10-06T12:00:00Z") + + def test_missing_row_is_none(self, store): + assert store.sched.get_execution_failure_stamp("exec-nope-000") is None + + def test_facade_delegates(self, store): + from database import DatabaseManager + _insert_exec(store, STALE, "failed", error=None, completed_at="2026-10-06T12:00:00Z") + assert DatabaseManager.get_execution_failure_stamp( + SimpleNamespace(_schedule_ops=store.sched), STALE) == ( + "failed", None, "2026-10-06T12:00:00Z") + + +# --------------------------------------------------------------------------- +# Service: execution_liveness / discard_replay_of +# --------------------------------------------------------------------------- + +T0 = "2026-10-06T12:00:00Z" # completed_at the slot reaper wrote +TIMEOUT = 600 # agent timeout; window = 600 + 300 = 900s +TAGGED = "lease_expired: Stale execution — slot TTL expired for agent 'agent-3245'" + + +def _at(seconds): + from datetime import datetime, timedelta, timezone + return datetime(2026, 10, 6, 12, 0, 0, tzinfo=timezone.utc) + timedelta(seconds=seconds) + + +class _Spy: + """Counts calls; optionally raises.""" + + def __init__(self, fn, raises=None): + self.fn, self.raises, self.calls = fn, raises, 0 + + def __call__(self, *a, **kw): + self.calls += 1 + if self.raises is not None: + raise self.raises + return self.fn(*a, **kw) + + +@pytest.fixture +def svc(store, monkeypatch): + """`idempotency_service` (the very module the admission seam reads) over a + db stand-in whose every method is the REAL db op on the temp file, wrapped + in a call-counting spy. `timeout` is the one non-table value.""" + import services.dispatch_admission_service as das + isvc = das.idempotency_service + state = SimpleNamespace(timeout=TIMEOUT) + fake = SimpleNamespace( + idempotency_claim=store.idem.claim, + idempotency_complete=store.idem.complete, + idempotency_release=store.idem.release, + idempotency_attach_execution=store.idem.attach_execution, + idempotency_discard_completed=store.idem.discard_completed, + idempotency_discard_completed_if_execution=_Spy( + getattr(store.idem, "discard_completed_if_execution", None)), + get_execution=store.sched.get_execution, + get_execution_gate_state=_Spy(store.sched.get_execution_gate_state), + get_execution_failure_stamp=_Spy( + getattr(store.sched, "get_execution_failure_stamp", None)), + get_execution_timeout=_Spy(lambda name: state.timeout), + ) + monkeypatch.setattr(isvc, "db", fake) + return SimpleNamespace(isvc=isvc, das=das, db=fake, state=state, store=store) + + +def _verdict(svc, execution_id=STALE, *, now=None, agent=AGENT): + return svc.isvc.execution_liveness(execution_id, agent, now=now).verdict + + +class TestExecutionLivenessVerdicts: + @pytest.mark.parametrize("status,expected", [ + ("queued", "live"), ("running", "live"), ("pending_retry", "live"), + ("success", "succeeded"), + ("cancelled", "ended"), ("skipped", "ended"), + ("error", "indeterminate"), # legacy / unknown status + ]) + def test_status_matrix(self, svc, status, expected): + _insert_exec(svc.store, STALE, status, completed_at=T0) + assert _verdict(svc, now=_at(1)) == expected + + @pytest.mark.parametrize("error", [ + "boom", None, "Marked as failed by cleanup: exceeded 3600s stale timeout", + ]) + def test_untagged_failed_is_ended_at_once(self, svc, error): + _insert_exec(svc.store, STALE, "failed", error=error, completed_at=T0) + assert _verdict(svc, now=_at(1)) == "ended" + + def test_no_row_is_gone(self, svc): + assert _verdict(svc, "exec-vanished-404") == "gone" + + def test_resolver_unknown_is_verdict_gone(self, svc): + """The effect resolver's word for "no such execution" is `unknown`; + the liveness verdict for the same fact is `gone` (decision 3).""" + assert svc.isvc._resolve_execution_with_reason( + "exec-vanished-404", AGENT, log=False)[1] == "unknown" + assert _verdict(svc, "exec-vanished-404") == "gone" + + def test_foreign_owner_is_indeterminate(self, svc): + _insert_exec(svc.store, STALE, "failed", agent="some-other-agent", completed_at=T0) + assert _verdict(svc) == "indeterminate" + + @pytest.mark.parametrize("execution_id", [None, "", "manual"]) + def test_no_usable_id_is_indeterminate_without_a_read(self, svc, execution_id): + assert _verdict(svc, execution_id) == "indeterminate" + assert svc.db.get_execution_gate_state.calls == 0 + + def test_gate_read_raising_is_indeterminate(self, svc): + svc.db.get_execution_gate_state.raises = RuntimeError("db down") + assert _verdict(svc) == "indeterminate" + + def test_only_a_failed_row_pays_the_second_read(self, svc): + _insert_exec(svc.store, STALE, "running") + _verdict(svc) + assert svc.db.get_execution_failure_stamp.calls == 0 + + +class TestLeaseExpiredWindow: + """Decision 1 (UC1): a `lease_expired` FAILED may still report success, so it + replays until completed_at + timeout + SLOT_TTL_BUFFER, then goes fresh.""" + + def test_time_axis_boundary(self, svc): + _insert_exec(svc.store, STALE, "failed", error=TAGGED, completed_at=T0) + held = svc.isvc.execution_liveness(STALE, AGENT, now=_at(899)) + assert held.verdict == "maybe_alive" and held.hold_until == _at(900) + assert _verdict(svc, now=_at(900)) == "ended" + assert _verdict(svc, now=_at(901)) == "ended" + + def test_window_uses_the_real_slot_buffer(self, svc): + from services.slot_service import SLOT_TTL_BUFFER + _insert_exec(svc.store, STALE, "failed", error=TAGGED, completed_at=T0) + assert svc.isvc.execution_liveness(STALE, AGENT, now=_at(1)).hold_until == _at( + TIMEOUT + SLOT_TTL_BUFFER) + + def test_late_success_inside_the_window_is_succeeded(self, svc): + _insert_exec(svc.store, STALE, "failed", error=TAGGED, completed_at=T0) + assert _verdict(svc, now=_at(10)) == "maybe_alive" + _set_exec(svc.store, STALE, status="success") + assert _verdict(svc, now=_at(10)) == "succeeded" + + def test_status_flip_between_the_two_reads_is_indeterminate(self, svc): + _insert_exec(svc.store, STALE, "success", error=TAGGED, completed_at=T0) + svc.db.get_execution_gate_state.fn = lambda eid: (AGENT, "failed") + assert _verdict(svc, now=_at(1)) == "indeterminate" + + @pytest.mark.parametrize("completed_at", [None, "not-a-timestamp"]) + def test_tagged_without_a_usable_stamp_is_indeterminate(self, svc, completed_at): + _insert_exec(svc.store, STALE, "failed", error=TAGGED, completed_at=completed_at) + assert _verdict(svc, now=_at(10_000)) == "indeterminate" + + def test_timeout_lookup_raising_is_indeterminate(self, svc): + _insert_exec(svc.store, STALE, "failed", error=TAGGED, completed_at=T0) + svc.db.get_execution_timeout.raises = RuntimeError("settings unreadable") + assert _verdict(svc, now=_at(10_000)) == "indeterminate" + + def test_second_read_raising_is_indeterminate(self, svc): + _insert_exec(svc.store, STALE, "failed", error=TAGGED, completed_at=T0) + svc.db.get_execution_failure_stamp.raises = RuntimeError("db down") + assert _verdict(svc, now=_at(10_000)) == "indeterminate" + + def test_tag_is_checked_before_the_stamp_or_the_timeout(self, svc): + """Codex finding 4: an ordinary failure is never held by a broken + timestamp or a timeout lookup error.""" + _insert_exec(svc.store, STALE, "failed", error="boom", completed_at=None) + svc.db.get_execution_timeout.raises = RuntimeError("settings unreadable") + assert _verdict(svc, now=_at(1)) == "ended" + assert svc.db.get_execution_timeout.calls == 0 + + def test_naive_stored_completed_at_is_read_as_utc(self, svc): + _insert_exec(svc.store, STALE, "failed", error=TAGGED, + completed_at="2026-10-06T12:00:00") + assert _verdict(svc, now=_at(899)) == "maybe_alive" + assert _verdict(svc, now=_at(900)) == "ended" + + def test_horizon_uses_the_current_timeout(self, svc): + """Documented choice: the agent's timeout NOW, not at dispatch.""" + _insert_exec(svc.store, STALE, "failed", error=TAGGED, completed_at=T0) + assert _verdict(svc, now=_at(899)) == "maybe_alive" + svc.state.timeout = 60 # window shrinks to 360s + assert _verdict(svc, now=_at(899)) == "ended" + + def test_clock_hook_is_used_only_when_no_clock_is_given(self, svc, monkeypatch): + _insert_exec(svc.store, STALE, "failed", error=TAGGED, completed_at=T0) + clock = _Spy(lambda: _at(899)) + monkeypatch.setattr(svc.isvc, "_now", clock) + assert svc.isvc.execution_liveness(STALE, AGENT).verdict == "maybe_alive" + assert clock.calls == 1 + assert _verdict(svc, now=_at(901)) == "ended" + assert clock.calls == 1 + + +class TestDiscardReplayOf: + S, K = f"agent:{AGENT}", "k-discard" + + def test_true_false_and_none(self, svc, caplog): + svc.store.idem.claim(self.S, self.K) + svc.store.idem.complete(self.S, self.K, STALE, _receipt(STALE)) + assert svc.isvc.discard_replay_of(self.S, self.K, FRESH) is False + assert svc.isvc.discard_replay_of(self.S, self.K, STALE) is True + svc.db.idempotency_discard_completed_if_execution.raises = RuntimeError("locked") + with caplog.at_level("WARNING"): + assert svc.isvc.discard_replay_of(self.S, self.K, STALE) is None + assert "discard_replay_of" in caplog.text and self.K not in caplog.text + + +# --------------------------------------------------------------------------- +# Guards: the `lease_expired` tag the window keys on +# --------------------------------------------------------------------------- + +def _backend_root(): + import pathlib + import services + return pathlib.Path(services.__file__).resolve().parent.parent + + +def test_lease_expired_tag_parity(): + """`cleanup_service._LEASE_EXPIRED_TAG` (private, the writer's spelling) must + equal the enum value the liveness check derives its prefix from.""" + import ast + from services.execution_envelope import TaskExecutionErrorCode + tree = ast.parse((_backend_root() / "services" / "cleanup_service.py").read_text()) + values = [n.value.value for n in ast.walk(tree) + if isinstance(n, ast.Assign) + and any(isinstance(t, ast.Name) and t.id == "_LEASE_EXPIRED_TAG" for t in n.targets)] + assert values == [TaskExecutionErrorCode.LEASE_EXPIRED.value] + + +def test_every_slot_reaper_failure_carries_the_tag(): + """UC1 covers exactly the rows whose error starts `lease_expired: `. A new + `fail_stale_slot_execution(` writer without the tag would silently drop out + of the hold window — fail it here instead.""" + import ast + root = _backend_root() + found = 0 + for path in sorted(root.rglob("*.py")): + if path.name == "database.py" or "tests" in path.parts: + continue + tree = ast.parse(path.read_text()) + for node in ast.walk(tree): + if not (isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) + and node.func.attr == "fail_stale_slot_execution"): + continue + found += 1 + err = next((k.value for k in node.keywords if k.arg == "error"), None) + where = f"{path.relative_to(root)}:{node.lineno}" + assert isinstance(err, ast.JoinedStr), where + first, second = err.values[0], err.values[1] + assert isinstance(first, ast.FormattedValue) and isinstance(first.value, ast.Name) \ + and first.value.id == "_LEASE_EXPIRED_TAG", where + assert isinstance(second, ast.Constant) and second.value.startswith(": "), where + assert found >= 2, "the slot-reaper writers were not found — the guard went blind" + + +# --------------------------------------------------------------------------- +# Admission seam: _reclaim_ended_receipt in begin_task_idempotency (/task) and +# admit_chat_request (/chat) +# --------------------------------------------------------------------------- + +SCOPE = f"agent:{AGENT}" +RECEIPT_STATUSES = ["accepted", "queued", "queued_timeout"] + + +def _store_receipt(svc, key, execution_id=STALE, status="accepted"): + assert svc.store.idem.claim(SCOPE, key)["state"] == "new" + svc.store.idem.complete(SCOPE, key, execution_id, _receipt(execution_id, status)) + + +def _row(svc, key): + """Current (state, execution_id) of the key's row — read without claiming.""" + from sqlalchemy import select + k = svc.store.keys + with svc.store.engine.connect() as conn: + r = conn.execute(select(k.c.status, k.c.execution_id).where( + (k.c.scope == SCOPE) & (k.c.idempotency_key == key))).first() + return (r.status, r.execution_id) if r else None + + +def _task(svc, key): + return svc.das.begin_task_idempotency(name=AGENT, idempotency_key=key) + + +def _user(): + return SimpleNamespace(id=1, email="u1@example.com", username="u1", agent_name=None, + mcp_key_id=None, mcp_key_name=None, role="user") + + +def _admit(svc, monkeypatch, key, *, pilot=False, gate=None, gate_exc=None): + import asyncio + from unittest.mock import AsyncMock, MagicMock + from models import ChatMessageRequest + from services.skill_gate_service import GateDecision + das = svc.das + monkeypatch.setenv("PULL_MODE_PILOT_AGENTS", AGENT if pilot else "") + monkeypatch.setattr(das, "enforce_inter_agent_depth", AsyncMock(return_value=None)) + monkeypatch.setattr(das.skill_gate_service, "enforce", AsyncMock( + side_effect=gate_exc, return_value=gate or GateDecision())) + monkeypatch.setattr(das, "dispatch_breaker_active", lambda name: False) + cap = MagicMock(acquire=AsyncMock(return_value=SimpleNamespace( + state="admitted", queue_position=0))) + monkeypatch.setattr(das, "get_capacity_manager", lambda: cap) + monkeypatch.setattr(das, "platform_audit_service", MagicMock(log=AsyncMock())) + monkeypatch.setattr(das, "db", SimpleNamespace( + get_execution_timeout=lambda name: 900, get_max_parallel_tasks=lambda name: 3)) + return asyncio.run(das.admit_chat_request( + name=AGENT, request=ChatMessageRequest(message="hi"), current_user=_user(), + x_source_agent=None, x_via_mcp="true", idempotency_key=key, + )) + + +class TestTaskSeamAC1: + @pytest.mark.parametrize("receipt_status", RECEIPT_STATUSES) + @pytest.mark.parametrize("status", ["failed", "cancelled", "skipped"]) + def test_ended_run_dispatches_fresh(self, svc, status, receipt_status): + _insert_exec(svc.store, STALE, status, error="boom", completed_at=T0) + _store_receipt(svc, "k-ac1", status=receipt_status) + idem, replay = _task(svc, "k-ac1") + assert replay is None and idem.enabled and not idem.replay + assert _row(svc, "k-ac1") == ("in_flight", None) + assert svc.db.get_execution_gate_state.calls == 1 + + @pytest.mark.parametrize("receipt_status", RECEIPT_STATUSES) + @pytest.mark.parametrize("status", ["queued", "running", "pending_retry", "success"]) + def test_live_or_succeeded_run_replays(self, svc, status, receipt_status): + """Guard (green on dev too): a run that can still succeed replays.""" + _insert_exec(svc.store, STALE, status, completed_at=T0) + _store_receipt(svc, "k-ac1", status=receipt_status) + idem, replay = _task(svc, "k-ac1") + assert replay is not None and replay.execution_id == STALE and not replay.in_flight + assert _row(svc, "k-ac1") == ("completed", STALE) + + def test_gone_row_dispatches_fresh(self, svc): + _store_receipt(svc, "k-gone") + assert _task(svc, "k-gone")[1] is None + assert svc.db.get_execution_gate_state.calls == 1 + + def test_pull_routed_task_receipt_shapes_dispatch_fresh(self, svc): + """#946's pull-routed /task stores the same accepted/queued receipt.""" + _insert_exec(svc.store, STALE, "failed", error="boom", completed_at=T0) + for key, st in (("k-946-a", "accepted"), ("k-946-q", "queued")): + _store_receipt(svc, key, status=st) + assert _task(svc, key)[1] is None + + @pytest.mark.parametrize("case", ["raises", "foreign", "legacy_status"]) + def test_cannot_tell_replays(self, svc, case): + if case == "raises": + svc.db.get_execution_gate_state.raises = RuntimeError("db down") + elif case == "foreign": + _insert_exec(svc.store, STALE, "failed", agent="other-agent", completed_at=T0) + else: + _insert_exec(svc.store, STALE, "error", completed_at=T0) + _store_receipt(svc, "k-edge") + assert _task(svc, "k-edge")[1].execution_id == STALE + assert _row(svc, "k-edge") == ("completed", STALE) + + def test_manual_execution_id_replays(self, svc): + _store_receipt(svc, "k-manual", execution_id="manual") + assert _task(svc, "k-manual")[1].execution_id == "manual" + assert svc.db.get_execution_gate_state.calls == 0 + + @pytest.mark.parametrize("snapshot", [ + {"response": "done", "task_execution_id": STALE}, + {"execution": {"task_execution_id": STALE}}, + {"status": "accepted", "execution_id": STALE, "async_mode": False}, + {"status": "success", "execution_id": STALE, "async_mode": True}, + ]) + def test_non_receipt_snapshot_replays_without_a_read(self, svc, snapshot): + _insert_exec(svc.store, STALE, "failed", error="boom", completed_at=T0) + svc.store.idem.claim(SCOPE, "k-sync") + svc.store.idem.complete(SCOPE, "k-sync", STALE, snapshot) + assert _task(svc, "k-sync")[1].snapshot == snapshot + assert svc.db.get_execution_gate_state.calls == 0 + + def test_pre_fix_row_written_by_raw_sql_dispatches_fresh(self, svc): + import json + _insert_exec(svc.store, STALE, "failed", error="boom", completed_at=T0) + with svc.store.engine.begin() as conn: + conn.execute(insert(svc.store.keys).values( + scope=SCOPE, idempotency_key="k-legacy", execution_id=STALE, + status="completed", response_snapshot=json.dumps(_receipt(STALE)), + created_at="2026-10-06T11:00:00Z", updated_at="2026-10-06T11:00:00Z")) + assert _task(svc, "k-legacy")[1] is None + + def test_scheduler_key_is_never_reclaimed(self, svc): + _insert_exec(svc.store, STALE, "failed", error="boom", completed_at=T0) + _store_receipt(svc, f"sched:{STALE}") + assert _task(svc, f"sched:{STALE}")[1].execution_id == STALE + assert svc.db.get_execution_gate_state.calls == 0 + + +class TestLeaseExpiredAtTheSeam: + @pytest.mark.parametrize("offset,fresh", [(899, False), (900, True), (901, True)]) + def test_time_axis(self, svc, monkeypatch, caplog, offset, fresh): + _insert_exec(svc.store, STALE, "failed", error=TAGGED, completed_at=T0) + _store_receipt(svc, "k-uc1") + monkeypatch.setattr(svc.isvc, "_now", lambda: _at(offset)) + with caplog.at_level("INFO"): + idem, replay = _task(svc, "k-uc1") + assert (replay is None) is fresh + if not fresh: + assert replay.execution_id == STALE + assert "receipt_hold" in caplog.text + assert "verdict=maybe_alive" in caplog.text + assert f"until={_at(900).isoformat()}" in caplog.text + assert "k-uc1" not in caplog.text + + @pytest.mark.parametrize("error", [ + "boom", None, "Marked as failed by cleanup: exceeded 3600s stale timeout"]) + def test_untagged_failure_is_fresh_at_once(self, svc, monkeypatch, error): + _insert_exec(svc.store, STALE, "failed", error=error, completed_at=T0) + _store_receipt(svc, "k-uc1") + monkeypatch.setattr(svc.isvc, "_now", lambda: _at(1)) + assert _task(svc, "k-uc1")[1] is None + + +class TestChatSeam: + def test_failed_queued_timeout_receipt_admits_fresh(self, svc, monkeypatch): + from services.chat_signals import ChatAdmission + _insert_exec(svc.store, STALE, "failed", error="boom", completed_at=T0) + _store_receipt(svc, "k-chat", status="queued_timeout") + admission = _admit(svc, monkeypatch, "k-chat") + assert isinstance(admission, ChatAdmission) + assert admission.idem.enabled and not admission.idem.replay + assert _row(svc, "k-chat") == ("in_flight", None) + + def test_running_receipt_still_replays(self, svc, monkeypatch): + from services.chat_signals import ChatAdmissionReplay + _insert_exec(svc.store, STALE, "running") + _store_receipt(svc, "k-chat", status="queued_timeout") + out = _admit(svc, monkeypatch, "k-chat") + assert isinstance(out, ChatAdmissionReplay) and out.execution_id == STALE + + def test_scheduler_key_is_never_reclaimed(self, svc, monkeypatch): + from services.chat_signals import ChatAdmissionReplay + _insert_exec(svc.store, STALE, "failed", error="boom", completed_at=T0) + _store_receipt(svc, f"sched:{STALE}", status="queued_timeout") + assert isinstance(_admit(svc, monkeypatch, f"sched:{STALE}"), ChatAdmissionReplay) + + @pytest.mark.parametrize("which", ["decided", "pending"]) + def test_gate_record_answers_before_the_receipt(self, svc, monkeypatch, which): + from services.skill_gate_errors import SkillApprovalRequired, SkillGateRefused + exc = (SkillGateRefused(409, "request_denied", "denied") if which == "decided" + else SkillApprovalRequired(request_id="req-1", agent_name=AGENT, + skills=["deploy"], approver_role="owner", + expires_at=None)) + _insert_exec(svc.store, STALE, "failed", error="boom", completed_at=T0) + _store_receipt(svc, "k-gated", status="queued_timeout") + with pytest.raises(type(exc)): + _admit(svc, monkeypatch, "k-gated", gate_exc=exc) + assert _row(svc, "k-gated") == ("completed", STALE) + assert svc.db.get_execution_gate_state.calls == 0 + + def test_self_approver_reaches_the_reclaim(self, svc, monkeypatch): + from services.skill_gate_service import GateDecision + gate = GateDecision(skills=("deploy",), self_approved_by="u1") + _insert_exec(svc.store, STALE, "failed", error="boom", completed_at=T0) + _store_receipt(svc, "k-gated", status="queued_timeout") + admission = _admit(svc, monkeypatch, "k-gated", gate=gate) + assert admission.gate is gate and not admission.idem.replay + + def test_pulled_chat_timeout_receipt_then_failure_admits_fresh(self, svc, monkeypatch): + """#3145 (required): the receipt is written by the REAL + `run_pulled_chat_turn` wait-timeout branch through the real idempotency + layer; once its row fails, an identical /chat on the pilot is admitted + fresh onto the queue (`capacity=None`).""" + import asyncio + import importlib + from unittest.mock import AsyncMock, MagicMock + from models import ChatMessageRequest + from services.chat_signals import ChatAdmission, ChatDispatchError + from services.execution_envelope import TaskExecutionErrorCode, TaskExecutionResult + + first = _admit(svc, monkeypatch, "k-pulled", pilot=True) + assert isinstance(first, ChatAdmission) and first.capacity is None + eid = first.execution_id + _insert_exec(svc.store, eid, "running") + + ce = importlib.import_module("services.chat_execution_service") + tes = importlib.import_module("services.task_execution_service") + monkeypatch.setattr(ce, "idempotency_service", svc.isvc) + monkeypatch.setattr(ce, "db", SimpleNamespace( + get_chat_session_claude_id=lambda sid: None)) + timed_out = TaskExecutionResult( + execution_id=eid, status="failed", response="", error="wait timed out", + error_code=TaskExecutionErrorCode.TIMEOUT) + monkeypatch.setattr(ce.session_turn_service, "run_resumable_turn", + AsyncMock(return_value=SimpleNamespace(result=timed_out))) + monkeypatch.setattr(tes, "result_from_execution_row", lambda execution_id: None) + with pytest.raises(ChatDispatchError) as exc: + asyncio.run(ce.run_pulled_chat_turn( + name=AGENT, request=ChatMessageRequest(message="hi"), current_user=_user(), + x_source_agent=None, triggered_by="mcp", task_execution_id=eid, + _chat_subscription_id=None, chat_activity_id="act-chat", + collaboration_activity_id=None, session=SimpleNamespace(id="cs-1"), + execution=SimpleNamespace(id="q1"), queue_result="queued", idem=first.idem, + )) + assert exc.value.status_code == 504 + assert _row(svc, "k-pulled") == ("completed", eid) + + _set_exec(svc.store, eid, status="failed", error="boom", completed_at=T0) + second = _admit(svc, monkeypatch, "k-pulled", pilot=True) + assert isinstance(second, ChatAdmission) and second.capacity is None + assert second.execution_id != eid and not second.idem.replay + + +class TestRetryGenerationsAndRaces: + def test_each_dead_generation_is_reclaimed_once(self, svc): + e1, e2 = "exec-gen1-c11", "exec-gen2-c22" + _insert_exec(svc.store, e1, "failed", error="boom", completed_at=T0) + _store_receipt(svc, "k-gen", execution_id=e1) + a, _ = _task(svc, "k-gen") # A: E1 → E2 + svc.isvc.complete(a, e2, _receipt(e2)) + _insert_exec(svc.store, e2, "running") + assert _task(svc, "k-gen")[1].execution_id == e2 # E2 live → replay + _set_exec(svc.store, e2, status="failed", error="boom", completed_at=T0) + b, replay = _task(svc, "k-gen") # B: E2 → E3 + assert replay is None and not b.replay + + def _two_observers(self, svc): + """A and B both begin() and see the completed row naming the dead STALE.""" + _insert_exec(svc.store, STALE, "failed", error="boom", completed_at=T0) + _store_receipt(svc, "k-race") + isvc = svc.isvc + a = isvc.begin(SCOPE, "k-race") + b = isvc.begin(SCOPE, "k-race") + assert a.replay and b.replay and a.execution_id == b.execution_id == STALE + return a, b + + def _new_count(self, *decisions): + return sum(1 for d in decisions if d.enabled and not d.replay) + + def test_a_completes_before_b_reclaims(self, svc): + a, b = self._two_observers(svc) + a2 = svc.das._reclaim_ended_receipt(a, name=AGENT) + svc.isvc.complete(a2, FRESH, _receipt(FRESH)) + _insert_exec(svc.store, FRESH, "running") + b2 = svc.das._reclaim_ended_receipt(b, name=AGENT) + assert self._new_count(a2, b2) == 1 + assert b2.replay and not b2.in_flight and b2.execution_id == FRESH + + def test_b_reclaims_while_a_is_in_flight(self, svc): + a, b = self._two_observers(svc) + a2 = svc.das._reclaim_ended_receipt(a, name=AGENT) + b2 = svc.das._reclaim_ended_receipt(b, name=AGENT) + assert self._new_count(a2, b2) == 1 + assert b2.replay and b2.in_flight + + def test_an_unconditional_delete_would_start_two_runs(self, svc, monkeypatch): + """Teeth: swap the CAS for #2040's unconditional `discard_completed` + and the first interleave starts two runs — what the CAS prevents.""" + a, b = self._two_observers(svc) + svc.db.idempotency_discard_completed_if_execution.fn = ( + lambda s, k, eid: svc.store.idem.discard_completed(s, k) or True) + a2 = svc.das._reclaim_ended_receipt(a, name=AGENT) + svc.isvc.complete(a2, FRESH, _receipt(FRESH)) + _insert_exec(svc.store, FRESH, "running") + b2 = svc.das._reclaim_ended_receipt(b, name=AGENT) + assert self._new_count(a2, b2) == 2 + + +class TestReclaimLog: + def _log(self, caplog): + lines = [r.getMessage() for r in caplog.records if "receipt_reclaim" in r.getMessage()] + assert len(lines) == 1, lines + assert "k-log" not in lines[0] + return lines[0] + + def _setup(self, svc): + _insert_exec(svc.store, STALE, "failed", error="boom", completed_at=T0) + _store_receipt(svc, "k-log") + return svc.isvc.begin(SCOPE, "k-log") + + def test_deleted_new(self, svc, caplog): + idem = self._setup(svc) + with caplog.at_level("INFO"): + svc.das._reclaim_ended_receipt(idem, name=AGENT) + line = self._log(caplog) + assert f"agent={AGENT}" in line and f"stale_exec={STALE}" in line + assert "verdict=ended" in line and "cas=deleted" in line and "outcome=new" in line + + def test_no_match_in_flight(self, svc, caplog): + idem = self._setup(svc) + svc.store.idem.discard_completed(SCOPE, "k-log") + svc.store.idem.claim(SCOPE, "k-log") # another retry won + with caplog.at_level("INFO"): + out = svc.das._reclaim_ended_receipt(idem, name=AGENT) + assert out.in_flight + line = self._log(caplog) + assert "cas=no_match" in line and "outcome=in_flight" in line + + def test_no_match_completed_names_the_winner(self, svc, caplog): + idem = self._setup(svc) + svc.store.idem.discard_completed(SCOPE, "k-log") + svc.store.idem.claim(SCOPE, "k-log") + svc.store.idem.complete(SCOPE, "k-log", FRESH, _receipt(FRESH)) + with caplog.at_level("INFO"): + out = svc.das._reclaim_ended_receipt(idem, name=AGENT) + assert out.execution_id == FRESH + line = self._log(caplog) + assert "cas=no_match" in line and "outcome=completed" in line + assert f"replay_exec={FRESH}" in line + + def test_delete_error_replays_the_original_without_a_second_begin(self, svc, caplog, + monkeypatch): + idem = self._setup(svc) + svc.db.idempotency_discard_completed_if_execution.raises = RuntimeError("locked") + claims = _Spy(svc.store.idem.claim) + monkeypatch.setattr(svc.db, "idempotency_claim", claims) + with caplog.at_level("INFO"): + out = svc.das._reclaim_ended_receipt(idem, name=AGENT) + assert out is idem and claims.calls == 0 + line = self._log(caplog) + assert "cas=error" in line and "outcome=replay_original" in line + assert f"replay_exec={STALE}" in line + + def test_second_begin_failing_open_is_no_dedup(self, svc, caplog, monkeypatch): + idem = self._setup(svc) + monkeypatch.setattr(svc.db, "idempotency_claim", _Spy(None, raises=RuntimeError("x"))) + with caplog.at_level("INFO"): + out = svc.das._reclaim_ended_receipt(idem, name=AGENT) + assert not out.enabled and not out.replay + assert "outcome=no_dedup" in self._log(caplog) + + +class TestAC2EndToEnd: + """AC2 over the real `/task` endpoint (`routers.chat.execute_parallel_task`), + collaborators mocked as in test_1483, the idempotency layer real.""" + + def test_dispatch_fail_resend_new_id_then_replay_the_new_run(self, svc, monkeypatch): + import asyncio + import sys + from unittest.mock import AsyncMock, MagicMock + from fastapi.responses import JSONResponse + from routers.chat import execute_parallel_task + from models import ParallelTaskRequest + from services import chat_persistence_service + import services.chat_execution_service as ce + + router_mod = sys.modules[execute_parallel_task.__module__] + ids = iter(["exec-first-a1", "exec-second-b2"]) + + def _create(*a, **kw): + eid = next(ids) + _insert_exec(svc.store, eid, "running") + return MagicMock(id=eid) + + db = MagicMock() + db.get_execution_timeout.return_value = 3600 + db.get_max_parallel_tasks.return_value = 3 + db.get_agent_subscription_id.return_value = None + db.create_task_execution.side_effect = _create + cap = MagicMock(acquire=AsyncMock(return_value=MagicMock(state="admitted", + queue_position=0)), + release=AsyncMock(), force_release=AsyncMock()) + monkeypatch.setattr(router_mod, "get_agent_container", + lambda name: MagicMock(status="running")) + monkeypatch.setattr(router_mod, "db", db) + monkeypatch.setattr(ce, "db", db) + monkeypatch.setattr(ce, "idempotency_service", svc.isvc) + monkeypatch.setattr(svc.das, "platform_audit_service", MagicMock(log=AsyncMock())) + monkeypatch.setattr(ce, "get_capacity_manager", lambda: cap) + monkeypatch.setattr(ce, "dispatch_breaker_active", lambda name: False) + monkeypatch.setattr(ce, "activity_service", MagicMock( + track_activity=AsyncMock(return_value="act1"), complete_activity=AsyncMock())) + monkeypatch.setattr(chat_persistence_service, "persist_chat_session", + AsyncMock(return_value="cs1")) + monkeypatch.setattr(ce, "run_async_task", AsyncMock()) + + def send(): + return asyncio.run(execute_parallel_task( + request=ParallelTaskRequest(message="do it", async_mode=True), name=AGENT, + current_user=MagicMock(id=1, email="u@e.com", username="u", role="user", + agent_name=None), + x_source_agent=None, x_via_mcp=None, idempotency_key="k-ac2", + x_event_trigger=None, x_internal_secret=None, + )) + + first = send() + assert first["execution_id"] == "exec-first-a1" and first["async_mode"] is True + _set_exec(svc.store, "exec-first-a1", status="failed", error="boom", completed_at=T0) + second = send() + assert not isinstance(second, JSONResponse) # no X-Idempotent-Replay + assert second["execution_id"] == "exec-second-b2" + third = send() # exec-second-b2 is running + assert isinstance(third, JSONResponse) + assert third.headers.get("X-Idempotent-Replay") == "true" + import json + assert json.loads(third.body)["execution_id"] == "exec-second-b2" + + +class TestTaskGateSeam: + """The `/task` twin of `TestChatSeam::test_gate_record_answers_before_the_receipt`: + `dispatch_parallel_task` runs `skill_gate_service.enforce` before + `begin_task_idempotency`, so a gated request gets the gate's answer and never + reaches the receipt reclaim. Driven through the real + `routers.chat.execute_parallel_task`, the idempotency layer real.""" + + @pytest.mark.parametrize("which", ["decided", "pending"]) + def test_gate_record_answers_before_the_receipt(self, svc, monkeypatch, which): + import asyncio + import sys + from unittest.mock import AsyncMock, MagicMock + from routers.chat import execute_parallel_task + from models import ParallelTaskRequest + from services.skill_gate_errors import SkillApprovalRequired, SkillGateRefused + import services.chat_execution_service as ce + + exc = (SkillGateRefused(409, "request_denied", "denied") if which == "decided" + else SkillApprovalRequired(request_id="req-task-9c4", agent_name=AGENT, + skills=["deploy"], approver_role="owner", + expires_at=None)) + _insert_exec(svc.store, STALE, "failed", error="boom", completed_at=T0) + _store_receipt(svc, "k-task-gated", status="accepted") + + router_mod = sys.modules[execute_parallel_task.__module__] + db = MagicMock() + db.get_execution_timeout.return_value = 3600 + db.get_max_parallel_tasks.return_value = 3 + monkeypatch.setattr(router_mod, "get_agent_container", + lambda name: MagicMock(status="running")) + monkeypatch.setattr(router_mod, "db", db) + monkeypatch.setattr(ce, "db", db) + monkeypatch.setattr(ce, "idempotency_service", svc.isvc) + monkeypatch.setattr(svc.das, "platform_audit_service", MagicMock(log=AsyncMock())) + enforce = AsyncMock(side_effect=exc) + monkeypatch.setattr(ce.skill_gate_service, "enforce", enforce) + + with pytest.raises(type(exc)): + asyncio.run(execute_parallel_task( + request=ParallelTaskRequest(message="deploy it", async_mode=True), name=AGENT, + current_user=MagicMock(id=1, email="u@e.com", username="u", role="user", + agent_name=None), + x_source_agent=None, x_via_mcp=None, idempotency_key="k-task-gated", + x_event_trigger=None, x_internal_secret=None, + )) + assert enforce.await_count == 1 + assert _row(svc, "k-task-gated") == ("completed", STALE) + assert svc.db.get_execution_gate_state.calls == 0 + db.create_task_execution.assert_not_called() + + +# --------------------------------------------------------------------------- +# Structural guards (supplementary pins over the behaviour above) +# --------------------------------------------------------------------------- + +def _fn(tree, name): + import ast + return next(n for n in ast.walk(tree) + if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) and n.name == name) + + +def test_both_seams_reclaim_right_after_begin(): + import ast + tree = ast.parse((_backend_root() / "services" / "dispatch_admission_service.py").read_text()) + for seam in ("admit_chat_request", "begin_task_idempotency"): + body = _fn(tree, seam).body + idx = next(i for i, s in enumerate(body) + if isinstance(s, ast.Assign) and isinstance(s.value, ast.Call) + and ast.unparse(s.value.func) == "idempotency_service.begin") + nxt = body[idx + 1] + assert isinstance(nxt, ast.Assign) and ast.unparse(nxt) == \ + "idem = _reclaim_ended_receipt(idem, name=name)", seam + + +_CLAIM_WRITERS = {"complete", "fail", "upgrade_snapshot", "attach_execution"} + + +def _claim_writes(node): + import ast + return [c for c in ast.walk(node) if isinstance(c, ast.Call) + and isinstance(c.func, ast.Attribute) and c.func.attr in _CLAIM_WRITERS + and isinstance(c.func.value, ast.Name) and c.func.value.id == "idempotency_service"] + + +def test_nothing_writes_a_claim_after_its_receipt(): + """The compare-and-delete is safe only because no late writer touches a + claim once its receipt is stored (a late `complete` would resurrect or + overwrite it).""" + import ast + root = _backend_root() / "services" + ce = ast.parse((root / "chat_execution_service.py").read_text()) + assert _claim_writes(_fn(ce, "run_async_task")) == [] + for mod in ("backlog_service.py", "pull_coordination_service.py"): + assert _claim_writes(ast.parse((root / mod).read_text())) == [], mod + receipts = 0 + for node in ast.walk(ce): + for field in ("body", "orelse", "finalbody"): + stmts = getattr(node, field, None) + if not isinstance(stmts, list): + continue + for i, s in enumerate(stmts): + if (isinstance(s, ast.Expr) and _claim_writes(s) + and ast.unparse(s.value).endswith(", receipt)")): + receipts += 1 + rest = stmts[i + 1:] + assert isinstance(rest[-1], ast.Raise), ast.unparse(s) + assert all(isinstance(r, (ast.Assign, ast.Raise)) and not _claim_writes(r) + for r in rest), ast.unparse(s) + assert receipts == 2, "the sync /task and pulled /chat receipt writers moved" diff --git a/tests/unit/test_ent568_delegation_contract.py b/tests/unit/test_ent568_delegation_contract.py index 2d8dbd958c..96108b4809 100644 --- a/tests/unit/test_ent568_delegation_contract.py +++ b/tests/unit/test_ent568_delegation_contract.py @@ -53,8 +53,11 @@ # chars on `chat_with_agent`, one name-only line on a `chat_with_` tool). # The MCP suite pins the published descriptions themselves; this keeps the # shared text from eating that headroom. Raising it is a decision, not an -# accident — the #1535 / ent#536 context-budget discipline. -MAX_CONTRACT_CHARS = 1650 +# accident — the #1535 / ent#536 context-budget discipline. #3245 raised it +# 1650 → 1730 for the confirmed-failure retry bullet (1,720 chars), and trimmed +# `chat_with_agent`'s lead by 36 chars to keep its published description inside +# the cap. +MAX_CONTRACT_CHARS = 1730 # Every backticked token in the contract, classified. A token outside these # sets fails the build until someone decides what it is and adds it to the set @@ -82,6 +85,9 @@ "pending_approval": ["src/mcp-server/src/client.ts"], "agent_busy": ["src/mcp-server/src/tools/chat.ts"], "running": ["src/backend/models.py"], + # #3245: the terminal statuses `get_execution_result` reports. + "failed": ["src/backend/models.py"], + "cancelled": ["src/backend/models.py"], } # field -> a file whose answer carries it. FIELD_PRODUCERS = { @@ -252,6 +258,21 @@ def test_the_contract_states_each_acceptance_term(term): assert term in DELEGATION_CONTRACT +def test_a_confirmed_failure_is_retried_word_for_word(): + """#3245 (UC2): the server now dispatches a verbatim re-send after a run + ended `failed`/`cancelled`, so the contract must say so — in ONE bullet, + together with what to do when the same `execution_id` comes back (a + `lease_expired` failure is held until its window ends): set a reminder and + end the turn, rather than re-sending in a loop.""" + bullets = [b for b in DELEGATION_CONTRACT.split("\n") + if "`failed`" in b and "`cancelled`" in b] + assert len(bullets) == 1, bullets + bullet = bullets[0] + assert "re-send word for word to retry" in bullet + assert "same `execution_id` back" in bullet and "`set_reminder`" in bullet + assert "end your turn" in bullet + + def test_every_backticked_token_is_classified(): known = (TOOLS | set(TOOL_GLOBS) | set(STATUS_PRODUCERS) | set(FIELD_PRODUCERS) | EVENTS | set(ARGUMENT_FORMS))