From 7aea6dfdce75ce224e9bd98f8235549ec2cb2bef Mon Sep 17 00:00:00 2001 From: Brian Krabach Date: Tue, 20 Oct 2026 22:34:51 -0700 Subject: [PATCH 1/2] Preserve provider ordinary-tool history options in native adapter --- .github/workflows/test.yml | 2 +- .../amplifier_loop_live_cli/bundle_astra.py | 7 +- .../computer_results.py | 8 ++- tests/test_native_history_options.py | 65 +++++++++++++++++++ 4 files changed, 75 insertions(+), 7 deletions(-) create mode 100644 tests/test_native_history_options.py diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 51d48dc..b015699 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -53,4 +53,4 @@ jobs: - name: Require the native provider and verify adapter boundaries offline run: | uv run --no-sync python -c 'from amplifier_module_provider_openai.native import NativeResponsesProvider' - uv run --no-sync pytest -q tests/test_native_provider_adapter.py tests/test_native_provider_budget_compatibility.py tests/test_native_job_attribution.py tests/test_native_adapter_fallback.py tests/test_image_budget.py + uv run --no-sync pytest -q tests/test_native_provider_adapter.py tests/test_native_history_options.py tests/test_native_provider_budget_compatibility.py tests/test_native_job_attribution.py tests/test_native_adapter_fallback.py tests/test_image_budget.py diff --git a/adapters/cli/amplifier_loop_live_cli/bundle_astra.py b/adapters/cli/amplifier_loop_live_cli/bundle_astra.py index c732da5..d2cfb0a 100644 --- a/adapters/cli/amplifier_loop_live_cli/bundle_astra.py +++ b/adapters/cli/amplifier_loop_live_cli/bundle_astra.py @@ -21,7 +21,8 @@ def wrap(cls, original): def _convert_messages(self, messages, **kwargs): from amplifier_loop_live_cli.computer_results import failed_computer_history,validate_computer_outputs - messages,_=failed_computer_history(messages) + messages,_=failed_computer_history( + messages, ordinary_names=kwargs.get("ordinary_computer_names", frozenset())) return validate_computer_outputs(super()._convert_messages(messages,**kwargs)) async def complete(self, request, **kwargs): @@ -47,9 +48,9 @@ def wrap(cls, original): return super().wrap(original, owner_getter=LIVE_OWNER.get, trace=trace, trace_context=trace_native_context) - def _prepare_native_messages(self, messages): + def _prepare_native_messages(self, messages, *, ordinary_names=frozenset()): from .computer_results import failed_computer_history - return failed_computer_history(messages)[0] + return failed_computer_history(messages, ordinary_names=ordinary_names)[0] def _validate_native_items(self, items): from .computer_results import validate_computer_outputs diff --git a/adapters/cli/amplifier_loop_live_cli/computer_results.py b/adapters/cli/amplifier_loop_live_cli/computer_results.py index 3c0bb13..43355de 100644 --- a/adapters/cli/amplifier_loop_live_cli/computer_results.py +++ b/adapters/cli/amplifier_loop_live_cli/computer_results.py @@ -39,23 +39,25 @@ def __init__(self, message): super().__init__(self.public_message, provider="openai", retryable=False) -def failed_computer_history(messages): +def failed_computer_history(messages, *, ordinary_names=frozenset()): """Represent an old failure as evidence only after a subsequent user message. The saved transcript is untouched. A failed native call cannot receive a made-up screenshot; the next user turn starts a new Responses lineage with an explicit failure record carrying the original identity and result. """ + # The provider owns native eligibility. An ordinary tool may share the + # computer name and actions-shaped arguments without returning screenshots. native = set() for message in messages: for call in message.get("tool_calls") or []: name=call.get("tool") or call.get("name") args=call.get("arguments") or call.get("input") or {} - if name=="computer" and isinstance(args,dict) and isinstance(args.get("actions"),list): + if name=="computer" and name not in ordinary_names and isinstance(args,dict) and isinstance(args.get("actions"),list): native.add(call.get("id")) content=message.get("content") for block in content if isinstance(content,list) else []: - if isinstance(block,dict) and block.get("type")=="tool_call" and block.get("name")=="computer" and isinstance((block.get("input") or {}).get("actions"),list): + if isinstance(block,dict) and block.get("type")=="tool_call" and block.get("name")=="computer" and block.get("name") not in ordinary_names and isinstance((block.get("input") or {}).get("actions"),list): native.add(block.get("id")) last_user=max((i for i,m in enumerate(messages) if m.get("role")=="user" and not (m.get("metadata") or {}).get("ephemeral")),default=-1) failures={} diff --git a/tests/test_native_history_options.py b/tests/test_native_history_options.py new file mode 100644 index 0000000..62a5de8 --- /dev/null +++ b/tests/test_native_history_options.py @@ -0,0 +1,65 @@ +"""Provider-owned ordinary-tool classification survives the optional CLI adapter.""" +import copy +import json +import sys +from pathlib import Path + +import pytest + +pytest.importorskip('amplifier_module_provider_openai.native') +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'adapters/cli')) +from amplifier_loop_live_cli.bundle_astra import BundleAstraProvider, SafeOpenAIProvider +from amplifier_loop_live_cli.computer_results import ComputerResultError, failed_computer_history +from amplifier_module_provider_openai import OpenAIProvider + + +def history(*, block, later_user): + call = {'id': 'ordinary-call', 'name': 'computer', 'arguments': {'actions': []}} + assistant = {'role': 'assistant', 'content': '', 'tool_calls': [call]} + if block: + assistant = {'role': 'assistant', 'content': [ + {'type': 'tool_call', 'id': call['id'], 'name': call['name'], 'input': call['arguments']}]} + messages = [{'role': 'user', 'content': 'Use the ordinary tool'}, assistant, + {'role': 'tool', 'tool_call_id': call['id'], + 'content': json.dumps({'success': False, 'error': 'ordinary tool failure'})}] + if later_user: + messages.append({'role': 'user', 'content': 'Explain that result'}) + return messages + + +@pytest.mark.parametrize('provider_type', [BundleAstraProvider, SafeOpenAIProvider]) +@pytest.mark.parametrize('block', [False, True]) +@pytest.mark.parametrize('later_user', [False, True]) +def test_actual_provider_serializer_preserves_ordinary_computer_result(provider_type, block, later_user): + messages = history(block=block, later_user=later_user) + original = copy.deepcopy(messages) + provider = provider_type.wrap(OpenAIProvider(api_key='fixture', config={'default_model': 'gpt-6-astra'})) + converted = provider._convert_messages(messages, ordinary_computer_names=frozenset({'computer'})) + output = [item for item in converted if item.get('type') == 'function_call_output'] + assert len(output) == 1 and output[0]['call_id'] == 'ordinary-call' + assert 'ordinary tool failure' in str(output[0]['output']) + assert not any(item.get('type') == 'computer_call_output' for item in converted) + assert messages == original + + +@pytest.mark.parametrize('block', [False, True]) +def test_native_failure_still_fails_closed_in_current_turn(block): + with pytest.raises(ComputerResultError, match='ordinary tool failure'): + failed_computer_history(history(block=block, later_user=False)) + + +@pytest.mark.parametrize('block', [False, True]) +def test_later_native_failure_is_reference_only_and_does_not_edit_transcript(block): + messages = history(block=block, later_user=True) + original = copy.deepcopy(messages) + projected, changed = failed_computer_history(messages) + assert changed and messages == original + assert any('Do not replay the prior action automatically' in str(m.get('content')) for m in projected) + assert not any(m.get('role') == 'tool' for m in projected) + +@pytest.mark.parametrize('block', [False, True]) +@pytest.mark.parametrize('later_user', [False, True]) +def test_native_projection_accepts_provider_owned_ordinary_names(block, later_user): + messages = history(block=block, later_user=later_user) + provider = BundleAstraProvider.wrap(OpenAIProvider(api_key='fixture', config={'default_model': 'gpt-6-astra'})) + assert provider._prepare_native_messages(messages, ordinary_names=frozenset({'computer'})) is messages From 62a4a0a7d01660b3bfcf52f4ae9f9676077ae4a2 Mon Sep 17 00:00:00 2001 From: Brian Krabach Date: Tue, 20 Oct 2026 22:31:09 -0700 Subject: [PATCH 2/2] Settle live owners when an inner turn is cancelled --- README.md | 6 + amplifier_module_loop_live/orchestrator.py | 6 + tests/test_turn_cancellation.py | 181 +++++++++++++++++++++ 3 files changed, 193 insertions(+) create mode 100644 tests/test_turn_cancellation.py diff --git a/README.md b/README.md index 045d354..0b3489e 100644 --- a/README.md +++ b/README.md @@ -57,6 +57,12 @@ approval policy, transport authentication, persistence location, and shutdown. Runtime history is bounded observation, not a replay log. Saved interrupted jobs are evidence and are never automatically dispatched again. +If a provider, hook, or host guard cancels an active turn, the live owner settles +through the same cancellation cleanup as Stop. It emits `generation.detached` +with cancellation unconfirmed and `session.closed` with cancelled status; it +does not report success or replay completed tools. Hosts must still reconcile +external effects before resuming interrupted work. + `JobStore.recover(transcript)` compares each saved call and result with the checkpoint. A returned job whose exact evidence is already present reconciles silently; the absence of an old display notice is not a recovery event. Missing diff --git a/amplifier_module_loop_live/orchestrator.py b/amplifier_module_loop_live/orchestrator.py index 9a5bc17..a52647c 100644 --- a/amplifier_module_loop_live/orchestrator.py +++ b/amplifier_module_loop_live/orchestrator.py @@ -241,6 +241,12 @@ async def turn(command, *, continuation=False): "" if continuation else self._text(command), turn_context, providers, tools, hooks, coordinator) await runtime.inbox.put(("bundle_turn", (result, None))) except asyncio.CancelledError: + # Cancellation can originate inside a provider, hook or host + # admission guard, without a Stop command in the live inbox. + # Wake the owner so awaiting this exact task propagates the + # cancellation through normal terminal cleanup instead of + # leaving an active generation waiting forever. + runtime.inbox.put_nowait(("bundle_turn", (None, None))) raise except Exception as exc: await runtime.inbox.put(("bundle_turn", (None, (exc, turn_failure(exc, stage))))) diff --git a/tests/test_turn_cancellation.py b/tests/test_turn_cancellation.py new file mode 100644 index 0000000..d6ef36b --- /dev/null +++ b/tests/test_turn_cancellation.py @@ -0,0 +1,181 @@ +"""Actual finite/live loop cancellation boundaries without a model service.""" +import asyncio +import json +from types import SimpleNamespace + +import pytest +from amplifier_core import AmplifierSession, ToolResult +from amplifier_core.message_models import ChatResponse, TextBlock, ToolCall +from amplifier_module_loop_live.runtime import Input, Runtime + + +class Provider: + name = 'fixture' + config = {} + + def __init__(self, *, completed_tool=False): + self.completed_tool = completed_tool + self.calls = 0 + self.entered = asyncio.Event() + self.release = asyncio.Event() + self.turn_done = asyncio.Event() + + async def complete(self, request, **kwargs): + self.calls += 1 + if self.completed_tool and self.calls == 1: + return ChatResponse(content=[], tool_calls=[ + ToolCall(id='completed-effect', name='record', arguments={})]) + self.entered.set() + await self.release.wait() + # The inner task really terminates; a timeout in the assertion below + # must not cancel the outer owner and manufacture terminal evidence. + asyncio.current_task().add_done_callback(lambda _task: self.turn_done.set()) + raise asyncio.CancelledError('private admission detail') + + def parse_tool_calls(self, response): + return response.tool_calls or [] + + def get_info(self): + return SimpleNamespace(default_model='fixture') + + +class Record: + name = 'record' + description = 'Record one completed effect.' + input_schema = {'type': 'object', 'properties': {}} + + def __init__(self): + self.calls = 0 + + async def execute(self, _input): + self.calls += 1 + return ToolResult(success=True, output={'saved': 'completed-effect'}) + + +async def open_session(*, live, completed_tool): + runtime = Runtime() + session = AmplifierSession({'session': { + 'orchestrator': {'module': 'loop-live', 'config': { + 'configured_bundle': True, 'min_delay_between_calls_ms': 0}}, + 'context': {'module': 'context-simple'}}, 'providers': []}, + session_id=runtime.session_id) + await session.initialize() + provider, tool = Provider(completed_tool=completed_tool), Record() + await session.coordinator.mount('providers', provider, name='fixture') + await session.coordinator.mount('tools', tool, name='record') + if live: + session.coordinator.register_capability('live.runtime', runtime) + return session, runtime, provider, tool + + +@pytest.mark.parametrize('completed_tool', [False, True]) +async def test_inner_cancel_settles_live_owner_without_success_or_replay(completed_tool): + session, runtime, provider, tool = await open_session(live=True, completed_tool=completed_tool) + task = asyncio.create_task(session.execute('')) + try: + await runtime.wait_for(lambda e: e['type'] == 'session.ready', 2) + await runtime.submit(Input('user', 'Do this once', id='original')) + await asyncio.wait_for(provider.entered.wait(), 2) + generation = runtime.generation['id'] + await runtime.submit_steering(Input('steer', 'Do not lose my guidance', id='held'), generation) + provider.release.set() + await asyncio.wait_for(provider.turn_done.wait(), 2) + done, _ = await asyncio.wait({task}, timeout=0.2) + assert task in done, 'The inner turn ended but its live owner is still waiting on an empty inbox' + with pytest.raises((asyncio.CancelledError, RuntimeError), match='private admission detail'): + await task + terminal = [e for e in runtime.events if e['type'] == 'generation.detached'] + assert len(terminal) == 1 + assert terminal[0]['generation_id'] == generation + assert terminal[0]['input_ids'] == ['original'] + assert terminal[0]['cancellation'] == 'unconfirmed' + closed = [e for e in runtime.events if e['type'] == 'session.closed'] + assert len(closed) == 1 and closed[0]['status'] == 'cancelled' + assert not any(e['type'] in {'generation.finished', 'generation.failed'} for e in runtime.events) + assert runtime.generation is None and runtime.closed + assert runtime.steering_outcomes['held']['disposition'] == 'held' + assert tool.calls == int(completed_tool) + assert provider.calls == 1 + int(completed_tool) + messages = await session.coordinator.get('context').get_messages() + if completed_tool: + assert any(m.get('role') == 'tool' and 'completed-effect' in str(m.get('content')) for m in messages) + assert 'private admission detail' not in json.dumps(runtime.events) + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + await session.cleanup() + + +async def test_explicit_stop_wins_cancelled_turn_race_without_another_turn(): + session, runtime, provider, tool = await open_session(live=True, completed_tool=True) + task = asyncio.create_task(session.execute('')) + try: + await runtime.wait_for(lambda e: e['type'] == 'session.ready', 2) + await runtime.submit(Input('user', 'Do this once', id='original')) + await asyncio.wait_for(provider.entered.wait(), 2) + await runtime.submit(Input('stop', target='cancel')) + provider.release.set() + await asyncio.wait_for(task, 2) + assert tool.calls == 1 and provider.calls == 2 + assert len([e for e in runtime.events if e['type'] == 'generation.detached']) == 1 + assert len([e for e in runtime.events if e['type'] == 'session.closed']) == 1 + assert runtime.events[-1]['status'] == 'cancelled' + assert not any(e['type'] == 'generation.finished' for e in runtime.events) + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + await session.cleanup() + + +async def test_finite_turn_still_propagates_cancellation_once(): + session, runtime, provider, tool = await open_session(live=False, completed_tool=True) + task = asyncio.create_task(session.execute('Do this once')) + try: + await asyncio.wait_for(provider.entered.wait(), 2) + provider.release.set() + with pytest.raises((asyncio.CancelledError, RuntimeError), match='private admission detail'): + await asyncio.wait_for(task, 2) + assert tool.calls == 1 and provider.calls == 2 + assert not runtime.events + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + await session.cleanup() + + + +async def test_cancelled_turn_checkpoint_retains_exact_host_activation(): + from test_activation import Ownership + + ownership = Ownership() + original = ownership.activate() + session, runtime, provider, tool = await open_session(live=True, completed_tool=True) + runtime.capture_activation = ownership.current + checkpoints = [] + + async def checkpoint(status=None): + ownership.check(ownership.current()) + checkpoints.append((ownership.current(), status)) + + session.coordinator.register_capability('live.activation', ownership) + session.coordinator.register_capability('live.checkpoint', checkpoint) + task = asyncio.create_task(session.execute('')) + try: + await runtime.wait_for(lambda e: e['type'] == 'session.ready', 2) + await runtime.submit(Input('user', 'Do this once', id='owned', activation=original)) + await asyncio.wait_for(provider.entered.wait(), 2) + provider.release.set() + await asyncio.wait_for(provider.turn_done.wait(), 2) + done, _ = await asyncio.wait({task}, timeout=0.2) + assert task in done + with pytest.raises((asyncio.CancelledError, RuntimeError), match='private admission detail'): + await task + assert checkpoints == [(original, 'cancelled')] + assert ownership.current_valid + assert tool.calls == 1 and provider.calls == 2 + assert [e['input_ids'] for e in runtime.events if e['type'] == 'generation.detached'] == [['owned']] + assert all('activation' not in e for e in runtime.events) + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + await session.cleanup()