From 8e2d80dbd88f5c9bfe02566e0bf59d3a702cee0e Mon Sep 17 00:00:00 2001 From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> Date: Wed, 2 Sep 2026 13:40:09 -0700 Subject: [PATCH 1/4] fix(spawn): checkpoint sub-session transcripts during the run so a timed-out delegate is resumable store.save() ran 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 could not be resumed ("Session not found. May have expired or never existed."). The resume path had the identical defect. 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. store.save() is also synchronous, so an asyncio.wait_for around it could not interrupt the disk write at all. So the cancellation path is left entirely untouched: no new code, no new await, no new write. Instead the transcript is checkpointed during NORMAL execution: - once before execute() (pre-registration), so the advertised session_id resolves even if the timeout fires during the first LLM call; - then on provider:request, throttled by AMPLIFIER_SPAWN_CHECKPOINT_INTERVAL_S (default 30s, chosen not measured; negative disables). provider:request is the only point in the orchestrator loop where the message list is guaranteed tool-pair-balanced -- checkpointing after a response would persist an assistant message with unmatched tool_calls, and resuming that reproduces "No tool call found for function call output". Checkpoints are labelled status="in_progress"; only the post-run save writes status="complete", so a caller can tell a rescued checkpoint from a finished session. Tests (11 new) pin: the end-to-end resume round trip; that nothing is awaited on the cancellation path (a hanging get_messages does not delay the unwind, plus an inverted control proving that probe can fail); the provider:request boundary choice; best-effort behaviour when a checkpoint or the hook registry fails; and the throttle and disable escape hatch. tests/test_session_spawner.py: FakeHooks modelled the hook registry as a single handler slot ignoring the event name; the real registry is event-keyed. Made it honour the event so a second registration no longer hides the first. Refs: model_performance-3yc (follow-up to model_performance-37n, DESIGN.md 3) --- .../DONE-NOTE.md | 206 ++++++ amplifier_app_cli/session_spawner.py | 278 +++++++- tests/test_session_spawner.py | 14 + tests/test_timedout_session_resumable.py | 599 ++++++++++++++++++ 4 files changed, 1064 insertions(+), 33 deletions(-) create mode 100644 ai_working/3yc-timedout-session-resumable/DONE-NOTE.md create mode 100644 tests/test_timedout_session_resumable.py 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 00000000..8c54d9e2 --- /dev/null +++ b/ai_working/3yc-timedout-session-resumable/DONE-NOTE.md @@ -0,0 +1,206 @@ +# 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**. + +### 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 + +`TestTimedOutSessionIsResumable` — two tests: + +* `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. This is the acceptance criterion's first branch + ("the advertised session_id genuinely resumes and returns the preserved transcript"), executed. + +--- + +## 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`) | **1571 passed, 1 skipped, 13 deselected, 1 xfailed** | +| full suite, patched (default random order) | **1571 passed, 1 skipped, 13 deselected, 1 xfailed** | +| new test file alone | **11 passed** | +| new test file against the UNPATCHED spawner (falsifier check) | **8 failed, 2 passed** | +| `ruff check` / `ruff format --check` on every file this PR touches | clean | + +**The falsifier check matters, and its two passes are disclosed, not hidden.** Reverting only +`session_spawner.py` and re-running the new file gives 8 failures. The two that still pass are +`test_hanging_get_messages_does_not_delay_the_unwind` (correct — the unpatched cancellation path is +also await-free; this is the regression guard discussed in §3.2) and +`test_negative_interval_disables_checkpointing_entirely` (correct — it pins that the escape hatch +reproduces pre-fix behaviour). Neither is a defect reproduction and neither is claimed as one. + +**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 NOT covered.** `session_spawner.py:637` returns before any + checkpointing, so a subprocess-mode delegate that times out is still unresumable. Not in scope + here; this is where option (c)'s explicit "not resumable" labelling is still the right answer. +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 | +| `tests/test_timedout_session_resumable.py` | 11 new tests — the 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 | diff --git a/amplifier_app_cli/session_spawner.py b/amplifier_app_cli/session_spawner.py index 2200b35a..8459020e 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") @@ -1428,15 +1625,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 e304ea2d..a92ff7d1 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 00000000..681abd3f --- /dev/null +++ b/tests/test_timedout_session_resumable.py @@ -0,0 +1,599 @@ +"""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" + ) + + +# --------------------------------------------------------------------------- +# 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 From 36a8ebd7c71615707748a04ae3d3d2ec53c2141c Mon Sep 17 00:00:00 2001 From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> Date: Wed, 2 Sep 2026 13:46:42 -0700 Subject: [PATCH 2/4] =?UTF-8?q?test(spawn):=20cover=20the=20acceptance's?= =?UTF-8?q?=20SECOND=20branch=20=E2=80=94=20non-resumability=20stated=20ex?= =?UTF-8?q?plicitly?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The acceptance contract is disjunctive: either the advertised session_id genuinely resumes, 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: the subprocess spawn path (session_spawner.py:637 returns before any checkpointing) and checkpointing disabled via the env knob — plus the pre-existing expired / never-existed cases. resume_sub_session's not-found error now says so in words and names the correct move, instead of leaving "retry the resume" as a plausible reading: "... This session is NOT resumable -- re-delegate to start a fresh session instead of retrying the resume." Two new tests pin it, including an end-to-end run where the disjunction resolves the OTHER way: spawn -> time out with checkpointing off -> the id is genuinely absent AND asking to resume it says so explicitly. SCOPE, verified at file:line rather than assumed: tool-delegate's `except FileNotFoundError` handler (foundation modules/tool-delegate/.../__init__.py:2148-2166) uses str(e) only in the delegate:error event and returns a HARDCODED result message that discards it. So this text reaches the event, the logs, and every non-foundation caller, but not the model-facing ToolResult. Closing that half is a one-line foundation change (error={"message": str(e)}), specified in the PR body. Deliberately not worked around by raising a different exception type to hit foundation's generic handler — that trades a correctly-typed error branch for a string. 13 tests in the file (was 11); suite 1573 green (was 1571). Refs: model_performance-3yc --- .../DONE-NOTE.md | 93 +++++++++++++++---- amplifier_app_cli/session_spawner.py | 15 ++- tests/test_timedout_session_resumable.py | 86 +++++++++++++++++ 3 files changed, 174 insertions(+), 20 deletions(-) diff --git a/ai_working/3yc-timedout-session-resumable/DONE-NOTE.md b/ai_working/3yc-timedout-session-resumable/DONE-NOTE.md index 8c54d9e2..c9eea941 100644 --- a/ai_working/3yc-timedout-session-resumable/DONE-NOTE.md +++ b/ai_working/3yc-timedout-session-resumable/DONE-NOTE.md @@ -130,17 +130,64 @@ Proved by `TestNoUnboundedAwaitOnCancellationPath` — two assertions plus a con `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 +### The contract the acceptance names — BOTH branches are tested -`TestTimedOutSessionIsResumable` — two tests: +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. This is the acceptance criterion's first branch - ("the advertised session_id genuinely resumes and returns the preserved transcript"), executed. + 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. + +**SCOPE OF BRANCH 2, STATED HONESTLY — one half is not ours, and this was verified at file:line, not +assumed.** `tool-delegate`'s handler +(`modules/tool-delegate/amplifier_module_tool_delegate/__init__.py:2148-2166`, foundation cache +`c909465861f9d6ce`) catches `FileNotFoundError`, uses `str(e)` **only** in the `delegate:error` +event, and returns a **hardcoded** result message that discards it: + +```python +return ToolResult( + success=False, + error={"message": f"Agent session '{session_id}' not found. May have expired or never existed."}, +) +``` + +So the app-side text reaches the `delegate:error` event, the logs, and every non-foundation caller +(recipes, programmatic callers) — but **not** the model-facing `ToolResult`. Closing that last half +is a one-line foundation change, specified here so it is trivially landable: + +```python +error={"message": str(e)}, # instead of the hardcoded sentence +``` + +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` (`:2146`, 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. --- @@ -149,18 +196,23 @@ Proved by `TestNoUnboundedAwaitOnCancellationPath` — two assertions plus a con | check | result | |---|---| | full suite, baseline at `963d793` before any change | **1560 passed, 1 skipped, 13 deselected, 1 xfailed** | -| full suite, patched (`-p no:randomly`) | **1571 passed, 1 skipped, 13 deselected, 1 xfailed** | -| full suite, patched (default random order) | **1571 passed, 1 skipped, 13 deselected, 1 xfailed** | -| new test file alone | **11 passed** | -| new test file against the UNPATCHED spawner (falsifier check) | **8 failed, 2 passed** | +| 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 two passes are disclosed, not hidden.** Reverting only -`session_spawner.py` and re-running the new file gives 8 failures. The two that still pass are -`test_hanging_get_messages_does_not_delay_the_unwind` (correct — the unpatched cancellation path is -also await-free; this is the regression guard discussed in §3.2) and -`test_negative_interval_disables_checkpointing_entirely` (correct — it pins that the escape hatch -reproduces pre-fix behaviour). Neither is a defect reproduction and neither is claimed as one. +**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, @@ -179,9 +231,12 @@ event name. The test's intent and assertions are untouched. 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 NOT covered.** `session_spawner.py:637` returns before any - checkpointing, so a subprocess-mode delegate that times out is still unresumable. Not in scope - here; this is where option (c)'s explicit "not resumable" labelling is still the right answer. +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. @@ -200,7 +255,7 @@ event name. The test's intent and assertions are untouched. | 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 | -| `tests/test_timedout_session_resumable.py` | 11 new tests — the contract, the invariant, the inverted control, the boundary choice, best-effort behaviour, the throttle and the escape hatch | +| `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 | diff --git a/amplifier_app_cli/session_spawner.py b/amplifier_app_cli/session_spawner.py index 8459020e..c7507fc8 100644 --- a/amplifier_app_cli/session_spawner.py +++ b/amplifier_app_cli/session_spawner.py @@ -1148,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: diff --git a/tests/test_timedout_session_resumable.py b/tests/test_timedout_session_resumable.py index 681abd3f..a4f3aa9f 100644 --- a/tests/test_timedout_session_resumable.py +++ b/tests/test_timedout_session_resumable.py @@ -316,6 +316,92 @@ async def execute_impl(instruction): ) +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 # --------------------------------------------------------------------------- From 453d89567de4200a1331993ecf807bd48f300444 Mon Sep 17 00:00:00 2001 From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> Date: Wed, 2 Sep 2026 13:50:44 -0700 Subject: [PATCH 3/4] fix(delegate): ship the foundation half of branch 2 as a verified patch + cross-repo round trip The acceptance's second branch requires the result to state explicitly that a session is NOT resumable and direct the caller to re-delegate. The previous commit made amplifier-app-cli raise exactly that -- but verification at file:line showed tool-delegate's `except FileNotFoundError` handler (foundation modules/tool-delegate/.../__init__.py:1314-1332 @ cc7e23a) uses str(e) only in the delegate:error event and returns a HARDCODED message that discards it. So the wording reached the event, the logs and non-foundation callers, but NOT the ToolResult the model reads. Branch 2 was therefore incomplete in the system the model actually sees. Adds: PATCH-foundation-surface-resume-detail.diff "message": str(e) or # detail-preserving, original retained as the empty-detail fallback git apply --check @ cc7e23a: exit 0 tool-delegate suite: 48 passed unpatched, 48 passed patched (scratch copy; the foundation repo was never modified -- verified clean after) test_resume_message_roundtrip.py Wires the REAL app-side resume_sub_session into the REAL tool-delegate resume path and reads the resulting ToolResult. PATCHED: 2 passed UNPATCHED: 1 failed, 1 passed -- the failure prints the exact string the model would otherwise see. That asymmetry is the result; it is what makes the diff necessary rather than cosmetic. Both live in ai_working/, NOT tests/, because the round trip depends on an unlanded foundation 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; land the foundation diff to complete the model-facing half. Suite unchanged at 1573 green (ai_working is not collected; testpaths=["tests"]). Refs: model_performance-3yc --- .../DONE-NOTE.md | 51 ++++--- ...ATCH-foundation-surface-resume-detail.diff | 21 +++ .../test_resume_message_roundtrip.py | 126 ++++++++++++++++++ 3 files changed, 182 insertions(+), 16 deletions(-) create mode 100644 ai_working/3yc-timedout-session-resumable/PATCH-foundation-surface-resume-detail.diff create mode 100644 ai_working/3yc-timedout-session-resumable/test_resume_message_roundtrip.py diff --git a/ai_working/3yc-timedout-session-resumable/DONE-NOTE.md b/ai_working/3yc-timedout-session-resumable/DONE-NOTE.md index c9eea941..ca244224 100644 --- a/ai_working/3yc-timedout-session-resumable/DONE-NOTE.md +++ b/ai_working/3yc-timedout-session-resumable/DONE-NOTE.md @@ -163,29 +163,46 @@ resume" as a plausible reading: **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. -**SCOPE OF BRANCH 2, STATED HONESTLY — one half is not ours, and this was verified at file:line, not -assumed.** `tool-delegate`'s handler -(`modules/tool-delegate/amplifier_module_tool_delegate/__init__.py:2148-2166`, foundation cache -`c909465861f9d6ce`) catches `FileNotFoundError`, uses `str(e)` **only** in the `delegate:error` -event, and returns a **hardcoded** result message that discards it: +**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 -return ToolResult( - success=False, - error={"message": f"Agent session '{session_id}' not found. May have expired or never existed."}, -) +"message": str(e) +or f"Agent session '{session_id}' not found. May have expired or never existed." ``` -So the app-side text reaches the `delegate:error` event, the logs, and every non-foundation caller -(recipes, programmatic callers) — but **not** the model-facing `ToolResult`. Closing that last half -is a one-line foundation change, specified here so it is trivially landable: +Detail-preserving, with the original sentence retained as the empty-detail fallback. -```python -error={"message": str(e)}, # instead of the hardcoded sentence -``` +| 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` (`:2146`, which *does* pass +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. @@ -259,3 +276,5 @@ event name. The test's intent and assertions are untouched. | `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 00000000..6003a3f2 --- /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 00000000..a803f200 --- /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}" + ) From 03a9bf3e672e80338f89e60aca355d0b3a244796 Mon Sep 17 00:00:00 2001 From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> Date: Wed, 2 Sep 2026 13:52:43 -0700 Subject: [PATCH 4/4] docs(done-note): state explicitly how option (b) satisfies the unbounded-await invariant The PR body argued (a) and (c) down but left (b)'s own justification against the hard invariant implicit. Names the specific danger (an await reached while unwinding a fired deadline can block past that deadline), and why (b) discharges it by construction rather than by a bound that has to hold. Refs: model_performance-3yc --- .../3yc-timedout-session-resumable/DONE-NOTE.md | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/ai_working/3yc-timedout-session-resumable/DONE-NOTE.md b/ai_working/3yc-timedout-session-resumable/DONE-NOTE.md index ca244224..bf2addaa 100644 --- a/ai_working/3yc-timedout-session-resumable/DONE-NOTE.md +++ b/ai_working/3yc-timedout-session-resumable/DONE-NOTE.md @@ -34,6 +34,18 @@ await** — `context.get_messages()` — and (b) everything else it needs is ava **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.**