From 59ce58b413ff9582af5ccac49ff8cc3e1cfdf8c5 Mon Sep 17 00:00:00 2001 From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> Date: Sun, 25 Oct 2026 23:18:15 -0700 Subject: [PATCH] fix: keep interactive input resilient through background errors Keep CLI-owned, bounded loop diagnostics active through initialization, idle and steering input, and cleanup without prompt-toolkit modal error prompts. Carry prompt-child interrupts to the parent task and preserve release-wins handoff behavior after child settlement. Add privacy and lifecycle regressions. Generated with Amplifier Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- AGENTS.md | 13 ++ amplifier_app_cli/loop_errors.py | 234 +++++++++++++++++++ amplifier_app_cli/main.py | 20 +- amplifier_app_cli/session_handoff.py | 28 ++- amplifier_app_cli/steering_input.py | 3 + docs/INTERACTIVE_MODE.md | 6 + tests/test_interactive_chat_tty_teardown.py | 80 ++++++- tests/test_loop_errors.py | 240 ++++++++++++++++++++ tests/test_loop_errors_bounds.py | 176 ++++++++++++++ tests/test_session_handoff.py | 142 ++++++++++++ tests/test_steering.py | 82 ++++++- 11 files changed, 1012 insertions(+), 12 deletions(-) create mode 100644 amplifier_app_cli/loop_errors.py create mode 100644 tests/test_loop_errors.py create mode 100644 tests/test_loop_errors_bounds.py diff --git a/AGENTS.md b/AGENTS.md index cff6eb06..d4f3bf7f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -250,6 +250,19 @@ and `tests/test_env_vars.py` for this boundary. REPL exit commands return an action from `CommandProcessor`; only the normal REPL loop may terminate so its shared `finally` runs cleanup and closes TTY input once. +`CLIHandoff.prompt` carries prompt-child `KeyboardInterrupt` as an ordinary +outcome, then raises it in the parent so the REPL can confirm exit. Preserve +release-wins EOF and child cancellation/await behavior; run `tests/test_session_handoff.py`. + +`loop_errors.report_loop_errors` owns loop diagnostics from interactive initialization +through cleanup. Both idle and steering prompts must set `set_exception_handler=False` +to prevent prompt-toolkit's modal error prompt; preserve steering's SIGINT policy. +Diagnostics contain bounded safe type/frame metadata, never raw exception messages, +task/context representations or tracebacks. Keep output off the loop callback. +Final draining follows replacement workers under one absolute 500 ms deadline; +it bypasses normal output rate delay, not cancellation or privacy controls. +Run `tests/test_loop_errors.py`, `tests/test_loop_errors_bounds.py`, `tests/test_steering.py` and +`tests/test_interactive_chat_tty_teardown.py` when changing this boundary. ## Highway watchdog regression diff --git a/amplifier_app_cli/loop_errors.py b/amplifier_app_cli/loop_errors.py new file mode 100644 index 00000000..66188391 --- /dev/null +++ b/amplifier_app_cli/loop_errors.py @@ -0,0 +1,234 @@ +"""Non-modal, privacy-bounded loop diagnostics owned by interactive CLI sessions. + +Only safe snapshots cross into the output worker. Never format an exception, +task, loop context, traceback source, locals, notes, or exception chain. +The drain budget bounds this scope, not a blocked executor thread or subsequent +asyncio.run() executor shutdown. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import logging +import sys +from collections import deque +from contextvars import ContextVar +from functools import partial, wraps + +logger = logging.getLogger(__name__) +_REPORTING = ContextVar("cli_loop_error_reporting", default=False) +_WORKERS: set[asyncio.Future] = set() # Retain cancelled workers until observed. +_INTERVAL = 0.5 +_DRAIN_TIMEOUT = 0.5 +_MAX_COUNT = (1 << 63) - 1 + + +def _token(value: str, limit: int = 64) -> str: + if type(value) is not str: + return "unknown" + return ( + "".join( + c if c.isascii() and (c.isalnum() or c in "_.-<>") else "_" + for c in value[:limit] + ) + or "unknown" + ) + + +def _snapshot(context: dict) -> tuple[object, str]: + """Use built-in descriptors, never exception overrides or context repr.""" + error = dict.get(context, "exception") if type(context) is dict else None + if not isinstance(error, BaseException): + return object(), "Background loop error; message withheld." + name = _token(type.__dict__["__name__"].__get__(type(error))) + message = f"Background loop error: {name}; message withheld." + if type(error) is AttributeError: + field = AttributeError.name.__get__(error) + if type(field) is str and field == "_async_httpx_client": + message += " SDK cleanup is missing _async_httpx_client." + tb = BaseException.__traceback__.__get__(error) + for _ in range(8): + if tb is None: + break + code = tb.tb_frame.f_code + basename = code.co_filename[-128:].replace("\\", "/").rsplit("/", 1)[-1] + message += f"\n at {_token(basename)}:{tb.tb_lineno} in {_token(code.co_name)}" + tb = tb.tb_next + if tb is not None: + message += "\n additional frames withheld" + message = message[:1800] # ASCII snapshot; leave room for a count below. + + # Private equality only. Do not retain raw args or publish their digest. + # Unsupported/large args deliberately do not coalesce. + args = BaseException.args.__get__(error) + if len(args) <= 4 and all(type(a) is str and len(a) <= 256 for a in args): + digest = hashlib.blake2b(digest_size=16) + for arg in args: + data = arg.encode("utf-8", errors="surrogatepass") + digest.update(len(data).to_bytes(4, "big")) + digest.update(data) + return (id(type(error)), message, digest.digest()), message + return object(), message + + +def _observe(task: asyncio.Future) -> None: + _WORKERS.discard(task) + if not task.cancelled(): + task.exception() # Observe outcomes without formatting or re-reporting. + + +class _Reporter: + def __init__(self, loop): + self.loop = loop + self.pending = deque() + self.overflow = 0 + self.worker = None + self.next_output = 0.0 + self.closed = False + self.draining = False + self.busy = False + self.fallback_sent = False + self.last_key = None + self.handler = self.handle # Stable identity for ownership restoration. + + def handle(self, loop, context): + """Snapshot and enqueue only: no terminal/logging I/O in this callback.""" + if self.closed or self.busy or _REPORTING.get(): + return + self.busy = True + try: + try: + key, message = _snapshot(context) + except Exception: # noqa: BLE001 - never expose raw loop context. + # Fail closed rather than letting asyncio format raw context. + key, message = object(), "Background loop error; metadata withheld." + if self.pending and self.last_key == key and self.pending[-1][0] == key: + self.pending[-1][2] = min(_MAX_COUNT, self.pending[-1][2] + 1) + elif len(self.pending) < 16: + self.pending.append([key, message, 1]) + else: + self.overflow = min(_MAX_COUNT, self.overflow + 1) + self.last_key = key + if self.worker is None or self.worker.done(): + coroutine = self.run() + try: + self.worker = self.loop.create_task(coroutine) + except Exception: # noqa: BLE001 - task factories are external. + coroutine.close() # No unawaited-coroutine warning or raw context. + self.worker = None # Keep snapshots for a later callback/close. + return + _WORKERS.add(self.worker) + self.worker.add_done_callback(_observe) + finally: + self.busy = False + + def fallback(self): + """One fixed notice, called only from an output thread.""" + if self.fallback_sent: + return + self.fallback_sent = True + try: + if sys.__stderr__ is not None: + sys.__stderr__.write( + "Background error reporting failed; diagnostic details withheld.\n" + ) + sys.__stderr__.flush() + except Exception: # noqa: BLE001 - the last sink cannot report recursively. + return + + def emit(self, batch, overflow): + token = _REPORTING.set(True) + try: + for _, message, count in batch: + count_text = f">={count}" if count == _MAX_COUNT else f"{count}" + logger.error("%s", f"{message}\n count={count_text}", exc_info=None) + if overflow: + count_text = ( + f">={overflow}" if overflow == _MAX_COUNT else f"{overflow}" + ) + logger.error( + "Background loop error queue overflow: %s additional errors; " + "details withheld.", + count_text, + exc_info=None, + ) + except Exception: # noqa: BLE001 - sink errors must not escape reporting. + self.fallback() + finally: + _REPORTING.reset(token) + + async def run(self): + try: + # Public export; keep import failures inside the safe fallback. + from prompt_toolkit.application import run_in_terminal + + while self.pending or self.overflow: + delay = self.next_output - self.loop.time() + if delay > 0 and not self.draining: + await asyncio.sleep(delay) + batch = list(self.pending) + self.pending.clear() + overflow, self.overflow = self.overflow, 0 + await run_in_terminal( + partial(self.emit, batch, overflow), in_executor=True + ) + self.next_output = self.loop.time() + _INTERVAL + except Exception: # noqa: BLE001 - terminal/import errors stay private. + # Terminal rendering failed before reaching the log sink. Do not + # call the loop handler recursively or emit a raw worker traceback. + await asyncio.to_thread(self.fallback) + + async def close(self): + deadline = self.loop.time() + _DRAIN_TIMEOUT + self.draining = True # Do not spend the final drain budget on rate delay. + try: + while ( + self.pending + or self.overflow + or (self.worker is not None and not self.worker.done()) + ): + if self.loop.time() >= deadline: + break + if self.worker is None or self.worker.done(): + # Scheduling/output failed with retained snapshots. Report + # one fixed notice, within the same absolute drain budget. + try: + self.worker = self.loop.run_in_executor(None, self.fallback) + except Exception: # noqa: BLE001 - no safe sink remains. + return + _WORKERS.add(self.worker) + self.worker.add_done_callback(_observe) + self.pending.clear() + self.overflow = 0 + worker = self.worker + # Unlike wait_for, wait does not await cancellation completion. + await asyncio.wait({worker}, timeout=max(0, deadline - self.loop.time())) + # Recheck the current worker/queue: ready callbacks can replace + # the worker we just waited for before this coroutine resumes. + finally: + self.closed = True # Accept/count loop errors until the drain ends. + if self.worker is not None and not self.worker.done(): + self.worker.cancel() + self.pending.clear() + + +def report_loop_errors(function): + """Own loop reporting for the entire awaited lifetime, including cleanup.""" + + @wraps(function) + async def wrapped(*args, **kwargs): + loop = asyncio.get_running_loop() + previous = loop.get_exception_handler() + reporter = _Reporter(loop) + loop.set_exception_handler(reporter.handler) + try: + return await function(*args, **kwargs) + finally: + try: + await reporter.close() + finally: + if loop.get_exception_handler() is reporter.handler: + loop.set_exception_handler(previous) + + return wrapped diff --git a/amplifier_app_cli/main.py b/amplifier_app_cli/main.py index 8cd88319..e9e8e184 100644 --- a/amplifier_app_cli/main.py +++ b/amplifier_app_cli/main.py @@ -13,6 +13,7 @@ import uuid from collections.abc import Callable from datetime import UTC, datetime +from functools import partial, wraps from pathlib import Path from threading import RLock from typing import TYPE_CHECKING, Any @@ -3691,6 +3692,19 @@ def on_text_insert(buffer: Buffer) -> None: return session +def _with_loop_error_reporting(function): + """Keep passive imports lazy and the public wrapper in main's namespace.""" + + @wraps(function) + async def wrapped(*args, **kwargs): + from .loop_errors import report_loop_errors + + return await report_loop_errors(function)(*args, **kwargs) + + return wrapped + + +@_with_loop_error_reporting async def interactive_chat( config: dict, search_paths: list[Path], @@ -4183,7 +4197,11 @@ def sigint_handler(signum, frame): # freeze risk applies to any background Rich writes that # land while the user is composing input. with patch_stdout(): - user_input = await handoff.prompt(prompt_session.prompt_async) + user_input = await handoff.prompt( + partial( + prompt_session.prompt_async, set_exception_handler=False + ) + ) if user_input.lower() in ["exit", "quit"]: break diff --git a/amplifier_app_cli/session_handoff.py b/amplifier_app_cli/session_handoff.py index 85de1edb..8b5c99dc 100644 --- a/amplifier_app_cli/session_handoff.py +++ b/amplifier_app_cli/session_handoff.py @@ -199,20 +199,34 @@ async def prepare(self, request): async def prompt(self, factory): if self.requested.is_set(): raise EOFError - prompt = asyncio.create_task(factory()) + interrupted = object() + + async def read_prompt(): + try: + return await factory() + except KeyboardInterrupt: + # Task re-raises KeyboardInterrupt into the event loop before + # its awaiter can catch it. Carry an ordinary outcome instead. + return interrupted + + prompt = asyncio.create_task(read_prompt()) release = asyncio.create_task(self.requested.wait()) try: - done, _ = await asyncio.wait( - {prompt, release}, return_when=asyncio.FIRST_COMPLETED - ) - if release in done: - raise EOFError - return await prompt + await asyncio.wait({prompt, release}, return_when=asyncio.FIRST_COMPLETED) finally: for task in (prompt, release): if not task.done(): task.cancel() await asyncio.gather(prompt, release, return_exceptions=True) + # The release event can be set after wait's completion snapshot, or + # while the finally above awaits child cleanup. Deliver no prompt + # outcome until both children are observed and release is rechecked. + if self.requested.is_set(): + raise EOFError + result = prompt.result() + if result is interrupted: + raise KeyboardInterrupt + return result async def finish(self, save=None, after_cleanup=None): """Complete all persistence before either normal release or handoff.""" diff --git a/amplifier_app_cli/steering_input.py b/amplifier_app_cli/steering_input.py index 9e7be34d..523d7908 100644 --- a/amplifier_app_cli/steering_input.py +++ b/amplifier_app_cli/steering_input.py @@ -453,6 +453,9 @@ async def run(self) -> None: # _execute_with_interrupt and prevent the cancellation # token from being updated when Ctrl-C is pressed. handle_sigint=False, + # The CLI owns non-modal loop diagnostics throughout + # interactive_chat, including this prompt and cleanup. + set_exception_handler=False, ) ) diff --git a/docs/INTERACTIVE_MODE.md b/docs/INTERACTIVE_MODE.md index 50e5d7a6..97de3ba1 100644 --- a/docs/INTERACTIVE_MODE.md +++ b/docs/INTERACTIVE_MODE.md @@ -14,6 +14,12 @@ amplifier run Within an interactive session, you compose prompts and issue slash commands. Runtime mode switching is provided by the [`amplifier-bundle-modes`](https://github.com/microsoft/amplifier-bundle-modes) bundle (`/mode `, `/modes`, `/mode off`); other commands are app-CLI built-ins documented below. +Background errors during initialization, input, or cleanup produce bounded +diagnostics without opening a second prompt that asks you to press Enter. +Diagnostics show the exception type and source-frame locations; arbitrary error +messages and task details are withheld. Repeated errors are counted, and queue +overflow is reported. This reports errors; it does not repair SDK cleanup failures. + ## Slash Commands | Command | Purpose | Notes | diff --git a/tests/test_interactive_chat_tty_teardown.py b/tests/test_interactive_chat_tty_teardown.py index c989f1e8..5d80440f 100644 --- a/tests/test_interactive_chat_tty_teardown.py +++ b/tests/test_interactive_chat_tty_teardown.py @@ -29,6 +29,7 @@ from __future__ import annotations +import asyncio import contextlib from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch @@ -190,7 +191,7 @@ async def test_close_dedicated_tty_input_called_on_normal_exit( bundle_name="test-bundle", ) - mock_ps.prompt_async.assert_awaited_once() + mock_ps.prompt_async.assert_awaited_once_with(set_exception_handler=False) mock_close.assert_called_once() @pytest.mark.asyncio @@ -455,3 +456,80 @@ async def test_interactive_chat_teardown_does_not_raise_when_fd_never_opened( # having been opened -- the close call didn't short-circuit anything. mock_ps.prompt_async.assert_awaited_once() initialized.cleanup.assert_awaited_once() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("cleanup_fails", [False, True]) +async def test_loop_error_reporter_spans_public_chat_lifetime( + tmp_path: Path, cleanup_fails: bool +): + """The same owner must cover initialization, idle input, and cleanup.""" + from amplifier_app_cli.main import interactive_chat + + loop = asyncio.get_running_loop() + previous = loop.get_exception_handler() + observer = MagicMock() + owners = {} + session = _make_mock_session() + initialized = _make_initialized(session) + failure = RuntimeError("synthetic cleanup failure") + + async def initialize(*_args, **_kwargs): + owners["initialize"] = loop.get_exception_handler() + return initialized + + async def prompt(*_args, **_kwargs): + owners["prompt"] = loop.get_exception_handler() + raise EOFError + + async def cleanup(): + owners["cleanup"] = loop.get_exception_handler() + if cleanup_fails: + raise failure + + initialized.cleanup.side_effect = cleanup + mock_ps = MagicMock() + mock_ps.prompt_async = AsyncMock(side_effect=prompt) + mock_close = MagicMock() + + loop.set_exception_handler(observer) + try: + with ( + patch(f"{_MODULE}.create_initialized_session", new=initialize), + patch(f"{_MODULE}._create_prompt_session", return_value=mock_ps), + patch("amplifier_app_cli.incremental_save.register_incremental_save"), + patch(f"{_MODULE}.SessionStore") as MockStore, + patch(f"{_MODULE}.AppSettings"), + patch(f"{_MODULE}.console"), + patch(f"{_MODULE}.get_effective_config_summary"), + patch(f"{_MODULE}.close_dedicated_tty_input", new=mock_close), + ): + MockStore.return_value.get_metadata_if_exists.return_value = {} + chat = interactive_chat( + config={}, + search_paths=[tmp_path], + verbose=False, + bundle_name="test-bundle", + ) + if cleanup_fails: + with pytest.raises(RuntimeError) as caught: + await chat + assert caught.value is failure + else: + await chat + + assert set(owners) == {"initialize", "prompt", "cleanup"} + reporter = owners["initialize"] + assert reporter is not None and reporter is not observer, ( + "interactive_chat must own loop error reporting before initialization" + ) + assert owners["prompt"] is reporter + assert owners["cleanup"] is reporter, ( + "The reporter must remain active while session cleanup runs" + ) + assert loop.get_exception_handler() is observer + initialized.cleanup.assert_awaited_once() + session.execute.assert_not_awaited() + mock_close.assert_called_once() + finally: + loop.set_exception_handler(previous) diff --git a/tests/test_loop_errors.py b/tests/test_loop_errors.py new file mode 100644 index 00000000..0450775b --- /dev/null +++ b/tests/test_loop_errors.py @@ -0,0 +1,240 @@ +"""Contract tests for the planned CLI-owned loop reporter. + +The planned module is imported only during test execution. Its absence is an +explicit ExpectedMissingModule failure, not collection failure or a skip. +Actual pre-existing prompt regressions live in test_steering.py and +test_interactive_chat_tty_teardown.py; missing-module failures are not that RED. +""" + +from __future__ import annotations + +import asyncio +import gc +import importlib +import logging +import re +from unittest.mock import MagicMock + +import pytest + +_MODULE = "amplifier_app_cli.loop_errors" + + +@pytest.fixture +def report_loop_errors(): + """Public API assumption: @report_loop_errors wraps an async callable.""" + try: + module = importlib.import_module(_MODULE) + except ModuleNotFoundError as exc: + if exc.name != _MODULE: + raise + pytest.fail( + f"ExpectedMissingModule: {_MODULE}; reporter is not implemented yet", + pytrace=False, + ) + reporter = getattr(module, "report_loop_errors", None) + assert callable(reporter), "loop_errors must export report_loop_errors" + return reporter + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "outcome", ["return", "failure", "eof", "cancel", "new_owner"] +) +async def test_reporter_restores_only_its_own_handler(report_loop_errors, outcome): + loop = asyncio.get_running_loop() + original = loop.get_exception_handler() + previous = MagicMock() + newer = MagicMock() + value = object() + failure = ( + EOFError("synthetic EOF") + if outcome == "eof" + else RuntimeError("synthetic failure") + ) + entered = asyncio.Event() + + @report_loop_errors + async def operation(*, result): + assert loop.get_exception_handler() is not None + assert loop.get_exception_handler() is not previous + entered.set() + if outcome in {"failure", "eof"}: + raise failure + if outcome == "cancel": + await asyncio.Future() + if outcome == "new_owner": + loop.set_exception_handler(newer) + return result + + loop.set_exception_handler(previous) + task = asyncio.create_task(operation(result=value)) + try: + if outcome == "cancel": + await asyncio.wait_for(entered.wait(), timeout=3) + task.cancel("synthetic cancellation") + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, timeout=3) + elif outcome in {"failure", "eof"}: + with pytest.raises(type(failure)) as caught: + await asyncio.wait_for(task, timeout=3) + assert caught.value is failure + else: + assert await asyncio.wait_for(task, timeout=3) is value + assert loop.get_exception_handler() is ( + newer if outcome == "new_owner" else previous + ) + finally: + if not task.done(): + task.cancel() + try: + await asyncio.gather(task, return_exceptions=True) + finally: + loop.set_exception_handler(original) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("known_attribute", [False, True]) +@pytest.mark.parametrize("strip_exc_info", [False, True]) +async def test_reporter_logs_safe_type_and_origin_without_raw_context( + report_loop_errors, caplog, known_attribute, strip_exc_info +): + """Safe diagnostics must not depend on a traceback surviving log filters.""" + payload = "SYNTHETIC_PRIVATE_PAYLOAD\x1b[2J\r\n[bold]not markup[/bold]" + repr_calls = [] + + class PrivateContext: + def __repr__(self): + repr_calls.append(True) + raise AssertionError("raw loop context must never be rendered") + + class StripTraceback(logging.Filter): + def filter(self, record): + record.exc_info = None + record.exc_text = None + return True + + error = ( + AttributeError(payload, name="_async_httpx_client") + if known_attribute + else RuntimeError(payload) + ) + + async def aclose(): + raise error + + @report_loop_errors + async def operation(): + task = asyncio.create_task(aclose(), name=payload) + await asyncio.gather(task, return_exceptions=True) + asyncio.get_running_loop().call_exception_handler( + { + "message": payload, + "exception": error, + "future": task, + "transport": PrivateContext(), + } + ) + + caplog.set_level(logging.DEBUG) + log_filter = StripTraceback() + if strip_exc_info: + caplog.handler.addFilter(log_filter) + try: + await asyncio.wait_for(operation(), timeout=3) + finally: + if strip_exc_info: + caplog.handler.removeFilter(log_filter) + + records = [record for record in caplog.records if record.name == _MODULE] + assert records, "A background cleanup error must produce a diagnostic" + text = "\n".join(record.getMessage() for record in records) + assert type(error).__name__ in text + assert "aclose" in text, "Retain the safe coroutine origin" + if known_attribute: + assert "_async_httpx_client" in text, "Recognize the fixed safe SDK symptom" + assert "SYNTHETIC_PRIVATE_PAYLOAD" not in caplog.text + assert "\x1b" not in text and "\r" not in text + assert not repr_calls + assert all(record.exc_info is None for record in records) + assert all(record.exc_text is None for record in records) + + +@pytest.mark.asyncio +async def test_reporter_coalesces_a_repeat_flood_with_bounded_workers( + report_loop_errors, caplog +): + """A flood must retain a count, not schedule one worker/log per error.""" + error = RuntimeError("synthetic private detail") + + @report_loop_errors + async def operation(): + loop = asyncio.get_running_loop() + before = asyncio.all_tasks() + for _ in range(1000): + loop.call_exception_handler( + {"message": "synthetic private detail", "exception": error} + ) + # At most one reporter worker, before any scheduled worker can run. + assert len(asyncio.all_tasks() - before) <= 1 + + caplog.set_level(logging.DEBUG) + await asyncio.wait_for(operation(), timeout=3) + records = [record for record in caplog.records if record.name == _MODULE] + assert records, "Repeated failures must remain visible" + assert len(records) < 1000, "Repeated errors must be coalesced" + text = "\n".join(record.getMessage() for record in records) + assert re.search(r"\b(?:999|1000)\b", text), ( + "Report either the repeated count or the total count when the scope drains" + ) + assert "synthetic private detail" not in caplog.text + + +@pytest.mark.asyncio +async def test_reporter_observes_worker_failure_when_log_sink_raises( + report_loop_errors, monkeypatch, caplog +): + """A broken sink must not leave an unobserved failed reporting task.""" + loop = asyncio.get_running_loop() + original = loop.get_exception_handler() + observer = MagicMock() + workers = set() + sink_calls = [] + + class BrokenSink(logging.Handler): + def emit(self, record): + sink_calls.append(True) + raise RuntimeError("synthetic reporting sink failure") + + logger = logging.getLogger(_MODULE) + monkeypatch.setattr(logger, "handlers", [BrokenSink()]) + monkeypatch.setattr(logger, "propagate", False) + caplog.set_level(logging.DEBUG, logger=_MODULE) + + @report_loop_errors + async def operation(): + before = asyncio.all_tasks() + handler = loop.get_exception_handler() + assert handler is not None and handler is not observer + # Call the installed callback directly so asyncio cannot swallow a + # synchronous callback failure and hide it behind its default handler. + handler(loop, {"exception": RuntimeError("synthetic private detail")}) + workers.update(asyncio.all_tasks() - before) + assert len(workers) <= 1 + + loop.set_exception_handler(observer) + try: + await asyncio.wait_for(operation(), timeout=3) + assert sink_calls, "Exercise the failing reporting sink" + assert all(task.done() for task in workers), "Drain reporter work on exit" + assert loop.get_exception_handler() is observer + # Drop our last references, without retrieving exceptions on the + # reporter's behalf. CPython then reports any unobserved task failure. + workers.clear() + gc.collect() + await asyncio.sleep(0) # Dispatch already-ready task finalizers. + for call in observer.call_args_list: + context = call.args[1] + assert context.get("message") != "Task exception was never retrieved" + finally: + loop.set_exception_handler(original) \ No newline at end of file diff --git a/tests/test_loop_errors_bounds.py b/tests/test_loop_errors_bounds.py new file mode 100644 index 00000000..a55380e5 --- /dev/null +++ b/tests/test_loop_errors_bounds.py @@ -0,0 +1,176 @@ +"""Additional privacy, overload, and shutdown bounds not in the frozen RED set.""" + +import asyncio +import logging + +import pytest + +from amplifier_app_cli.loop_errors import report_loop_errors + +_LOGGER = "amplifier_app_cli.loop_errors" + + +@pytest.mark.asyncio +async def test_exception_metadata_cannot_invoke_hostile_overrides(caplog): + calls = [] + + def forbidden(*_args): + calls.append(True) + raise AssertionError("Do not invoke exception-controlled accessors") + + class HostileMeta(type): + __name__ = property(forbidden) + + class HostileError(RuntimeError, metaclass=HostileMeta): + __str__ = forbidden + __repr__ = forbidden + __getattribute__ = forbidden + + error = HostileError("SYNTHETIC_PRIVATE_ARGUMENT") + + @report_loop_errors + async def operation(): + asyncio.get_running_loop().call_exception_handler({"exception": error}) + + caplog.set_level(logging.DEBUG) + await operation() + assert not calls + assert "HostileError" in caplog.text + assert "SYNTHETIC_PRIVATE_ARGUMENT" not in caplog.text + + +@pytest.mark.asyncio +async def test_distinct_private_arguments_fill_bounded_queue_and_count_overflow(caplog): + @report_loop_errors + async def operation(): + loop = asyncio.get_running_loop() + for i in range(100): + loop.call_exception_handler( + {"exception": RuntimeError(f"SYNTHETIC_PRIVATE_ARGUMENT_{i}")} + ) + + caplog.set_level(logging.DEBUG) + await operation() + records = [r.getMessage() for r in caplog.records if r.name == _LOGGER] + assert len(records) == 17 # 16 snapshots plus one counted overflow notice. + assert all("count=1" in message for message in records[:16]) + assert "84 additional errors" in records[-1] + assert "SYNTHETIC_PRIVATE_ARGUMENT" not in caplog.text + assert all(len(message.encode("utf-8")) <= 2048 for message in records) + + +@pytest.mark.asyncio +async def test_batches_use_terminal_executor_and_are_rate_limited(monkeypatch, caplog): + emitted = asyncio.Queue() + times = [] + + async def terminal(function, *, in_executor): + assert in_executor is True + times.append(asyncio.get_running_loop().time()) + await asyncio.to_thread(function) + emitted.put_nowait(None) + + monkeypatch.setattr("prompt_toolkit.application.run_in_terminal", terminal) + + @report_loop_errors + async def operation(): + loop = asyncio.get_running_loop() + for _ in range(3): + loop.call_exception_handler({"exception": RuntimeError("synthetic")}) + await asyncio.wait_for(emitted.get(), timeout=2) + + caplog.set_level(logging.DEBUG) + await operation() + assert len(times) == 3 + assert all(later - earlier >= 0.5 for earlier, later in zip(times, times[1:])) + + +@pytest.mark.asyncio +async def test_stalled_reporter_drain_restores_owner_without_awaiting_cancel( + monkeypatch, +): + """A cancellation-resistant renderer must not hold the CLI's finally open.""" + started = asyncio.Event() + release = asyncio.Event() + finished = asyncio.Event() + loop = asyncio.get_running_loop() + original = loop.get_exception_handler() + workers = [] + + async def terminal(function, *, in_executor): + workers.append(asyncio.current_task()) + started.set() + try: + await release.wait() + except asyncio.CancelledError: + await release.wait() + finally: + finished.set() + + monkeypatch.setattr("prompt_toolkit.application.run_in_terminal", terminal) + + @report_loop_errors + async def operation(): + loop.call_exception_handler({"exception": RuntimeError("synthetic")}) + await started.wait() + + try: + begin = loop.time() + await asyncio.wait_for(operation(), timeout=1.5) + assert loop.time() - begin < 1.0 + assert loop.get_exception_handler() is original + finally: + release.set() + await asyncio.wait_for(finished.wait(), timeout=2) + await asyncio.wait_for(asyncio.gather(*workers), timeout=2) + + +@pytest.mark.asyncio +async def test_close_drains_replacement_worker_from_ready_callback(monkeypatch, caplog): + """A ready callback can replace the completed worker before close resumes.""" + loop = asyncio.get_running_loop() + original = loop.get_exception_handler() + workers = [] + late_callback_ran = [] + + def report_late(_completed): + late_callback_ran.append(True) + loop.call_exception_handler({"exception": ValueError("late private detail")}) + + async def terminal(function, *, in_executor): + assert in_executor is True + worker = asyncio.current_task() + workers.append(worker) + if len(workers) == 1: + # close's asyncio.wait completion callback is already registered. + # This callback runs after it, but before the waiting parent resumes. + worker.add_done_callback(report_late) + function() # Test-only sink: no terminal or executor I/O. + + monkeypatch.setattr("prompt_toolkit.application.run_in_terminal", terminal) + + @report_loop_errors + async def operation(): + loop.call_exception_handler({"exception": RuntimeError("first private detail")}) + # No yield: the decorator enters close before the worker starts. + + caplog.set_level(logging.DEBUG) + begin = loop.time() + try: + await asyncio.wait_for(operation(), timeout=1.5) + assert late_callback_ran == [True], "Exercise the accepted late error" + messages = [r.getMessage() for r in caplog.records if r.name == _LOGGER] + assert any("RuntimeError" in message for message in messages) + assert any("ValueError" in message for message in messages), ( + "close cancelled a replacement worker before draining its accepted error" + ) + assert len(workers) == 2 + assert all(worker.done() and not worker.cancelled() for worker in workers) + assert loop.time() - begin < 1.0, "Do not restart the 500 ms drain budget" + assert loop.get_exception_handler() is original + assert "private detail" not in caplog.text + finally: + for worker in workers: + if not worker.done(): + worker.cancel() + await asyncio.wait_for(asyncio.gather(*workers, return_exceptions=True), timeout=2) \ No newline at end of file diff --git a/tests/test_session_handoff.py b/tests/test_session_handoff.py index 8d23c19a..e4999eb7 100644 --- a/tests/test_session_handoff.py +++ b/tests/test_session_handoff.py @@ -193,3 +193,145 @@ async def test_interactive_unsupported_takeover_keeps_owner_and_reports_failure( assert "Session acquired" not in output.getvalue() finally: root.release() + + +@pytest.mark.parametrize("release_wins", [False, True]) +async def test_prompt_keyboard_interrupt_is_caught_in_parent_task(tmp_path, release_wins): + """Run in a child process: the pre-fix Task re-raise would abort pytest.""" + source = """ +import asyncio +import sys +from types import SimpleNamespace +from amplifier_app_cli.session_handoff import CLIHandoff + +async def main(): + release_wins = sys.argv[1] == 'True' + control = CLIHandoff(SimpleNamespace(), SimpleNamespace()) + before = asyncio.all_tasks() + async def factory(): + if release_wins: + control.requested.set() + raise KeyboardInterrupt + try: + await control.prompt(factory) + except KeyboardInterrupt: + assert not release_wins + print('PARENT_CAUGHT') + except EOFError: + assert release_wins + print('RELEASE_WON') + else: + raise AssertionError('Prompt did not propagate its outcome') + assert not (asyncio.all_tasks() - before), 'Handoff left pending children' + +asyncio.run(main()) +""" + process = await asyncio.create_subprocess_exec( + sys.executable, + "-B", + "-c", + source, + str(release_wins), + cwd=tmp_path, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + try: + stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=3) + finally: + if process.returncode is None: + process.kill() + await asyncio.wait_for(process.wait(), timeout=3) + + assert process.returncode == 0, stderr.decode(errors="replace") + assert stdout.decode().strip() == ("RELEASE_WON" if release_wins else "PARENT_CAUGHT") + assert not stderr, stderr.decode(errors="replace") + + +@pytest.mark.parametrize("outcome", ["value", "eof", "cancel"]) +async def test_prompt_preserves_value_eof_and_parent_cancellation(outcome): + control = CLIHandoff(SimpleNamespace(), Mock()) + entered = asyncio.Event() + prompt_children = [] + value = object() + + async def factory(): + prompt_children.append(asyncio.current_task()) + entered.set() + if outcome == "cancel": + await asyncio.Future() + if outcome == "eof": + raise EOFError + return value + + before = asyncio.all_tasks() + task = asyncio.create_task(control.prompt(factory)) + if outcome == "cancel": + await asyncio.wait_for(entered.wait(), timeout=1) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, timeout=1) + assert prompt_children[0].cancelled() + elif outcome == "eof": + with pytest.raises(EOFError): + await asyncio.wait_for(task, timeout=1) + else: + assert await asyncio.wait_for(task, timeout=1) is value + assert not (asyncio.all_tasks() - before), "Handoff left pending children" + + +@pytest.mark.parametrize("timing", ["ready_callback", "child_cleanup"]) +@pytest.mark.parametrize("outcome", ["interrupt", "value", "error"]) +async def test_late_release_wins_after_prompt_completion(timing, outcome): + """Release takes priority after the wait snapshot and during child cleanup.""" + control = CLIHandoff(SimpleNamespace(), Mock()) + release_points = [] + children = [] + + class ReleaseDuringCleanup(asyncio.Event): + async def wait(self): + children.append(asyncio.current_task()) + try: + return await super().wait() + except asyncio.CancelledError: + release_points.append("child_cleanup") + self.set() + await asyncio.sleep(0) # Keep the parent suspended in its gather. + raise + + if timing == "child_cleanup": + control.requested = ReleaseDuringCleanup() + + def request_release(completed): + assert completed.done() + release_points.append("ready_callback") + control.requested.set() + + async def factory(): + child = asyncio.current_task() + children.append(child) + if timing == "ready_callback": + # asyncio.wait has registered its completion callback first. That + # wakes the parent; this callback sets the event before it resumes, + # but the release-wait child is still in the ready queue. + child.add_done_callback(request_release) + if outcome == "interrupt": + raise KeyboardInterrupt + if outcome == "error": + raise RuntimeError("synthetic prompt failure") + return "must not be delivered after release" + + before = asyncio.all_tasks() + try: + await control.prompt(factory) + except EOFError: + assert control.requested.is_set() + except KeyboardInterrupt: + pytest.fail("Late release must prevent parent interrupt/exit confirmation") + except RuntimeError: + pytest.fail("Late release must take priority over the prompt error") + else: + pytest.fail("Late release must take priority over the prompt value") + assert release_points == [timing] + assert all(child.done() for child in children) + assert not (asyncio.all_tasks() - before), "Handoff left pending children" diff --git a/tests/test_steering.py b/tests/test_steering.py index c6c44282..b258812f 100644 --- a/tests/test_steering.py +++ b/tests/test_steering.py @@ -1,9 +1,9 @@ """Unit tests for mid-turn steering UX (app-cli side). Tests the SteeringInputManager introduced in the anchored-input + queued-badge -feature. All tests cover the logic layer (_enqueue, on_steering_injected, -_prompt_message, run() with injected input) so no real TTY or prompt_toolkit -app is required. +feature. Most tests cover the logic layer with injected input. The loop-handler +regression runs the real prompt_toolkit app with pipe input and DummyOutput; +no real TTY is required. Spec coverage ~~~~~~~~~~~~~ @@ -36,6 +36,7 @@ from __future__ import annotations import asyncio +from types import SimpleNamespace from typing import Any from unittest.mock import MagicMock, patch @@ -1757,3 +1758,78 @@ async def identity_expander(session, t): else: steer_cap.assert_not_called() assert manager.pending_count == 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("check", ["prompt_flags", "live_loop_handler"]) +async def test_real_steering_prompt_preserves_loop_error_owner(monkeypatch, check): + """Exercise the real Application, not the injected-input shortcut. + + prompt_toolkit's default exception handler opens a second, modal prompt + for background errors. Steering must leave the CLI's handler installed. + """ + import prompt_toolkit + from prompt_toolkit.input.defaults import create_pipe_input + from prompt_toolkit.output import DummyOutput + + from amplifier_app_cli.steering_input import _CtrlCInterrupt + + loop = asyncio.get_running_loop() + previous = loop.get_exception_handler() + observer = MagicMock() + ready = asyncio.Event() + calls = [] + constructors = [] + real_session = prompt_toolkit.PromptSession + real_prompt_async = real_session.prompt_async + stop = asyncio.Event() + manager = SteeringInputManager( + steer_cap=MagicMock(), + arbiter=None, + stop_event=stop, + console=MagicMock(), + session=SimpleNamespace( + coordinator=SimpleNamespace(cancellation=MagicMock()) + ), + ) + + async def prompt_async(self, *args, **kwargs): + calls.append(kwargs.copy()) + return await real_prompt_async(self, *args, **kwargs) + + def create_session(*args, **kwargs): + constructors.append(kwargs.copy()) + session = real_session(*args, output=DummyOutput(), **kwargs) + session.app.before_render += lambda _app: ready.set() + return session + + with create_pipe_input() as pipe: + monkeypatch.setattr( + "amplifier_app_cli.dedicated_tty_input.get_dedicated_tty_input", + lambda: pipe, + ) + monkeypatch.setattr(prompt_toolkit, "PromptSession", create_session) + monkeypatch.setattr(real_session, "prompt_async", prompt_async) + loop.set_exception_handler(observer) + task = asyncio.create_task(manager.run()) + try: + await asyncio.wait_for(ready.wait(), timeout=3) + assert not task.done(), "The real steering prompt must be active" + assert constructors[0]["interrupt_exception"] is _CtrlCInterrupt + assert calls[0]["handle_sigint"] is False + if check == "prompt_flags": + assert calls[0].get("set_exception_handler") is False + else: + assert loop.get_exception_handler() is observer, ( + "The active steering Application replaced the CLI loop " + "handler with prompt_toolkit's modal error handler" + ) + finally: + stop.set() + task.cancel() + try: + await asyncio.wait_for( + asyncio.gather(task, return_exceptions=True), timeout=3 + ) + finally: + loop.set_exception_handler(previous)