fix: repair tool-call/result pairing on load to prevent unrecoverable 400 errors

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"],

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"