"""Cancelling a queued admission settles every observer kind in the authority, not per caller.""" import asyncio from types import SimpleNamespace import pytest from gateway.session_authority import LiveSession, SessionAuthority from gateway.session_contract import Principal, SessionRef, Submission from hermes_state import SessionDB from hermes_state_runtime import RuntimeStoreError, begin_runtime_epoch, list_session_admissions ACTOR = Principal('human', 'owned', frozenset({'session:submit', 'session:control'}), 'cli') REF = SessionRef('owned', 's') def _authority(tmp_path, monkeypatch, platform=None): monkeypatch.setenv('HERMES_HOME', str(tmp_path)) db = SessionDB(db_path=tmp_path / 'state.db') db.create_session('s', source='test') epoch = begin_runtime_epoch(db, instance_id='current') runner = SimpleNamespace(_draining=False, config=SimpleNamespace(multiplex_profiles=False), _adapter_for_source=lambda source: None) authority = SessionAuthority(runner, profile_id='owned', instance_id='current', db=db, epoch=epoch) authority.sessions['s'] = LiveSession(SimpleNamespace(platform=platform, user_id='human'), 'route') return db, authority def _submit(authority, request_id, text=None): return authority.submit(ACTOR, Submission(request_id=request_id, ref=REF, payload={'text': text or request_id}, intent='queue')) @pytest.mark.asyncio async def test_cancel_queued_settles_native_waiter_and_publishes_terminal_completion(tmp_path, monkeypatch): db, authority = _authority(tmp_path, monkeypatch) monkeypatch.setattr(authority, '_schedule', lambda ref: None) frames = [] with db: receipt = await _submit(authority, 'queued-then-cancelled') # The native ingress delivery waiter (session_ingress.admit_message) and an # event-stream observer (ACP prompt(), API run projections) both wait on this row. authority.native_waiters.add(receipt.admission_id) waiter = authority.waiters.setdefault(receipt.admission_id, asyncio.get_running_loop().create_future()) authority.sessions['s'].event_stream.observers.add(frames.append) cancelled = await authority.cancel_queued(ACTOR, REF, receipt.admission_id) assert (cancelled.status, cancelled.outcome) == ('terminal', 'cancelled') assert waiter.done() and receipt.admission_id not in authority.native_waiters assert receipt.admission_id not in authority.waiters terminal = [f['params'] for f in frames if f['params']['type'] == 'message.complete'] assert len(terminal) == 1, [f['params']['type'] for f in frames] assert terminal[0]['admission_id'] == receipt.admission_id assert terminal[0]['payload']['outcome'] == 'cancelled' assert terminal[0]['payload']['admission_id'] == receipt.admission_id # Idempotent: a repeated cancel of the terminal row is not a second completion. await authority.cancel_queued(ACTOR, REF, receipt.admission_id) assert len([f for f in frames if f['params']['type'] == 'message.complete']) == 1 @pytest.mark.asyncio async def test_cancelling_the_head_of_a_paused_fifo_resumes_its_successor(tmp_path, monkeypatch): from gateway.config import Platform from gateway import session_api_turn, session_finite db, authority = _authority(tmp_path, monkeypatch, platform=Platform.API_SERVER) executed = [] def check_api_turn(authority, ref, payload): if payload['text'] == 'blocked': raise RuntimeStoreError('permission_denied') monkeypatch.setattr(session_api_turn, 'check_api_turn', check_api_turn) async def execute(authority, ref, row): executed.append(row['payload']['text']) return 'done' monkeypatch.setattr(session_finite, 'execute_finite_admission', execute) with db: head = await _submit(authority, 'blocked') await _submit(authority, 'follower') # The drain pauses on the head's preclaim refusal and its task ends. paused = authority.sessions['s'].task await asyncio.wait_for(paused, 5) assert executed == [] await authority.cancel_queued(ACTOR, REF, head.admission_id) # cancel_queued awaits its off-loop media release after rescheduling, so the new drain # may already have finished: assert it is a NEW task, not that it is still running. task = authority.sessions['s'].task assert task is not None and task is not paused, 'cancelling the blocking head must reschedule the drain' await asyncio.wait_for(task, 5) assert executed == ['follower'] statuses = {r['request_id']: (r['status'], r['outcome']) for r in list_session_admissions(db, session_id='s', pending_only=False)} assert statuses == {'blocked': ('terminal', 'cancelled'), 'follower': ('terminal', 'completed')} @pytest.mark.asyncio async def test_hosted_head_cancelled_during_preclaim_does_not_let_successor_skip_its_check(tmp_path, monkeypatch): import threading from gateway import session_finite, session_hosted_transport, session_local_recovery from gateway.config import Platform db, authority = _authority(tmp_path, monkeypatch, platform=Platform.LOCAL) monkeypatch.setattr(session_local_recovery, 'restore_local_session', lambda authority, sid: None) validated, executed = [], [] head_checking, release_head = threading.Event(), threading.Event() def check_remote_hosted_admission(authority, ref, row): validated.append(row['request_id']) if row['request_id'] == 'hosted:A': head_checking.set() release_head.wait(5) return True # B's own reauthorization is revoked; it must be asked, never inherit A's verdict. raise RuntimeStoreError('permission_denied') monkeypatch.setattr(session_hosted_transport, 'check_remote_hosted_admission', check_remote_hosted_admission) async def execute(authority, ref, row): executed.append(row['request_id']) return 'done' monkeypatch.setattr(session_finite, 'execute_finite_admission', execute) with db: # Both rows are queued before the drain starts (admission commits off-loop, so the # first submit's drain would otherwise take the session lock before B is admitted). schedule, authority._schedule = authority._schedule, lambda ref: None head = await _submit(authority, 'hosted:A') await _submit(authority, 'hosted:B') authority._schedule = schedule schedule(REF) await asyncio.get_running_loop().run_in_executor(None, head_checking.wait, 5) await authority.cancel_queued(ACTOR, REF, head.admission_id) release_head.set() await asyncio.wait_for(authority.sessions['s'].task, 5) assert executed == [], 'the successor reached execution without its own hosted validation' assert validated == ['hosted:A', 'hosted:B'] statuses = {r['request_id']: r['status'] for r in list_session_admissions(db, session_id='s', pending_only=False)} assert statuses == {'hosted:A': 'terminal', 'hosted:B': 'queued'} @pytest.mark.asyncio async def test_head_cancelled_after_its_last_preclaim_check_does_not_hand_the_claim_to_its_successor( tmp_path, monkeypatch): """Cancellation commits from a worker thread, so it can land after the head passed every preclaim check (and was re-read as queued) while the off-loop claim waits for the writer: the claim takes the validated head or nothing, never the successor.""" import threading from gateway import session_authority, session_finite, session_hosted_transport, session_local_recovery from gateway.config import Platform db, authority = _authority(tmp_path, monkeypatch, platform=Platform.LOCAL) monkeypatch.setattr(session_local_recovery, 'restore_local_session', lambda authority, sid: None) monkeypatch.setattr(session_local_recovery, 'reopen_local_session', lambda authority, ref: None) validated, executed = [], [] claim_waiting, head_cancelled = threading.Event(), threading.Event() def check_remote_hosted_admission(authority, ref, row): validated.append(row['request_id']) if row['request_id'] == 'hosted:B': raise RuntimeStoreError('permission_denied') # B's own source binding is revoked return True monkeypatch.setattr(session_hosted_transport, 'check_remote_hosted_admission', check_remote_hosted_admission) claim = session_authority.claim_session_input def claim_behind_the_cancellation(*args, **kwargs): if not head_cancelled.is_set(): claim_waiting.set() head_cancelled.wait(5) return claim(*args, **kwargs) monkeypatch.setattr(session_authority, 'claim_session_input', claim_behind_the_cancellation) async def execute(authority, ref, row): executed.append(row['request_id']) return 'done' monkeypatch.setattr(session_finite, 'execute_finite_admission', execute) with db: schedule, authority._schedule = authority._schedule, lambda ref: None head = await _submit(authority, 'hosted:A') await _submit(authority, 'hosted:B') authority._schedule = schedule schedule(REF) assert await asyncio.get_running_loop().run_in_executor(None, claim_waiting.wait, 5) await authority.cancel_queued(ACTOR, REF, head.admission_id) head_cancelled.set() await asyncio.wait_for(authority.sessions['s'].task, 5) assert executed == [], 'the successor reached execution without its own hosted validation' assert validated == ['hosted:A', 'hosted:B'] statuses = {r['request_id']: r['status'] for r in list_session_admissions(db, session_id='s', pending_only=False)} assert statuses == {'hosted:A': 'terminal', 'hosted:B': 'queued'} def _wire_turn_agent(authority, generation, agent): """Run the real per-turn agent wiring with this authority as the turn's approval owner.""" from gateway.run_turn_runner import TurnRunner ctx = SimpleNamespace( source=SimpleNamespace(platform=None), # notification presentation reads ctx.source.platform mute_notification_reply=False, progress_callback=None, native_tool_start_callback=None, voice_ack_callback=None, _voice_ack_guild=[None], _native_slack_task_cards=False, native_tool_complete_callback=None, _step_callback_sync=None, _hooks_ref=SimpleNamespace(loaded_hooks=[]), _status_callback_sync=None, _event_callback_sync=None, _status_adapter=None, session_key='route', user_config={}, _thinking_enabled=False, agent_holder=[None], tools_holder=[None], process_task_id=None, process_baseline=None, run_generation=0, voice_turn=False) holder = SimpleNamespace( _ctx=ctx, _approval_owner=(authority, 's', generation), _runner=SimpleNamespace(_service_tier=None, _consume_pending_turn_sidecar_notes=lambda key: []), _make_bg_review_callbacks=lambda: (lambda message: None, lambda: None), _merge_turn_request_overrides=TurnRunner._merge_turn_request_overrides, _clarify_callback_sync=lambda *a, **k: None, _notice_callback_sync=lambda *a, **k: None, _attach_session_title_callback=lambda agent, ctx: None, combined_tool_start_callback=None, combined_tool_complete_callback=None) TurnRunner._wire_turn_agent_callbacks(holder, agent, {}, None, None, None, False) @pytest.mark.asyncio async def test_stop_before_the_turn_agent_exists_reaches_the_agent_once_it_is_wired(tmp_path, monkeypatch): from gateway import session_finite db, authority = _authority(tmp_path, monkeypatch) cache = {} authority.runner._cached_agent_for = cache.get agent = SimpleNamespace(interrupted=0) agent.interrupt = lambda *a, **k: setattr(agent, 'interrupted', agent.interrupted + 1) stopped = asyncio.Event() async def execute(authority, ref, row): # First turn: the claim is committed and message.start published, but agent # construction has not finished. Stop arrives in exactly this window. await asyncio.wait_for(stopped.wait(), 5) cache['route'] = agent _wire_turn_agent(authority, row['generation'], agent) return 'done' monkeypatch.setattr(session_finite, 'execute_finite_admission', execute) with db: await _submit(authority, 'first') task = authority.sessions['s'].task while authority._handle(REF).execution_state != 'running': await asyncio.sleep(0) handle = await authority.interrupt(ACTOR, REF, authority._handle(REF).execution_generation) assert handle.execution_state == 'running' stopped.set() await asyncio.wait_for(task, 5) assert agent.interrupted == 1, 'a Stop accepted for the running generation must reach its agent' # The latch belongs to that exact generation: the next turn's agent is not interrupted. agent.interrupted = 0 with db: await _submit(authority, 'second') stopped.set() await asyncio.wait_for(authority.sessions['s'].task, 5) assert agent.interrupted == 0 @pytest.mark.asyncio async def test_settlement_failure_is_logged_and_does_not_kill_the_drain(tmp_path, monkeypatch, caplog): """finish_result can raise (a concurrent reset/compression moved runtime_generation). The pump must survive that: the failure is logged with the admission id, the row is marked unknown rather than silently re-settled, and the drain task ends without an unretrieved exception so the FIFO can be re-armed. """ import logging from gateway import session_finite db, authority = _authority(tmp_path, monkeypatch) async def execute(authority, ref, row): # Simulate a concurrent lineage move while the turn ran: the settle fence # (settle_session_input) compares sessions.runtime_generation to the claim. authority.db._execute_write(lambda conn: conn.execute( "UPDATE sessions SET runtime_generation=runtime_generation+1 WHERE id='s'")) return 'done' monkeypatch.setattr(session_finite, 'execute_finite_admission', execute) with db, caplog.at_level(logging.ERROR, logger='gateway.session_authority'): head = await _submit(authority, 'moved-under-us') task = authority.sessions['s'].task await asyncio.wait_for(task, 5) assert task.exception() is None, 'a settlement failure must not escape the drain task' assert any(head.admission_id in rec.getMessage() for rec in caplog.records), \ [rec.getMessage() for rec in caplog.records] row, = [r for r in list_session_admissions(db, session_id='s', pending_only=False) if r['admission_id'] == head.admission_id] assert row['status'] == 'unknown', 'explicit recovery is possible without inventing a terminal result' assert authority.sessions['s'].event_stream.execution == {}, 'stamp cleared even on failure' @pytest.mark.asyncio async def test_storage_lock_during_claim_retries_in_place_and_keeps_waiters(tmp_path, monkeypatch, caplog): """A writer held past the claim's patience (or a full disk) is transient. The drain must not die with the committed head queued and its waiter parked forever: every _schedule caller is event-driven, so nothing would re-arm it. It retries in place and runs the head once.""" import logging import sqlite3 from gateway import session_authority, session_finite from tests.waiting import async_wait_until caplog.set_level(logging.WARNING, logger='gateway.session_authority') db, authority = _authority(tmp_path, monkeypatch) monkeypatch.setattr(db, '_WRITE_PATIENCE_S', 0.2) monkeypatch.setattr(session_authority, '_CLAIM_RETRY_MIN_S', 0.05, raising=False) ran = [] async def execute(authority, ref, row): ran.append(row['request_id']) return 'done' monkeypatch.setattr(session_finite, 'execute_finite_admission', execute) schedule = authority._schedule monkeypatch.setattr(authority, '_schedule', lambda ref: None) with db: head = await _submit(authority, 'head') waiter = authority.waiters.setdefault(head.admission_id, asyncio.get_running_loop().create_future()) blocker = sqlite3.connect(tmp_path / 'state.db', isolation_level=None) blocker.execute('BEGIN IMMEDIATE') try: schedule(REF) task = authority.sessions['s'].task # Two deferred claims: the drain hit the held writer and retried in place at least once. await async_wait_until(lambda: caplog.text.count('claim deferred by storage') >= 2 or task.done(), timeout=20, what='the drain to retry its claim behind the held writer') assert not task.done(), task.exception() if task.done() else None assert not waiter.done(), 'a transient storage error must not release accepted work' finally: blocker.execute('ROLLBACK') blocker.close() assert await asyncio.wait_for(waiter, 10) == 'done' await asyncio.wait_for(task, 5) assert ran == ['head'] row, = list_session_admissions(db, session_id='s', pending_only=False) assert (row['status'], row['outcome']) == ('terminal', 'completed') @pytest.mark.asyncio async def test_cancel_committed_before_a_failed_media_cleanup_still_settles_observers_once(tmp_path, monkeypatch): """Media collection is a separate write after the cancellation commits. Its failure must not withhold observer settlement, and an exact retry (which collects again) must neither leave the waiter parked nor publish a second completion.""" from gateway import session_ingress_media db, authority = _authority(tmp_path, monkeypatch) scheduled, frames = [], [] monkeypatch.setattr(authority, '_schedule', scheduled.append) collect = session_ingress_media.release_admission_media def fail_once(db, admission_id): monkeypatch.setattr(session_ingress_media, 'release_admission_media', collect) raise OSError('media cleanup write failed') with db: receipt = await _submit(authority, 'cancel-then-cleanup-fails') scheduled.clear() authority.native_waiters.add(receipt.admission_id) waiter = authority.waiters.setdefault(receipt.admission_id, asyncio.get_running_loop().create_future()) authority.sessions['s'].event_stream.observers.add(frames.append) monkeypatch.setattr(session_ingress_media, 'release_admission_media', fail_once) with pytest.raises(OSError, match='media cleanup'): await authority.cancel_queued(ACTOR, REF, receipt.admission_id) retried = await authority.cancel_queued(ACTOR, REF, receipt.admission_id) await authority.cancel_queued(ACTOR, REF, receipt.admission_id) assert (retried.status, retried.outcome) == ('terminal', 'cancelled') assert waiter.done() and receipt.admission_id not in authority.native_waiters completions = [f for f in frames if f['params']['type'] == 'message.complete'] assert len(completions) == 1 and completions[0]['params']['payload']['outcome'] == 'cancelled' assert len(scheduled) == 1, 'the successors of a paused head need exactly one fresh drain'