diff --git a/src/backend/services/chat_history_service.py b/src/backend/services/chat_history_service.py index 359fa770..af990463 100644 --- a/src/backend/services/chat_history_service.py +++ b/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( { diff --git a/src/backend/services/chat_prompt_queue_dispatch_service.py b/src/backend/services/chat_prompt_queue_dispatch_service.py index 67ad3bfb..5bedac76 100644 --- a/src/backend/services/chat_prompt_queue_dispatch_service.py +++ b/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.""" diff --git a/src/backend/services/chat_prompt_queue_service.py b/src/backend/services/chat_prompt_queue_service.py index 7533b9a0..a7f69ae0 100644 --- a/src/backend/services/chat_prompt_queue_service.py +++ b/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 diff --git a/src/backend/services/debug_payload_store.py b/src/backend/services/debug_payload_store.py index ac190563..fe51f66c 100644 --- a/src/backend/services/debug_payload_store.py +++ b/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", diff --git a/tests/backend/test_chat_history_blob_backfill.py b/tests/backend/test_chat_history_blob_backfill.py index a57fb333..a71d4910 100644 --- a/tests/backend/test_chat_history_blob_backfill.py +++ b/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" diff --git a/tests/backend/test_chat_prompt_queue_dispatch_service.py b/tests/backend/test_chat_prompt_queue_dispatch_service.py index 126352d0..b2348d33 100644 --- a/tests/backend/test_chat_prompt_queue_dispatch_service.py +++ b/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", diff --git a/tests/backend/test_chat_prompt_queue_service.py b/tests/backend/test_chat_prompt_queue_service.py index 6c924592..a885f71d 100644 --- a/tests/backend/test_chat_prompt_queue_service.py +++ b/tests/backend/test_chat_prompt_queue_service.py @@ -2737,3 +2737,92 @@ def test_attachment_only_queue_preserves_exact_descriptor_ids(): assert record["prompt_raw"] == "" persisted = mongo_client.get_chat_prompt_queue_collection().find_one({"queue_id": record["queue_id"]}) assert persisted["execution_envelope"]["image_attachment_ids"] == ["#V#file"] + + +@pytest.mark.parametrize( + "process_state,receipt_status,history_present,expected", + [ + ("dead", None, False, "failed"), + ("dead", "completed", True, "completed"), + ("dead", "failed", True, "failed"), + ("dead", "completed", False, "failed"), + ("live", None, False, None), + ("denied", None, False, None), + ("foreign", None, False, None), + ("renewed", None, False, None), + ], +) +def test_stopped_server_recovery_never_replays_or_steals_live_work( + monkeypatch, process_state, receipt_status, history_present, expected +): + from src.backend.services import ( + chat_history_service as h, + turn_execution_record_service as records, + ) + from unittest.mock import MagicMock + + now = datetime.now(timezone.utc) + host = queue_service.socket.gethostname() + owner = ( + f"{host}:987654:old" if process_state != "foreign" else "other-host:987654:old" + ) + row = { + "queue_id": "dead-turn", + "user_concept_id": "#V#u", + "namespace": "#V#u", + "organisation_concept_id": None, + "session_id": "session", + "client_request_id": "request", + "dispatch_mode": "server", + "status": "in_progress", + "attempt_id": "attempt", + "server_instance_id": owner, + "active_conversation_key": "conversation", + "lease_expires_at": ( + now + timedelta(seconds=90) + if process_state == "renewed" + else now - timedelta(seconds=90) + ), + } + coll = queue_service._collection() + coll.insert_one(row) + + def probe(pid, signal): + if process_state == "live": + return + if process_state == "denied": + raise PermissionError() + raise ProcessLookupError() + + monkeypatch.setattr(queue_service.os, "kill", probe) + history = MagicMock() + history.find_one.return_value = {"_id": 1} if history_present else None + monkeypatch.setattr(h, "get_chat_history_collection_service", lambda **kw: history) + monkeypatch.setattr( + records, + "get_turn_execution_record_projection", + lambda **kw: { + "user_id": "#V#u", + "session_id": "session", + "execution_terminal_status": receipt_status, + }, + ) + result = queue_service.terminalise_one_stopped_server_attempt( + current_server_instance_id=f"{host}:1:current" + ) + if expected is None: + assert result is None + assert coll.find_one({"queue_id": "dead-turn"})["status"] == "in_progress" + else: + assert result["status"] == expected + stored = coll.find_one({"queue_id": "dead-turn"}) + assert "active_conversation_key" not in stored + assert coll.count_documents({}) == 1 + if expected == "failed": + assert "No automatic replay" in stored["last_error"] + assert ( + queue_service.terminalise_one_stopped_server_attempt( + current_server_instance_id=f"{host}:1:current" + ) + is None + )