diff --git a/ai_working/3yc-timedout-session-resumable/DONE-NOTE.md b/ai_working/3yc-timedout-session-resumable/DONE-NOTE.md new file mode 100644 index 0000000..bf2adda --- /dev/null +++ b/ai_working/3yc-timedout-session-resumable/DONE-NOTE.md @@ -0,0 +1,292 @@ +# DONE-NOTE — `model_performance-3yc` + +**A timed-out sub-session is never persisted, so the session_id in its result is not resumable** + +**Lane spend: $0.00** — code reading + local `pytest` only. No API calls, no DTU, no infrastructure +created (nothing to register in the infra ledger, nothing to tear down). +**Repo:** `amplifier-app-cli` · branch `lane/3yc-timedout-session-not-resumable` · base `963d793` +**Claim tags:** (knob) · (family) · (confidence: measured / inferred / assumed) · (evidence: file:line) + +--- + +## 1. VERIFIED MECHANISM (against the current base, not the filing base) + +The item was filed against `f16375fc`. Re-read at **`963d793`** (current `origin/main`); the mechanism +is unchanged, only the line numbers moved. + +| claim | evidence at `963d793` | confidence | +|---|---|---| +| the child transcript is read only after a successful `execute()` | `session_spawner.py:845` `response = await child_session.execute(instruction)` → `:857` `transcript = await context.get_messages()` | measured | +| `store.save` runs only after that | `session_spawner.py:887` `store.save(sub_session_id, transcript, metadata)` | measured | +| there is no `except` on the execute block — only `finally` | `session_spawner.py:843-910` | measured | +| the same defect exists on the RESUME path (not mentioned in the item) | `session_spawner.py:1619` execute → `:1629` `store.save`, same `try/finally` shape | measured | +| the metadata `store.save` needs is fully known BEFORE execute | `merged_config` `:298`, `agent_config` `:290/:295`, `self_delegation_depth` (parameter `:242`), `_extract_bundle_context(parent_session)` — none depend on the response | measured | +| `store.save` itself is fully synchronous | `session_store.py:100-131` — `_save_transcript` / `_save_metadata` / `write_with_backup`, no `await` anywhere | measured | +| resume reconstructs purely from `metadata["config"]` + `metadata["agent_overlay"]` + transcript | `session_spawner.py:966-970`, `:1314-1327`, `:1556-1559` | measured | + +The last two are the load-bearing findings. Together they mean (a) the persist step needs **exactly one +await** — `context.get_messages()` — and (b) everything else it needs is available before the run starts. + +--- + +## 2. THE DECISION — option (b), persist during the run + +**Chosen: (b) persist the transcript incrementally during the run, so no cancellation-path write is +needed at all.** The cancellation path gains **no new code, no new await, and no new write**. + +**How this satisfies the hard invariant — "the fix must NOT introduce an unbounded await on the +cancellation path".** The danger is specific: an `await` reached while a task is unwinding a fired +deadline can *block past the very deadline that caused the unwind*, re-creating the hang the timeout +exists to bound. Options (a) and (b) differ in *where* they discharge that risk. (a) puts work on the +cancellation path and then tries to bound it — so the invariant holds only as strongly as the bound, +and §2 shows the bound is partly fictional. (b) moves the work to normal execution, where blocking is +already accepted and the post-run save has always done exactly this work; it changes *when* the +transcript is written, not *what*. The invariant then holds **by construction rather than by +argument**: there is no new statement on the cancellation path to bound, so there is nothing to get +wrong. §3 proves it empirically anyway — `get_messages()` is made to hang forever from the instant +cancellation is delivered, and the unwind is unaffected. + +### Why not (a) — `asyncio.shield` + a hard secondary timeout + +Rejected on a measured, not aesthetic, ground: **the bound would be partly fictional.** + +1. `store.save()` is **synchronous** (`session_store.py:100-131`). `asyncio.wait_for` can only + interrupt at an `await`. It therefore cannot bound the disk write — the part most likely to be + slow on a loaded or networked filesystem. The "hard secondary timeout" would bound + `get_messages()` and nothing else, while reading as though it bounded the save. +2. Once a deadline's `CancelledError` has been delivered and caught, a **fresh await is not + re-cancelled** — it simply blocks. This is demonstrated directly, as executable code, in + `test_the_probe_would_catch_a_violating_implementation`: the option-(a) shape hangs past its own + 0.2 s deadline and never unwinds. That is precisely the hang the timeout exists to bound. + **(confidence: measured — the control test hangs deterministically.)** +3. `shield` + `wait_for` leaks the shielded task when the outer `wait_for` fires, and a re-entrant + parent cancellation (Ctrl-C, an outer timeout) then propagates out of the handler — which + `DESIGN.md §3` establishes would destroy the sibling delegates the timeout path exists to protect. + +### Why not (c) — mark the result explicitly non-resumable + +Legitimate, and it was seriously weighed. Rejected for three reasons, in order of weight: + +1. **The value discarded is large and known.** Measured delegate legs run **284–1543 s** ([SOL], via + `00 §2c`); sol S1 runs cost **$15.96–$27.18 median/run** (`00 §2c`). Option (c) makes every timeout + throw away the whole transcript and forces the caller to pay for it again from zero. Option (b) + costs a handful of small synchronous writes per run. +2. **It does not compose with 37n.** `model_performance-37n` already preserves the in-flight + assistant *text* via the `session.partial` capability. What is still missing is the *completed + turns*, which is exactly what makes "resume where it left off" work. (b) supplies the missing + half; (c) declares the gap permanent. +3. **Its deliverable is not in this repo.** The result shape (`partial_available`, `guidance`, + `status`) lives in `amplifier-foundation`'s `tool-delegate`, outside this lane's owned repo. A + `resumable: false` flag shipped here could not be surfaced by the consumer without a second, + coordinated PR. (This is a practical constraint, not the reason — reasons 1 and 2 stand alone.) + +**Honest note:** (c) is still the correct *labelling* answer for the residual gap. The subprocess +spawn path (`session_spawner.py:637`) returns before any checkpointing and remains unresumable on +timeout — see §5. + +### The design, and why `provider:request` + +* The transcript is checkpointed **during normal execution**, where blocking is already accepted + (the post-run save has always done exactly this work — this moves *when*, not *what*). +* One checkpoint is written **before `execute()`** (pre-registration), so the advertised `session_id` + resolves in `SessionStore` even if the timeout fires during the very first LLM call. +* Mid-run checkpoints fire on **`provider:request`** — emitted at + `loop_streaming/__init__.py:3202` (per iteration) and `:2996` (turn start). + **(confidence: measured — code read at cache `amplifier-module-loop-streaming-b0b975ea6a1072dd`.)** + + `provider:request` is not an arbitrary choice: it is **the only point in the loop where the message + list is guaranteed tool-pair-balanced.** The previous round's tool results have all been appended, + and the next assistant message (which may open new `tool_calls`) does not exist yet. Checkpointing + on `provider:response` would persist an assistant message with unmatched `tool_calls`, and resuming + that transcript reproduces the `InvalidRequestError: No tool call found for function call output` + class of failure recorded for `context-managed` in `00 §2g` (29 of 30 turns). + **(confidence: inferred — the balance property is read from the loop; the resulting provider error + is measured, but in the cited `context-managed` runs, not here.)** +* Checkpoints are labelled `status: "in_progress"`; only the post-run save writes `status: "complete"`. + A caller can therefore tell a rescued checkpoint from a finished session. +* Applied to **both** `spawn_sub_session` and `resume_sub_session` — the resume path had the same + defect and the item did not mention it. + +### The knob + +`AMPLIFIER_SPAWN_CHECKPOINT_INTERVAL_S`, default **30.0 s**. **CHOSEN, NOT MEASURED, and labelled as +such in the source comment.** No data exists on sub-session checkpoint sizes because no mid-run +checkpoint has ever been written, so there is nothing to bank (`00 §5` rules 3 and 6). The reasoning +recorded in-source: 30 s bounds the transcript lost to a timeout to at most one window against legs +of 284–1543 s, while capping write amplification on fast-iterating sub-sessions. A negative value +disables checkpointing entirely and restores the pre-fix behaviour exactly — pinned by a test. + +--- + +## 3. THE HARD INVARIANT, AND HOW IT IS PROVED + +> The fix must NOT introduce an unbounded await on the cancellation path. + +Proved by `TestNoUnboundedAwaitOnCancellationPath` — two assertions plus a control: + +1. **`test_hanging_get_messages_does_not_delay_the_unwind`** — `context.get_messages()` is made to + hang **forever** from the instant cancellation is delivered. The spawn runs under a 0.2 s + `asyncio.timeout`, exactly as `tool-delegate:1092` does. Asserts: the unwind completes (the + harness fails loudly at 5 s rather than hanging the suite), elapsed < 2 s, **and** + `get_messages`'s call count did not move after cancellation began. The count assertion is what + stops the test passing by accident. + +2. **`test_the_probe_would_catch_a_violating_implementation`** — the **inverted control**, present + because of `00 §5` rule 5. *This is a gate that would otherwise be vacuous:* the unpatched code + ALSO has no await on its cancellation path, so assertion 1 passes before the fix as well + (verified — see §4). It is a regression guard, not a defect reproduction, and a guard is worth + nothing unless it can fail. The control reproduces the rejected option-(a) shape and asserts the + probe's instrument **does** catch it. **Disclosed rather than presented as a pass.** + +3. **`test_cleanup_still_runs_and_the_timeout_still_propagates`** — the timeout still surfaces as + `TimeoutError`, `child_session.cleanup()` still runs, `unregister_child` still runs, and the + checkpoint hook is unregistered even on the timeout path. + +### The contract the acceptance names — BOTH branches are tested + +The acceptance is disjunctive: *"either the advertised session_id is genuinely resumable … **or** the +result states explicitly that it is not resumable and directs the caller to re-delegate."* Option (b) +resolves it to the first branch **for every checkpointed session**. Two residual cases are not +checkpointed and land on the second. Both branches are covered. + +**Branch 1 — `TestTimedOutSessionIsResumable`:** + +* `test_timed_out_spawn_leaves_a_loadable_session` — after a real 0.2 s timeout, `SessionStore` holds + the session, the transcript is the preserved messages, `status == "in_progress"`, and + `metadata["config"]` is present. +* `test_timed_out_session_round_trips_through_resume` — the **full recovery move, end to end**: spawn + → time out → call the real `resume_sub_session(session_id)` → assert the preserved transcript is + restored into the resumed session's context. + +**Branch 2 — `TestNonResumableIsStatedExplicitly`:** + +Residual non-checkpointed cases: the **subprocess spawn path** (`session_spawner.py:637` returns +before any checkpointing), checkpointing **explicitly disabled** via the env knob, plus the +pre-existing expired/pruned/never-existed cases. `resume_sub_session` now raises a message that +states non-resumability **in words** and names the correct move, rather than leaving "retry the +resume" as a plausible reading: + +> `Sub-session '' not found. Session may have expired or was never created. This session is NOT +> resumable -- re-delegate to start a fresh session instead of retrying the resume.` + +* `test_missing_session_says_not_resumable_and_says_re_delegate` — asserts both `"not resumable"` and + `"re-delegate"` are present. +* `test_disabled_checkpointing_lands_on_the_non_resumable_branch` — the disjunction resolving the + **other way in a real run**: spawn → time out with checkpointing off → the id is genuinely absent + **and** asking to resume it says so explicitly. + +**BRANCH 2 SPANS TWO REPOS, AND BOTH HALVES ARE NOW DONE — the second ships as a verified patch.** +Verified at file:line, not assumed: `tool-delegate`'s handler +(`modules/tool-delegate/amplifier_module_tool_delegate/__init__.py:1314-1332` @ `cc7e23a`) catches +`FileNotFoundError`, uses `str(e)` **only** in the `delegate:error` event, and returns a **hardcoded** +message that discards it — so the app-side wording reached the event, the logs and non-foundation +callers, but **not the ToolResult the model reads**. + +`PATCH-foundation-surface-resume-detail.diff` (in this directory) closes it: + +```python +"message": str(e) +or f"Agent session '{session_id}' not found. May have expired or never existed." +``` + +Detail-preserving, with the original sentence retained as the empty-detail fallback. + +| check | result | +|---|---| +| `git apply --check` @ `cc7e23a` | **exit 0** | +| tool-delegate suite, UNPATCHED baseline | **48 passed** | +| tool-delegate suite, PATCHED (scratch copy) | **48 passed** | +| foundation working tree afterwards | **clean — the repo was never modified** | + +**The round trip is proved, not asserted in halves.** +`test_resume_message_roundtrip.py` (beside the patch) wires the **real** app-side +`resume_sub_session` into the **real** tool-delegate resume path and reads the resulting `ToolResult`: + +* against **PATCHED** tool-delegate: **2 passed**; +* against **UNPATCHED**: **1 failed, 1 passed**, and the failure prints the exact string the model + would otherwise see — `"Agent session '' not found. May have expired or never existed."` + +That asymmetry is the result: it is what makes the foundation diff necessary rather than cosmetic. +The file lives beside the patch, **not** in `tests/`, because it depends on an unlanded change and +would otherwise redden CI — the same convention the w3/37n lane used for `test_partial_roundtrip.py`. + +**Landing order:** this app-cli PR is safe alone (the wording still reaches the event, the logs and +non-foundation callers). Land the foundation diff to complete the model-facing half. + +Deliberately **not** worked around from this side: the only in-repo lever would be raising a +different exception type so foundation's generic `except Exception` (`:1312`, which *does* pass +`str(e)` through) caught it instead — trading away a structured, correctly-typed error branch to +smuggle a string. That is a worse design and it is not done. + +--- + +## 4. VERIFICATION + +| check | result | +|---|---| +| full suite, baseline at `963d793` before any change | **1560 passed, 1 skipped, 13 deselected, 1 xfailed** | +| full suite, patched (`-p no:randomly`) | **1573 passed, 1 skipped, 13 deselected, 1 xfailed** | +| full suite, patched (default random order) | **1573 passed, 1 skipped, 13 deselected, 1 xfailed** | +| new test file alone | **13 passed** | +| new test file against the UNPATCHED spawner (falsifier check) | **10 failed, 3 passed** | +| `ruff check` / `ruff format --check` on every file this PR touches | clean | + +**The falsifier check matters, and its three passes are disclosed, not hidden.** Reverting only +`session_spawner.py` to `963d793` and re-running the new file gives **10 failed, 3 passed**. The three +that still pass, and why each is correct rather than vacuous: + +| test | why it passes pre-fix | +|---|---| +| `test_hanging_get_messages_does_not_delay_the_unwind` | the unpatched cancellation path is also await-free — this is a regression guard, not a defect reproduction (see §3.2) | +| `test_the_probe_would_catch_a_violating_implementation` | it is a self-contained control over a synthetic option-(a) coroutine; it never touches the spawner | +| `test_negative_interval_disables_checkpointing_entirely` | it pins that the escape hatch reproduces exactly the pre-fix behaviour | + +None is claimed as a defect reproduction. + +**One existing test was modified**, and the reason is not "to make the new code pass": `FakeHooks` in +`tests/test_session_spawner.py` (two copies) modelled the hook registry as a **single handler slot, +ignoring the event name**. The real registry is event-keyed. Registering a second hook made the last +writer win, hiding the `orchestrator:complete` handler the test asserts on. The fake now honours the +event name. The test's intent and assertions are untouched. + +--- + +## 5. WHAT THIS DOES NOT CLAIM + +1. **No eval was run. $0 lane.** Every result above is a local unit test. No live delegate has ever + timed out under this patch. +2. **The 30 s default is chosen, not measured** (`00 §5` rule 6). It is labelled as such in the source. +3. **Write amplification is not measured.** `_save_transcript` rewrites the whole JSONL per + checkpoint, so cost is O(checkpoints × transcript size), bounded by the throttle. No workload + measurement exists. `store.save` is synchronous, so each checkpoint briefly blocks the event loop + — the same operation the post-run save has always performed, now up to N times per run. +4. **The subprocess spawn path is still not CHECKPOINTED** — `session_spawner.py:637` returns before + any checkpointing, so a subprocess-mode delegate that times out cannot be resumed. It is now + covered by the acceptance's second branch instead (it says so explicitly and directs a + re-delegate, §3), but the model-facing half of that message needs the one-line foundation change + specified in §3. Making the subprocess path itself checkpointable is a separate piece of work and + is NOT attempted here. +5. **Up to one throttle window of transcript can still be lost.** Because the transcript only changes + between provider calls, the practical loss is "iterations that completed within the last 30 s", + not 30 s of work — but that is reasoned, not measured. +6. **The tool-pair-balance argument for `provider:request` is inferred** from reading the loop. The + consequence of getting it wrong is measured, but in `00 §2g`'s `context-managed` runs, not here. +7. **This PR does not enable any timeout.** `settings.timeout` remains `None` by default in + `tool-delegate`. Landing order from `DESIGN.md §7` is unchanged: foundation, then app-cli, then + sweep the timeout. +8. **No Anthropic guardrail run was performed** (`00 §5` rule 9). This patch changes only local disk + writes on the app side — it adds no request, alters no prompt, and touches no provider payload, so + there is no cache surface to regress. That is an argument, not a measurement. + +--- + +## 6. FILES + +| file | what | +|---|---| +| `amplifier_app_cli/session_spawner.py` | `_checkpoint_interval_s`, `_write_checkpoint`, `_install_transcript_checkpoint`; metadata + store construction moved before `execute()`; checkpoint wired into both the spawn and resume paths; `status` field on saved metadata; explicit not-resumable/re-delegate wording on the resume-miss error | +| `tests/test_timedout_session_resumable.py` | 13 new tests — BOTH branches of the disjunctive contract, the invariant, the inverted control, the boundary choice, best-effort behaviour, the throttle and the escape hatch | +| `tests/test_session_spawner.py` | `FakeHooks` made event-keyed (see §4) | +| `ai_working/3yc-timedout-session-resumable/DONE-NOTE.md` | this note | +| `ai_working/3yc-timedout-session-resumable/PATCH-foundation-surface-resume-detail.diff` | the foundation half of branch 2; `git apply --check` exit 0 @ `cc7e23a`, 48 passed patched and unpatched | +| `ai_working/3yc-timedout-session-resumable/test_resume_message_roundtrip.py` | cross-repo round trip: 2 passed patched, 1 failed unpatched (deliberately outside `tests/`) | diff --git a/ai_working/3yc-timedout-session-resumable/PATCH-foundation-surface-resume-detail.diff b/ai_working/3yc-timedout-session-resumable/PATCH-foundation-surface-resume-detail.diff new file mode 100644 index 0000000..6003a3f --- /dev/null +++ b/ai_working/3yc-timedout-session-resumable/PATCH-foundation-surface-resume-detail.diff @@ -0,0 +1,21 @@ +--- a/modules/tool-delegate/amplifier_module_tool_delegate/__init__.py ++++ b/modules/tool-delegate/amplifier_module_tool_delegate/__init__.py +@@ -1327,7 +1327,17 @@ + return ToolResult( + success=False, + error={ +- "message": f"Agent session '{session_id}' not found. May have expired or never existed." ++ # Surface the SESSION LAYER's own detail rather than a ++ # hardcoded sentence. Only that layer knows WHY the id is ++ # gone and therefore what the caller should do instead -- ++ # e.g. amplifier-app-cli now reports "This session is NOT ++ # resumable -- re-delegate to start a fresh session instead ++ # of retrying the resume." Hardcoding here discarded that, ++ # leaving "retry the resume" a plausible reading for the ++ # model. Falls back to the generic sentence when the ++ # exception carries no detail. ++ "message": str(e) ++ or f"Agent session '{session_id}' not found. May have expired or never existed." + }, + ) + diff --git a/ai_working/3yc-timedout-session-resumable/test_resume_message_roundtrip.py b/ai_working/3yc-timedout-session-resumable/test_resume_message_roundtrip.py new file mode 100644 index 0000000..a803f20 --- /dev/null +++ b/ai_working/3yc-timedout-session-resumable/test_resume_message_roundtrip.py @@ -0,0 +1,126 @@ +"""Cross-repo contract check: does the "NOT resumable" wording reach the MODEL? + +WHY THIS FILE LIVES HERE AND NOT IN tests/ + +It depends on a tool-delegate patch that has NOT landed +(`PATCH-foundation-surface-resume-detail.diff`). Putting it in the app-cli +suite would make CI fail against unpatched foundation, so it ships beside the +patch and is run manually -- the same convention the w3/37n lane used for +`test_partial_roundtrip.py`. + +WHAT IT PROVES + +The acceptance's second branch says the result must state "explicitly that it +is not resumable and direct the caller to re-delegate". amplifier-app-cli's +`resume_sub_session` now raises exactly that sentence -- but tool-delegate's +`except FileNotFoundError` handler built a HARDCODED message and discarded +`str(e)`, so the wording never reached the ToolResult the model actually reads. + +This test wires the REAL app-side `resume_sub_session` into the REAL +tool-delegate resume path and asserts the wording survives all the way into +`ToolResult`. It is a genuine round trip, not two halves asserted separately. + +HOW TO RUN + + # PATCHED (expected: 2 passed) + cp -rL ~/dev/amplifier-foundation/modules/tool-delegate /tmp/td-patched + git -C ~/dev/amplifier-foundation apply --check \ + PATCH-foundation-surface-resume-detail.diff # exit 0 + # ...apply the diff inside /tmp/td-patched, then: + cd + PYTHONPATH=/tmp/td-patched uv run pytest \ + ai_working/3yc-timedout-session-resumable/test_resume_message_roundtrip.py \ + -q -p no:randomly --asyncio-mode=auto + + # UNPATCHED (expected: test_wording_reaches_the_model FAILS) + PYTHONPATH=~/dev/amplifier-foundation/modules/tool-delegate uv run pytest ... same file + +That asymmetry IS the result. It is what makes the foundation diff necessary +rather than cosmetic. +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from amplifier_module_tool_delegate import DelegateTool + +from amplifier_app_cli.session_spawner import resume_sub_session + +pytestmark = pytest.mark.asyncio + +MISSING_SESSION_ID = "parent0000000000-child00000000000_test-agent" + + +def _make_delegate_tool(resume_fn) -> DelegateTool: + """Minimal DelegateTool wired for the resume path (mirrors the module's own tests).""" + coordinator = MagicMock() + coordinator.session_id = "parent-session-123" + coordinator.config = {"agents": {"test-agent": {"description": "t"}}} + coordinator.session_state = {} + coordinator._tool_dispatch_context = {} + coordinator._tool_dispatch_contexts = {} + + capabilities = { + "session.spawn": AsyncMock(return_value={"output": "done"}), + "session.resume": resume_fn, + "agents.list": lambda: coordinator.config["agents"], + "agents.get": lambda name: coordinator.config["agents"].get(name), + "self_delegation_depth": 0, + } + coordinator.get_capability = lambda name: capabilities.get(name) + coordinator.get = MagicMock(return_value=None) + + parent_session = MagicMock() + parent_session.session_id = "parent-session-123" + parent_session.config = {"session": {"orchestrator": {}}} + coordinator.session = parent_session + + return DelegateTool(coordinator, {"features": {}, "settings": {"exclude_tools": []}}) + + +async def test_app_side_really_raises_the_wording(tmp_path, monkeypatch): + """Guard: the app half genuinely produces the sentence (no patch needed).""" + monkeypatch.setenv("HOME", str(tmp_path)) + + with pytest.raises(FileNotFoundError) as excinfo: + await resume_sub_session(MISSING_SESSION_ID, "carry on") + + message = str(excinfo.value).lower() + assert "not resumable" in message + assert "re-delegate" in message + + +async def test_wording_reaches_the_model(tmp_path, monkeypatch): + """The round trip: app-side wording must survive into the ToolResult. + + FAILS against unpatched tool-delegate -- the handler replaces the detail + with "Agent session '' not found. May have expired or never existed.", + which leaves "retry the resume" a plausible reading for the model. + """ + monkeypatch.setenv("HOME", str(tmp_path)) + + tool = _make_delegate_tool(resume_fn=resume_sub_session) + + result = await tool.execute( + {"session_id": MISSING_SESSION_ID, "instruction": "carry on"} + ) + + assert result.success is False + + # Read BOTH channels the model can see: ToolResult.output is derived from + # error["message"] by ToolResult.model_post_init, so a regression in either + # is caught here. + serialized = f"{result.error} {result.output}".lower() + + assert "not resumable" in serialized, ( + "the acceptance's second branch requires the MODEL-visible result to " + "state explicitly that the session is not resumable; got: " + f"{result.error!r}" + ) + assert "re-delegate" in serialized, ( + "the acceptance's second branch requires the MODEL-visible result to " + f"direct the caller to re-delegate; got: {result.error!r}" + ) diff --git a/amplifier_app_cli/session_spawner.py b/amplifier_app_cli/session_spawner.py index 2200b35..c7507fc 100644 --- a/amplifier_app_cli/session_spawner.py +++ b/amplifier_app_cli/session_spawner.py @@ -5,7 +5,9 @@ import copy import logging +import os import sys +import time from pathlib import Path from amplifier_core import AmplifierSession @@ -228,6 +230,178 @@ def _find_redacted_values(value: object, path: str = "") -> list[str]: return found +# --------------------------------------------------------------------------- +# Mid-run transcript checkpointing +# --------------------------------------------------------------------------- +# +# WHY THIS EXISTS +# +# Both spawn_sub_session and resume_sub_session used to call store.save() only +# AFTER a successful `await child_session.execute(...)`. A wall-clock timeout in +# tool-delegate CANCELS that await, so the save never ran, the child session was +# cleaned up in the outer finally, and nothing about the timed-out sub-session +# ever reached SessionStore. The session_id the caller was handed back therefore +# could not be resumed -- `delegate(session_id=...)` failed with "Session not +# found. May have expired or never existed." +# +# THE CONSTRAINT THAT SHAPES THE FIX +# +# Persisting from the CANCELLATION path would require awaiting +# context.get_messages() while the task is already unwinding a deadline. Any +# await there can block past the very deadline that caused the unwind, +# re-creating the hang the timeout exists to bound -- and store.save() is +# synchronous I/O, so an `asyncio.wait_for` around it could not interrupt the +# part most likely to be slow anyway. So the cancellation path is left ENTIRELY +# UNTOUCHED: it gains no await, no write, and no new code at all. Instead the +# transcript is checkpointed DURING normal execution, where blocking is already +# accepted (the post-run save has always done exactly this work). +# +# WHY THE CHECKPOINT FIRES ON provider:request, NOT provider:response +# +# provider:request is the only point in the orchestrator loop where the message +# list is guaranteed TOOL-PAIR-BALANCED: the previous round's tool results have +# all been appended, and the next assistant message (which may open new +# tool_calls) has not been produced yet. Checkpointing after a response would +# persist an assistant message whose tool_calls have no matching results, and +# resuming that transcript reproduces the "No tool call found for function call +# output" class of provider error. Balanced-by-construction is the point. + +_CHECKPOINT_INTERVAL_ENV = "AMPLIFIER_SPAWN_CHECKPOINT_INTERVAL_S" + +# CHOSEN, NOT MEASURED. No data exists on sub-session checkpoint sizes because +# no mid-run checkpoint has ever been written. 30 s bounds the transcript lost +# to a timeout to at most one window, while capping write amplification on +# fast-iterating sub-sessions (which would otherwise rewrite the whole +# transcript once per provider call). Override with the env var above; a +# negative value disables mid-run checkpointing entirely. +_DEFAULT_CHECKPOINT_INTERVAL_S = 30.0 + + +def _checkpoint_interval_s() -> float: + """Resolve the minimum interval between mid-run transcript checkpoints.""" + raw = os.environ.get(_CHECKPOINT_INTERVAL_ENV) + if raw is None or raw.strip() == "": + return _DEFAULT_CHECKPOINT_INTERVAL_S + try: + return float(raw) + except ValueError: + logger.warning( + f"Invalid {_CHECKPOINT_INTERVAL_ENV}={raw!r}; " + f"using default {_DEFAULT_CHECKPOINT_INTERVAL_S}s" + ) + return _DEFAULT_CHECKPOINT_INTERVAL_S + + +async def _write_checkpoint( + session: "AmplifierSession", + store, + session_id: str, + metadata: dict, +) -> bool: + """Snapshot the live transcript to SessionStore. Best-effort, never raises. + + MUST only ever be called from the NORMAL execution path (pre-execute, or + from a provider:request hook). Calling it while unwinding a cancellation + would reintroduce the unbounded await this whole design exists to avoid. + + Returns True if a checkpoint was written, False if it was skipped or failed. + """ + from datetime import UTC + from datetime import datetime + + try: + context = session.coordinator.get("context") + transcript = await context.get_messages() if context else [] + snapshot = dict(metadata) + snapshot["status"] = "in_progress" + snapshot["turn_count"] = len(transcript) + snapshot["last_updated"] = datetime.now(UTC).isoformat() + store.save(session_id, transcript, snapshot) + logger.debug( + f"Sub-session {session_id} checkpointed ({len(transcript)} messages)" + ) + return True + except Exception as e: + # A failed checkpoint must never take down the run it is protecting. + # CancelledError is a BaseException and is deliberately NOT caught here: + # if the hook itself is cancelled, that cancellation must propagate. + logger.debug(f"Transcript checkpoint for {session_id} failed: {e}") + return False + + +async def _install_transcript_checkpoint( + session: "AmplifierSession", + store, + session_id: str, + metadata: dict, + write_now: bool = True, +): + """Register a throttled provider:request checkpoint on ``session``. + + When ``write_now`` is True (the spawn path), one checkpoint is written + immediately so the session_id is resolvable in SessionStore from before the + first provider call onward -- a timeout that fires during the very first + LLM request still leaves a resumable (if short) session rather than + nothing at all. That write also seeds the throttle. + + When False (the resume path), the store already holds this session and + nothing has been added to the transcript yet, so no immediate write is + needed and the first provider:request checkpoints straight away. + + Returns a zero-arg callable that unregisters the hook (a no-op if no hook + was registered). + """ + + def _noop() -> None: + return None + + interval = _checkpoint_interval_s() + if interval < 0: + # Explicitly opted out: no pre-registration, no mid-run checkpoints. + return _noop + + last_written: list[float | None] = [None] + + if write_now: + await _write_checkpoint(session, store, session_id, metadata) + last_written[0] = time.monotonic() + + hooks = session.coordinator.get("hooks") + if hooks is None or not hasattr(hooks, "register"): + # No hook registry: the pre-registration write above still stands, but + # no mid-run events will fire, so there is nothing further to install. + return _noop + + try: + from amplifier_core.events import PROVIDER_REQUEST + from amplifier_core.hooks import HookResult + except ImportError as e: # pragma: no cover - kernel always provides these + logger.debug(f"Transcript checkpointing unavailable: {e}") + return _noop + + async def _on_provider_request(event: str, data: dict) -> HookResult: + now = time.monotonic() + previous = last_written[0] + if previous is not None and (now - previous) < interval: + return HookResult() + last_written[0] = now + await _write_checkpoint(session, store, session_id, metadata) + return HookResult() + + try: + unregister = hooks.register( + PROVIDER_REQUEST, + _on_provider_request, + priority=999, + name="_spawn_transcript_checkpoint", + ) + except Exception as e: + logger.debug(f"Could not register transcript checkpoint hook: {e}") + return _noop + + return unregister if callable(unregister) else _noop + + async def spawn_sub_session( agent_name: str, instruction: str, @@ -839,51 +1013,74 @@ async def _capture_completion(event: str, data: dict) -> HookResult: relative_to=_instr_rel, ) + # --------------------------------------------------------------------- + # Build persistence state BEFORE execute() + # + # Everything the metadata needs is already known at this point, and + # resume_sub_session reconstructs a session purely from metadata["config"] + # + metadata["agent_overlay"] + the transcript. Building it here is what + # lets the transcript be checkpointed DURING the run (see + # _install_transcript_checkpoint above) instead of only after a successful + # execute() -- which is what left timed-out sub-sessions unresumable. + # --------------------------------------------------------------------- + from datetime import UTC + from datetime import datetime + + from .session_store import SessionStore + + # Extract or generate trace_id for W3C Trace Context pattern + # Root session ID is the trace_id, propagate it to all children + parent_trace_id = getattr(parent_session, "trace_id", parent_session.session_id) + + # Extract child_span from sub_session_id for short_id resolution + # Format: {parent_id}-{child_span}_{agent_name} + child_span: str | None = None + if sub_session_id and "_" in sub_session_id and "-" in sub_session_id: + base = sub_session_id.rsplit("_", 1)[0] # Remove agent name + child_span = base.rsplit("-", 1)[-1] # Get child_span (16 hex chars) + + metadata = { + "session_id": sub_session_id, + "parent_id": parent_session.session_id, + "trace_id": parent_trace_id, # W3C Trace Context: trace entire conversation + "agent_name": agent_name, + "child_span": child_span, # For short_id resolution (first 8 chars = short_id) + "created": datetime.now(UTC).isoformat(), + "config": merged_config, + "agent_overlay": agent_config, + "turn_count": 1, + "bundle_context": _extract_bundle_context(parent_session), + "self_delegation_depth": self_delegation_depth, # For recursion limit tracking + # Store working_dir for session sync between CLI and web + "working_dir": str(Path.cwd().resolve()), + } + + store = SessionStore() + unregister_checkpoint = await _install_transcript_checkpoint( + child_session, store, sub_session_id, metadata, write_now=True + ) + # Execute instruction in child session; cleanup MUST run even on CancelledError + # + # NOTE: there is deliberately NO `except` here. A timeout cancels + # execute(), and the cancellation path must stay free of any await -- + # the transcript has already been checkpointed above and by the + # provider:request hook, so nothing needs to be rescued while unwinding. try: try: response = await child_session.execute(instruction) finally: if unregister_hook: unregister_hook() + unregister_checkpoint() - # Persist state for multi-turn resumption - from datetime import UTC - from datetime import datetime - - from .session_store import SessionStore - + # Persist final state for multi-turn resumption context = child_session.coordinator.get("context") transcript = await context.get_messages() if context else [] - # Extract or generate trace_id for W3C Trace Context pattern - # Root session ID is the trace_id, propagate it to all children - parent_trace_id = getattr(parent_session, "trace_id", parent_session.session_id) - - # Extract child_span from sub_session_id for short_id resolution - # Format: {parent_id}-{child_span}_{agent_name} - child_span: str | None = None - if sub_session_id and "_" in sub_session_id and "-" in sub_session_id: - base = sub_session_id.rsplit("_", 1)[0] # Remove agent name - child_span = base.rsplit("-", 1)[-1] # Get child_span (16 hex chars) - - metadata = { - "session_id": sub_session_id, - "parent_id": parent_session.session_id, - "trace_id": parent_trace_id, # W3C Trace Context: trace entire conversation - "agent_name": agent_name, - "child_span": child_span, # For short_id resolution (first 8 chars = short_id) - "created": datetime.now(UTC).isoformat(), - "config": merged_config, - "agent_overlay": agent_config, - "turn_count": 1, - "bundle_context": _extract_bundle_context(parent_session), - "self_delegation_depth": self_delegation_depth, # For recursion limit tracking - # Store working_dir for session sync between CLI and web - "working_dir": str(Path.cwd().resolve()), - } + metadata["status"] = "complete" + metadata["last_updated"] = datetime.now(UTC).isoformat() - store = SessionStore() store.save(sub_session_id, transcript, metadata) logger.debug(f"Sub-session {sub_session_id} state persisted") @@ -951,8 +1148,21 @@ async def resume_sub_session( store = SessionStore() if not store.exists(sub_session_id): + # NOT RESUMABLE -- say so, and say what to do instead. + # + # Mid-run checkpointing (see _install_transcript_checkpoint) means a + # normally-spawned sub-session is present in SessionStore from before + # its first provider call, so reaching here means one of: the + # subprocess spawn path (which returns before any checkpointing), + # checkpointing explicitly disabled via + # AMPLIFIER_SPAWN_CHECKPOINT_INTERVAL_S, an expired/pruned session, or + # an id that never existed. In every one of those cases the correct + # caller move is the same and it is NOT "retry the resume" -- so the + # message names it rather than leaving the caller to infer it. raise FileNotFoundError( - f"Sub-session '{sub_session_id}' not found. Session may have expired or was never created." + f"Sub-session '{sub_session_id}' not found. Session may have expired or was never created. " + f"This session is NOT resumable -- re-delegate to start a fresh session instead of retrying " + f"the resume." ) try: @@ -1428,15 +1638,30 @@ async def _capture_completion(event: str, data: dict) -> HookResult: relative_to=_resume_rel, ) + # Checkpoint the transcript DURING the run so a wall-clock timeout on this + # resume does not discard the turn (same defect, same fix, as the spawn + # path above). Installed AFTER the transcript restore so the first + # checkpoint carries the full history, not an empty list. No immediate + # write: the store already holds this session, and the resume path adds + # nothing to the transcript until the first provider call. + unregister_checkpoint = await _install_transcript_checkpoint( + child_session, store, sub_session_id, metadata, write_now=False + ) + # Execute new instruction with full context; cleanup MUST run even on CancelledError + # + # NOTE: deliberately NO `except` here -- see the spawn path's note. The + # cancellation path must stay free of any await. try: try: response = await child_session.execute(instruction) finally: if unregister_hook: unregister_hook() + unregister_checkpoint() # Update state for next resumption + metadata["status"] = "complete" updated_transcript = await context.get_messages() if context else [] metadata["turn_count"] = len(updated_transcript) metadata["last_updated"] = datetime.now(UTC).isoformat() diff --git a/tests/test_session_spawner.py b/tests/test_session_spawner.py index e304ea2..a92ff7d 100644 --- a/tests/test_session_spawner.py +++ b/tests/test_session_spawner.py @@ -639,6 +639,13 @@ async def test_spawn_result_includes_status_and_turn_count( class FakeHooks: def register(self, event, handler, priority=0, name=None): + # The real registry is keyed by event; this fake has a single + # slot, so it must ignore registrations for other events (the + # spawner also registers a provider:request transcript + # checkpoint) rather than let the last writer win. + if event != "orchestrator:complete": + return lambda: None + nonlocal captured_handler captured_handler = handler @@ -849,6 +856,13 @@ async def test_resume_result_includes_status_and_turn_count( class FakeHooks: def register(self, event, handler, priority=0, name=None): + # The real registry is keyed by event; this fake has a single + # slot, so it must ignore registrations for other events (the + # spawner also registers a provider:request transcript + # checkpoint) rather than let the last writer win. + if event != "orchestrator:complete": + return lambda: None + nonlocal captured_handler captured_handler = handler diff --git a/tests/test_timedout_session_resumable.py b/tests/test_timedout_session_resumable.py new file mode 100644 index 0000000..a4f3aa9 --- /dev/null +++ b/tests/test_timedout_session_resumable.py @@ -0,0 +1,685 @@ +"""A timed-out sub-session must still be resumable -- without adding an +unbounded await to the cancellation path. + +THE DEFECT + spawn_sub_session persisted the child transcript only AFTER a successful + ``await child_session.execute(...)``. tool-delegate's wall-clock timeout + CANCELS that await, so the save never ran and nothing about the timed-out + sub-session reached SessionStore. The session_id handed back to the caller + therefore could not be resumed: + "Session not found. May have expired or never existed." + +THE CONSTRAINT + Rescuing the transcript from the CANCELLATION path would mean awaiting + ``context.get_messages()`` while already unwinding a deadline -- an await + that can block past the very deadline that caused the unwind, recreating + the hang the timeout exists to bound. + +THE FIX UNDER TEST (option (b)) + Checkpoint the transcript DURING normal execution, at ``provider:request`` + -- the only point in the orchestrator loop where the message list is + guaranteed tool-pair-balanced. The cancellation path gains no new code, + no new await, and no new write. + +Tests below pin, in order: + 1. the contract the acceptance names -- the advertised session_id resumes + and yields the preserved transcript; + 2. the hard invariant -- nothing is awaited on the cancellation path + (proved by hanging ``get_messages()`` and showing the unwind is + unaffected AND that it was never called after cancellation began); + 3. the balanced-boundary choice (provider:request, never provider:response); + 4. that a failing checkpoint never takes down the run it protects; + 5. the throttle and the disable escape hatch. +""" + +from __future__ import annotations + +import asyncio +import time +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from amplifier_app_cli.session_store import SessionStore + +pytestmark = pytest.mark.anyio + +SUB_SESSION_ID = "parent0000000000-child00000000000_test-agent" + + +@pytest.fixture(scope="module") +def anyio_backend(): + return "asyncio" + + +# --------------------------------------------------------------------------- +# Fakes +# --------------------------------------------------------------------------- + + +class FakeHooks: + """Event-keyed hook registry (the real one is keyed; MagicMock is not).""" + + def __init__(self) -> None: + self.handlers: dict[str, list] = {} + + def register(self, event, handler, priority=0, name=None): + self.handlers.setdefault(event, []).append(handler) + + def _unregister(): + try: + self.handlers.get(event, []).remove(handler) + except ValueError: + pass + + return _unregister + + async def emit(self, event, data): + for handler in list(self.handlers.get(event, [])): + await handler(event, data) + + +class FakeContext: + """Minimal context module. + + ``hang_get_messages`` flips the behaviour of get_messages() to "never + returns" -- the instrument for proving the cancellation path does not + await it. + """ + + def __init__(self, messages: list[dict] | None = None) -> None: + self.messages: list[dict] = list(messages or []) + self.hang_get_messages = False + self.get_messages_calls = 0 + self.factory = None + + async def set_system_prompt_factory(self, factory) -> None: + self.factory = factory + + async def add_message(self, message: dict) -> None: + self.messages.append(message) + + async def get_messages(self) -> list[dict]: + self.get_messages_calls += 1 + if self.hang_get_messages: + await asyncio.Event().wait() # never returns + return list(self.messages) + + +def _make_parent_session() -> MagicMock: + parent = MagicMock() + parent.config = { + "session": {"orchestrator": "loop-basic", "context": "context-simple"}, + "agents": {}, + } + parent.session_id = "parent-123" + parent.trace_id = "trace-abc" + parent.loader = None + parent.coordinator = MagicMock() + parent.coordinator.config = {"agents": {}} + parent.coordinator.get = MagicMock(return_value=None) + parent.coordinator.get_capability = MagicMock(return_value=None) + parent.coordinator.display_system = MagicMock() + parent.coordinator.approval_system = MagicMock() + parent.coordinator.cancellation = MagicMock() + return parent + + +def _make_child_session( + context: FakeContext, hooks: FakeHooks, execute_impl +) -> MagicMock: + child = MagicMock() + child.session_id = SUB_SESSION_ID + + def _get(name): + if name == "hooks": + return hooks + if name == "context": + return context + return None + + child.coordinator = MagicMock() + child.coordinator.get = MagicMock(side_effect=_get) + child.coordinator.get_capability = MagicMock(return_value=None) + child.coordinator.register_capability = MagicMock() + child.coordinator.mount = AsyncMock() + child.coordinator.collect_contributions = AsyncMock(return_value=[]) + child.coordinator.display_system = MagicMock() + child.coordinator.approval_system = MagicMock() + child.coordinator.cancellation = MagicMock() + child.initialize = AsyncMock() + child.execute = AsyncMock(side_effect=execute_impl) + child.cleanup = AsyncMock() + return child + + +def _spawn_patches(child_session: MagicMock): + """The heavy-dependency patch stack shared by every spawn in this file.""" + return ( + patch( + "amplifier_app_cli.session_spawner.AmplifierSession", + return_value=child_session, + ), + patch( + "amplifier_app_cli.session_spawner.generate_sub_session_id", + return_value=SUB_SESSION_ID, + ), + patch( + "amplifier_app_cli.session_spawner.bridge_child_cost", + new_callable=AsyncMock, + ), + patch( + "amplifier_app_cli.session_spawner._extract_bundle_context", + return_value=None, + ), + patch("amplifier_app_cli.paths.create_foundation_resolver"), + ) + + +async def _spawn(child_session: MagicMock, parent_session: MagicMock): + from amplifier_app_cli.session_spawner import spawn_sub_session + + p1, p2, p3, p4, p5 = _spawn_patches(child_session) + with p1, p2, p3, p4, p5: + return await spawn_sub_session( + agent_name="test-agent", + instruction="Do something long", + parent_session=parent_session, + agent_configs={"test-agent": {"description": "A test agent"}}, + ) + + +async def _spawn_until_timeout( + child_session: MagicMock, parent_session: MagicMock, timeout_s: float = 0.2 +) -> float: + """Spawn under a wall-clock timeout, exactly as tool-delegate does. + + Returns elapsed seconds. Fails loudly (rather than hanging the suite) + if the unwind never completes. + """ + + async def _under_timeout(): + with pytest.raises(TimeoutError): + async with asyncio.timeout(timeout_s): + await _spawn(child_session, parent_session) + + started = time.monotonic() + task = asyncio.ensure_future(_under_timeout()) + done, pending = await asyncio.wait({task}, timeout=5.0) + elapsed = time.monotonic() - started + if task in pending: + task.cancel() + pytest.fail( + "spawn_sub_session did not unwind within 5s of a 0.2s timeout -- " + "the cancellation path awaited something unbounded" + ) + exc = task.exception() + assert exc is None, f"unexpected error unwinding the timeout: {exc!r}" + return elapsed + + +# --------------------------------------------------------------------------- +# 1. The contract the acceptance names +# --------------------------------------------------------------------------- + + +class TestTimedOutSessionIsResumable: + async def test_timed_out_spawn_leaves_a_loadable_session( + self, tmp_path, monkeypatch + ): + """The advertised session_id exists in SessionStore after a timeout.""" + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("AMPLIFIER_SPAWN_CHECKPOINT_INTERVAL_S", "0") + + hooks = FakeHooks() + context = FakeContext() + + async def execute_impl(instruction): + # iteration 1: nothing in the transcript yet + await hooks.emit("provider:request", {"iteration": 1}) + context.messages.append({"role": "user", "content": "Do something long"}) + context.messages.append({"role": "assistant", "content": "partial work"}) + # iteration 2: tool-pair-balanced boundary -> checkpoint + await hooks.emit("provider:request", {"iteration": 2}) + await asyncio.Event().wait() # straggler + + child = _make_child_session(context, hooks, execute_impl) + await _spawn_until_timeout(child, _make_parent_session()) + + store = SessionStore() + assert store.exists(SUB_SESSION_ID), ( + "a timed-out sub-session must still be present in SessionStore" + ) + + transcript, metadata = store.load(SUB_SESSION_ID) + assert [m["content"] for m in transcript] == [ + "Do something long", + "partial work", + ] + assert metadata["status"] == "in_progress", ( + "a checkpoint must be labelled in_progress, never complete" + ) + assert metadata["config"], "resume needs metadata['config'] to reconstruct" + assert metadata["agent_name"] == "test-agent" + + async def test_timed_out_session_round_trips_through_resume( + self, tmp_path, monkeypatch + ): + """The full recovery move: resume the advertised id, get the transcript. + + This is the acceptance criterion end to end -- spawn, time out, then + resume_sub_session(session_id) and observe the preserved messages + restored into the resumed session's context. + """ + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("AMPLIFIER_SPAWN_CHECKPOINT_INTERVAL_S", "0") + + hooks = FakeHooks() + context = FakeContext() + + async def execute_impl(instruction): + await hooks.emit("provider:request", {"iteration": 1}) + context.messages.append({"role": "user", "content": "Do something long"}) + context.messages.append({"role": "assistant", "content": "partial work"}) + await hooks.emit("provider:request", {"iteration": 2}) + await asyncio.Event().wait() + + child = _make_child_session(context, hooks, execute_impl) + await _spawn_until_timeout(child, _make_parent_session()) + + # --- now resume the session_id the caller was handed ----------------- + from amplifier_app_cli.session_spawner import resume_sub_session + + resumed_context = FakeContext() + resumed_hooks = FakeHooks() + resumed = _make_child_session( + resumed_context, resumed_hooks, AsyncMock(return_value="resumed response") + ) + resumed.execute = AsyncMock(return_value="resumed response") + + with ( + patch( + "amplifier_app_cli.session_spawner.AmplifierSession", + return_value=resumed, + ), + patch("amplifier_app_cli.ui.CLIApprovalSystem"), + patch("amplifier_app_cli.ui.CLIDisplaySystem"), + patch("amplifier_app_cli.paths.create_foundation_resolver"), + ): + result = await resume_sub_session(SUB_SESSION_ID, "carry on") + + assert result["session_id"] == SUB_SESSION_ID + assert result["output"] == "resumed response" + restored = [m.get("content") for m in resumed_context.messages] + assert "Do something long" in restored and "partial work" in restored, ( + "resume must restore the transcript preserved by the checkpoint" + ) + + +class TestNonResumableIsStatedExplicitly: + """The acceptance's SECOND branch, for the cases option (b) cannot cover. + + The acceptance is disjunctive -- "either the advertised session_id is + genuinely resumable ... OR the result states explicitly that it is not + resumable and directs the caller to re-delegate". Option (b) satisfies the + first branch for every checkpointed session (covered above). + + Two residual cases are NOT checkpointed and so land on the second branch: + * the subprocess spawn path (session_spawner.py:637 returns before any + checkpointing); + * checkpointing explicitly disabled via the env knob. + Plus the pre-existing cases: an expired/pruned session, or an id that never + existed. + + For all of them the app-side resume surface must say, in words, that the + session is not resumable and that the caller should re-delegate -- never + leave "retry the resume" as a plausible reading. + + SCOPE, STATED HONESTLY: this pins the boundary THIS repo owns + (``resume_sub_session``'s raised error, which reaches the + ``delegate:error`` event, the logs, and every non-foundation caller such + as recipes and programmatic callers). The MODEL-facing half is not ours: + tool-delegate's ``except FileNotFoundError`` handler builds its own + message and discards ``str(e)``, so making this text reach the ToolResult + needs a one-line foundation change, specified in the PR body. That is a + real, disclosed gap -- not something this test pretends to cover. + """ + + async def test_missing_session_says_not_resumable_and_says_re_delegate( + self, tmp_path, monkeypatch + ): + from amplifier_app_cli.session_spawner import resume_sub_session + + monkeypatch.setenv("HOME", str(tmp_path)) + + with pytest.raises(FileNotFoundError) as excinfo: + await resume_sub_session("never-existed-session-id", "carry on") + + message = str(excinfo.value) + assert "not resumable" in message.lower(), ( + "the second branch of the acceptance requires the result to state " + f"EXPLICITLY that the session is not resumable; got: {message!r}" + ) + assert "re-delegate" in message.lower(), ( + "the second branch of the acceptance requires the caller be " + f"directed to re-delegate; got: {message!r}" + ) + + async def test_disabled_checkpointing_lands_on_the_non_resumable_branch( + self, tmp_path, monkeypatch + ): + """End to end: a timeout with checkpointing off yields branch 2, in words. + + This is the disjunction resolving the other way in a real run -- spawn, + time out, and observe that the advertised session_id is genuinely + absent AND that asking to resume it says so explicitly. + """ + from amplifier_app_cli.session_spawner import resume_sub_session + + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("AMPLIFIER_SPAWN_CHECKPOINT_INTERVAL_S", "-1") + + hooks = FakeHooks() + context = FakeContext() + + async def execute_impl(instruction): + await hooks.emit("provider:request", {"iteration": 1}) + context.messages.append({"role": "assistant", "content": "partial work"}) + await asyncio.Event().wait() + + child = _make_child_session(context, hooks, execute_impl) + await _spawn_until_timeout(child, _make_parent_session()) + + assert not SessionStore().exists(SUB_SESSION_ID) + + with pytest.raises(FileNotFoundError) as excinfo: + await resume_sub_session(SUB_SESSION_ID, "carry on") + + message = str(excinfo.value).lower() + assert "not resumable" in message and "re-delegate" in message, ( + "a timed-out, non-checkpointed session must report its own " + f"non-resumability in words; got: {message!r}" + ) + + +# --------------------------------------------------------------------------- +# 2. The hard invariant +# --------------------------------------------------------------------------- + + +class TestNoUnboundedAwaitOnCancellationPath: + async def test_hanging_get_messages_does_not_delay_the_unwind( + self, tmp_path, monkeypatch + ): + """The invariant, proved two ways. + + ``context.get_messages()`` is made to hang forever from the moment + cancellation begins. If ANY code awaited it while unwinding, the + 0.2s timeout would never complete -- the harness fails loudly at 5s + instead of hanging the suite. We additionally assert the call count + did not move after cancellation started, so the test cannot pass by + accident (e.g. via a short-circuit that skipped the await for an + unrelated reason). + """ + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("AMPLIFIER_SPAWN_CHECKPOINT_INTERVAL_S", "0") + + hooks = FakeHooks() + context = FakeContext() + calls_when_cancelled: list[int] = [] + + async def execute_impl(instruction): + await hooks.emit("provider:request", {"iteration": 1}) + context.messages.append({"role": "user", "content": "work"}) + await hooks.emit("provider:request", {"iteration": 2}) + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + # The deadline has fired. From here on, any await on + # get_messages() would never return. + context.hang_get_messages = True + calls_when_cancelled.append(context.get_messages_calls) + raise + + child = _make_child_session(context, hooks, execute_impl) + elapsed = await _spawn_until_timeout(child, _make_parent_session()) + + assert calls_when_cancelled, "the child was never actually cancelled" + assert context.get_messages_calls == calls_when_cancelled[0], ( + "get_messages() was awaited on the cancellation path -- that is the " + "unbounded await this design exists to forbid" + ) + assert elapsed < 2.0, ( + f"unwinding a 0.2s timeout took {elapsed:.2f}s; the cancellation " + "path is doing work it must not do" + ) + + async def test_the_probe_would_catch_a_violating_implementation(self): + """Inverted control -- proof the probe above is not vacuous. + + The unpatched code ALSO has no await on its cancellation path, so the + probe passes before the fix as well. That makes it a regression guard + rather than a defect reproduction, and a guard is worthless unless it + can fail. This control reproduces the shape of the REJECTED option + (a) -- a best-effort save while unwinding -- and shows the probe's + instrument (a hanging ``get_messages``) does catch it. + + It also demonstrates concretely why option (a) is unsafe: once the + deadline's CancelledError has been delivered and caught, a fresh + await is NOT re-cancelled. It simply blocks -- past the very deadline + that caused the unwind. + """ + context = FakeContext() + + async def violating_unwind(): + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + context.hang_get_messages = True + await context.get_messages() # option (a): the forbidden await + raise + + async def _under_timeout(): + async with asyncio.timeout(0.2): + await violating_unwind() + + task = asyncio.ensure_future(_under_timeout()) + _done, pending = await asyncio.wait({task}, timeout=1.0) + try: + assert task in pending, ( + "the violating control completed -- the probe's instrument does " + "not actually detect an await on the cancellation path, so the " + "invariant test above proves nothing" + ) + finally: + task.cancel() + try: + await task + except BaseException: # noqa: BLE001 - teardown of a cancelled probe + pass + + async def test_cleanup_still_runs_and_the_timeout_still_propagates( + self, tmp_path, monkeypatch + ): + """Checkpointing must not swallow the timeout or skip child cleanup.""" + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("AMPLIFIER_SPAWN_CHECKPOINT_INTERVAL_S", "0") + + hooks = FakeHooks() + context = FakeContext() + + async def execute_impl(instruction): + await hooks.emit("provider:request", {"iteration": 1}) + await asyncio.Event().wait() + + child = _make_child_session(context, hooks, execute_impl) + parent = _make_parent_session() + await _spawn_until_timeout(child, parent) + + child.cleanup.assert_awaited() + parent.coordinator.cancellation.unregister_child.assert_called() + assert hooks.handlers.get("provider:request") == [], ( + "the checkpoint hook must be unregistered even on the timeout path" + ) + + +# --------------------------------------------------------------------------- +# 3. The balanced-boundary choice +# --------------------------------------------------------------------------- + + +class TestCheckpointBoundary: + async def test_checkpoint_is_wired_to_provider_request_only( + self, tmp_path, monkeypatch + ): + """provider:request is the only tool-pair-balanced point in the loop. + + Checkpointing after a response would persist an assistant message + whose tool_calls have no matching results; resuming that transcript + reproduces "No tool call found for function call output". + """ + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("AMPLIFIER_SPAWN_CHECKPOINT_INTERVAL_S", "0") + + hooks = FakeHooks() + context = FakeContext() + seen: dict[str, list[str]] = {} + + async def execute_impl(instruction): + seen["events"] = sorted(e for e, hs in hooks.handlers.items() if hs) + return "done" + + child = _make_child_session(context, hooks, execute_impl) + await _spawn(child, _make_parent_session()) + + assert "provider:request" in seen["events"] + assert "provider:response" not in seen["events"] + + +# --------------------------------------------------------------------------- +# 4. A failing checkpoint never takes down the run it protects +# --------------------------------------------------------------------------- + + +class TestCheckpointIsBestEffort: + async def test_failing_checkpoint_does_not_break_the_run( + self, tmp_path, monkeypatch + ): + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("AMPLIFIER_SPAWN_CHECKPOINT_INTERVAL_S", "0") + + hooks = FakeHooks() + context = FakeContext() + attempts: list[int] = [] + + real_save = SessionStore.save + + def exploding_save(self, session_id, transcript, metadata): + attempts.append(1) + if metadata.get("status") == "in_progress": + raise OSError("disk is on fire") + return real_save(self, session_id, transcript, metadata) + + async def execute_impl(instruction): + await hooks.emit("provider:request", {"iteration": 1}) + context.messages.append({"role": "assistant", "content": "ok"}) + return "agent response" + + child = _make_child_session(context, hooks, execute_impl) + + with patch.object(SessionStore, "save", exploding_save): + result = await _spawn(child, _make_parent_session()) + + assert result["output"] == "agent response" + assert len(attempts) >= 2, "checkpoint and final save should both be attempted" + transcript, metadata = SessionStore().load(SUB_SESSION_ID) + assert metadata["status"] == "complete" + + async def test_missing_hook_registry_still_pre_registers_the_session( + self, tmp_path, monkeypatch + ): + """No hooks -> no mid-run checkpoints, but the id must still resolve.""" + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("AMPLIFIER_SPAWN_CHECKPOINT_INTERVAL_S", "0") + + context = FakeContext() + + async def execute_impl(instruction): + await asyncio.Event().wait() + + child = _make_child_session(context, FakeHooks(), execute_impl) + child.coordinator.get = MagicMock( + side_effect=lambda name: context if name == "context" else None + ) + + await _spawn_until_timeout(child, _make_parent_session()) + assert SessionStore().exists(SUB_SESSION_ID) + + +# --------------------------------------------------------------------------- +# 5. Throttle and escape hatch +# --------------------------------------------------------------------------- + + +class TestThrottleAndEscapeHatch: + async def test_interval_throttles_mid_run_checkpoints(self, tmp_path, monkeypatch): + """A large interval collapses N provider calls to the one pre-registration.""" + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("AMPLIFIER_SPAWN_CHECKPOINT_INTERVAL_S", "3600") + + hooks = FakeHooks() + context = FakeContext() + checkpoints: list[str] = [] + + real_save = SessionStore.save + + def counting_save(self, session_id, transcript, metadata): + checkpoints.append(metadata.get("status", "?")) + return real_save(self, session_id, transcript, metadata) + + async def execute_impl(instruction): + for i in range(5): + await hooks.emit("provider:request", {"iteration": i + 1}) + await asyncio.Event().wait() + + child = _make_child_session(context, hooks, execute_impl) + with patch.object(SessionStore, "save", counting_save): + await _spawn_until_timeout(child, _make_parent_session()) + + assert checkpoints == ["in_progress"], ( + f"expected exactly one (pre-registration) checkpoint, got {checkpoints}" + ) + + async def test_negative_interval_disables_checkpointing_entirely( + self, tmp_path, monkeypatch + ): + """The documented escape hatch, and its cost, pinned in one test.""" + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("AMPLIFIER_SPAWN_CHECKPOINT_INTERVAL_S", "-1") + + hooks = FakeHooks() + context = FakeContext() + + async def execute_impl(instruction): + await hooks.emit("provider:request", {"iteration": 1}) + await asyncio.Event().wait() + + child = _make_child_session(context, hooks, execute_impl) + await _spawn_until_timeout(child, _make_parent_session()) + + assert not SessionStore().exists(SUB_SESSION_ID), ( + "with checkpointing disabled the pre-fix behaviour is restored: " + "a timed-out sub-session leaves no store record at all" + ) + + async def test_invalid_interval_falls_back_to_the_default(self, monkeypatch): + from amplifier_app_cli.session_spawner import ( + _DEFAULT_CHECKPOINT_INTERVAL_S, + _checkpoint_interval_s, + ) + + monkeypatch.setenv("AMPLIFIER_SPAWN_CHECKPOINT_INTERVAL_S", "not-a-number") + assert _checkpoint_interval_s() == _DEFAULT_CHECKPOINT_INTERVAL_S + + monkeypatch.delenv("AMPLIFIER_SPAWN_CHECKPOINT_INTERVAL_S") + assert _checkpoint_interval_s() == _DEFAULT_CHECKPOINT_INTERVAL_S