From 0c79ffc632d90b7e431f43d379411a6dfa15aa2a Mon Sep 17 00:00:00 2001 From: Shakti Prasad Mohapatra Date: Thu, 30 Jul 2026 20:59:29 +0530 Subject: [PATCH 1/3] fix: repair tool-call/result pairing on load to prevent unrecoverable 400 errors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a turn is interrupted at the wrong moment, the append-only JSONL can end up with a user message between an assistant tool_calls block and its tool result. Providers reject this ordering (Anthropic 400/2013, OpenAI 'tool_call_ids did not have response messages'), making the session permanently unrecoverable from the UI. _repair_tool_pairing() runs in ConversationStore.load() and: - moves a real tool result found later in the thread to sit right after its call - synthesises a placeholder result for a call with no matching tool message - is idempotent — well-formed threads pass through unchanged Fixes #331 --- coworker/conversations.py | 95 +++++++++++++++ tests/test_conversations_repair.py | 181 +++++++++++++++++++++++++++++ 2 files changed, 276 insertions(+) create mode 100644 tests/test_conversations_repair.py diff --git a/coworker/conversations.py b/coworker/conversations.py index e67ef2fb..81463572 100644 --- a/coworker/conversations.py +++ b/coworker/conversations.py @@ -118,6 +118,97 @@ def _read_jsonl(self, sid: str) -> Optional[list[dict]]: if line.strip() ] + # -- tool-call/result pairing repair --------------------------------------- + @staticmethod + def _repair_tool_pairing(messages: list[dict]) -> list[dict]: + """Reorder messages so every tool result immediately follows its call. + + Append-only persistence means an interrupted turn can leave a user + message between an assistant ``tool_calls`` block and the matching + ``tool`` result. Providers reject this ordering (Anthropic 400/2013, + OpenAI "tool_call_ids did not have response messages"), making the + session permanently unrecoverable. + + This pass: + * Moves a real ``tool`` result found later in the thread to sit right + after its call. + * Synthesises a placeholder result for a call with no matching tool + message, so the replay is well-formed. + * Is idempotent — a well-formed thread passes through unchanged. + """ + if not messages: + return messages + + # Collect tool_call ids from assistant messages. + pending_calls: dict[str, int] = {} # call_id → index of the assistant msg + for i, m in enumerate(messages): + if m.get("role") == "assistant" and m.get("tool_calls"): + for tc in m["tool_calls"]: + call_id = tc.get("id") + if call_id: + pending_calls[call_id] = i + + if not pending_calls: + return messages # no tool calls at all + + # Find tool results and where they sit relative to their calls. + # call_id → index of the tool result message (if found) + found_results: dict[str, int] = {} + for i, m in enumerate(messages): + if m.get("role") == "tool": + call_id = m.get("tool_call_id") + if call_id and call_id in pending_calls: + # Only keep the first result for each call. + if call_id not in found_results: + found_results[call_id] = i + + # Calls that have a result already immediately following the assistant + # message are fine — no work needed. We only need to act when a result + # is missing or out-of-order. + needs_repair = False + for call_id, call_idx in pending_calls.items(): + if call_id in found_results: + result_idx = found_results[call_id] + if result_idx != call_idx + 1: + needs_repair = True # result exists but not immediately after + else: + needs_repair = True # no result at all + if not needs_repair: + return messages # already well-formed + + # Build the repaired list. We iterate through the original messages, + # and after each assistant message we emit its tool results (moved from + # their original position or synthesised if missing). + consumed_result_indices: set[int] = set() + repaired: list[dict] = [] + + for i, m in enumerate(messages): + if m.get("role") == "assistant" and m.get("tool_calls"): + repaired.append(m) + # Emit results for each tool call in this block, in order. + for tc in m["tool_calls"]: + call_id = tc.get("id") + if not call_id: + continue + if call_id in found_results: + result_idx = found_results[call_id] + if result_idx not in consumed_result_indices: + repaired.append(messages[result_idx]) + consumed_result_indices.add(result_idx) + else: + # Synthesise a placeholder so the thread is well-formed. + repaired.append({ + "role": "tool", + "tool_call_id": call_id, + "content": '{"error": "tool result was lost during an interrupted turn"}', + }) + elif i in consumed_result_indices: + continue # already moved this tool result up + else: + repaired.append(m) + + return repaired + def _count(self, sid: str) -> int: path = self._file(sid) if not path.exists(): @@ -230,6 +321,10 @@ def load(self, session_id: str) -> Optional[SessionRecord]: messages = json.loads(row["messages"] or "[]") except json.JSONDecodeError: messages = [] + # Self-heal: ensure every tool result immediately follows its call. + # An interrupted turn can persist a user message between an assistant + # tool_calls block and its tool result, which providers reject (400). + messages = self._repair_tool_pairing(messages) return SessionRecord( session_id=session_id, workspace=row["workspace"], diff --git a/tests/test_conversations_repair.py b/tests/test_conversations_repair.py new file mode 100644 index 00000000..5db0e249 --- /dev/null +++ b/tests/test_conversations_repair.py @@ -0,0 +1,181 @@ +"""Regression tests for ConversationStore tool-call/result pairing repair. + +When a turn is interrupted at the wrong moment, the append-only JSONL can end +up with a user message between an assistant ``tool_calls`` block and its +``tool`` result. Providers reject this ordering (Anthropic 400/2013, OpenAI +"tool_call_ids did not have response messages"), making the session permanently +unrecoverable. + +``ConversationStore._repair_tool_pairing`` reorders messages on load so every +tool result immediately follows its call, synthesising a placeholder when no +result exists. +""" + +from __future__ import annotations + +import json + +from coworker.conversations import ConversationStore + + +# ── helpers ────────────────────────────────────────────────────────────────── + +def _assistant_with_calls(call_id: str = "c1") -> dict: + return { + "role": "assistant", + "content": None, + "tool_calls": [ + {"id": call_id, "type": "function", "function": {"name": "run_shell", "arguments": "{}"}} + ], + } + + +def _tool_result(call_id: str = "c1", content: str = '{"ok": true}') -> dict: + return {"role": "tool", "tool_call_id": call_id, "content": content} + + +def _user(text: str = "continue") -> dict: + return {"role": "user", "content": text} + + +# ── tests ──────────────────────────────────────────────────────────────────── + +def test_passthrough_well_formed(): + """A well-formed thread with tool results immediately after calls is unchanged.""" + messages = [ + _user("go"), + _assistant_with_calls("c1"), + _tool_result("c1"), + _assistant_with_calls("c2"), + _tool_result("c2"), + _user("done"), + ] + assert ConversationStore._repair_tool_pairing(messages) == messages + + +def test_interleaved_user_between_call_and_result(): + """A user message between a tool call and its result is moved after the result.""" + messages = [ + _user("go"), + _assistant_with_calls("c1"), + _user("continue"), # interleaved — this caused the 400 + _tool_result("c1"), + ] + repaired = ConversationStore._repair_tool_pairing(messages) + # The tool result must come right after the assistant message. + assert repaired[1]["role"] == "assistant" + assert repaired[2]["role"] == "tool" + assert repaired[2]["tool_call_id"] == "c1" + # The user message is pushed after the result. + assert repaired[3]["role"] == "user" + assert repaired[3]["content"] == "continue" + + +def test_dangling_call_gets_placeholder(): + """A tool call with no matching result gets a synthesised placeholder.""" + messages = [ + _user("go"), + _assistant_with_calls("c1"), + _user("next"), + ] + repaired = ConversationStore._repair_tool_pairing(messages) + assert repaired[0]["role"] == "user" + assert repaired[1]["role"] == "assistant" + assert repaired[2]["role"] == "tool" + assert repaired[2]["tool_call_id"] == "c1" + assert "error" in repaired[2]["content"] + assert repaired[3]["role"] == "user" + + +def test_multi_call_block(): + """Multiple tool calls in one assistant message each get their result in order.""" + messages = [ + _user("go"), + { + "role": "assistant", + "content": None, + "tool_calls": [ + {"id": "a", "type": "function", "function": {"name": "run_shell", "arguments": "{}"}}, + {"id": "b", "type": "function", "function": {"name": "read_file", "arguments": "{}"}}, + ], + }, + _user("wait"), + _tool_result("b", '{"file": true}'), + _tool_result("a", '{"shell": true}'), + ] + repaired = ConversationStore._repair_tool_pairing(messages) + # Both results must follow the assistant message, in call order. + assert repaired[1]["role"] == "assistant" + assert repaired[2]["role"] == "tool" + assert repaired[2]["tool_call_id"] == "a" + assert repaired[3]["role"] == "tool" + assert repaired[3]["tool_call_id"] == "b" + # The user message comes after both results. + assert repaired[4]["role"] == "user" + + +def test_idempotent(): + """Running repair twice produces the same output as running it once.""" + messages = [ + _user("go"), + _assistant_with_calls("c1"), + _user("continue"), + _tool_result("c1"), + _assistant_with_calls("c2"), + _user("again"), + _tool_result("c2"), + ] + once = ConversationStore._repair_tool_pairing(messages) + twice = ConversationStore._repair_tool_pairing(once) + assert once == twice + + +def test_no_tool_calls_unchanged(): + """A thread with no tool calls passes through unchanged.""" + messages = [ + _user("hello"), + {"role": "assistant", "content": "hi there"}, + _user("bye"), + ] + assert ConversationStore._repair_tool_pairing(messages) == messages + + +def test_empty_list(): + """An empty message list passes through unchanged.""" + assert ConversationStore._repair_tool_pairing([]) == [] + + +def test_load_repairs_corrupt_jsonl(tmp_path): + """End-to-end: a corrupt JSONL file is repaired on load().""" + store = ConversationStore(tmp_path) + sid = "test-sid" + + # Persist a corrupt thread: user message between call and result. + corrupt_messages = [ + _user("go"), + _assistant_with_calls("c1"), + _user("continue"), + _tool_result("c1"), + ] + # Insert the session row so load() can find it. + store._conn.execute( + "INSERT INTO sessions (session_id, workspace, model, mode, n_msgs) " + "VALUES (?, ?, ?, ?, ?)", + (sid, str(tmp_path), "test-model", "interactive", len(corrupt_messages)), + ) + store._conn.commit() + # Write the corrupt JSONL. + with open(store._file(sid), "w", encoding="utf-8") as f: + for m in corrupt_messages: + f.write(json.dumps(m) + "\n") + + record = store.load(sid) + assert record is not None + msgs = record.messages + # The tool result must immediately follow the assistant message. + assert msgs[1]["role"] == "assistant" + assert msgs[2]["role"] == "tool" + assert msgs[2]["tool_call_id"] == "c1" + # The user message is after the result. + assert msgs[3]["role"] == "user" + assert msgs[3]["content"] == "continue" \ No newline at end of file From b82b5971cae3ccd531743b375da5c601ae9e6193 Mon Sep 17 00:00:00 2001 From: Shakti Prasad Mohapatra Date: Fri, 31 Jul 2026 12:37:43 +0530 Subject: [PATCH 2/3] chore: trigger CI re-run From 4dd12fd9b3fa82cd057ba043183fd08514cbd75c Mon Sep 17 00:00:00 2001 From: Shakti Prasad Mohapatra Date: Fri, 31 Jul 2026 16:55:56 +0530 Subject: [PATCH 3/3] fix: don't inject placeholder for trailing pending tool calls MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The repair was synthesising placeholder tool results for any dangling call, including calls in the last assistant message that are simply pending (interrupted for approval/question). This broke durable resume because the provider saw the placeholder and thought the tool already ran, so resolve_inbox could not re-execute the tool. Now trailing calls (assistant tool_calls as the last message with no result) are left untouched — the engine will resume them. Placeholders are only injected when the thread has moved past the call, proving it is corrupt rather than pending. Fixes test_durable_resume_question and test_durable_resume_approval_executes_tool. --- coworker/conversations.py | 25 +++++++++++++++++++++---- tests/test_conversations_repair.py | 16 ++++++++++++++-- 2 files changed, 35 insertions(+), 6 deletions(-) diff --git a/coworker/conversations.py b/coworker/conversations.py index 81463572..03756f53 100644 --- a/coworker/conversations.py +++ b/coworker/conversations.py @@ -133,7 +133,11 @@ def _repair_tool_pairing(messages: list[dict]) -> list[dict]: * Moves a real ``tool`` result found later in the thread to sit right after its call. * Synthesises a placeholder result for a call with no matching tool - message, so the replay is well-formed. + message — but **only** when the thread has moved past the call + (i.e. there are messages after the assistant block). A trailing + assistant ``tool_calls`` with no result is a pending/interrupted + call that the engine will resume; injecting a placeholder there + would break durable resume. * Is idempotent — a well-formed thread passes through unchanged. """ if not messages: @@ -162,11 +166,23 @@ def _repair_tool_pairing(messages: list[dict]) -> list[dict]: if call_id not in found_results: found_results[call_id] = i + # Determine which calls are "trailing" — the assistant block is the + # last message in the thread (nothing after it). These are pending + # calls that the engine will resume; we must not inject placeholders. + last_msg_idx = len(messages) - 1 + trailing_calls: set[str] = set() + for call_id, call_idx in pending_calls.items(): + if call_idx == last_msg_idx: + trailing_calls.add(call_id) + # Calls that have a result already immediately following the assistant # message are fine — no work needed. We only need to act when a result - # is missing or out-of-order. + # is missing or out-of-order. Trailing calls without results are + # skipped (they're pending, not corrupt). needs_repair = False for call_id, call_idx in pending_calls.items(): + if call_id in trailing_calls and call_id not in found_results: + continue # pending call — engine will resume if call_id in found_results: result_idx = found_results[call_id] if result_idx != call_idx + 1: @@ -174,7 +190,7 @@ def _repair_tool_pairing(messages: list[dict]) -> list[dict]: else: needs_repair = True # no result at all if not needs_repair: - return messages # already well-formed + return messages # already well-formed (or only pending calls) # Build the repaired list. We iterate through the original messages, # and after each assistant message we emit its tool results (moved from @@ -195,8 +211,9 @@ def _repair_tool_pairing(messages: list[dict]) -> list[dict]: if result_idx not in consumed_result_indices: repaired.append(messages[result_idx]) consumed_result_indices.add(result_idx) - else: + elif call_id not in trailing_calls: # Synthesise a placeholder so the thread is well-formed. + # Skip trailing calls — they're pending, not corrupt. repaired.append({ "role": "tool", "tool_call_id": call_id, diff --git a/tests/test_conversations_repair.py b/tests/test_conversations_repair.py index 5db0e249..ce6bff3f 100644 --- a/tests/test_conversations_repair.py +++ b/tests/test_conversations_repair.py @@ -72,11 +72,12 @@ def test_interleaved_user_between_call_and_result(): def test_dangling_call_gets_placeholder(): - """A tool call with no matching result gets a synthesised placeholder.""" + """A tool call with no matching result gets a synthesised placeholder — but + only when the thread has moved past the call (not a trailing pending call).""" messages = [ _user("go"), _assistant_with_calls("c1"), - _user("next"), + _user("next"), # thread moved past → call is corrupt, not pending ] repaired = ConversationStore._repair_tool_pairing(messages) assert repaired[0]["role"] == "user" @@ -87,6 +88,17 @@ def test_dangling_call_gets_placeholder(): assert repaired[3]["role"] == "user" +def test_trailing_pending_call_not_repaired(): + """A tool call as the last message is a pending/interrupted call — the + engine will resume it, so we must not inject a placeholder.""" + messages = [ + _user("go"), + _assistant_with_calls("c1"), + ] + repaired = ConversationStore._repair_tool_pairing(messages) + assert repaired == messages # unchanged — no placeholder injected + + def test_multi_call_block(): """Multiple tool calls in one assistant message each get their result in order.""" messages = [