diff --git a/coworker/conversations.py b/coworker/conversations.py index e67ef2fb..03756f53 100644 --- a/coworker/conversations.py +++ b/coworker/conversations.py @@ -118,6 +118,114 @@ 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 — 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: + 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 + + # 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. 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: + 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 (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 + # 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) + 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, + "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 +338,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..ce6bff3f --- /dev/null +++ b/tests/test_conversations_repair.py @@ -0,0 +1,193 @@ +"""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 — but + only when the thread has moved past the call (not a trailing pending call).""" + messages = [ + _user("go"), + _assistant_with_calls("c1"), + _user("next"), # thread moved past → call is corrupt, not pending + ] + 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_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 = [ + _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