Repair recurring naming history overflow and dead server queue fences

src/backend/services/chat_history_service.py

@@ -8191,12 +8191,34 @@ def backfill_chat_history_blob_payloads(
                 continue
 
             try:
-                chat_history_coll.update_one(
-                    {"_id": doc.get("_id")},
-                    {"$set": {"history": updated_history}},
-                )
-                sessions_updated += 1
-                entries_updated += session_candidate_entries
+                # Replace only entries we actually observed. A full-array write
+                # could erase a concurrent append, edit, or deletion.
+                changed = 0
+                for index, (before, after) in enumerate(zip(history, updated_history)):
+                    if before == after:
+                        continue
+                    field = f"history.{index}"
+                    result = chat_history_coll.update_one(
+                        {
+                            "_id": doc.get("_id"),
+                            "user_id": doc.get("user_id"),
+                            "namespace": doc.get("namespace"),
+                            field: before,
+                        },
+                        {"$set": {field: after}},
+                    )
+                    if result.matched_count:
+                        changed += 1
+                    else:
+                        errors.append(
+                            {
+                                "type": "entry_changed_concurrently",
+                                "session_id": doc.get("session_id"),
+                                "history_index": index,
+                            }
+                        )
+                sessions_updated += int(changed > 0)
+                entries_updated += changed
             except Exception as exc:
                 errors.append(
                     {

src/backend/services/chat_prompt_queue_dispatch_service.py

@@ -338,6 +338,8 @@ def run_once(self) -> bool:
 
         if self._reconcile_one_linked_terminal():
             return True
+        if self._reconcile_one_stopped_server_attempt():
+            return True
         if self._reconcile_one_interrupted_cancellation():
             return True
         if self._reconcile_one_task_launch():
@@ -435,6 +437,22 @@ def run_once(self) -> bool:
             )
         return True
 
+    def _reconcile_one_stopped_server_attempt(self) -> bool:
+        record = chat_prompt_queue_service.terminalise_one_stopped_server_attempt(
+            current_server_instance_id=SERVER_INSTANCE_ID,
+        )
+        if record is None:
+            return False
+        # finish_prompt_record retains a durable reconciliation marker, so a
+        # projection failure here is retried by the existing dispatcher.
+        reconcile_linked_task_execution_terminal(
+            record,
+            status=record["status"],
+            source="stopped_server_reconciliation",
+            error=record.get("last_error"),
+        )
+        return True
+
     def _reconcile_one_interrupted_cancellation(self) -> bool:
         """Terminalise one cancellation left by a stopped server process."""
 

src/backend/services/chat_prompt_queue_service.py

@@ -11,6 +11,7 @@
 import math
 import os
 import re
+import socket
 import threading
 from datetime import datetime, timedelta, timezone
 from typing import Any, Mapping, Sequence
@@ -3780,6 +3781,108 @@ def is_prompt_cancellation_requested(
     return _collection().find_one(query, {"_id": 1}) is not None
 
 
+def terminalise_one_stopped_server_attempt(
+    *,
+    current_server_instance_id: str,
+) -> dict[str, Any] | None:
+    """Release a dead local process's fence without replaying uncertain effects.
+
+    Lease age alone is insufficient. A missing local PID proves this execution
+    process has stopped; other hosts, live/reused PIDs and permission failures
+    remain for explicit reconciliation. Terminal history wins over interruption.
+    """
+    host = socket.gethostname()
+    coll = _collection()
+    candidates = (
+        coll.find(
+            {
+                "status": STATUS_IN_PROGRESS,
+                "dispatch_mode": DISPATCH_MODE_SERVER,
+                "server_instance_id": {"$regex": "^" + re.escape(host) + ":"},
+                "lease_expires_at": {"$lte": _now()},
+                "attempt_id": {"$type": "string", "$ne": ""},
+            }
+        )
+        .sort("lease_expires_at", ASCENDING)
+        .limit(50)
+    )
+    for row in candidates:
+        owner = row.get("server_instance_id")
+        if owner == current_server_instance_id:
+            continue
+        parts = str(owner).split(":")
+        if len(parts) != 3 or not parts[1].isdigit() or int(parts[1]) <= 0:
+            continue
+        try:
+            os.kill(int(parts[1]), 0)
+        except ProcessLookupError:
+            pass
+        except OSError:
+            continue
+        else:
+            continue
+
+        from .chat_history_service import (
+            build_chat_history_query,
+            get_chat_history_collection_service,
+        )
+
+        history = get_chat_history_collection_service(read_only=True)
+        if history is None:
+            raise ChatPromptQueueUnavailable("History reconciliation is unavailable")
+        request_id = row.get("client_request_id")
+        completed = request_id and history.find_one(
+            {
+                **build_chat_history_query(
+                    user_id=row["user_concept_id"],
+                    session_id=row.get("session_id"),
+                    namespace=row.get("namespace"),
+                    include_legacy=False,
+                ),
+                "history": {
+                    "$elemMatch": {
+                        "role": "assistant",
+                        "llm_debug_data.request_id": request_id,
+                    }
+                },
+            },
+            {"_id": 1},
+        )
+        from .turn_execution_record_service import get_turn_execution_record_projection
+
+        receipt = get_turn_execution_record_projection(
+            request_id=request_id,
+            namespace=row.get("namespace"),
+        )
+        completed = bool(
+            completed
+            and isinstance(receipt, Mapping)
+            and receipt.get("user_id") == row["user_concept_id"]
+            and receipt.get("session_id") == row.get("session_id")
+            and receipt.get("execution_terminal_status") == STATUS_COMPLETED
+        )
+        # This is an execution failure, not a verdict that external effects
+        # failed. Keep the original prompt and request locator for recovery;
+        # do not resubmit a potentially state-changing request automatically.
+        return finish_prompt_record(
+            scope=_queue_scope_values(row),
+            queue_id=row["queue_id"],
+            attempt_id=row["attempt_id"],
+            expected_server_instance_id=owner,
+            status=STATUS_COMPLETED if completed else STATUS_FAILED,
+            error=(
+                None
+                if completed
+                else (
+                    "Execution interrupted: owning server process stopped before a "
+                    "terminal response was persisted. Effect outcomes may be unknown; "
+                    f"reconcile exact turn {request_id} before continuing. No automatic replay."
+                )
+            ),
+        )
+    return None
+
+
 def terminalise_one_interrupted_cancellation(
     *,
     current_server_instance_id: str,
@@ -3818,6 +3921,7 @@ def finish_prompt_record(
     status: str = STATUS_COMPLETED,
     error: str | None = None,
     attempt_id: str | None = None,
+    expected_server_instance_id: str | None = None,
 ) -> dict[str, Any]:
     if status not in TERMINAL_STATUSES:
         raise InvalidChatPromptQueueInput("status must be a terminal queue status")
@@ -3841,6 +3945,9 @@ def finish_prompt_record(
         "status": {"$in": ACTIVE_STATUSES},
         **ownership_guard,
     }
+    if expected_server_instance_id is not None:
+        transition_query["server_instance_id"] = expected_server_instance_id
+        transition_query["lease_expires_at"] = {"$lte": now}
     existing_for_terminal = coll.find_one(transition_query)
     if not attempt_id_clean:
         existing = existing_for_terminal

src/backend/services/debug_payload_store.py

@@ -36,6 +36,9 @@
     "stage_diagnostics",
     "tool_history",
     "turn_execution_diagnostics",
+    "turn_execution_record",
+    "turn_execution_record_tool_invocations",
+    "tool_invocations",
     "workflow_discovery",
     "workflow_routing",
     "workflow_routing_diagnostics",

tests/backend/test_chat_history_blob_backfill.py

@@ -11,6 +11,15 @@ def __init__(self, docs: list[dict[str, Any]]):
 
     def _matches_query(self, doc: dict[str, Any], query: dict[str, Any]) -> bool:
         for key, value in query.items():
+            actual = doc
+            for part in key.split("."):
+                actual = (
+                    actual[int(part)] if isinstance(actual, list) else actual.get(part)
+                )
+            if key.startswith("history."):
+                if actual != value:
+                    return False
+                continue
             if isinstance(value, dict) and "$in" in value:
                 if doc.get(key) not in value["$in"]:
                     return False
@@ -33,7 +42,10 @@ def update_one(self, query: dict[str, Any], update: dict[str, Any]):
             if not self._matches_query(doc, query):
                 continue
             for key, value in (update.get("$set") or {}).items():
-                doc[key] = value
+                if key.startswith("history."):
+                    doc["history"][int(key.split(".")[1])] = value
+                else:
+                    doc[key] = value
 
             class _Result:
                 matched_count = 1
@@ -210,3 +222,92 @@ def test_chat_history_blob_backfill_is_idempotent(monkeypatch):
     assert first["entries_updated"] == 1
     assert second["entries_updated"] == 0
     assert second["candidate_entries"] == 0
+
+
+def test_forensic_records_offload_losslessly_and_preserve_concurrent_append(
+    monkeypatch,
+):
+    from copy import deepcopy
+    from src.backend.services import chat_history_service as h
+    from src.backend.services.debug_payload_store import hydrate_debug_payload_blob_refs
+
+    entry = {
+        "role": "assistant",
+        "content": "Done",
+        "llm_debug_data": {
+            "request_id": "exact-request",
+            "turn_execution_record": {"observations": ["x" * 1024] * 70},
+            "turn_execution_record_tool_invocations": [{"args": "y" * 1024}] * 40,
+        },
+    }
+    original = deepcopy(entry)
+    docs = [
+        {
+            "_id": "one",
+            "user_id": "#V#u",
+            "namespace": "#V#u",
+            "session_id": "s",
+            "history": [entry],
+        }
+    ]
+    collection = _FakeCollection(docs)
+    update = collection.update_one
+
+    def concurrent_update(query, changes):
+        docs[0]["history"].append({"role": "user", "content": "New message"})
+        return update(query, changes)
+
+    monkeypatch.setattr(collection, "update_one", concurrent_update)
+    monkeypatch.setattr(
+        h, "get_chat_history_collection_service", lambda **kw: collection
+    )
+    store = _FakeBlobStore()
+    monkeypatch.setattr(
+        "src.backend.services.blob_store.get_blob_store_from_env", lambda: store
+    )
+    result = h.backfill_chat_history_blob_payloads(
+        user_concept_id="#V#u", namespace="#V#u", session_ids=["s"], dry_run=False
+    )
+    assert result["entries_updated"] == 1
+    assert docs[0]["history"][1]["content"] == "New message"
+    assert docs[0]["history"][0]["llm_debug_data"]["request_id"] == "exact-request"
+    assert (
+        hydrate_debug_payload_blob_refs(docs[0]["history"][0], fail_soft=False).payload
+        == original
+    )
+    assert result["bytes_saved_estimate"] > 80000
+
+
+def test_backfill_does_not_overwrite_concurrent_entry_edit(monkeypatch):
+    from src.backend.services import chat_history_service as h
+
+    docs = [
+        {
+            "_id": "one",
+            "user_id": "#V#u",
+            "namespace": "#V#u",
+            "session_id": "s",
+            "history": [{"role": "assistant", "content": "x" * 100000}],
+        }
+    ]
+    collection = _FakeCollection(docs)
+    update = collection.update_one
+
+    def concurrent_edit(query, changes):
+        docs[0]["history"][0] = {"role": "assistant", "content": "Edited by user"}
+        return update(query, changes)
+
+    monkeypatch.setattr(collection, "update_one", concurrent_edit)
+    monkeypatch.setattr(
+        h, "get_chat_history_collection_service", lambda **kw: collection
+    )
+    monkeypatch.setattr(
+        "src.backend.services.blob_store.get_blob_store_from_env",
+        lambda: _FakeBlobStore(),
+    )
+    result = h.backfill_chat_history_blob_payloads(
+        user_concept_id="#V#u", dry_run=False
+    )
+    assert result["entries_updated"] == 0
+    assert result["errors"][0]["type"] == "entry_changed_concurrently"
+    assert docs[0]["history"][0]["content"] == "Edited by user"

tests/backend/test_chat_prompt_queue_dispatch_service.py

@@ -34,6 +34,11 @@ def isolate_dispatcher_queue_reconciliation(monkeypatch):
         "reconcile_next_server_dispatch_handoff",
         lambda: None,
     )
+    monkeypatch.setattr(
+        dispatch_service.chat_prompt_queue_service,
+        "terminalise_one_stopped_server_attempt",
+        lambda **_kwargs: None,
+    )
     monkeypatch.setattr(
         dispatch_service.chat_prompt_queue_service,
         "terminalise_one_interrupted_cancellation",
@@ -99,6 +104,11 @@ def test_dispatcher_terminalises_requested_cancellation_before_new_work(
         "queue_id": "queue-cancelled",
         "status": "cancelled",
     }
+    monkeypatch.setattr(
+        dispatch_service.chat_prompt_queue_service,
+        "terminalise_one_stopped_server_attempt",
+        lambda **_kwargs: None,
+    )
     monkeypatch.setattr(
         dispatch_service.chat_prompt_queue_service,
         "terminalise_one_interrupted_cancellation",