From 4c4fd6e31c40fa12ac9fff4f2b2fc1699d2bd9a0 Mon Sep 17 00:00:00 2001 From: Yichuan Wang <133667618+Charlie-Wang-03@users.noreply.github.com> Date: Sat, 5 Sep 2026 13:12:03 +0800 Subject: [PATCH 01/10] feat: journal tool execution lifecycle --- model_tools.py | 264 +++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 255 insertions(+), 9 deletions(-) diff --git a/model_tools.py b/model_tools.py index 8674ecc..50fed0e 100644 --- a/model_tools.py +++ b/model_tools.py @@ -2,9 +2,21 @@ from __future__ import annotations +import hashlib +import json import traceback +import uuid from typing import Dict, List, Optional, Sequence +from moonshine.utils import shorten, utc_now + + +TOOL_EXECUTION_STARTED = "tool_execution_started" +TOOL_EXECUTION_FINISHED = "tool_execution_finished" +TOOL_EXECUTION_AMBIGUOUS = "tool_execution_ambiguous" +TOOL_EXECUTION_BLOCKED = "tool_execution_blocked" + + def collect_tool_schemas( registry, mode: Optional[str] = None, @@ -16,19 +28,252 @@ def collect_tool_schemas( return registry.schemas(mode=mode, include=include, exclude=exclude) +def _execution_store(runtime: Dict[str, object]): + """Return the session store and id used for durable execution journaling.""" + runtime = dict(runtime or {}) + store = runtime.get("session_store") + session_id = str(runtime.get("session_id") or "").strip() + if store is None or not session_id: + return None, "" + return store, session_id + + +def _execution_payload(event: Dict[str, object]) -> Dict[str, object]: + """Return one execution-event payload defensively.""" + payload = event.get("payload") or {} + return dict(payload) if isinstance(payload, dict) else {} + + +def _unresolved_tool_executions(runtime: Dict[str, object]) -> List[Dict[str, object]]: + """Return tool executions that started but have no durable terminal record. + + ``tool_execution_ambiguous`` remains blocking by design: Moonshine cannot know + whether a side-effecting handler completed before the interruption, so retrying + automatically would risk duplicate external effects. + """ + store, session_id = _execution_store(runtime) + if store is None or not hasattr(store, "get_conversation_events"): + return [] + + active: Dict[str, Dict[str, object]] = {} + for event in store.get_conversation_events(session_id): + kind = str(event.get("event_kind") or "") + if kind not in {TOOL_EXECUTION_STARTED, TOOL_EXECUTION_FINISHED, TOOL_EXECUTION_AMBIGUOUS}: + continue + payload = _execution_payload(event) + execution_id = str(payload.get("execution_id") or "").strip() + if not execution_id: + continue + if kind == TOOL_EXECUTION_FINISHED: + active.pop(execution_id, None) + continue + record = dict(payload) + record["state"] = "ambiguous" if kind == TOOL_EXECUTION_AMBIGUOUS else "started" + record["event_id"] = event.get("id") + active[execution_id] = record + return list(active.values()) + + +def _append_execution_event( + runtime: Dict[str, object], + *, + event_kind: str, + content: str, + payload: Dict[str, object], +) -> None: + """Persist one execution lifecycle event when session storage is available.""" + store, session_id = _execution_store(runtime) + if store is None or not hasattr(store, "append_conversation_event"): + return + store.append_conversation_event( + session_id, + event_kind=event_kind, + role="tool", + content=content, + payload=dict(payload), + ) + + +def _begin_tool_execution(call: object, runtime: Dict[str, object]) -> str: + """Write a durable intent record immediately before dispatch.""" + execution_id = "tool-exec-%s" % uuid.uuid4().hex[:12] + tool_name = str(getattr(call, "name", "") or "") + call_id = str(getattr(call, "call_id", "") or "") + arguments = dict(getattr(call, "arguments", {}) or {}) + payload = { + "execution_id": execution_id, + "tool": tool_name, + "call_id": call_id, + "arguments": arguments, + "tool_round": runtime.get("_current_tool_round", ""), + "started_at": utc_now(), + } + _append_execution_event( + runtime, + event_kind=TOOL_EXECUTION_STARTED, + content="Tool execution started: %s (%s)" % (tool_name, call_id or execution_id), + payload=payload, + ) + return execution_id + + +def _finish_tool_execution( + call: object, + runtime: Dict[str, object], + execution_id: str, + *, + output: object, + error: Optional[str], +) -> None: + """Write the terminal lifecycle record before the next tool is dispatched.""" + rendered_output = json.dumps(output, ensure_ascii=False, sort_keys=True, default=str) + payload = { + "execution_id": execution_id, + "tool": str(getattr(call, "name", "") or ""), + "call_id": str(getattr(call, "call_id", "") or ""), + "tool_round": runtime.get("_current_tool_round", ""), + "outcome": "error" if error else "ok", + "error": shorten(str(error or ""), 500), + "output_preview": shorten(rendered_output, 1200), + "output_sha256": hashlib.sha256(rendered_output.encode("utf-8")).hexdigest(), + "finished_at": utc_now(), + } + _append_execution_event( + runtime, + event_kind=TOOL_EXECUTION_FINISHED, + content="Tool execution finished: %s (%s)" % (payload["tool"], payload["call_id"] or execution_id), + payload=payload, + ) + + +def _mark_tool_execution_ambiguous( + call: object, + runtime: Dict[str, object], + execution_id: str, + exc: BaseException, +) -> None: + """Record an interrupted dispatch whose external completion is unknowable.""" + now = utc_now() + payload = { + "execution_id": execution_id, + "tool": str(getattr(call, "name", "") or ""), + "call_id": str(getattr(call, "call_id", "") or ""), + "arguments": dict(getattr(call, "arguments", {}) or {}), + "tool_round": runtime.get("_current_tool_round", ""), + "state": "ambiguous", + "interruption_type": type(exc).__name__, + "interruption": shorten(str(exc), 500), + "interrupted_at": now, + } + _append_execution_event( + runtime, + event_kind=TOOL_EXECUTION_AMBIGUOUS, + content="Tool execution became ambiguous after interruption: %s (%s)" + % (payload["tool"], payload["call_id"] or execution_id), + payload=payload, + ) + + store, session_id = _execution_store(runtime) + if store is None: + return + if hasattr(store, "update_session_meta"): + store.update_session_meta( + session_id, + status="interrupted", + updated_at=now, + interrupted_tool_execution={ + "execution_id": execution_id, + "tool": payload["tool"], + "call_id": payload["call_id"], + "state": "ambiguous", + }, + ) + db = getattr(store, "db", None) + if db is not None and hasattr(db, "update_session"): + db.update_session(session_id, updated_at=now, status="interrupted") + + +def _blocked_results(calls: List[object], runtime: Dict[str, object], blockers: Sequence[Dict[str, object]]) -> List[Dict[str, object]]: + """Fail closed instead of dispatching new tools after an ambiguous execution.""" + first = dict(blockers[0]) if blockers else {} + blocker_tool = str(first.get("tool") or "unknown") + blocker_call_id = str(first.get("call_id") or "unknown") + blocker_execution_id = str(first.get("execution_id") or "unknown") + message = ( + "Tool dispatch is blocked because this session contains an interrupted tool execution " + "with ambiguous completion: tool=%s, call_id=%s, execution_id=%s. " + "Moonshine will not replay or dispatch additional tools automatically because the prior " + "handler may already have produced external side effects. Inspect the session records and " + "continue in a fresh session once the ambiguity is resolved." + % (blocker_tool, blocker_call_id, blocker_execution_id) + ) + results: List[Dict[str, object]] = [] + for call in calls: + result = { + "name": getattr(call, "name", ""), + "call_id": getattr(call, "call_id", ""), + "arguments": getattr(call, "arguments", {}), + "output": { + "status": "blocked_interrupted_execution", + "message": message, + "ambiguous_executions": [dict(item) for item in blockers], + }, + "error": message, + } + results.append(result) + runtime.setdefault("_tool_results_in_round", []).append(result) + _append_execution_event( + runtime, + event_kind=TOOL_EXECUTION_BLOCKED, + content="Blocked tool dispatch: %s" % str(getattr(call, "name", "") or ""), + payload={ + "tool": str(getattr(call, "name", "") or ""), + "call_id": str(getattr(call, "call_id", "") or ""), + "blocked_at": utc_now(), + "ambiguous_execution_ids": [str(item.get("execution_id") or "") for item in blockers], + }, + ) + return results + + def handle_function_calls(registry, calls: List[object], runtime: Dict[str, object]) -> List[Dict[str, object]]: - """Dispatch provider tool calls through the registry.""" + """Dispatch provider tool calls through the registry with crash-safe journaling. + + Each call is journaled immediately before dispatch and receives a durable + terminal record before the next call begins. If execution is interrupted by a + process-level exception such as ``KeyboardInterrupt``, the call is marked + ambiguous and the exception is re-raised. Future tool batches in the same + session fail closed rather than risk replaying a side effect whose completion + cannot be proven. + """ + blockers = _unresolved_tool_executions(runtime) + if blockers: + return _blocked_results(calls, runtime, blockers) + results = [] for call in calls: + execution_id = _begin_tool_execution(call, runtime) try: - result = registry.dispatch(call.name, call.arguments, runtime) - error = None - except Exception as exc: - result = { - "error": str(exc), - "traceback": traceback.format_exc(limit=3), - } - error = str(exc) + try: + result = registry.dispatch(call.name, call.arguments, runtime) + error = None + except Exception as exc: + result = { + "error": str(exc), + "traceback": traceback.format_exc(limit=3), + } + error = str(exc) + _finish_tool_execution( + call, + runtime, + execution_id, + output=result, + error=error, + ) + except BaseException as exc: + _mark_tool_execution_ambiguous(call, runtime, execution_id, exc) + raise + results.append( { "name": call.name, @@ -36,6 +281,7 @@ def handle_function_calls(registry, calls: List[object], runtime: Dict[str, obje "arguments": call.arguments, "output": result, "error": error, + "execution_id": execution_id, } ) runtime.setdefault("_tool_results_in_round", []).append(results[-1]) From 1e0c7e8cb5e955c2161ac317f609f56311a3f963 Mon Sep 17 00:00:00 2001 From: Yichuan Wang <133667618+Charlie-Wang-03@users.noreply.github.com> Date: Sat, 5 Sep 2026 13:12:53 +0800 Subject: [PATCH 02/10] test: cover interrupted tool recovery contract --- tests/test_interrupted_tool_recovery.py | 223 ++++++++++++++++++++++++ 1 file changed, 223 insertions(+) create mode 100644 tests/test_interrupted_tool_recovery.py diff --git a/tests/test_interrupted_tool_recovery.py b/tests/test_interrupted_tool_recovery.py new file mode 100644 index 0000000..3141585 --- /dev/null +++ b/tests/test_interrupted_tool_recovery.py @@ -0,0 +1,223 @@ +"""Regression tests for crash-safe tool execution journaling.""" + +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from moonshine.model_tools import ( + TOOL_EXECUTION_AMBIGUOUS, + TOOL_EXECUTION_BLOCKED, + TOOL_EXECUTION_FINISHED, + TOOL_EXECUTION_STARTED, + handle_function_calls, +) +from moonshine.moonshine_constants import MoonshinePaths +from moonshine.providers import ProviderToolCall +from moonshine.storage.session_store import SessionStore + + +class ScriptedRegistry(object): + """Minimal deterministic registry for execution-lifecycle tests.""" + + def __init__(self, handlers): + self.handlers = dict(handlers) + self.dispatches = [] + + def dispatch(self, name, arguments, runtime): + self.dispatches.append((name, dict(arguments or {}))) + handler = self.handlers[name] + return handler(runtime, **dict(arguments or {})) + + +class InterruptedToolRecoveryTest(unittest.TestCase): + def setUp(self): + self.temp_dir = tempfile.TemporaryDirectory() + self.addCleanup(self.temp_dir.cleanup) + self.paths = MoonshinePaths(Path(self.temp_dir.name)) + self.store = SessionStore(self.paths) + self.session_id = self.store.create_session("chat", "tool-recovery-test") + + def _runtime(self, store=None): + return { + "session_store": store or self.store, + "session_id": self.session_id, + "_tool_results_in_round": [], + } + + def _execution_events(self, store=None): + store = store or self.store + return [ + item + for item in store.get_conversation_events(self.session_id) + if str(item.get("event_kind") or "").startswith("tool_execution_") + ] + + def test_completed_call_is_terminal_before_later_call_is_interrupted(self): + side_effects = [] + + def complete(runtime, value): + side_effects.append("complete:%s" % value) + return {"value": value, "status": "done"} + + def interrupt(runtime): + side_effects.append("interrupt-side-effect") + raise KeyboardInterrupt("simulated process interruption") + + registry = ScriptedRegistry({"complete": complete, "interrupt": interrupt}) + runtime = self._runtime() + calls = [ + ProviderToolCall(name="complete", arguments={"value": 7}, call_id="call-complete"), + ProviderToolCall(name="interrupt", arguments={}, call_id="call-interrupt"), + ] + + with self.assertRaises(KeyboardInterrupt): + handle_function_calls(registry, calls, runtime) + + self.assertEqual(side_effects, ["complete:7", "interrupt-side-effect"]) + self.assertEqual([item["name"] for item in runtime["_tool_results_in_round"]], ["complete"]) + + events = self._execution_events() + complete_started = [ + item for item in events + if item["event_kind"] == TOOL_EXECUTION_STARTED + and item["payload"].get("call_id") == "call-complete" + ] + complete_finished = [ + item for item in events + if item["event_kind"] == TOOL_EXECUTION_FINISHED + and item["payload"].get("call_id") == "call-complete" + ] + interrupted_started = [ + item for item in events + if item["event_kind"] == TOOL_EXECUTION_STARTED + and item["payload"].get("call_id") == "call-interrupt" + ] + interrupted_ambiguous = [ + item for item in events + if item["event_kind"] == TOOL_EXECUTION_AMBIGUOUS + and item["payload"].get("call_id") == "call-interrupt" + ] + + self.assertEqual(len(complete_started), 1) + self.assertEqual(len(complete_finished), 1) + self.assertEqual( + complete_started[0]["payload"]["execution_id"], + complete_finished[0]["payload"]["execution_id"], + ) + self.assertEqual(complete_finished[0]["payload"]["outcome"], "ok") + self.assertTrue(complete_finished[0]["payload"]["output_sha256"]) + self.assertIn('"status": "done"', complete_finished[0]["payload"]["output_preview"]) + + self.assertEqual(len(interrupted_started), 1) + self.assertEqual(len(interrupted_ambiguous), 1) + self.assertEqual( + interrupted_started[0]["payload"]["execution_id"], + interrupted_ambiguous[0]["payload"]["execution_id"], + ) + self.assertEqual(interrupted_ambiguous[0]["payload"]["state"], "ambiguous") + self.assertEqual(self.store.get_session_meta(self.session_id)["status"], "interrupted") + + def test_restart_blocks_new_tool_dispatch_after_ambiguous_execution(self): + execution_id = "tool-exec-hard-crash" + self.store.append_conversation_event( + self.session_id, + event_kind=TOOL_EXECUTION_STARTED, + role="tool", + content="Tool execution started before simulated hard crash", + payload={ + "execution_id": execution_id, + "tool": "external_side_effect", + "call_id": "call-before-crash", + "arguments": {"value": 1}, + }, + ) + + restarted_store = SessionStore(self.paths) + dispatch_count = [] + + def must_not_run(runtime): + dispatch_count.append(1) + return {"unexpected": True} + + registry = ScriptedRegistry({"must_not_run": must_not_run}) + runtime = self._runtime(restarted_store) + results = handle_function_calls( + registry, + [ProviderToolCall(name="must_not_run", arguments={}, call_id="call-after-restart")], + runtime, + ) + + self.assertEqual(dispatch_count, []) + self.assertEqual(registry.dispatches, []) + self.assertEqual(len(results), 1) + self.assertEqual(results[0]["output"]["status"], "blocked_interrupted_execution") + self.assertIn(execution_id, results[0]["error"]) + self.assertEqual(restarted_store.get_session_meta(self.session_id)["status"], "interrupted") + blocked = [ + item for item in self._execution_events(restarted_store) + if item["event_kind"] == TOOL_EXECUTION_BLOCKED + ] + self.assertEqual(len(blocked), 1) + self.assertIn(execution_id, blocked[0]["payload"]["ambiguous_execution_ids"]) + + def test_ordinary_tool_error_is_terminal_and_does_not_poison_future_dispatch(self): + def fail(runtime): + raise RuntimeError("deterministic tool failure") + + failing_registry = ScriptedRegistry({"fail": fail}) + first = handle_function_calls( + failing_registry, + [ProviderToolCall(name="fail", arguments={}, call_id="call-fail")], + self._runtime(), + ) + self.assertEqual(len(first), 1) + self.assertIn("deterministic tool failure", first[0]["error"]) + + finish_events = [ + item for item in self._execution_events() + if item["event_kind"] == TOOL_EXECUTION_FINISHED + and item["payload"].get("call_id") == "call-fail" + ] + self.assertEqual(len(finish_events), 1) + self.assertEqual(finish_events[0]["payload"]["outcome"], "error") + + later_effects = [] + + def succeed(runtime): + later_effects.append("ran") + return {"ok": True} + + succeeding_registry = ScriptedRegistry({"succeed": succeed}) + second = handle_function_calls( + succeeding_registry, + [ProviderToolCall(name="succeed", arguments={}, call_id="call-succeed")], + self._runtime(), + ) + self.assertEqual(later_effects, ["ran"]) + self.assertIsNone(second[0]["error"]) + + def test_dispatch_without_session_store_keeps_legacy_behavior(self): + effects = [] + + def succeed(runtime, value): + effects.append(value) + return {"value": value} + + registry = ScriptedRegistry({"succeed": succeed}) + runtime = {"_tool_results_in_round": []} + results = handle_function_calls( + registry, + [ProviderToolCall(name="succeed", arguments={"value": 3}, call_id="call-no-store")], + runtime, + ) + + self.assertEqual(effects, [3]) + self.assertEqual(results[0]["output"], {"value": 3}) + self.assertIsNone(results[0]["error"]) + self.assertTrue(results[0]["execution_id"].startswith("tool-exec-")) + + +if __name__ == "__main__": + unittest.main() From a375df198abf0369349e56b1ab643991052240c5 Mon Sep 17 00:00:00 2001 From: Yichuan Wang <133667618+Charlie-Wang-03@users.noreply.github.com> Date: Sat, 5 Sep 2026 13:13:28 +0800 Subject: [PATCH 03/10] fix: fail closed on unresolved tool executions --- model_tools.py | 63 ++++++++++++++++++++++++++++++++++---------------- 1 file changed, 43 insertions(+), 20 deletions(-) diff --git a/model_tools.py b/model_tools.py index 50fed0e..0eb4954 100644 --- a/model_tools.py +++ b/model_tools.py @@ -94,6 +94,24 @@ def _append_execution_event( ) +def _mark_session_interrupted(runtime: Dict[str, object], interruption: Dict[str, object]) -> None: + """Expose interrupted execution state in both session metadata stores.""" + store, session_id = _execution_store(runtime) + if store is None: + return + now = utc_now() + if hasattr(store, "update_session_meta"): + store.update_session_meta( + session_id, + status="interrupted", + updated_at=now, + interrupted_tool_execution=dict(interruption), + ) + db = getattr(store, "db", None) + if db is not None and hasattr(db, "update_session"): + db.update_session(session_id, updated_at=now, status="interrupted") + + def _begin_tool_execution(call: object, runtime: Dict[str, object]) -> str: """Write a durable intent record immediately before dispatch.""" execution_id = "tool-exec-%s" % uuid.uuid4().hex[:12] @@ -172,25 +190,15 @@ def _mark_tool_execution_ambiguous( % (payload["tool"], payload["call_id"] or execution_id), payload=payload, ) - - store, session_id = _execution_store(runtime) - if store is None: - return - if hasattr(store, "update_session_meta"): - store.update_session_meta( - session_id, - status="interrupted", - updated_at=now, - interrupted_tool_execution={ - "execution_id": execution_id, - "tool": payload["tool"], - "call_id": payload["call_id"], - "state": "ambiguous", - }, - ) - db = getattr(store, "db", None) - if db is not None and hasattr(db, "update_session"): - db.update_session(session_id, updated_at=now, status="interrupted") + _mark_session_interrupted( + runtime, + { + "execution_id": execution_id, + "tool": payload["tool"], + "call_id": payload["call_id"], + "state": "ambiguous", + }, + ) def _blocked_results(calls: List[object], runtime: Dict[str, object], blockers: Sequence[Dict[str, object]]) -> List[Dict[str, object]]: @@ -199,6 +207,15 @@ def _blocked_results(calls: List[object], runtime: Dict[str, object], blockers: blocker_tool = str(first.get("tool") or "unknown") blocker_call_id = str(first.get("call_id") or "unknown") blocker_execution_id = str(first.get("execution_id") or "unknown") + _mark_session_interrupted( + runtime, + { + "execution_id": blocker_execution_id, + "tool": blocker_tool, + "call_id": blocker_call_id, + "state": str(first.get("state") or "ambiguous"), + }, + ) message = ( "Tool dispatch is blocked because this session contains an interrupted tool execution " "with ambiguous completion: tool=%s, call_id=%s, execution_id=%s. " @@ -271,7 +288,13 @@ def handle_function_calls(registry, calls: List[object], runtime: Dict[str, obje error=error, ) except BaseException as exc: - _mark_tool_execution_ambiguous(call, runtime, execution_id, exc) + try: + _mark_tool_execution_ambiguous(call, runtime, execution_id, exc) + except Exception: + # Never replace the process-level interruption with a best-effort + # journaling failure. A durable start record, when it was written, + # is itself enough for the next process to fail closed. + pass raise results.append( From bd89c0eaa30f53d21897eda22162bff147d64006 Mon Sep 17 00:00:00 2001 From: Yichuan Wang <133667618+Charlie-Wang-03@users.noreply.github.com> Date: Sat, 5 Sep 2026 13:16:12 +0800 Subject: [PATCH 04/10] refactor: guard interrupted tool-bearing turns --- model_tools.py | 166 ++++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 136 insertions(+), 30 deletions(-) diff --git a/model_tools.py b/model_tools.py index 0eb4954..790cbc3 100644 --- a/model_tools.py +++ b/model_tools.py @@ -8,7 +8,7 @@ import uuid from typing import Dict, List, Optional, Sequence -from moonshine.utils import shorten, utc_now +from moonshine.utils import read_jsonl, shorten, utc_now TOOL_EXECUTION_STARTED = "tool_execution_started" @@ -38,6 +38,20 @@ def _execution_store(runtime: Dict[str, object]): return store, session_id +def _render_json(value: object) -> str: + """Render a deterministic JSON-ish representation for hashes and previews.""" + return json.dumps(value, ensure_ascii=False, sort_keys=True, default=str) + + +def _fingerprint(value: object, preview_chars: int = 800) -> Dict[str, str]: + """Return bounded trace metadata without duplicating large tool payloads.""" + rendered = _render_json(value) + return { + "sha256": hashlib.sha256(rendered.encode("utf-8")).hexdigest(), + "preview": shorten(rendered, preview_chars), + } + + def _execution_payload(event: Dict[str, object]) -> Dict[str, object]: """Return one execution-event payload defensively.""" payload = event.get("payload") or {} @@ -45,7 +59,7 @@ def _execution_payload(event: Dict[str, object]) -> Dict[str, object]: def _unresolved_tool_executions(runtime: Dict[str, object]) -> List[Dict[str, object]]: - """Return tool executions that started but have no durable terminal record. + """Return executions that started but cannot be proven terminal. ``tool_execution_ambiguous`` remains blocking by design: Moonshine cannot know whether a side-effecting handler completed before the interruption, so retrying @@ -74,6 +88,79 @@ def _unresolved_tool_executions(runtime: Dict[str, object]) -> List[Dict[str, ob return list(active.values()) +def _prior_interrupted_tool_turn(runtime: Dict[str, object]) -> Optional[Dict[str, object]]: + """Return a prior open turn that had tool execution before a later turn began. + + A finished handler is not enough to declare the *turn* durable. The process can + die after the handler returns but before Moonshine records the tool result and + final assistant message. Once a later turn has started, such a prior tool-bearing + open turn is treated as interrupted and all new tool dispatch fails closed. + """ + store, session_id = _execution_store(runtime) + paths = getattr(store, "paths", None) if store is not None else None + if store is None or paths is None or not hasattr(store, "get_conversation_events"): + return None + + turn_events = [ + item + for item in read_jsonl(paths.session_turn_events_file(session_id)) + if isinstance(item, dict) and str(item.get("type") or "") in {"turn_started", "turn_completed"} + ] + open_turns: List[Dict[str, object]] = [] + all_starts: List[Dict[str, object]] = [] + for item in turn_events: + if str(item.get("type") or "") == "turn_started": + open_turns.append(item) + all_starts.append(item) + elif open_turns: + # A completion belongs to the most recently started live turn. This + # preserves an older interrupted turn if a later resumed turn completes. + open_turns.pop() + + # During a normal dispatch the current turn itself is open. Only older open + # turns are recovery hazards. + if len(open_turns) <= 1: + return None + + conversation_events = store.get_conversation_events(session_id) + execution_starts = [ + item + for item in conversation_events + if str(item.get("event_kind") or "") == TOOL_EXECUTION_STARTED + ] + start_times = [str(item.get("created_at") or "") for item in all_starts] + + for open_turn in open_turns[:-1]: + started_at = str(open_turn.get("created_at") or "") + if not started_at: + continue + try: + start_index = start_times.index(started_at) + except ValueError: + continue + next_started_at = start_times[start_index + 1] if start_index + 1 < len(start_times) else "" + matching = [] + for event in execution_starts: + created_at = str(event.get("created_at") or "") + if created_at < started_at: + continue + if next_started_at and created_at >= next_started_at: + continue + matching.append(event) + if not matching: + continue + first_payload = _execution_payload(matching[0]) + return { + "state": "interrupted_turn", + "execution_id": str(first_payload.get("execution_id") or "turn:%s" % started_at), + "tool": str(first_payload.get("tool") or "unknown"), + "call_id": str(first_payload.get("call_id") or ""), + "turn_started_at": started_at, + "tool_execution_count": len(matching), + } + return None + + def _append_execution_event( runtime: Dict[str, object], *, @@ -118,11 +205,13 @@ def _begin_tool_execution(call: object, runtime: Dict[str, object]) -> str: tool_name = str(getattr(call, "name", "") or "") call_id = str(getattr(call, "call_id", "") or "") arguments = dict(getattr(call, "arguments", {}) or {}) + arguments_fingerprint = _fingerprint(arguments) payload = { "execution_id": execution_id, "tool": tool_name, "call_id": call_id, - "arguments": arguments, + "arguments_sha256": arguments_fingerprint["sha256"], + "arguments_preview": arguments_fingerprint["preview"], "tool_round": runtime.get("_current_tool_round", ""), "started_at": utc_now(), } @@ -144,7 +233,7 @@ def _finish_tool_execution( error: Optional[str], ) -> None: """Write the terminal lifecycle record before the next tool is dispatched.""" - rendered_output = json.dumps(output, ensure_ascii=False, sort_keys=True, default=str) + output_fingerprint = _fingerprint(output, preview_chars=1200) payload = { "execution_id": execution_id, "tool": str(getattr(call, "name", "") or ""), @@ -152,8 +241,8 @@ def _finish_tool_execution( "tool_round": runtime.get("_current_tool_round", ""), "outcome": "error" if error else "ok", "error": shorten(str(error or ""), 500), - "output_preview": shorten(rendered_output, 1200), - "output_sha256": hashlib.sha256(rendered_output.encode("utf-8")).hexdigest(), + "output_preview": output_fingerprint["preview"], + "output_sha256": output_fingerprint["sha256"], "finished_at": utc_now(), } _append_execution_event( @@ -172,11 +261,13 @@ def _mark_tool_execution_ambiguous( ) -> None: """Record an interrupted dispatch whose external completion is unknowable.""" now = utc_now() + arguments_fingerprint = _fingerprint(dict(getattr(call, "arguments", {}) or {})) payload = { "execution_id": execution_id, "tool": str(getattr(call, "name", "") or ""), "call_id": str(getattr(call, "call_id", "") or ""), - "arguments": dict(getattr(call, "arguments", {}) or {}), + "arguments_sha256": arguments_fingerprint["sha256"], + "arguments_preview": arguments_fingerprint["preview"], "tool_round": runtime.get("_current_tool_round", ""), "state": "ambiguous", "interruption_type": type(exc).__name__, @@ -202,28 +293,42 @@ def _mark_tool_execution_ambiguous( def _blocked_results(calls: List[object], runtime: Dict[str, object], blockers: Sequence[Dict[str, object]]) -> List[Dict[str, object]]: - """Fail closed instead of dispatching new tools after an ambiguous execution.""" + """Fail closed instead of dispatching new tools after an interrupted execution.""" first = dict(blockers[0]) if blockers else {} blocker_tool = str(first.get("tool") or "unknown") blocker_call_id = str(first.get("call_id") or "unknown") blocker_execution_id = str(first.get("execution_id") or "unknown") + blocker_state = str(first.get("state") or "ambiguous") _mark_session_interrupted( runtime, { "execution_id": blocker_execution_id, "tool": blocker_tool, "call_id": blocker_call_id, - "state": str(first.get("state") or "ambiguous"), + "state": blocker_state, }, ) + if blocker_state == "interrupted_turn": + reason = ( + "a prior tool-bearing turn was interrupted before Moonshine durably completed the turn" + ) + else: + reason = "a prior tool execution has ambiguous completion" message = ( - "Tool dispatch is blocked because this session contains an interrupted tool execution " - "with ambiguous completion: tool=%s, call_id=%s, execution_id=%s. " - "Moonshine will not replay or dispatch additional tools automatically because the prior " - "handler may already have produced external side effects. Inspect the session records and " - "continue in a fresh session once the ambiguity is resolved." - % (blocker_tool, blocker_call_id, blocker_execution_id) + "Tool dispatch is blocked because %s: tool=%s, call_id=%s, execution_id=%s. " + "Moonshine will not replay or dispatch additional tools automatically because prior " + "handlers may already have produced external side effects. Inspect the session records " + "and continue in a fresh session once the ambiguity is resolved." + % (reason, blocker_tool, blocker_call_id, blocker_execution_id) ) + public_blockers = [ + { + key: item.get(key) + for key in ("state", "execution_id", "tool", "call_id", "turn_started_at", "started_at", "interrupted_at") + if item.get(key) not in {None, ""} + } + for item in blockers + ] results: List[Dict[str, object]] = [] for call in calls: result = { @@ -233,7 +338,7 @@ def _blocked_results(calls: List[object], runtime: Dict[str, object], blockers: "output": { "status": "blocked_interrupted_execution", "message": message, - "ambiguous_executions": [dict(item) for item in blockers], + "ambiguous_executions": public_blockers, }, "error": message, } @@ -247,6 +352,7 @@ def _blocked_results(calls: List[object], runtime: Dict[str, object], blockers: "tool": str(getattr(call, "name", "") or ""), "call_id": str(getattr(call, "call_id", "") or ""), "blocked_at": utc_now(), + "blocker_states": [str(item.get("state") or "") for item in blockers], "ambiguous_execution_ids": [str(item.get("execution_id") or "") for item in blockers], }, ) @@ -259,11 +365,13 @@ def handle_function_calls(registry, calls: List[object], runtime: Dict[str, obje Each call is journaled immediately before dispatch and receives a durable terminal record before the next call begins. If execution is interrupted by a process-level exception such as ``KeyboardInterrupt``, the call is marked - ambiguous and the exception is re-raised. Future tool batches in the same - session fail closed rather than risk replaying a side effect whose completion - cannot be proven. + ambiguous and the exception is re-raised. A later tool-bearing turn also stays + blocked if an earlier tool-bearing turn never reached ``turn_completed``. """ blockers = _unresolved_tool_executions(runtime) + interrupted_turn = _prior_interrupted_tool_turn(runtime) + if interrupted_turn is not None: + blockers.append(interrupted_turn) if blockers: return _blocked_results(calls, runtime, blockers) @@ -297,15 +405,13 @@ def handle_function_calls(registry, calls: List[object], runtime: Dict[str, obje pass raise - results.append( - { - "name": call.name, - "call_id": getattr(call, "call_id", ""), - "arguments": call.arguments, - "output": result, - "error": error, - "execution_id": execution_id, - } - ) - runtime.setdefault("_tool_results_in_round", []).append(results[-1]) + result_record = { + "name": call.name, + "call_id": getattr(call, "call_id", ""), + "arguments": call.arguments, + "output": result, + "error": error, + } + results.append(result_record) + runtime.setdefault("_tool_results_in_round", []).append(result_record) return results From a93a262bf6148fd06fc3453f93958ad68e9ab10e Mon Sep 17 00:00:00 2001 From: Yichuan Wang <133667618+Charlie-Wang-03@users.noreply.github.com> Date: Sat, 5 Sep 2026 13:17:01 +0800 Subject: [PATCH 05/10] fix: use monotonic turn sequence for recovery --- model_tools.py | 107 ++++++++++++++++++++++--------------------------- 1 file changed, 49 insertions(+), 58 deletions(-) diff --git a/model_tools.py b/model_tools.py index 790cbc3..ba69b10 100644 --- a/model_tools.py +++ b/model_tools.py @@ -58,6 +58,30 @@ def _execution_payload(event: Dict[str, object]) -> Dict[str, object]: return dict(payload) if isinstance(payload, dict) else {} +def _turn_lifecycle(runtime: Dict[str, object]) -> Dict[str, object]: + """Return monotonically numbered turn starts and currently open sequences.""" + store, session_id = _execution_store(runtime) + paths = getattr(store, "paths", None) if store is not None else None + if paths is None: + return {"sequence": 0, "open_sequences": []} + turn_events = [ + item + for item in read_jsonl(paths.session_turn_events_file(session_id)) + if isinstance(item, dict) and str(item.get("type") or "") in {"turn_started", "turn_completed"} + ] + sequence = 0 + open_sequences: List[int] = [] + for item in turn_events: + if str(item.get("type") or "") == "turn_started": + sequence += 1 + open_sequences.append(sequence) + elif open_sequences: + # A resumed turn can complete while an older interrupted turn remains + # unresolved, so pair completions with the most recent live start. + open_sequences.pop() + return {"sequence": sequence, "open_sequences": open_sequences} + + def _unresolved_tool_executions(runtime: Dict[str, object]) -> List[Dict[str, object]]: """Return executions that started but cannot be proven terminal. @@ -89,74 +113,40 @@ def _unresolved_tool_executions(runtime: Dict[str, object]) -> List[Dict[str, ob def _prior_interrupted_tool_turn(runtime: Dict[str, object]) -> Optional[Dict[str, object]]: - """Return a prior open turn that had tool execution before a later turn began. + """Return a prior open turn that executed tools before a later turn began. A finished handler is not enough to declare the *turn* durable. The process can die after the handler returns but before Moonshine records the tool result and - final assistant message. Once a later turn has started, such a prior tool-bearing - open turn is treated as interrupted and all new tool dispatch fails closed. + final assistant message. Each execution intent stores its monotonic turn + sequence, so recovery does not rely on coarse wall-clock timestamps. """ store, session_id = _execution_store(runtime) - paths = getattr(store, "paths", None) if store is not None else None - if store is None or paths is None or not hasattr(store, "get_conversation_events"): + if store is None or not hasattr(store, "get_conversation_events"): return None + lifecycle = _turn_lifecycle(runtime) + open_sequences = [int(item) for item in list(lifecycle.get("open_sequences") or [])] - turn_events = [ - item - for item in read_jsonl(paths.session_turn_events_file(session_id)) - if isinstance(item, dict) and str(item.get("type") or "") in {"turn_started", "turn_completed"} - ] - open_turns: List[Dict[str, object]] = [] - all_starts: List[Dict[str, object]] = [] - for item in turn_events: - if str(item.get("type") or "") == "turn_started": - open_turns.append(item) - all_starts.append(item) - elif open_turns: - # A completion belongs to the most recently started live turn. This - # preserves an older interrupted turn if a later resumed turn completes. - open_turns.pop() - - # During a normal dispatch the current turn itself is open. Only older open + # During normal dispatch the current turn itself is open. Only older open # turns are recovery hazards. - if len(open_turns) <= 1: + if len(open_sequences) <= 1: return None - - conversation_events = store.get_conversation_events(session_id) - execution_starts = [ - item - for item in conversation_events - if str(item.get("event_kind") or "") == TOOL_EXECUTION_STARTED - ] - start_times = [str(item.get("created_at") or "") for item in all_starts] - - for open_turn in open_turns[:-1]: - started_at = str(open_turn.get("created_at") or "") - if not started_at: + prior_sequences = set(open_sequences[:-1]) + for event in store.get_conversation_events(session_id): + if str(event.get("event_kind") or "") != TOOL_EXECUTION_STARTED: continue + payload = _execution_payload(event) try: - start_index = start_times.index(started_at) - except ValueError: - continue - next_started_at = start_times[start_index + 1] if start_index + 1 < len(start_times) else "" - matching = [] - for event in execution_starts: - created_at = str(event.get("created_at") or "") - if created_at < started_at: - continue - if next_started_at and created_at >= next_started_at: - continue - matching.append(event) - if not matching: + turn_sequence = int(payload.get("turn_sequence") or 0) + except (TypeError, ValueError): + turn_sequence = 0 + if turn_sequence not in prior_sequences: continue - first_payload = _execution_payload(matching[0]) return { "state": "interrupted_turn", - "execution_id": str(first_payload.get("execution_id") or "turn:%s" % started_at), - "tool": str(first_payload.get("tool") or "unknown"), - "call_id": str(first_payload.get("call_id") or ""), - "turn_started_at": started_at, - "tool_execution_count": len(matching), + "execution_id": str(payload.get("execution_id") or "turn:%s" % turn_sequence), + "tool": str(payload.get("tool") or "unknown"), + "call_id": str(payload.get("call_id") or ""), + "turn_sequence": turn_sequence, } return None @@ -212,6 +202,7 @@ def _begin_tool_execution(call: object, runtime: Dict[str, object]) -> str: "call_id": call_id, "arguments_sha256": arguments_fingerprint["sha256"], "arguments_preview": arguments_fingerprint["preview"], + "turn_sequence": int(_turn_lifecycle(runtime).get("sequence") or 0), "tool_round": runtime.get("_current_tool_round", ""), "started_at": utc_now(), } @@ -238,6 +229,7 @@ def _finish_tool_execution( "execution_id": execution_id, "tool": str(getattr(call, "name", "") or ""), "call_id": str(getattr(call, "call_id", "") or ""), + "turn_sequence": int(_turn_lifecycle(runtime).get("sequence") or 0), "tool_round": runtime.get("_current_tool_round", ""), "outcome": "error" if error else "ok", "error": shorten(str(error or ""), 500), @@ -268,6 +260,7 @@ def _mark_tool_execution_ambiguous( "call_id": str(getattr(call, "call_id", "") or ""), "arguments_sha256": arguments_fingerprint["sha256"], "arguments_preview": arguments_fingerprint["preview"], + "turn_sequence": int(_turn_lifecycle(runtime).get("sequence") or 0), "tool_round": runtime.get("_current_tool_round", ""), "state": "ambiguous", "interruption_type": type(exc).__name__, @@ -309,9 +302,7 @@ def _blocked_results(calls: List[object], runtime: Dict[str, object], blockers: }, ) if blocker_state == "interrupted_turn": - reason = ( - "a prior tool-bearing turn was interrupted before Moonshine durably completed the turn" - ) + reason = "a prior tool-bearing turn was interrupted before Moonshine durably completed the turn" else: reason = "a prior tool execution has ambiguous completion" message = ( @@ -324,7 +315,7 @@ def _blocked_results(calls: List[object], runtime: Dict[str, object], blockers: public_blockers = [ { key: item.get(key) - for key in ("state", "execution_id", "tool", "call_id", "turn_started_at", "started_at", "interrupted_at") + for key in ("state", "execution_id", "tool", "call_id", "turn_sequence", "started_at", "interrupted_at") if item.get(key) not in {None, ""} } for item in blockers From b116a4f78be36e536c814c1a742699fc6ab1a60d Mon Sep 17 00:00:00 2001 From: Yichuan Wang <133667618+Charlie-Wang-03@users.noreply.github.com> Date: Sat, 5 Sep 2026 13:17:27 +0800 Subject: [PATCH 06/10] test: cover interrupted turn recovery guard --- tests/test_interrupted_tool_recovery.py | 78 +++++++++++++++++++++++-- 1 file changed, 74 insertions(+), 4 deletions(-) diff --git a/tests/test_interrupted_tool_recovery.py b/tests/test_interrupted_tool_recovery.py index 3141585..4e34550 100644 --- a/tests/test_interrupted_tool_recovery.py +++ b/tests/test_interrupted_tool_recovery.py @@ -109,6 +109,8 @@ def interrupt(runtime): self.assertEqual(complete_finished[0]["payload"]["outcome"], "ok") self.assertTrue(complete_finished[0]["payload"]["output_sha256"]) self.assertIn('"status": "done"', complete_finished[0]["payload"]["output_preview"]) + self.assertNotIn("arguments", complete_started[0]["payload"]) + self.assertIn("arguments_sha256", complete_started[0]["payload"]) self.assertEqual(len(interrupted_started), 1) self.assertEqual(len(interrupted_ambiguous), 1) @@ -119,7 +121,7 @@ def interrupt(runtime): self.assertEqual(interrupted_ambiguous[0]["payload"]["state"], "ambiguous") self.assertEqual(self.store.get_session_meta(self.session_id)["status"], "interrupted") - def test_restart_blocks_new_tool_dispatch_after_ambiguous_execution(self): + def test_restart_blocks_new_tool_dispatch_after_orphaned_execution_start(self): execution_id = "tool-exec-hard-crash" self.store.append_conversation_event( self.session_id, @@ -130,7 +132,9 @@ def test_restart_blocks_new_tool_dispatch_after_ambiguous_execution(self): "execution_id": execution_id, "tool": "external_side_effect", "call_id": "call-before-crash", - "arguments": {"value": 1}, + "arguments_sha256": "deadbeef", + "arguments_preview": '{"value": 1}', + "turn_sequence": 0, }, ) @@ -162,6 +166,69 @@ def must_not_run(runtime): self.assertEqual(len(blocked), 1) self.assertIn(execution_id, blocked[0]["payload"]["ambiguous_execution_ids"]) + def test_later_turn_blocks_after_prior_tool_bearing_turn_never_completed(self): + self.store.append_turn_event( + self.session_id, + {"type": "turn_started", "text": "first", "created_at": "2026-09-05T00:00:00Z"}, + ) + execution_id = "tool-exec-finished-before-crash" + self.store.append_conversation_event( + self.session_id, + event_kind=TOOL_EXECUTION_STARTED, + role="tool", + content="Tool execution started", + payload={ + "execution_id": execution_id, + "tool": "external_side_effect", + "call_id": "call-first-turn", + "turn_sequence": 1, + "arguments_sha256": "abc", + "arguments_preview": "{}", + }, + ) + self.store.append_conversation_event( + self.session_id, + event_kind=TOOL_EXECUTION_FINISHED, + role="tool", + content="Tool execution finished", + payload={ + "execution_id": execution_id, + "tool": "external_side_effect", + "call_id": "call-first-turn", + "turn_sequence": 1, + "outcome": "ok", + "output_sha256": "def", + "output_preview": '{"ok": true}', + }, + ) + # Simulate process restart followed by a new user turn. The prior turn has + # no turn_completed marker even though its handler returned successfully. + self.store.append_turn_event( + self.session_id, + {"type": "turn_started", "text": "resumed", "created_at": "2026-09-05T00:00:00Z"}, + ) + + dispatch_count = [] + + def must_not_run(runtime): + dispatch_count.append(1) + return {"unexpected": True} + + registry = ScriptedRegistry({"must_not_run": must_not_run}) + results = handle_function_calls( + registry, + [ProviderToolCall(name="must_not_run", arguments={}, call_id="call-resumed")], + self._runtime(), + ) + + self.assertEqual(dispatch_count, []) + self.assertEqual(registry.dispatches, []) + self.assertEqual(results[0]["output"]["status"], "blocked_interrupted_execution") + self.assertIn("prior tool-bearing turn was interrupted", results[0]["error"]) + blocker = results[0]["output"]["ambiguous_executions"][0] + self.assertEqual(blocker["state"], "interrupted_turn") + self.assertEqual(blocker["turn_sequence"], 1) + def test_ordinary_tool_error_is_terminal_and_does_not_poison_future_dispatch(self): def fail(runtime): raise RuntimeError("deterministic tool failure") @@ -198,7 +265,7 @@ def succeed(runtime): self.assertEqual(later_effects, ["ran"]) self.assertIsNone(second[0]["error"]) - def test_dispatch_without_session_store_keeps_legacy_behavior(self): + def test_dispatch_without_session_store_keeps_legacy_result_shape(self): effects = [] def succeed(runtime, value): @@ -216,7 +283,10 @@ def succeed(runtime, value): self.assertEqual(effects, [3]) self.assertEqual(results[0]["output"], {"value": 3}) self.assertIsNone(results[0]["error"]) - self.assertTrue(results[0]["execution_id"].startswith("tool-exec-")) + self.assertEqual( + set(results[0]), + {"name", "call_id", "arguments", "output", "error"}, + ) if __name__ == "__main__": From a7a7c5b88369e0cabbafb27fb921207253974e4f Mon Sep 17 00:00:00 2001 From: Yichuan Wang <133667618+Charlie-Wang-03@users.noreply.github.com> Date: Sat, 5 Sep 2026 13:18:41 +0800 Subject: [PATCH 07/10] feat: add durable tool execution journal --- agent_runtime/execution_journal.py | 298 +++++++++++++++++++++++++++++ 1 file changed, 298 insertions(+) create mode 100644 agent_runtime/execution_journal.py diff --git a/agent_runtime/execution_journal.py b/agent_runtime/execution_journal.py new file mode 100644 index 0000000..500b72a --- /dev/null +++ b/agent_runtime/execution_journal.py @@ -0,0 +1,298 @@ +"""Crash-safe execution journal for provider-requested tool calls.""" + +from __future__ import annotations + +import hashlib +import json +import uuid +from typing import Dict, List, Optional, Sequence + +from moonshine.utils import read_jsonl, shorten, utc_now + + +TOOL_EXECUTION_STARTED = "tool_execution_started" +TOOL_EXECUTION_FINISHED = "tool_execution_finished" +TOOL_EXECUTION_AMBIGUOUS = "tool_execution_ambiguous" +TOOL_EXECUTION_BLOCKED = "tool_execution_blocked" + + +class ToolExecutionJournal(object): + """Persist tool dispatch boundaries and fail closed after interrupted turns. + + The journal does not claim exactly-once execution. Instead it establishes a + conservative contract: write intent before dispatch, write a terminal marker + before the next call begins, and never automatically dispatch more tools in a + session when an earlier tool execution or tool-bearing turn has ambiguous + completion. + """ + + def __init__(self, runtime: Dict[str, object]): + self.runtime = runtime + self.store = runtime.get("session_store") if isinstance(runtime, dict) else None + self.session_id = str(runtime.get("session_id") or "").strip() if isinstance(runtime, dict) else "" + self.paths = getattr(self.store, "paths", None) if self.store is not None else None + self.turn_sequence, self.open_turn_sequences = self._turn_lifecycle() + + @property + def enabled(self) -> bool: + """Return whether durable session journaling is available.""" + return bool(self.store is not None and self.session_id and hasattr(self.store, "append_conversation_event")) + + def _render_json(self, value: object) -> str: + """Render a deterministic JSON-ish representation for hashes and previews.""" + return json.dumps(value, ensure_ascii=False, sort_keys=True, default=str) + + def _fingerprint(self, value: object, preview_chars: int = 800) -> Dict[str, str]: + """Return bounded trace metadata without duplicating large tool payloads.""" + rendered = self._render_json(value) + return { + "sha256": hashlib.sha256(rendered.encode("utf-8")).hexdigest(), + "preview": shorten(rendered, preview_chars), + } + + def _turn_lifecycle(self): + """Return monotonically numbered turn starts and currently open sequences.""" + if self.paths is None or not self.session_id: + return 0, [] + turn_events = [ + item + for item in read_jsonl(self.paths.session_turn_events_file(self.session_id)) + if isinstance(item, dict) and str(item.get("type") or "") in {"turn_started", "turn_completed"} + ] + sequence = 0 + open_sequences: List[int] = [] + for item in turn_events: + if str(item.get("type") or "") == "turn_started": + sequence += 1 + open_sequences.append(sequence) + elif open_sequences: + # A later resumed turn can complete while an older interrupted + # turn remains unresolved, so pair completion with the newest start. + open_sequences.pop() + return sequence, open_sequences + + def _payload(self, event: Dict[str, object]) -> Dict[str, object]: + payload = event.get("payload") or {} + return dict(payload) if isinstance(payload, dict) else {} + + def _events(self) -> List[Dict[str, object]]: + if self.store is None or not self.session_id or not hasattr(self.store, "get_conversation_events"): + return [] + return list(self.store.get_conversation_events(self.session_id)) + + def blockers(self) -> List[Dict[str, object]]: + """Return unresolved executions plus any prior interrupted tool-bearing turn.""" + events = self._events() + active: Dict[str, Dict[str, object]] = {} + for event in events: + kind = str(event.get("event_kind") or "") + if kind not in {TOOL_EXECUTION_STARTED, TOOL_EXECUTION_FINISHED, TOOL_EXECUTION_AMBIGUOUS}: + continue + payload = self._payload(event) + execution_id = str(payload.get("execution_id") or "").strip() + if not execution_id: + continue + if kind == TOOL_EXECUTION_FINISHED: + active.pop(execution_id, None) + continue + record = dict(payload) + record["state"] = "ambiguous" if kind == TOOL_EXECUTION_AMBIGUOUS else "started" + record["event_id"] = event.get("id") + active[execution_id] = record + + blockers = list(active.values()) + prior_turn = self._prior_interrupted_tool_turn(events) + if prior_turn is not None: + blockers.append(prior_turn) + return blockers + + def _prior_interrupted_tool_turn(self, events: Sequence[Dict[str, object]]) -> Optional[Dict[str, object]]: + """Return a prior open turn that executed tools before a later turn began.""" + # During a normal dispatch the current turn itself is open. Older open + # sequences represent turns that survived into a later user turn. + if len(self.open_turn_sequences) <= 1: + return None + prior_sequences = set(self.open_turn_sequences[:-1]) + for event in events: + if str(event.get("event_kind") or "") != TOOL_EXECUTION_STARTED: + continue + payload = self._payload(event) + try: + turn_sequence = int(payload.get("turn_sequence") or 0) + except (TypeError, ValueError): + turn_sequence = 0 + if turn_sequence not in prior_sequences: + continue + return { + "state": "interrupted_turn", + "execution_id": str(payload.get("execution_id") or "turn:%s" % turn_sequence), + "tool": str(payload.get("tool") or "unknown"), + "call_id": str(payload.get("call_id") or ""), + "turn_sequence": turn_sequence, + } + return None + + def _append(self, event_kind: str, content: str, payload: Dict[str, object]) -> None: + if not self.enabled: + return + self.store.append_conversation_event( + self.session_id, + event_kind=event_kind, + role="tool", + content=content, + payload=dict(payload), + ) + + def _mark_session_interrupted(self, interruption: Dict[str, object]) -> None: + if self.store is None or not self.session_id: + return + now = utc_now() + if hasattr(self.store, "update_session_meta"): + self.store.update_session_meta( + self.session_id, + status="interrupted", + updated_at=now, + interrupted_tool_execution=dict(interruption), + ) + db = getattr(self.store, "db", None) + if db is not None and hasattr(db, "update_session"): + db.update_session(self.session_id, updated_at=now, status="interrupted") + + def begin(self, call: object) -> str: + """Write a durable intent immediately before dispatch.""" + execution_id = "tool-exec-%s" % uuid.uuid4().hex[:12] + tool_name = str(getattr(call, "name", "") or "") + call_id = str(getattr(call, "call_id", "") or "") + arguments = dict(getattr(call, "arguments", {}) or {}) + arguments_fingerprint = self._fingerprint(arguments) + payload = { + "execution_id": execution_id, + "tool": tool_name, + "call_id": call_id, + "arguments_sha256": arguments_fingerprint["sha256"], + "arguments_preview": arguments_fingerprint["preview"], + "turn_sequence": self.turn_sequence, + "tool_round": self.runtime.get("_current_tool_round", ""), + "started_at": utc_now(), + } + self._append( + TOOL_EXECUTION_STARTED, + "Tool execution started: %s (%s)" % (tool_name, call_id or execution_id), + payload, + ) + return execution_id + + def finish(self, call: object, execution_id: str, *, output: object, error: Optional[str]) -> None: + """Write a terminal marker before the next call is dispatched.""" + output_fingerprint = self._fingerprint(output, preview_chars=1200) + payload = { + "execution_id": execution_id, + "tool": str(getattr(call, "name", "") or ""), + "call_id": str(getattr(call, "call_id", "") or ""), + "turn_sequence": self.turn_sequence, + "tool_round": self.runtime.get("_current_tool_round", ""), + "outcome": "error" if error else "ok", + "error": shorten(str(error or ""), 500), + "output_preview": output_fingerprint["preview"], + "output_sha256": output_fingerprint["sha256"], + "finished_at": utc_now(), + } + self._append( + TOOL_EXECUTION_FINISHED, + "Tool execution finished: %s (%s)" % (payload["tool"], payload["call_id"] or execution_id), + payload, + ) + + def mark_ambiguous(self, call: object, execution_id: str, exc: BaseException) -> None: + """Record a process-level interruption whose completion is unknowable.""" + arguments_fingerprint = self._fingerprint(dict(getattr(call, "arguments", {}) or {})) + payload = { + "execution_id": execution_id, + "tool": str(getattr(call, "name", "") or ""), + "call_id": str(getattr(call, "call_id", "") or ""), + "arguments_sha256": arguments_fingerprint["sha256"], + "arguments_preview": arguments_fingerprint["preview"], + "turn_sequence": self.turn_sequence, + "tool_round": self.runtime.get("_current_tool_round", ""), + "state": "ambiguous", + "interruption_type": type(exc).__name__, + "interruption": shorten(str(exc), 500), + "interrupted_at": utc_now(), + } + self._append( + TOOL_EXECUTION_AMBIGUOUS, + "Tool execution became ambiguous after interruption: %s (%s)" + % (payload["tool"], payload["call_id"] or execution_id), + payload, + ) + self._mark_session_interrupted( + { + "execution_id": execution_id, + "tool": payload["tool"], + "call_id": payload["call_id"], + "state": "ambiguous", + } + ) + + def blocked_results(self, calls: List[object], blockers: Sequence[Dict[str, object]]) -> List[Dict[str, object]]: + """Return provider-visible errors without dispatching any requested tool.""" + first = dict(blockers[0]) if blockers else {} + blocker_tool = str(first.get("tool") or "unknown") + blocker_call_id = str(first.get("call_id") or "unknown") + blocker_execution_id = str(first.get("execution_id") or "unknown") + blocker_state = str(first.get("state") or "ambiguous") + self._mark_session_interrupted( + { + "execution_id": blocker_execution_id, + "tool": blocker_tool, + "call_id": blocker_call_id, + "state": blocker_state, + } + ) + reason = ( + "a prior tool-bearing turn was interrupted before Moonshine durably completed the turn" + if blocker_state == "interrupted_turn" + else "a prior tool execution has ambiguous completion" + ) + message = ( + "Tool dispatch is blocked because %s: tool=%s, call_id=%s, execution_id=%s. " + "Moonshine will not replay or dispatch additional tools automatically because prior " + "handlers may already have produced external side effects. Inspect the session records " + "and continue in a fresh session once the ambiguity is resolved." + % (reason, blocker_tool, blocker_call_id, blocker_execution_id) + ) + public_blockers = [ + { + key: item.get(key) + for key in ("state", "execution_id", "tool", "call_id", "turn_sequence", "started_at", "interrupted_at") + if item.get(key) not in {None, ""} + } + for item in blockers + ] + results: List[Dict[str, object]] = [] + for call in calls: + result = { + "name": getattr(call, "name", ""), + "call_id": getattr(call, "call_id", ""), + "arguments": getattr(call, "arguments", {}), + "output": { + "status": "blocked_interrupted_execution", + "message": message, + "ambiguous_executions": public_blockers, + }, + "error": message, + } + results.append(result) + self.runtime.setdefault("_tool_results_in_round", []).append(result) + self._append( + TOOL_EXECUTION_BLOCKED, + "Blocked tool dispatch: %s" % str(getattr(call, "name", "") or ""), + { + "tool": str(getattr(call, "name", "") or ""), + "call_id": str(getattr(call, "call_id", "") or ""), + "blocked_at": utc_now(), + "blocker_states": [str(item.get("state") or "") for item in blockers], + "ambiguous_execution_ids": [str(item.get("execution_id") or "") for item in blockers], + }, + ) + return results From 183fd7eab4b5f9957a8ca17105e11118fea3bde4 Mon Sep 17 00:00:00 2001 From: Yichuan Wang <133667618+Charlie-Wang-03@users.noreply.github.com> Date: Sat, 5 Sep 2026 13:18:51 +0800 Subject: [PATCH 08/10] refactor: isolate tool execution journal --- model_tools.py | 362 ++----------------------------------------------- 1 file changed, 8 insertions(+), 354 deletions(-) diff --git a/model_tools.py b/model_tools.py index ba69b10..217efaa 100644 --- a/model_tools.py +++ b/model_tools.py @@ -2,19 +2,10 @@ from __future__ import annotations -import hashlib -import json import traceback -import uuid from typing import Dict, List, Optional, Sequence -from moonshine.utils import read_jsonl, shorten, utc_now - - -TOOL_EXECUTION_STARTED = "tool_execution_started" -TOOL_EXECUTION_FINISHED = "tool_execution_finished" -TOOL_EXECUTION_AMBIGUOUS = "tool_execution_ambiguous" -TOOL_EXECUTION_BLOCKED = "tool_execution_blocked" +from moonshine.agent_runtime.execution_journal import ToolExecutionJournal def collect_tool_schemas( @@ -28,347 +19,16 @@ def collect_tool_schemas( return registry.schemas(mode=mode, include=include, exclude=exclude) -def _execution_store(runtime: Dict[str, object]): - """Return the session store and id used for durable execution journaling.""" - runtime = dict(runtime or {}) - store = runtime.get("session_store") - session_id = str(runtime.get("session_id") or "").strip() - if store is None or not session_id: - return None, "" - return store, session_id - - -def _render_json(value: object) -> str: - """Render a deterministic JSON-ish representation for hashes and previews.""" - return json.dumps(value, ensure_ascii=False, sort_keys=True, default=str) - - -def _fingerprint(value: object, preview_chars: int = 800) -> Dict[str, str]: - """Return bounded trace metadata without duplicating large tool payloads.""" - rendered = _render_json(value) - return { - "sha256": hashlib.sha256(rendered.encode("utf-8")).hexdigest(), - "preview": shorten(rendered, preview_chars), - } - - -def _execution_payload(event: Dict[str, object]) -> Dict[str, object]: - """Return one execution-event payload defensively.""" - payload = event.get("payload") or {} - return dict(payload) if isinstance(payload, dict) else {} - - -def _turn_lifecycle(runtime: Dict[str, object]) -> Dict[str, object]: - """Return monotonically numbered turn starts and currently open sequences.""" - store, session_id = _execution_store(runtime) - paths = getattr(store, "paths", None) if store is not None else None - if paths is None: - return {"sequence": 0, "open_sequences": []} - turn_events = [ - item - for item in read_jsonl(paths.session_turn_events_file(session_id)) - if isinstance(item, dict) and str(item.get("type") or "") in {"turn_started", "turn_completed"} - ] - sequence = 0 - open_sequences: List[int] = [] - for item in turn_events: - if str(item.get("type") or "") == "turn_started": - sequence += 1 - open_sequences.append(sequence) - elif open_sequences: - # A resumed turn can complete while an older interrupted turn remains - # unresolved, so pair completions with the most recent live start. - open_sequences.pop() - return {"sequence": sequence, "open_sequences": open_sequences} - - -def _unresolved_tool_executions(runtime: Dict[str, object]) -> List[Dict[str, object]]: - """Return executions that started but cannot be proven terminal. - - ``tool_execution_ambiguous`` remains blocking by design: Moonshine cannot know - whether a side-effecting handler completed before the interruption, so retrying - automatically would risk duplicate external effects. - """ - store, session_id = _execution_store(runtime) - if store is None or not hasattr(store, "get_conversation_events"): - return [] - - active: Dict[str, Dict[str, object]] = {} - for event in store.get_conversation_events(session_id): - kind = str(event.get("event_kind") or "") - if kind not in {TOOL_EXECUTION_STARTED, TOOL_EXECUTION_FINISHED, TOOL_EXECUTION_AMBIGUOUS}: - continue - payload = _execution_payload(event) - execution_id = str(payload.get("execution_id") or "").strip() - if not execution_id: - continue - if kind == TOOL_EXECUTION_FINISHED: - active.pop(execution_id, None) - continue - record = dict(payload) - record["state"] = "ambiguous" if kind == TOOL_EXECUTION_AMBIGUOUS else "started" - record["event_id"] = event.get("id") - active[execution_id] = record - return list(active.values()) - - -def _prior_interrupted_tool_turn(runtime: Dict[str, object]) -> Optional[Dict[str, object]]: - """Return a prior open turn that executed tools before a later turn began. - - A finished handler is not enough to declare the *turn* durable. The process can - die after the handler returns but before Moonshine records the tool result and - final assistant message. Each execution intent stores its monotonic turn - sequence, so recovery does not rely on coarse wall-clock timestamps. - """ - store, session_id = _execution_store(runtime) - if store is None or not hasattr(store, "get_conversation_events"): - return None - lifecycle = _turn_lifecycle(runtime) - open_sequences = [int(item) for item in list(lifecycle.get("open_sequences") or [])] - - # During normal dispatch the current turn itself is open. Only older open - # turns are recovery hazards. - if len(open_sequences) <= 1: - return None - prior_sequences = set(open_sequences[:-1]) - for event in store.get_conversation_events(session_id): - if str(event.get("event_kind") or "") != TOOL_EXECUTION_STARTED: - continue - payload = _execution_payload(event) - try: - turn_sequence = int(payload.get("turn_sequence") or 0) - except (TypeError, ValueError): - turn_sequence = 0 - if turn_sequence not in prior_sequences: - continue - return { - "state": "interrupted_turn", - "execution_id": str(payload.get("execution_id") or "turn:%s" % turn_sequence), - "tool": str(payload.get("tool") or "unknown"), - "call_id": str(payload.get("call_id") or ""), - "turn_sequence": turn_sequence, - } - return None - - -def _append_execution_event( - runtime: Dict[str, object], - *, - event_kind: str, - content: str, - payload: Dict[str, object], -) -> None: - """Persist one execution lifecycle event when session storage is available.""" - store, session_id = _execution_store(runtime) - if store is None or not hasattr(store, "append_conversation_event"): - return - store.append_conversation_event( - session_id, - event_kind=event_kind, - role="tool", - content=content, - payload=dict(payload), - ) - - -def _mark_session_interrupted(runtime: Dict[str, object], interruption: Dict[str, object]) -> None: - """Expose interrupted execution state in both session metadata stores.""" - store, session_id = _execution_store(runtime) - if store is None: - return - now = utc_now() - if hasattr(store, "update_session_meta"): - store.update_session_meta( - session_id, - status="interrupted", - updated_at=now, - interrupted_tool_execution=dict(interruption), - ) - db = getattr(store, "db", None) - if db is not None and hasattr(db, "update_session"): - db.update_session(session_id, updated_at=now, status="interrupted") - - -def _begin_tool_execution(call: object, runtime: Dict[str, object]) -> str: - """Write a durable intent record immediately before dispatch.""" - execution_id = "tool-exec-%s" % uuid.uuid4().hex[:12] - tool_name = str(getattr(call, "name", "") or "") - call_id = str(getattr(call, "call_id", "") or "") - arguments = dict(getattr(call, "arguments", {}) or {}) - arguments_fingerprint = _fingerprint(arguments) - payload = { - "execution_id": execution_id, - "tool": tool_name, - "call_id": call_id, - "arguments_sha256": arguments_fingerprint["sha256"], - "arguments_preview": arguments_fingerprint["preview"], - "turn_sequence": int(_turn_lifecycle(runtime).get("sequence") or 0), - "tool_round": runtime.get("_current_tool_round", ""), - "started_at": utc_now(), - } - _append_execution_event( - runtime, - event_kind=TOOL_EXECUTION_STARTED, - content="Tool execution started: %s (%s)" % (tool_name, call_id or execution_id), - payload=payload, - ) - return execution_id - - -def _finish_tool_execution( - call: object, - runtime: Dict[str, object], - execution_id: str, - *, - output: object, - error: Optional[str], -) -> None: - """Write the terminal lifecycle record before the next tool is dispatched.""" - output_fingerprint = _fingerprint(output, preview_chars=1200) - payload = { - "execution_id": execution_id, - "tool": str(getattr(call, "name", "") or ""), - "call_id": str(getattr(call, "call_id", "") or ""), - "turn_sequence": int(_turn_lifecycle(runtime).get("sequence") or 0), - "tool_round": runtime.get("_current_tool_round", ""), - "outcome": "error" if error else "ok", - "error": shorten(str(error or ""), 500), - "output_preview": output_fingerprint["preview"], - "output_sha256": output_fingerprint["sha256"], - "finished_at": utc_now(), - } - _append_execution_event( - runtime, - event_kind=TOOL_EXECUTION_FINISHED, - content="Tool execution finished: %s (%s)" % (payload["tool"], payload["call_id"] or execution_id), - payload=payload, - ) - - -def _mark_tool_execution_ambiguous( - call: object, - runtime: Dict[str, object], - execution_id: str, - exc: BaseException, -) -> None: - """Record an interrupted dispatch whose external completion is unknowable.""" - now = utc_now() - arguments_fingerprint = _fingerprint(dict(getattr(call, "arguments", {}) or {})) - payload = { - "execution_id": execution_id, - "tool": str(getattr(call, "name", "") or ""), - "call_id": str(getattr(call, "call_id", "") or ""), - "arguments_sha256": arguments_fingerprint["sha256"], - "arguments_preview": arguments_fingerprint["preview"], - "turn_sequence": int(_turn_lifecycle(runtime).get("sequence") or 0), - "tool_round": runtime.get("_current_tool_round", ""), - "state": "ambiguous", - "interruption_type": type(exc).__name__, - "interruption": shorten(str(exc), 500), - "interrupted_at": now, - } - _append_execution_event( - runtime, - event_kind=TOOL_EXECUTION_AMBIGUOUS, - content="Tool execution became ambiguous after interruption: %s (%s)" - % (payload["tool"], payload["call_id"] or execution_id), - payload=payload, - ) - _mark_session_interrupted( - runtime, - { - "execution_id": execution_id, - "tool": payload["tool"], - "call_id": payload["call_id"], - "state": "ambiguous", - }, - ) - - -def _blocked_results(calls: List[object], runtime: Dict[str, object], blockers: Sequence[Dict[str, object]]) -> List[Dict[str, object]]: - """Fail closed instead of dispatching new tools after an interrupted execution.""" - first = dict(blockers[0]) if blockers else {} - blocker_tool = str(first.get("tool") or "unknown") - blocker_call_id = str(first.get("call_id") or "unknown") - blocker_execution_id = str(first.get("execution_id") or "unknown") - blocker_state = str(first.get("state") or "ambiguous") - _mark_session_interrupted( - runtime, - { - "execution_id": blocker_execution_id, - "tool": blocker_tool, - "call_id": blocker_call_id, - "state": blocker_state, - }, - ) - if blocker_state == "interrupted_turn": - reason = "a prior tool-bearing turn was interrupted before Moonshine durably completed the turn" - else: - reason = "a prior tool execution has ambiguous completion" - message = ( - "Tool dispatch is blocked because %s: tool=%s, call_id=%s, execution_id=%s. " - "Moonshine will not replay or dispatch additional tools automatically because prior " - "handlers may already have produced external side effects. Inspect the session records " - "and continue in a fresh session once the ambiguity is resolved." - % (reason, blocker_tool, blocker_call_id, blocker_execution_id) - ) - public_blockers = [ - { - key: item.get(key) - for key in ("state", "execution_id", "tool", "call_id", "turn_sequence", "started_at", "interrupted_at") - if item.get(key) not in {None, ""} - } - for item in blockers - ] - results: List[Dict[str, object]] = [] - for call in calls: - result = { - "name": getattr(call, "name", ""), - "call_id": getattr(call, "call_id", ""), - "arguments": getattr(call, "arguments", {}), - "output": { - "status": "blocked_interrupted_execution", - "message": message, - "ambiguous_executions": public_blockers, - }, - "error": message, - } - results.append(result) - runtime.setdefault("_tool_results_in_round", []).append(result) - _append_execution_event( - runtime, - event_kind=TOOL_EXECUTION_BLOCKED, - content="Blocked tool dispatch: %s" % str(getattr(call, "name", "") or ""), - payload={ - "tool": str(getattr(call, "name", "") or ""), - "call_id": str(getattr(call, "call_id", "") or ""), - "blocked_at": utc_now(), - "blocker_states": [str(item.get("state") or "") for item in blockers], - "ambiguous_execution_ids": [str(item.get("execution_id") or "") for item in blockers], - }, - ) - return results - - def handle_function_calls(registry, calls: List[object], runtime: Dict[str, object]) -> List[Dict[str, object]]: - """Dispatch provider tool calls through the registry with crash-safe journaling. - - Each call is journaled immediately before dispatch and receives a durable - terminal record before the next call begins. If execution is interrupted by a - process-level exception such as ``KeyboardInterrupt``, the call is marked - ambiguous and the exception is re-raised. A later tool-bearing turn also stays - blocked if an earlier tool-bearing turn never reached ``turn_completed``. - """ - blockers = _unresolved_tool_executions(runtime) - interrupted_turn = _prior_interrupted_tool_turn(runtime) - if interrupted_turn is not None: - blockers.append(interrupted_turn) + """Dispatch provider tool calls through the registry with crash-safe journaling.""" + journal = ToolExecutionJournal(runtime) + blockers = journal.blockers() if blockers: - return _blocked_results(calls, runtime, blockers) + return journal.blocked_results(calls, blockers) results = [] for call in calls: - execution_id = _begin_tool_execution(call, runtime) + execution_id = journal.begin(call) try: try: result = registry.dispatch(call.name, call.arguments, runtime) @@ -379,16 +39,10 @@ def handle_function_calls(registry, calls: List[object], runtime: Dict[str, obje "traceback": traceback.format_exc(limit=3), } error = str(exc) - _finish_tool_execution( - call, - runtime, - execution_id, - output=result, - error=error, - ) + journal.finish(call, execution_id, output=result, error=error) except BaseException as exc: try: - _mark_tool_execution_ambiguous(call, runtime, execution_id, exc) + journal.mark_ambiguous(call, execution_id, exc) except Exception: # Never replace the process-level interruption with a best-effort # journaling failure. A durable start record, when it was written, From 46deaa7f9a6e515ed914b5f7e8d6c72e44293f53 Mon Sep 17 00:00:00 2001 From: Yichuan Wang <133667618+Charlie-Wang-03@users.noreply.github.com> Date: Sat, 5 Sep 2026 13:19:14 +0800 Subject: [PATCH 09/10] test: import execution journal contract --- tests/test_interrupted_tool_recovery.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_interrupted_tool_recovery.py b/tests/test_interrupted_tool_recovery.py index 4e34550..0a64c70 100644 --- a/tests/test_interrupted_tool_recovery.py +++ b/tests/test_interrupted_tool_recovery.py @@ -6,13 +6,13 @@ import unittest from pathlib import Path -from moonshine.model_tools import ( +from moonshine.agent_runtime.execution_journal import ( TOOL_EXECUTION_AMBIGUOUS, TOOL_EXECUTION_BLOCKED, TOOL_EXECUTION_FINISHED, TOOL_EXECUTION_STARTED, - handle_function_calls, ) +from moonshine.model_tools import handle_function_calls from moonshine.moonshine_constants import MoonshinePaths from moonshine.providers import ProviderToolCall from moonshine.storage.session_store import SessionStore From 7cb61148e22ac3be1dc9d168c024f4cc7be32d5f Mon Sep 17 00:00:00 2001 From: Yichuan Wang <133667618+Charlie-Wang-03@users.noreply.github.com> Date: Sat, 5 Sep 2026 13:19:58 +0800 Subject: [PATCH 10/10] test: preserve older interrupted turn across resume --- tests/test_interrupted_tool_recovery.py | 20 ++++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/tests/test_interrupted_tool_recovery.py b/tests/test_interrupted_tool_recovery.py index 0a64c70..2acab43 100644 --- a/tests/test_interrupted_tool_recovery.py +++ b/tests/test_interrupted_tool_recovery.py @@ -166,7 +166,9 @@ def must_not_run(runtime): self.assertEqual(len(blocked), 1) self.assertIn(execution_id, blocked[0]["payload"]["ambiguous_execution_ids"]) - def test_later_turn_blocks_after_prior_tool_bearing_turn_never_completed(self): + def test_completed_resume_turn_does_not_hide_older_interrupted_tool_turn(self): + # Turn 1 starts and executes a tool, but the process dies before its + # turn_completed record is written. self.store.append_turn_event( self.session_id, {"type": "turn_started", "text": "first", "created_at": "2026-09-05T00:00:00Z"}, @@ -201,11 +203,21 @@ def test_later_turn_blocks_after_prior_tool_bearing_turn_never_completed(self): "output_preview": '{"ok": true}', }, ) - # Simulate process restart followed by a new user turn. The prior turn has - # no turn_completed marker even though its handler returned successfully. + + # Turn 2 is a resumed, tool-free turn that completes. LIFO pairing must + # close turn 2, not accidentally consume the older interrupted turn 1. + self.store.append_turn_event( + self.session_id, + {"type": "turn_started", "text": "resume-one", "created_at": "2026-09-05T00:00:00Z"}, + ) + self.store.append_turn_event( + self.session_id, + {"type": "turn_completed", "text": "resume-one done", "created_at": "2026-09-05T00:00:00Z"}, + ) + # Turn 3 begins and attempts another tool call. self.store.append_turn_event( self.session_id, - {"type": "turn_started", "text": "resumed", "created_at": "2026-09-05T00:00:00Z"}, + {"type": "turn_started", "text": "resume-two", "created_at": "2026-09-05T00:00:00Z"}, ) dispatch_count = []