Settle cancelled live turns and preserve native history options

.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

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

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

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={}

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)))))

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

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()