From 784e59974e31fbf67e7ec7f940b4a35db150581a Mon Sep 17 00:00:00 2001 From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> Date: Thu, 3 Sep 2026 01:18:47 -0700 Subject: [PATCH 1/2] feat(spawner): expose a timed-out sub-session's partial output via session.partial PRODUCER half of the per-delegate timeout partial-result path. The CONSUMER shipped in amplifier-foundation f42f48c (PR #353): on a wall-clock timeout the delegate now RETURNS rather than raises, and reads an optional app-layer `session.partial` capability. Nothing offered one, so `partial_available` was false for every real timeout. This is what offers one. A sub-session cancelled by its per-delegate timeout previously had everything it had produced discarded: `child_session.execute()` is cancelled, the post-run block never runs, the child is cleaned up. This accumulates the agent's own assistant text (content_block:end) and publishes it under the sub_session_id, so the delegate hands the caller INCOMPLETE-with-partial instead of an empty failure. The record is published EAGERLY and updated in place, not sealed from the child's unwind. That is not a stylistic choice. tool-delegate's `_await_child_with_deadline` deliberately does not wait for a slow unwind: it cancels the child, DETACHES it, and reads `session.partial` immediately. A partial published from the child's `except BaseException:` handler arrives after the consumer has already reported `partial_available: false` -- measured cross-repo, with both halves' unit tests passing. `test_partial_is_readable_before_the_child_has_unwound` pins the ordering in-repo, without needing foundation on the path. Normal completions are untouched: the returned dict gains no key and is byte-identical (identical sha256 of the serialized result, parent vs patched), and the registry is cleared on the success path. Absent/None/raising degrades to `partial_available: false`, never to an error -- raising out of the timeout path would discard the completed siblings that path exists to protect. Evidence, round-trip check and probe under docs/lanes/9w0-delegate-timeout-partial-producer/. --- amplifier_app_cli/session_runner.py | 5 + amplifier_app_cli/session_spawner.py | 202 +++++++++- .../DONE-NOTE.md | 178 +++++++++ .../conftest.py | 12 + .../00-parent-baseline-named-suites.txt | 11 + .../01-parent-baseline-full-suite.txt | 3 + .../02-original-patch-does-not-apply.txt | 4 + .../03-failbefore-roundtrip-parent.txt | 45 +++ ...ounterevidence-naive-shape-still-false.txt | 13 + .../05-normal-completion-byte-identical.txt | 21 ++ .../06-gd4-partial-available-true.txt | 55 +++ .../07-passafter-roundtrip-patched.txt | 13 + .../08-lint-and-full-suite-patched.txt | 14 + .../probe_partial_roundtrip.py | 186 ++++++++++ .../test_partial_roundtrip.py | 190 ++++++++++ tests/test_session_spawner_partial.py | 346 ++++++++++++++++++ 16 files changed, 1292 insertions(+), 6 deletions(-) create mode 100644 docs/lanes/9w0-delegate-timeout-partial-producer/DONE-NOTE.md create mode 100644 docs/lanes/9w0-delegate-timeout-partial-producer/conftest.py create mode 100644 docs/lanes/9w0-delegate-timeout-partial-producer/evidence/00-parent-baseline-named-suites.txt create mode 100644 docs/lanes/9w0-delegate-timeout-partial-producer/evidence/01-parent-baseline-full-suite.txt create mode 100644 docs/lanes/9w0-delegate-timeout-partial-producer/evidence/02-original-patch-does-not-apply.txt create mode 100644 docs/lanes/9w0-delegate-timeout-partial-producer/evidence/03-failbefore-roundtrip-parent.txt create mode 100644 docs/lanes/9w0-delegate-timeout-partial-producer/evidence/04-counterevidence-naive-shape-still-false.txt create mode 100644 docs/lanes/9w0-delegate-timeout-partial-producer/evidence/05-normal-completion-byte-identical.txt create mode 100644 docs/lanes/9w0-delegate-timeout-partial-producer/evidence/06-gd4-partial-available-true.txt create mode 100644 docs/lanes/9w0-delegate-timeout-partial-producer/evidence/07-passafter-roundtrip-patched.txt create mode 100644 docs/lanes/9w0-delegate-timeout-partial-producer/evidence/08-lint-and-full-suite-patched.txt create mode 100644 docs/lanes/9w0-delegate-timeout-partial-producer/probe_partial_roundtrip.py create mode 100644 docs/lanes/9w0-delegate-timeout-partial-producer/test_partial_roundtrip.py create mode 100644 tests/test_session_spawner_partial.py diff --git a/amplifier_app_cli/session_runner.py b/amplifier_app_cli/session_runner.py index ebaabdb..b750ab3 100644 --- a/amplifier_app_cli/session_runner.py +++ b/amplifier_app_cli/session_runner.py @@ -512,10 +512,14 @@ def register_session_spawning(session: AmplifierSession) -> None: The capabilities registered: - session.spawn: Create new agent sub-session - session.resume: Resume existing sub-session + - session.partial: Recover a sub-session's preserved partial output after it + was cancelled or timed out (tool-delegate reads this to return an + incomplete-with-partial result instead of discarding the work) Args: session: The AmplifierSession to register capabilities on """ + from .session_spawner import get_partial_output from .session_spawner import resume_sub_session from .session_spawner import spawn_sub_session @@ -569,6 +573,7 @@ async def resume_capability( session.coordinator.register_capability("session.spawn", spawn_capability) session.coordinator.register_capability("session.resume", resume_capability) + session.coordinator.register_capability("session.partial", get_partial_output) # ============================================================================= diff --git a/amplifier_app_cli/session_spawner.py b/amplifier_app_cli/session_spawner.py index 1585e73..6cbe456 100644 --- a/amplifier_app_cli/session_spawner.py +++ b/amplifier_app_cli/session_spawner.py @@ -21,6 +21,159 @@ logger = logging.getLogger(__name__) +# ============================================================================= +# Partial-output preservation for delegates that never finish +# ============================================================================= +# When tool-delegate's wall-clock timeout fires it cancels the spawn coroutine. +# `child_session.execute()` raises CancelledError, the post-run block below is +# skipped, and everything the agent produced is discarded -- turning a hang into +# data loss. This registry keeps the agent's own assistant text so the delegate +# tool can hand the caller an INCOMPLETE-with-partial result instead of an empty +# failure. +# +# This is the PRODUCER half of the contract whose CONSUMER shipped in +# amplifier-foundation f42f48c (tool-delegate). That side reads an optional +# `session.partial` capability: +# +# (sub_session_id: str) -> {"text": str, "segments": int, "source": str} | None +# +# and degrades to `partial_available: false` when it is absent, returns None, or +# raises. Nothing here may raise into the timeout path. +# +# NOTE: the transcript is separately checkpointed mid-run (see +# _install_transcript_checkpoint below) so a timed-out sub-session stays +# RESUMABLE. That is a different property from this one: the checkpoint makes +# the work reachable by a later resume, this registry makes it readable by the +# delegate call that timed out, at the moment it gives up. +# +# WHY THE RECORD IS PUBLISHED EAGERLY (and not only on the cancellation path) +# +# The obvious shape -- accumulate privately, publish from an `except +# BaseException:` around execute() -- DOES NOT WORK against the consumer that +# actually shipped, and fails silently rather than loudly. tool-delegate's +# `_await_child_with_deadline` deliberately does NOT wait for a child that is +# slow to unwind: at the deadline it calls `child_task.cancel()`, DETACHES the +# task, and raises immediately. Its `except _DelegateTimeoutExpired:` handler +# then calls `session.partial` straight away -- while the cancelled child task +# has not yet been scheduled to run its own exception handlers. A partial +# published from the child's unwind is therefore published AFTER the consumer +# has already read and reported `partial_available: false`. +# +# Measured, cross-repo, against foundation f42f48c: the app logged "preserved 2 +# partial text segment(s), 42 chars" AFTER the delegate had logged "No partial +# output could be recovered". Same run, wrong order, and both halves' own unit +# tests passed. This is exactly the drift the round-trip check exists to catch. +# +# So the record is entered in the registry when the accumulator is installed and +# is updated in place as text arrives -- readable at ANY instant, with no +# dependence on cancellation ordering -- and is REMOVED when the sub-session +# completes normally. `get_partial_output` reads a snapshot; nothing here awaits. + +_PARTIAL_OUTPUTS: dict[str, dict] = {} +_PARTIAL_MAX_SESSIONS = 64 + + +def get_partial_output(sub_session_id: str) -> dict | None: + """``session.partial`` capability: what a sub-session produced before it died. + + Returns ``{"text", "segments", "source"}`` for a sub-session that was + cancelled or timed out mid-flight, or ``None`` when nothing was preserved. + ``segments`` counts preserved assistant text segments, not turns. + + Reads are destructive -- one delegate call consumes one record -- so the + registry cannot grow without bound on a long-lived root session. A record + with no text yet reads as ``None``: "produced nothing" and "produced + nothing recoverable" are the same answer to the consumer. + """ + record = _PARTIAL_OUTPUTS.pop(sub_session_id, None) + if not record: + return None + chunks = list(record.get("chunks") or ()) + if not chunks: + return None + return { + "text": "".join(chunks), + "segments": len(chunks), + "source": "spawn-accumulator", + } + + +def _publish_partial(sub_session_id: str, record: dict) -> None: + """Enter an in-flight accumulator in the registry, evicting oldest-first.""" + while len(_PARTIAL_OUTPUTS) >= _PARTIAL_MAX_SESSIONS: + oldest = next(iter(_PARTIAL_OUTPUTS)) + _PARTIAL_OUTPUTS.pop(oldest, None) + logger.warning( + "Partial-output registry is at its %d-session cap; evicted %s to " + "make room for %s", + _PARTIAL_MAX_SESSIONS, + oldest, + sub_session_id, + ) + _PARTIAL_OUTPUTS[sub_session_id] = record + + +def _discard_partial(sub_session_id: str) -> None: + """Drop a record for a sub-session that completed normally.""" + _PARTIAL_OUTPUTS.pop(sub_session_id, None) + + +def _seal_partial(sub_session_id: str, record: dict) -> None: + """Confirm an accumulator is readable after the sub-session failed to finish. + + The record is normally already published (see the note above); this + re-publishes it if it was evicted under the cap, and logs what survived. + Synchronous by design: awaiting anything while unwinding a timeout risks + blocking past the very deadline that caused the unwind. + """ + if not record.get("chunks"): + return + if sub_session_id not in _PARTIAL_OUTPUTS: + _publish_partial(sub_session_id, record) + logger.warning( + "Sub-session %s did not complete; preserved %d partial text segment(s), %d chars", + sub_session_id, + len(record["chunks"]), + sum(len(c) for c in record["chunks"]), + ) + + +def _open_partial(sub_session_id: str, hooks): + """Start accumulating assistant text, published from the first moment. + + Returns ``(record, unregister)``. ``unregister`` is ``None`` when there is + no hooks coordinator to register against -- in that case nothing can ever + be accumulated, so nothing is published either and the consumer correctly + degrades to ``partial_available: false``. + + The hook is registered at low priority so it observes blocks after the UI + has rendered them and never influences rendering. + """ + record: dict = {"chunks": []} + if not hooks: + return record, None + + from amplifier_core.events import CONTENT_BLOCK_END + from amplifier_core.hooks import HookResult + + async def _accumulate_partial(event: str, data: dict) -> HookResult: + block = data.get("block") + if isinstance(block, dict) and block.get("type") == "text": + text = block.get("text") or "" + if text: + record["chunks"].append(text) + return HookResult() + + unregister = hooks.register( + CONTENT_BLOCK_END, + _accumulate_partial, + priority=999, + name="_spawn_partial", + ) + _publish_partial(sub_session_id, record) + return record, unregister + + # Capture default sys.path entries at import time. # Used to filter out bundle-added paths when forwarding sys_paths to subprocess children. _DEFAULT_SYS_PATHS: frozenset[str] = frozenset(sys.path) @@ -932,6 +1085,8 @@ async def child_resume_capability( child_session.coordinator.register_capability( "session.resume", child_resume_capability ) + # Partial-output recovery for grandchildren that time out under this child. + child_session.coordinator.register_capability("session.partial", get_partial_output) # Approval provider (for hooks-approval module, if active) register_provider_fn = child_session.coordinator.get_capability( @@ -1006,6 +1161,10 @@ async def _capture_completion(event: str, data: dict) -> HookResult: name="_spawn_capture", ) + # Accumulate assistant text as it is produced, so a delegate killed by + # tool-delegate's wall-clock timeout still has recoverable output. + partial_record, unregister_partial = _open_partial(sub_session_id, hooks) + # Expand @-mentions in delegation instruction before executing. # Content lands inline as XML blocks prepended to the instruction. if instruction: @@ -1074,16 +1233,31 @@ async def _capture_completion(event: str, data: dict) -> HookResult: # 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. + # NOTE: the `except BaseException` below is SYNCHRONOUS ONLY and must stay + # that way. A timeout cancels execute(), and the cancellation path must + # remain free of any await -- the transcript has already been checkpointed + # above and by the provider:request hook, so nothing needs to be *fetched* + # while unwinding. The handler only confirms and logs an already-published + # accumulator; it is NOT what makes the partial readable (see the note at + # the top of this module -- the consumer often reads before this runs). try: try: response = await child_session.execute(instruction) + except BaseException: + # Timed out or cancelled: the post-run block below never runs, so + # the agent's own partial text stays published for the delegate + # tool to read. Synchronous only -- see _seal_partial. + _seal_partial(sub_session_id, partial_record) + raise + else: + # Completed normally: there is no partial to offer, and the + # registry must not carry one. + _discard_partial(sub_session_id) finally: if unregister_hook: unregister_hook() + if unregister_partial: + unregister_partial() unregister_checkpoint() # Persist final state for multi-turn resumption @@ -1743,6 +1917,8 @@ async def child_resume_capability( child_session.coordinator.register_capability( "session.resume", child_resume_capability ) + # Partial-output recovery for grandchildren that time out under this child. + child_session.coordinator.register_capability("session.partial", get_partial_output) # Approval provider (for hooks-approval module, if active) register_provider_fn = child_session.coordinator.get_capability( @@ -1885,6 +2061,10 @@ async def _capture_completion(event: str, data: dict) -> HookResult: name="_spawn_capture", ) + # Accumulate assistant text as it is produced, so a delegate killed by + # tool-delegate's wall-clock timeout still has recoverable output. + partial_record, unregister_partial = _open_partial(sub_session_id, hooks) + # Wire up cancellation propagation if parent session provided # Enables graceful Ctrl+C to stop the child after its current tool call if parent_session is not None: @@ -1929,14 +2109,24 @@ async def _capture_completion(event: str, data: dict) -> HookResult: # 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. + # NOTE: the `except BaseException` below is synchronous only -- see the + # spawn path's note. The cancellation path must stay free of any await. try: try: response = await child_session.execute(instruction) + except BaseException: + # Timed out or cancelled: the agent's own partial text stays + # published for the delegate tool to read. + # Synchronous only -- see _seal_partial. + _seal_partial(sub_session_id, partial_record) + raise + else: + _discard_partial(sub_session_id) finally: if unregister_hook: unregister_hook() + if unregister_partial: + unregister_partial() unregister_checkpoint() # Update state for next resumption diff --git a/docs/lanes/9w0-delegate-timeout-partial-producer/DONE-NOTE.md b/docs/lanes/9w0-delegate-timeout-partial-producer/DONE-NOTE.md new file mode 100644 index 0000000..3106694 --- /dev/null +++ b/docs/lanes/9w0-delegate-timeout-partial-producer/DONE-NOTE.md @@ -0,0 +1,178 @@ +# DONE-NOTE — lane 9w0 · `model_performance-9w0` + +**W3-PREREQ (2 of 2, PRODUCER): app-cli exposes a timed-out sub-session's partial output** + +| | | +|---|---| +| Repo | `microsoft/amplifier-app-cli` | +| Branch | `lane/9w0-delegate-timeout-partial-producer` | +| Parent commit | `ab47608fc989e13f9674c3f5f9efc5625a9d7673` | +| Consumer half it pairs with | amplifier-foundation `f42f48c` (`model_performance-bp0`, PR #353) — **merged** | +| Spend authority | **$0** (pure code change, no API/DTU spend authorized) | +| **Spend incurred** | **$0.00.** No API calls, no DTU, no infrastructure created, nothing to tear down. Local `pytest` + local reads only. The authority was sized for a code lane and it closed: every deliverable below landed inside it. | +| Outcome | **A — RESOLVED.** All deliverables DONE; none NOT-POSSIBLE. | + +--- + +## 1. The finding that mattered: the designed patch does not work against the shipped consumer + +37n's `PATCH-app-cli-session-spawner.diff` publishes the partial from an +`except BaseException:` handler around `child_session.execute()` — i.e. from the +**child's own unwind**. Applied faithfully against the **real merged consumer**, +that shape still reports `partial_available: false`, and it does so **silently**: +both halves' unit tests pass. + +**Root cause (measured, not inferred).** `tool-delegate`'s +`_await_child_with_deadline` (foundation `f42f48c`, `__init__.py:707-736`) +deliberately does **not** wait for a child that is slow to unwind. At the +deadline it calls `child_task.cancel()`, **detaches** the task, and raises +`_DelegateTimeoutExpired` immediately. The `except _DelegateTimeoutExpired:` +handler then calls `session.partial` straight away — before the cancelled child +task has been scheduled to run any exception handler of its own. + +Observed ordering in a single run (evidence `04`): + +``` +WARNING amplifier_module_tool_delegate Agent 'explorer' timed out after 1s … No partial output could be recovered. +WARNING amplifier_app_cli.session_spawner Sub-session …_explorer did not complete; preserved 2 partial text segment(s), 42 chars +``` + +The producer preserved the work — one line **after** the consumer had already +given up on it and reported `false`. + +37n was not wrong at the time: DESIGN.md was written against `asyncio.timeout` +semantics (`:1092`, "`async with asyncio.timeout(self.timeout)`"), which *does* +wait for the child to unwind. The consumer that actually shipped replaced that +with cancel-and-detach, and no unit test on either side could see the change. + +**The fix (this lane's only design deviation).** The accumulator record is +entered in the registry when it is installed and updated **in place** as text +arrives — readable at *any* instant, with no dependence on cancellation ordering +— and is removed when the sub-session completes normally. Nothing else about +37n's design changed: same capability name, same payload shape, same destructive +read, same 64-session cap, same synchronous-only cancellation path. + +This is exactly the drift the cross-repo round trip exists to catch, and it was +invisible to every test either repo owns on its own. + +--- + +## 2. Deliverables + +| # | Deliverable | State | Evidence | +|---|---|---|---| +| 1 | Partial retrievable through `get_partial_output` after a per-delegate timeout | **DONE** | `evidence/03` (fail-before), `evidence/07` (pass-after), `tests/test_session_spawner_partial.py` (12 tests) | +| 2 | **Cross-repo contract**: real app reader ↔ real foundation consumer, overlaid copies, against foundation `f42f48c` | **DONE** | `evidence/07` — 2 passed | +| 3 | `partial_available: true` actually reached (k64 gate **G-D4**) | **DONE** | `evidence/06` — side-by-side parent (`false`) vs patched (`true`, 42 chars, 2 segments, `source: spawn-accumulator`) | +| 4 | Normal completions unchanged — **byte-identical**, shown not asserted | **DONE** | `evidence/05` — `diff` empty, identical sha256 `ef8c86fd…` | +| 5 | The two `TestSpawnEnrichment` failures shown on the parent | **DONE, with a correction — see §4** | `evidence/00` | +| 6 | Fail-before evidence committed and pasted in the PR body | **DONE** | `evidence/03`, `evidence/04` | +| 7 | Draft PR on origin naming foundation `f42f48c` as the consumer half | **DONE** | see `publication` in `DONE.json` | +| 8 | This DONE-NOTE under the lane artifact root | **DONE** | this file | + +Nothing was dropped, and no deliverable was cap-bound. + +--- + +## 3. What changed + +`amplifier_app_cli/session_spawner.py` + +* `_PARTIAL_OUTPUTS` registry (cap 64, oldest-first eviction, eviction logged). +* `get_partial_output(sub_session_id)` — the `session.partial` capability. + Destructive read; a record with no text reads as `None`. +* `_open_partial` / `_publish_partial` / `_discard_partial` / `_seal_partial`. +* `content_block:end` accumulator registered at priority 999 on both the spawn + and the resume path; `session.partial` registered on every child coordinator + so a *grandchild* that times out is recoverable too. +* `except BaseException:` → `_seal_partial` (confirm + log; synchronous only), + `else:` → `_discard_partial` on normal completion. + +`amplifier_app_cli/session_runner.py` + +* `session.partial` registered on the root session in `register_session_spawning`. + +`tests/test_session_spawner_partial.py` — 12 tests, including +`test_partial_is_readable_before_the_child_has_unwound`, which pins the ordering +property **without** needing foundation on the path. That test is the in-repo +guard against silently regressing back into 37n's shape. + +`docs/lanes/9w0-delegate-timeout-partial-producer/` — round-trip test, probe, +evidence, this note. + +**Not touched:** amplifier-foundation, or any repo other than this one. +`settings.timeout` still defaults to `None`; k64's eval was not run. Both are +separately funded. + +--- + +## 4. Deviations from the goal text — stated, not absorbed + +**(a) The two "pre-existing failures" do not exist on this parent.** The goal +requires them to be shown failing on the parent commit, "patched and unpatched +alike". They do **not** fail: +`tests/test_session_spawner.py::TestSpawnEnrichment::test_spawn_result_includes_status_and_turn_count` +and its `test_resume_…` twin both **PASS** at `ab47608`, and the named-suite +baseline is **73 passed, 0 failed** — not 37n's "71 passed, 2 failed" at +`f16375fc`. The baseline improved between the two commits. Reported rather than +manufactured; the instruction's *purpose* (do not report someone else's failure +as yours) is satisfied — there is no failure to attribute in either direction. + +**(b) The patch was re-targeted by hand, as the goal predicted.** +`git apply --check` exits 1 at `ab47608` (`evidence/02`). The regions had moved +under the mid-run transcript-checkpoint work (`_install_transcript_checkpoint`, +`unregister_checkpoint`), which also *changed* the comment the patch edits: +main said "there is deliberately NO `except` here". That comment is now accurate +again, and says why the `except` that exists is safe. + +**(c) One design change, described in §1.** Eager publication instead of +publish-on-unwind. Everything else follows 37n's design. The counter-evidence +run (`evidence/04`) keeps the alternative honest: 37n's exact shape, everything +else identical, still `false`. + +**(d) `partial_max_chars` truncation is the consumer's job**, not this half's. +The producer hands over the full accumulated text; foundation caps it at 20 000 +chars keeping the tail. Confirmed reading `_collect_partial`, and pinned by +`partial_truncated: false` in `evidence/06`. + +--- + +## 5. Known limits (honest, not smuggled) + +* **Subprocess children.** `spawn_sub_session(use_subprocess=True)` runs the + child in another process, so the in-memory registry cannot see its text. Such + a timeout degrades to `partial_available: false` — the same graceful + degradation as no capability at all, never an error. Not in scope here. +* **Text blocks only.** The accumulator keys on + `block["type"] == "text"`. Thinking blocks and tool-call blocks are not + preserved. `segments` counts text segments, not turns — named for what it is. +* **No provider was driven.** Every check here is local: mocked child sessions, + the real spawner, the real delegate tool. `$0` authority permits nothing else, + and nothing else was needed for this contract. +* **Registry cap under extreme fan-out.** >64 concurrent live sub-sessions per + root process would evict the oldest live record. Measured fan-out is P50 2 / + P95 6 / max 7 (37n DESIGN.md §6), so this does not bind today; the eviction is + logged rather than silent. + +--- + +## 6. Reproduce + +```bash +# in-repo suite (12 new tests, full suite green) +.venv/bin/python -m pytest tests/ -q # 1659 passed, 1 skipped, 1 xfailed + +# cross-repo round trip — overlaid COPIES, neither repo mutated +git clone --depth 5 https://github.com/microsoft/amplifier-foundation /tmp/f # f42f48c +cp -rL /tmp/f/modules/tool-delegate/amplifier_module_tool_delegate /tmp/td/ +cp -rL ./amplifier_app_cli /tmp/itest/ +PYTHONPATH=/tmp/td:/tmp/itest .venv/bin/python -m pytest \ + docs/lanes/9w0-delegate-timeout-partial-producer/test_partial_roundtrip.py -q # 2 passed + +# G-D4 side by side +PYTHONPATH=/tmp/td:/tmp/itest .venv/bin/python \ + docs/lanes/9w0-delegate-timeout-partial-producer/probe_partial_roundtrip.py timeout +``` + +`conftest.py` for the out-of-repo runs supplies only an `anyio_backend` +fixture returning `"asyncio"`. diff --git a/docs/lanes/9w0-delegate-timeout-partial-producer/conftest.py b/docs/lanes/9w0-delegate-timeout-partial-producer/conftest.py new file mode 100644 index 0000000..7619eb1 --- /dev/null +++ b/docs/lanes/9w0-delegate-timeout-partial-producer/conftest.py @@ -0,0 +1,12 @@ +"""anyio backend for running this directory's cross-repo checks out of tree. + +The round-trip check is NOT part of the repo's own suite -- it needs +amplifier_module_tool_delegate on PYTHONPATH (see DONE-NOTE.md section 6). +""" + +import pytest + + +@pytest.fixture +def anyio_backend(): + return "asyncio" diff --git a/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/00-parent-baseline-named-suites.txt b/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/00-parent-baseline-named-suites.txt new file mode 100644 index 0000000..0d7f005 --- /dev/null +++ b/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/00-parent-baseline-named-suites.txt @@ -0,0 +1,11 @@ +# Baseline on PARENT commit ab47608fc989e13f9674c3f5f9efc5625a9d7673 (2026-09-03T08:07:50Z) + +## Named suites (37n's comparison set) +........................................................................ [ 98%] +. [100%] +73 passed in 0.27s + +## The two failures 37n named at f16375fc +tests/test_session_spawner.py::TestSpawnEnrichment::test_spawn_result_includes_status_and_turn_count PASSED [ 50%] +tests/test_session_spawner.py::TestSpawnEnrichment::test_resume_result_includes_status_and_turn_count PASSED [100%] +============================== 2 passed in 0.02s =============================== diff --git a/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/01-parent-baseline-full-suite.txt b/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/01-parent-baseline-full-suite.txt new file mode 100644 index 0000000..1d7839f --- /dev/null +++ b/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/01-parent-baseline-full-suite.txt @@ -0,0 +1,3 @@ +# FULL suite on PARENT commit ab47608fc989e13f9674c3f5f9efc5625a9d7673 +................................................................. [100%] +1647 passed, 1 skipped, 13 deselected, 1 xfailed in 8.44s diff --git a/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/02-original-patch-does-not-apply.txt b/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/02-original-patch-does-not-apply.txt new file mode 100644 index 0000000..ac401b3 --- /dev/null +++ b/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/02-original-patch-does-not-apply.txt @@ -0,0 +1,4 @@ +$ git apply --check -p1 PATCH-app-cli-session-spawner.diff # 37n's patch, cut at f16375fc +error: patch failed: amplifier_app_cli/session_spawner.py:843 +error: amplifier_app_cli/session_spawner.py: patch does not apply +exit=1 diff --git a/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/03-failbefore-roundtrip-parent.txt b/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/03-failbefore-roundtrip-parent.txt new file mode 100644 index 0000000..28a9bcf --- /dev/null +++ b/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/03-failbefore-roundtrip-parent.txt @@ -0,0 +1,45 @@ +# FAIL-BEFORE: the same cross-repo round trip, run against the PARENT app package +# app-cli parent : ab47608fc989e13f9674c3f5f9efc5625a9d7673 +# foundation (real): f42f48c (consumer half, merged as PR #353) +# tool-delegate : cp -rL from a fresh clone of foundation main; app pkg: cp -rL, overlaid with parent sources + + async def test_producer_and_consumer_agree_on_the_partial_contract(): + result, caps = await _run_timeout_leg() + + # The timeout invariant the consumer half pins: never success, on either channel. + assert result.success is False + assert result.output["status"] == "timeout" + assert "response" not in result.output + + # The producer half: the straggler's own work survived and was handed over. +> assert result.output["partial_available"] is True, ( + "session.partial produced nothing -- k64's G-D4 would record " + "PARTIAL-PATH-NOT-EXERCISED here" + ) +E AssertionError: session.partial produced nothing -- k64's G-D4 would record PARTIAL-PATH-NOT-EXERCISED here +E assert False is True + +test_partial_roundtrip.py:168: AssertionError +------------------------------ Captured log call ------------------------------- +WARNING amplifier_module_tool_delegate:__init__.py:2282 Agent 'explorer' timed out after 1s (delegate tool session-level timeout; elapsed 1.001s). No partial output could be recovered. Child cancellation cleanup is still in progress; do not resume this session until cleanup and persistence complete. +_________ test_a_second_timeout_leg_does_not_inherit_the_first_partial _________ + + async def test_a_second_timeout_leg_does_not_inherit_the_first_partial(): + """Two legs, two ids -- the second must carry only its own work.""" + first, _ = await _run_timeout_leg() + second, _ = await _run_timeout_leg() + + assert first.output["session_id"] != second.output["session_id"] +> assert second.output["partial_response"] == "".join(PARTIAL_CHUNKS) +E AssertionError: assert None == 'anchor A1 confirmed. anchor A2 confirmed. ' +E + where 'anchor A1 confirmed. anchor A2 confirmed. ' = (['anchor A1 confirmed. ', 'anchor A2 confirmed. ']) +E + where = ''.join + +test_partial_roundtrip.py:189: AssertionError +------------------------------ Captured log call ------------------------------- +WARNING amplifier_module_tool_delegate:__init__.py:2282 Agent 'explorer' timed out after 1s (delegate tool session-level timeout; elapsed 1.001s). No partial output could be recovered. Child cancellation cleanup is still in progress; do not resume this session until cleanup and persistence complete. +WARNING amplifier_module_tool_delegate:__init__.py:2282 Agent 'explorer' timed out after 1s (delegate tool session-level timeout; elapsed 1.001s). No partial output could be recovered. Child cancellation cleanup is still in progress; do not resume this session until cleanup and persistence complete. +=========================== short test summary info ============================ +FAILED test_partial_roundtrip.py::test_producer_and_consumer_agree_on_the_partial_contract +FAILED test_partial_roundtrip.py::test_a_second_timeout_leg_does_not_inherit_the_first_partial +2 failed in 3.29s diff --git a/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/04-counterevidence-naive-shape-still-false.txt b/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/04-counterevidence-naive-shape-still-false.txt new file mode 100644 index 0000000..fe70ffd --- /dev/null +++ b/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/04-counterevidence-naive-shape-still-false.txt @@ -0,0 +1,13 @@ +# COUNTER-EVIDENCE: 37n's ORIGINAL patch shape (publish only from the child's +# 'except BaseException' unwind) against the REAL merged consumer, foundation f42f48c. +# Everything else identical to the shipped patch. +# Result: still partial_available=false. The producer seals AFTER the consumer has read. + +E assert False is True +WARNING amplifier_module_tool_delegate:__init__.py:2282 Agent 'explorer' timed out after 1s (delegate tool session-level timeout; elapsed 1.001s). No partial output could be recovered. Child cancellation cleanup is still in progress; do not resume this session until cleanup and persistence complete. +WARNING amplifier_app_cli.session_spawner:session_spawner.py:133 Sub-session 0000000000000000-14995cfa065d4e59_explorer did not complete; preserved 2 partial text segment(s), 42 chars +WARNING amplifier_module_tool_delegate:__init__.py:2282 Agent 'explorer' timed out after 1s (delegate tool session-level timeout; elapsed 1.0s). No partial output could be recovered. Child cancellation cleanup is still in progress; do not resume this session until cleanup and persistence complete. +WARNING amplifier_app_cli.session_spawner:session_spawner.py:133 Sub-session 0000000000000000-31d194b3a158428e_explorer did not complete; preserved 2 partial text segment(s), 42 chars +WARNING amplifier_module_tool_delegate:__init__.py:2282 Agent 'explorer' timed out after 1s (delegate tool session-level timeout; elapsed 1.002s). No partial output could be recovered. Child cancellation cleanup is still in progress; do not resume this session until cleanup and persistence complete. +WARNING amplifier_app_cli.session_spawner:session_spawner.py:133 Sub-session 0000000000000000-977f57772dd8433b_explorer did not complete; preserved 2 partial text segment(s), 42 chars +2 failed in 3.30s diff --git a/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/05-normal-completion-byte-identical.txt b/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/05-normal-completion-byte-identical.txt new file mode 100644 index 0000000..6a5089a --- /dev/null +++ b/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/05-normal-completion-byte-identical.txt @@ -0,0 +1,21 @@ +# NORMAL COMPLETION -- byte-identity, PARENT vs PATCHED +# same probe, same fixed sub_session_id, only amplifier_app_cli differs + +$ diff <(probe normal @parent) <(probe normal @patched) +(no differences) +exit=0 + +$ sha256sum +ef8c86fdaf3b68e26e02e86e6a1bd59e0b638e4eb9b36f8f0a7f9f0138a35700 /tmp/9w0/normal-parent.json +ef8c86fdaf3b68e26e02e86e6a1bd59e0b638e4eb9b36f8f0a7f9f0138a35700 /tmp/9w0/normal-patched.json + +# the serialized result itself: +{ + "metadata": { + "o": "loop-basic" + }, + "output": "agent response", + "session_id": "parent-session-0000000000000000_explorer", + "status": "success", + "turn_count": 5 +} diff --git a/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/06-gd4-partial-available-true.txt b/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/06-gd4-partial-available-true.txt new file mode 100644 index 0000000..db0bb7f --- /dev/null +++ b/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/06-gd4-partial-available-true.txt @@ -0,0 +1,55 @@ +# G-D4: a real timeout that carries partial_available: true +# consumer = amplifier-foundation f42f48c tool-delegate (merged, PR #353), unmodified +# producer = amplifier_app_cli on PYTHONPATH; both overlaid via cp -rL, neither repo mutated + +================ PARENT (app-cli ab47608) ================ +app-registered capabilities: session.resume, session.spawn +session.partial registered: False + +ToolResult.success = False +{ + "agent": "explorer", + "completed": false, + "guidance": "INCOMPLETE: this delegate did not finish and no partial output could be recovered. Nothing here is a result. Re-delegate a narrower task or complete the work yourself; see metadata.recovery_message before considering this session for resumption.", + "metadata": { + "elapsed_s": 1.002, + "recovery_message": "Child cancellation cleanup is still in progress; do not resume this session until cleanup and persistence complete.", + "resumable": false, + "resume_status": "pending_child_cleanup", + "timeout_seconds": 1 + }, + "partial_available": false, + "partial_chars_total": 0, + "partial_response": null, + "partial_segments": 0, + "partial_source": "none", + "partial_truncated": false, + "session_id": "0000000000000000-b3d0212ab3024983_explorer", + "status": "timeout" +} + +================ PATCHED (this lane) ================ +app-registered capabilities: session.partial, session.resume, session.spawn +session.partial registered: True + +ToolResult.success = False +{ + "agent": "explorer", + "completed": false, + "guidance": "INCOMPLETE: this delegate did not finish. The text in 'partial_response' is unfinished work salvaged from the agent mid-flight -- it has NOT been checked, concluded, or self-reviewed by that agent. Do not report it as a completed result and do not treat its conclusions as final. Re-delegate a narrower task or complete the work yourself; see metadata.recovery_message before considering this session for resumption.", + "metadata": { + "elapsed_s": 1.001, + "recovery_message": "Child cancellation cleanup is still in progress; do not resume this session until cleanup and persistence complete.", + "resumable": false, + "resume_status": "pending_child_cleanup", + "timeout_seconds": 1 + }, + "partial_available": true, + "partial_chars_total": 42, + "partial_response": "anchor A1 confirmed. anchor A2 confirmed. ", + "partial_segments": 2, + "partial_source": "spawn-accumulator", + "partial_truncated": false, + "session_id": "0000000000000000-d3593d0d39d24acd_explorer", + "status": "timeout" +} diff --git a/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/07-passafter-roundtrip-patched.txt b/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/07-passafter-roundtrip-patched.txt new file mode 100644 index 0000000..f234049 --- /dev/null +++ b/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/07-passafter-roundtrip-patched.txt @@ -0,0 +1,13 @@ +# PASS-AFTER: cross-repo round trip, patched app-cli x foundation f42f48c +# PYTHONPATH=: (cp -rL; neither repo mutated) + +============================= test session starts ============================== +platform linux -- Python 3.13.11, pytest-9.0.3, pluggy-1.6.0 +rootdir: /tmp/9w0/run +plugins: anyio-4.12.0, amplifier-core-1.6.0, asyncio-1.3.0 +asyncio: mode=Mode.STRICT, debug=False, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function +collected 2 items + +test_partial_roundtrip.py .. [100%] + +============================== 2 passed in 3.15s =============================== diff --git a/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/08-lint-and-full-suite-patched.txt b/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/08-lint-and-full-suite-patched.txt new file mode 100644 index 0000000..c68c3a8 --- /dev/null +++ b/docs/lanes/9w0-delegate-timeout-partial-producer/evidence/08-lint-and-full-suite-patched.txt @@ -0,0 +1,14 @@ +# LINT + FULL SUITE on the patched tree + +$ ruff check amplifier_app_cli tests +1472 | """Create a minimal Click CLI with the run command registered. + | + +Found 14 errors. +[*] 9 fixable with the `--fix` option (1 hidden fix can be enabled with the `--unsafe-fixes` option). +$ ruff format --check amplifier_app_cli/session_spawner.py amplifier_app_cli/session_runner.py tests/test_session_spawner_partial.py +3 files already formatted + +$ pytest tests/ -q +..... [100%] +1659 passed, 1 skipped, 13 deselected, 1 xfailed in 10.50s diff --git a/docs/lanes/9w0-delegate-timeout-partial-producer/probe_partial_roundtrip.py b/docs/lanes/9w0-delegate-timeout-partial-producer/probe_partial_roundtrip.py new file mode 100644 index 0000000..633d40a --- /dev/null +++ b/docs/lanes/9w0-delegate-timeout-partial-producer/probe_partial_roundtrip.py @@ -0,0 +1,186 @@ +"""Behavioural probe for the partial-result path -- prints, does not assert. + +Two modes, both run against whatever `amplifier_app_cli` is on PYTHONPATH, so the +same script produces comparable output for the parent commit and the patched tree: + + normal -- spawn a sub-session that COMPLETES; print the returned dict as + canonical JSON. Byte-identity between parent and patched is the + "normal completions unchanged" deliverable, shown rather than + asserted. + timeout -- drive the REAL foundation consumer (tool-delegate f42f48c) over + the REAL app producer; print the delegate's model-visible output. + `partial_available` is k64 gate G-D4's deciding fact. + +Usage: + PYTHONPATH=: python probe_partial_roundtrip.py normal + PYTHONPATH=: python probe_partial_roundtrip.py timeout +""" + +import asyncio +import json +import sys +from unittest.mock import AsyncMock +from unittest.mock import MagicMock +from unittest.mock import patch + +from amplifier_core.events import CONTENT_BLOCK_END + +from amplifier_app_cli.session_runner import register_session_spawning +from amplifier_app_cli.session_spawner import spawn_sub_session + +FIXED_SUB_SESSION_ID = "parent-session-0000000000000000_explorer" +PARTIAL_CHUNKS = ["anchor A1 confirmed. ", "anchor A2 confirmed. "] + + +class FakeHooks: + def __init__(self): + self.handlers: dict[str, list] = {} + + def register(self, event, handler, priority=0, name=None): + self.handlers.setdefault(event, []).append(handler) + return lambda: self.handlers[event].remove(handler) + + async def emit(self, event, data): + for handler in list(self.handlers.get(event, [])): + await handler(event, data) + + async def text(self, chunk): + await self.emit(CONTENT_BLOCK_END, {"block": {"type": "text", "text": chunk}}) + + +def parent_session(): + parent = MagicMock() + parent.coordinator.get.return_value = None + parent.coordinator.get_capability.return_value = None + parent.coordinator.display_system = MagicMock() + parent.coordinator.cancellation = MagicMock() + parent.config = { + "session": {"orchestrator": "loop-basic", "context": "context-simple"} + } + parent.session_id = "parent-session" + parent.trace_id = "trace-abc" + parent.loader = None + return parent + + +def child_session(hooks, execute_body): + coordinator = MagicMock() + coordinator.register_capability = MagicMock() + coordinator.get_capability.return_value = None + coordinator.display_system = MagicMock() + coordinator.mount = AsyncMock() + coordinator.collect_contributions = AsyncMock(return_value=[]) + + def _get(name): + if name == "hooks": + return hooks + if name == "context": + ctx = AsyncMock() + ctx.get_messages = AsyncMock(return_value=[]) + ctx.add_message = AsyncMock() + return ctx + return None + + coordinator.get = _get + child = MagicMock() + child.coordinator = coordinator + child.initialize = AsyncMock() + child.execute = AsyncMock(side_effect=execute_body) + child.cleanup = AsyncMock() + child.session_id = "child-session" + return child + + +def patched_env(child): + return ( + patch( + "amplifier_app_cli.session_spawner.AmplifierSession", return_value=child + ), + patch( + "amplifier_app_cli.session_spawner.generate_sub_session_id", + return_value=FIXED_SUB_SESSION_ID, + ), + patch("amplifier_app_cli.paths.create_foundation_resolver"), + patch("amplifier_app_cli.session_store.SessionStore.save"), + ) + + +async def run_normal(): + hooks = FakeHooks() + + async def _completes(instruction): + await hooks.text("the finished answer") + await hooks.emit( + "orchestrator:complete", + {"status": "success", "turn_count": 5, "metadata": {"o": "loop-basic"}}, + ) + return "agent response" + + a, b, c, d = patched_env(child_session(hooks, _completes)) + with a, b, c, d: + result = await spawn_sub_session( + agent_name="explorer", + instruction="do the thing", + parent_session=parent_session(), + agent_configs={"explorer": {"description": "an agent"}}, + ) + print(json.dumps(result, indent=2, sort_keys=True, default=str)) + + +class FakeCoordinator: + def __init__(self, capabilities, parent): + self.session_id = parent.session_id + self.session = parent + self.config = {"agents": {"explorer": {}}} + self._capabilities = capabilities + + def get_capability(self, name): + return self._capabilities.get(name) + + +async def run_timeout(): + from amplifier_module_tool_delegate import DelegateTool + + registered: dict = {} + session = MagicMock() + session.coordinator.register_capability = MagicMock( + side_effect=lambda name, fn: registered.__setitem__(name, fn) + ) + register_session_spawning(session) + print( + "app-registered capabilities: " + + ", ".join(sorted(registered)) + + "\nsession.partial registered: " + + str("session.partial" in registered) + + "\n" + ) + + hooks = FakeHooks() + + async def _never_finishes(instruction): + for chunk in PARTIAL_CHUNKS: + await hooks.text(chunk) + await asyncio.sleep(3600) + + tool = DelegateTool( + FakeCoordinator(registered, parent_session()), {"settings": {"timeout": 1}} + ) + a, b, c, d = patched_env(child_session(hooks, _never_finishes)) + with a, b, c, d: + result = await tool._spawn_new_session( + agent_name="explorer", + instruction="find the anchors", + context_depth="none", + context_scope="conversation", + context_turns=0, + provider_preferences=None, + hooks=FakeHooks(), + agents={"explorer": {}}, + ) + print("ToolResult.success = " + str(result.success)) + print(json.dumps(result.output, indent=2, sort_keys=True, default=str)) + + +if __name__ == "__main__": + mode = sys.argv[1] if len(sys.argv) > 1 else "normal" + asyncio.run(run_normal() if mode == "normal" else run_timeout()) diff --git a/docs/lanes/9w0-delegate-timeout-partial-producer/test_partial_roundtrip.py b/docs/lanes/9w0-delegate-timeout-partial-producer/test_partial_roundtrip.py new file mode 100644 index 0000000..76425cc --- /dev/null +++ b/docs/lanes/9w0-delegate-timeout-partial-producer/test_partial_roundtrip.py @@ -0,0 +1,190 @@ +"""Cross-repo contract check for the partial-result path. + +amplifier-foundation `tool-delegate` is the CONSUMER of the partial; +amplifier-app-cli `session_spawner` is the PRODUCER. Each half is unit-tested in +its own repo against a fake counterpart; this check wires the two REAL halves +together so the contract cannot drift silently between them. + +Unlike 37n's original draft, this version does not import the producer's private +`_seal_partial` and hand the id over by side channel. It drives the REAL app-layer +`session.spawn` capability -- exactly the one `register_session_spawning()` puts on +a root session -- through the REAL `DelegateTool`, and lets the delegate mint the +`sub_session_id` itself. That is what makes the check meaningful: if the two sides +disagree about the key, the signature, or the payload shape, this fails. + +It therefore runs unchanged against BOTH the parent commit (no producer: +`partial_available: false` -- k64's G-D4 stop condition) and the patched tree +(`partial_available: true`). + +Run per APPLY.md, against overlaid COPIES of both packages, never the repos: + + PYTHONPATH=: python -m pytest \ + test_partial_roundtrip.py -q +""" + +import asyncio +from unittest.mock import AsyncMock +from unittest.mock import MagicMock +from unittest.mock import patch + +import pytest +from amplifier_core.events import CONTENT_BLOCK_END +from amplifier_module_tool_delegate import DelegateTool + +from amplifier_app_cli.session_runner import register_session_spawning + +pytestmark = pytest.mark.anyio + +PARTIAL_CHUNKS = ["anchor A1 confirmed. ", "anchor A2 confirmed. "] + + +class FakeHooks: + def __init__(self): + self.handlers: dict[str, list] = {} + self.events: list = [] + + def register(self, event, handler, priority=0, name=None): + self.handlers.setdefault(event, []).append(handler) + return lambda: self.handlers[event].remove(handler) + + async def emit(self, event, data): + self.events.append((event, data)) + for handler in list(self.handlers.get(event, [])): + await handler(event, data) + + +def _app_capabilities(): + """Exactly what the app registers on a real root session.""" + registered: dict = {} + session = MagicMock() + session.coordinator.register_capability = MagicMock( + side_effect=lambda name, fn: registered.__setitem__(name, fn) + ) + register_session_spawning(session) + return registered + + +def _parent_session(): + parent = MagicMock() + parent.coordinator.get.return_value = None + parent.coordinator.get_capability.return_value = None + parent.coordinator.display_system = MagicMock() + parent.coordinator.cancellation = MagicMock() + parent.config = { + "session": {"orchestrator": "loop-basic", "context": "context-simple"} + } + parent.session_id = "parent-session" + parent.trace_id = "trace-abc" + parent.loader = None + return parent + + +def _straggler_child(child_hooks): + """A child session that produces two text blocks and then never finishes.""" + coordinator = MagicMock() + coordinator.register_capability = MagicMock() + coordinator.get_capability.return_value = None + coordinator.display_system = MagicMock() + coordinator.mount = AsyncMock() + coordinator.collect_contributions = AsyncMock(return_value=[]) + + def _get(name): + if name == "hooks": + return child_hooks + if name == "context": + ctx = AsyncMock() + ctx.get_messages = AsyncMock(return_value=[]) + ctx.add_message = AsyncMock() + return ctx + return None + + coordinator.get = _get + + async def _never_finishes(instruction): + for chunk in PARTIAL_CHUNKS: + await child_hooks.emit( + CONTENT_BLOCK_END, {"block": {"type": "text", "text": chunk}} + ) + await asyncio.sleep(3600) + + child = MagicMock() + child.coordinator = coordinator + child.initialize = AsyncMock() + child.execute = AsyncMock(side_effect=_never_finishes) + child.cleanup = AsyncMock() + child.session_id = "child-session" + return child + + +class FakeCoordinator: + def __init__(self, capabilities, parent_session): + self.session_id = parent_session.session_id + self.session = parent_session + self.config = {"agents": {"explorer": {}}} + self._capabilities = capabilities + + def get_capability(self, name): + return self._capabilities.get(name) + + +async def _run_timeout_leg(): + parent = _parent_session() + caps = _app_capabilities() + tool = DelegateTool( + FakeCoordinator(caps, parent), {"settings": {"timeout": 1}} + ) + + child_hooks = FakeHooks() + with patch( + "amplifier_app_cli.session_spawner.AmplifierSession", + return_value=_straggler_child(child_hooks), + ): + with patch("amplifier_app_cli.paths.create_foundation_resolver"): + with patch("amplifier_app_cli.session_store.SessionStore.save"): + return ( + await tool._spawn_new_session( + agent_name="explorer", + instruction="find the anchors", + context_depth="none", + context_scope="conversation", + context_turns=0, + provider_preferences=None, + hooks=FakeHooks(), + agents={"explorer": {}}, + ), + caps, + ) + + +async def test_producer_and_consumer_agree_on_the_partial_contract(): + result, caps = await _run_timeout_leg() + + # The timeout invariant the consumer half pins: never success, on either channel. + assert result.success is False + assert result.output["status"] == "timeout" + assert "response" not in result.output + + # The producer half: the straggler's own work survived and was handed over. + assert result.output["partial_available"] is True, ( + "session.partial produced nothing -- k64's G-D4 would record " + "PARTIAL-PATH-NOT-EXERCISED here" + ) + assert result.output["partial_response"] == "".join(PARTIAL_CHUNKS) + assert result.output["partial_segments"] == len(PARTIAL_CHUNKS) + assert result.output["partial_source"] == "spawn-accumulator" + assert result.output["partial_truncated"] is False + assert result.output["partial_chars_total"] == len("".join(PARTIAL_CHUNKS)) + + # Reads are destructive: the registry does not leak across delegate calls. + sub_session_id = result.output["session_id"] + assert caps["session.partial"](sub_session_id) is None + + +async def test_a_second_timeout_leg_does_not_inherit_the_first_partial(): + """Two legs, two ids -- the second must carry only its own work.""" + first, _ = await _run_timeout_leg() + second, _ = await _run_timeout_leg() + + assert first.output["session_id"] != second.output["session_id"] + assert second.output["partial_response"] == "".join(PARTIAL_CHUNKS) + assert second.output["partial_segments"] == len(PARTIAL_CHUNKS) diff --git a/tests/test_session_spawner_partial.py b/tests/test_session_spawner_partial.py new file mode 100644 index 0000000..7d0934e --- /dev/null +++ b/tests/test_session_spawner_partial.py @@ -0,0 +1,346 @@ +"""Tests for the partial-output producer path (`session.partial`). + +PRODUCER half of a cross-repo contract. The CONSUMER shipped in +amplifier-foundation `f42f48c` (tool-delegate): on a per-delegate wall-clock +timeout it returns rather than raises, and calls an optional app-layer +``session.partial`` capability:: + + (sub_session_id: str) -> {"text": str, "segments": int, "source": str} | None + +Absent / None / malformed / raising -> ``partial_available: false``. Everything +here exists so that boolean can be TRUE for a real timeout: without a producer, +a cancelled sub-session's work is discarded and every timeout reports +``partial_available: false``. + +The two properties under test are opposites and both matter: + +* a CANCELLED sub-session's assistant text survives and is readable; +* a NORMALLY COMPLETED sub-session is untouched -- the result dict gains no + key, and nothing is left in the registry. +""" + +from unittest.mock import AsyncMock +from unittest.mock import MagicMock +from unittest.mock import patch + +import pytest +from amplifier_core.events import CONTENT_BLOCK_END + +from amplifier_app_cli import session_spawner +from amplifier_app_cli.session_spawner import _seal_partial +from amplifier_app_cli.session_spawner import get_partial_output +from amplifier_app_cli.session_spawner import spawn_sub_session + +pytestmark = pytest.mark.anyio + + +# --------------------------------------------------------------------------- +# Harness +# --------------------------------------------------------------------------- + + +class FakeHooks: + """Hook registry keyed by event, with multiple handlers per event. + + The spawner registers three handlers (orchestrator:complete, the + provider:request transcript checkpoint, and the content_block:end partial + accumulator), so a single-slot fake would let the last writer win. + """ + + def __init__(self): + self.handlers: dict[str, list] = {} + + def register(self, event, handler, priority=0, name=None): + self.handlers.setdefault(event, []).append(handler) + + def _unregister(): + if handler in self.handlers.get(event, []): + self.handlers[event].remove(handler) + + return _unregister + + async def emit(self, event, data): + for handler in list(self.handlers.get(event, [])): + await handler(event, data) + + async def fire_text_block(self, text: str): + await self.emit(CONTENT_BLOCK_END, {"block": {"type": "text", "text": text}}) + + +def _parent_session(): + parent_coordinator = MagicMock() + parent_coordinator.get.return_value = None + parent_coordinator.get_capability.return_value = None + parent_coordinator.display_system = MagicMock() + parent_coordinator.cancellation = MagicMock() + parent_coordinator.cancellation.register_child = MagicMock() + parent_coordinator.cancellation.unregister_child = MagicMock() + + parent_session = MagicMock() + parent_session.coordinator = parent_coordinator + parent_session.config = { + "session": {"orchestrator": "loop-basic", "context": "context-simple"}, + } + parent_session.session_id = "parent-123" + parent_session.trace_id = "trace-abc" + parent_session.loader = None + return parent_session + + +def _child_session(hooks, execute_body): + """A child session whose execute() runs ``execute_body(hooks)``.""" + child_coordinator = MagicMock() + child_coordinator.registered_capabilities = {} + + def _register_capability(name, fn): + child_coordinator.registered_capabilities[name] = fn + + child_coordinator.register_capability = MagicMock(side_effect=_register_capability) + child_coordinator.get_capability.return_value = None + child_coordinator.display_system = MagicMock() + + def child_get(name): + if name == "hooks": + return hooks + if name == "context": + ctx = AsyncMock() + ctx.get_messages = AsyncMock(return_value=[]) + ctx.add_message = AsyncMock() + return ctx + return None + + child_coordinator.get = child_get + child_coordinator.mount = AsyncMock() + child_coordinator.collect_contributions = AsyncMock(return_value=[]) + + child_session = MagicMock() + child_session.coordinator = child_coordinator + child_session.initialize = AsyncMock() + child_session.execute = AsyncMock(side_effect=execute_body) + child_session.cleanup = AsyncMock() + child_session.session_id = "child-001" + return child_session + + +async def _spawn(child_session, sub_session_id="child-001"): + with patch( + "amplifier_app_cli.session_spawner.AmplifierSession", + return_value=child_session, + ): + with patch( + "amplifier_app_cli.session_spawner.generate_sub_session_id", + return_value=sub_session_id, + ): + with patch("amplifier_app_cli.paths.create_foundation_resolver"): + with patch("amplifier_app_cli.session_store.SessionStore.save"): + return await spawn_sub_session( + agent_name="test-agent", + instruction="Do something", + parent_session=_parent_session(), + agent_configs={"test-agent": {"description": "A test agent"}}, + ) + + +@pytest.fixture(autouse=True) +def _clean_registry(): + session_spawner._PARTIAL_OUTPUTS.clear() + yield + session_spawner._PARTIAL_OUTPUTS.clear() + + +# --------------------------------------------------------------------------- +# The property this whole item exists for +# --------------------------------------------------------------------------- + + +async def test_cancelled_spawn_preserves_partial_text(): + """FAIL-BEFORE: on the parent commit the cancelled agent's text is discarded. + + This is the single fact k64's gate G-D4 turns on: without it every timeout + carries ``partial_available: false``. + """ + hooks = FakeHooks() + + async def _cancelled_midway(instruction): + await hooks.fire_text_block("anchor A1 confirmed. ") + await hooks.fire_text_block("anchor A2 confirmed. ") + raise TimeoutError("wall clock") + + with pytest.raises(TimeoutError): + await _spawn(_child_session(hooks, _cancelled_midway)) + + partial = get_partial_output("child-001") + assert partial is not None, "the cancelled sub-session's work was discarded" + assert partial["text"] == "anchor A1 confirmed. anchor A2 confirmed. " + assert partial["segments"] == 2 + assert partial["source"] == "spawn-accumulator" + + +async def test_asyncio_cancellation_also_preserves_partial_text(): + """The real timeout path raises CancelledError, a BaseException.""" + import asyncio + + hooks = FakeHooks() + + async def _hard_cancelled(instruction): + await hooks.fire_text_block("half a finding") + raise asyncio.CancelledError() + + with pytest.raises(asyncio.CancelledError): + await _spawn(_child_session(hooks, _hard_cancelled)) + + partial = get_partial_output("child-001") + assert partial is not None + assert partial["text"] == "half a finding" + + +async def test_partial_is_readable_before_the_child_has_unwound(): + """The ordering property the cross-repo round trip caught. + + tool-delegate's `_await_child_with_deadline` cancels the child, DETACHES it, + and reads `session.partial` immediately -- it deliberately does not wait for + a slow unwind. So the partial must be readable at the instant of cancel, + before the child's own `except` handler has been scheduled. Publishing only + from that handler passes both halves' unit tests and still reports + `partial_available: false` in production. + """ + import asyncio + + hooks = FakeHooks() + started = asyncio.Event() + + async def _produces_then_hangs(instruction): + await hooks.fire_text_block("work in progress") + started.set() + await asyncio.sleep(3600) + + task = asyncio.ensure_future(_spawn(_child_session(hooks, _produces_then_hangs))) + await started.wait() + task.cancel() # the child has NOT yet run any exception handler + + partial = get_partial_output("child-001") + assert partial is not None, "unreadable until the child unwinds -- too late" + assert partial["text"] == "work in progress" + + with pytest.raises(asyncio.CancelledError): + await task + + +async def test_partial_capability_is_registered_on_the_child_session(): + """A grandchild that times out under this child must be recoverable too.""" + hooks = FakeHooks() + + async def _ok(instruction): + return "done" + + child = _child_session(hooks, _ok) + await _spawn(child) + + registered = child.coordinator.registered_capabilities + assert "session.partial" in registered + assert registered["session.partial"] is get_partial_output + + +# --------------------------------------------------------------------------- +# The inverse guard: normal completion is untouched +# --------------------------------------------------------------------------- + + +async def test_normal_completion_result_gains_no_key(): + """A sub-session that finishes normally returns exactly today's shape.""" + hooks = FakeHooks() + + async def _completes(instruction): + await hooks.fire_text_block("the finished answer") + await hooks.emit( + "orchestrator:complete", + {"status": "success", "turn_count": 5, "metadata": {"o": "loop-basic"}}, + ) + return "agent response" + + result = await _spawn(_child_session(hooks, _completes)) + + assert set(result) == {"output", "session_id", "status", "turn_count", "metadata"} + assert result["output"] == "agent response" + assert result["session_id"] == "child-001" + assert result["status"] == "success" + assert result["turn_count"] == 5 + + +async def test_normal_completion_leaves_nothing_in_the_registry(): + """No seal on the success path -- the registry does not accumulate.""" + hooks = FakeHooks() + + async def _completes(instruction): + await hooks.fire_text_block("the finished answer") + return "agent response" + + await _spawn(_child_session(hooks, _completes)) + + assert session_spawner._PARTIAL_OUTPUTS == {} + assert get_partial_output("child-001") is None + + +# --------------------------------------------------------------------------- +# Registry mechanics the consumer relies on +# --------------------------------------------------------------------------- + + +async def test_reads_are_destructive(): + _seal_partial("s1", {"chunks": ["a", "b"]}) + assert get_partial_output("s1")["text"] == "ab" + assert get_partial_output("s1") is None + + +async def test_unknown_session_returns_none(): + assert get_partial_output("never-existed") is None + + +async def test_empty_accumulator_is_not_sealed(): + """Nothing produced -> nothing to offer; the consumer degrades to false.""" + _seal_partial("s2", {"chunks": []}) + assert get_partial_output("s2") is None + + +async def test_registry_is_capped(): + """A long-lived root cannot leak unbounded partial records.""" + for i in range(session_spawner._PARTIAL_MAX_SESSIONS + 10): + _seal_partial(f"s{i}", {"chunks": ["x"]}) + assert ( + len(session_spawner._PARTIAL_OUTPUTS) <= session_spawner._PARTIAL_MAX_SESSIONS + ) + + +async def test_no_hooks_coordinator_does_not_break_spawn(): + """A session with no hooks registry still spawns; it just offers no partial.""" + + async def _completes(instruction): + return "agent response" + + child = _child_session(None, _completes) + child.coordinator.get = lambda name: ( + AsyncMock(get_messages=AsyncMock(return_value=[]), add_message=AsyncMock()) + if name == "context" + else None + ) + result = await _spawn(child) + assert result["output"] == "agent response" + + +# --------------------------------------------------------------------------- +# Root-session registration +# --------------------------------------------------------------------------- + + +async def test_root_session_registers_session_partial(): + from amplifier_app_cli.session_runner import register_session_spawning + + session = MagicMock() + registered: dict = {} + session.coordinator.register_capability = MagicMock( + side_effect=lambda name, fn: registered.__setitem__(name, fn) + ) + + register_session_spawning(session) + + assert registered["session.partial"] is get_partial_output From 7f749c0db23a4d8920f3b275880e4a04b2c77dfd Mon Sep 17 00:00:00 2001 From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> Date: Thu, 3 Sep 2026 01:19:40 -0700 Subject: [PATCH 2/2] docs(lane-9w0): record the draft PR link in the lane DONE-NOTE --- docs/lanes/9w0-delegate-timeout-partial-producer/DONE-NOTE.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/docs/lanes/9w0-delegate-timeout-partial-producer/DONE-NOTE.md b/docs/lanes/9w0-delegate-timeout-partial-producer/DONE-NOTE.md index 3106694..1fb8ae6 100644 --- a/docs/lanes/9w0-delegate-timeout-partial-producer/DONE-NOTE.md +++ b/docs/lanes/9w0-delegate-timeout-partial-producer/DONE-NOTE.md @@ -6,6 +6,7 @@ |---|---| | Repo | `microsoft/amplifier-app-cli` | | Branch | `lane/9w0-delegate-timeout-partial-producer` | +| Draft PR | https://github.com/microsoft/amplifier-app-cli/pull/297 | | Parent commit | `ab47608fc989e13f9674c3f5f9efc5625a9d7673` | | Consumer half it pairs with | amplifier-foundation `f42f48c` (`model_performance-bp0`, PR #353) — **merged** | | Spend authority | **$0** (pure code change, no API/DTU spend authorized) | @@ -67,7 +68,7 @@ invisible to every test either repo owns on its own. | 4 | Normal completions unchanged — **byte-identical**, shown not asserted | **DONE** | `evidence/05` — `diff` empty, identical sha256 `ef8c86fd…` | | 5 | The two `TestSpawnEnrichment` failures shown on the parent | **DONE, with a correction — see §4** | `evidence/00` | | 6 | Fail-before evidence committed and pasted in the PR body | **DONE** | `evidence/03`, `evidence/04` | -| 7 | Draft PR on origin naming foundation `f42f48c` as the consumer half | **DONE** | see `publication` in `DONE.json` | +| 7 | Draft PR on origin naming foundation `f42f48c` as the consumer half | **DONE** | [PR #297](https://github.com/microsoft/amplifier-app-cli/pull/297), draft; `publication` block in `DONE.json` read back from the remote | | 8 | This DONE-NOTE under the lane artifact root | **DONE** | this file | Nothing was dropped, and no deliverable was cap-bound.