diff --git a/loopx/chat_ingress.py b/loopx/chat_ingress.py index 19f6ff8598..f5e041051d 100644 --- a/loopx/chat_ingress.py +++ b/loopx/chat_ingress.py @@ -1,13 +1,182 @@ -"""Durable Chat ingress receipts shared by provider steering and conversation modes.""" +"""Durable queued inputs and ingress receipts for the existing Chat store.""" from __future__ import annotations +from datetime import datetime, timedelta, timezone import os +import uuid from typing import Any -from .chat import require_matching_replay +from .chat import require_matching_message_replay, require_matching_replay from .file_lock import exclusive_file_lock +SESSION_QUEUE_MAX_PENDING = 20 +SESSION_QUEUE_TTL_SECONDS = 3600 + + class ChatIngressStore: + def create_queued_turn( + self, + session_id: str, + *, + client_turn_id: str, + message: str, + attachments: list[dict[str, Any]] | None = None, + goal_instance_id: str | None = None, + ttl_seconds: int = SESSION_QUEUE_TTL_SECONDS, + origin: str = "external", + external_agent_target: dict[str, Any] | None = None, + ) -> tuple[dict[str, Any], bool]: + """Persist one bounded follow-up without replacing the active Turn.""" + + from .chat_store import _opaque_id, _instant, _atomic_write_json, utc_now, CHAT_TURN_SCHEMA_VERSION + + client_id = _opaque_id(client_turn_id, field="client_turn_id") + from .chat_attachments import normalize_chat_image_attachments, validate_chat_turn_envelope + normalized_attachments = normalize_chat_image_attachments(attachments) or None + if normalized_attachments: + validate_chat_turn_envelope({"message": message, "attachments": normalized_attachments}) + session_path = self._session_path(session_id) + with self._session_lock(session_id): + with exclusive_file_lock( + session_path, + agent_id="loopx-chat", + operation="create_chat_queued_turn", + ): + existing = self.turn_for_client(session_id, client_id) + if existing is not None: + require_matching_replay( + existing, + identity="client_turn_id", + request={ + "message": str(message), + "attachments": normalized_attachments, + "origin": _opaque_id(origin, field="origin"), + "external_agent_target": external_agent_target, + }, + ) + # Managed acceptance retains its own message identity and + # visible text. A queue-mode retry only replays that receipt; + # the original typed acceptance owner finishes its writes. + if "_acceptance" not in existing: + self._recover_queued_turn_input(session_id, existing) + return existing, False + session = self.load_session(session_id) + if session is None or session.get("status") == "closed": + raise KeyError("chat session was not found") + now = datetime.now(timezone.utc) + queued = [ + turn + for turn in self.queued_turns(session_id) + if (_instant(turn.get("expires_at")) or now + timedelta(seconds=1)) > now + ] + if len(queued) >= SESSION_QUEUE_MAX_PENDING: + raise RuntimeError("session_queue_full") + turn_id = uuid.uuid4().hex + now_text = now.isoformat().replace("+00:00", "Z") + payload = { + "schema_version": CHAT_TURN_SCHEMA_VERSION, + "turn_id": turn_id, + "session_id": session_id, + **( + { + "goal_instance_id": _opaque_id( + goal_instance_id, + field="goal_instance_id", + ) + } + if goal_instance_id is not None + else {} + ), + "client_turn_id": client_id, + "status": "queued", + **({"external_agent_target": external_agent_target} if external_agent_target is not None else {}), + "message": str(message), + **({"attachments": normalized_attachments} if normalized_attachments else {}), + "origin": _opaque_id(origin, field="origin"), + "upstream_turn_id": None, + "response": None, + "error_code": None, + "error": None, + "created_at": now_text, + "expires_at": (now + timedelta(seconds=max(1, ttl_seconds))) + .isoformat() + .replace("+00:00", "Z"), + "started_at": None, + "first_event_at": None, + "completed_at": None, + "last_activity_at": now_text, + "delta_count": 0, + "sse_reconnect_count": 0, + } + path = self._turn_path(session_id, turn_id) + _atomic_write_json(path, payload) + os.chmod(path, 0o600) + session.update( + { + "last_activity_at": now_text, + "updated_at": utc_now(), + } + ) + _atomic_write_json(session_path, session, preserve_mode=True) + # Keep the cross-process Session fence through publication: + # neither another retry nor a queue claim may overtake it. + self.append_message( + session_id, + role="user", + text=message, + turn_id=turn_id, + origin=origin, + attachments=normalized_attachments, + ) + self.append_event( + session_id, + turn_id, + kind="turn.queued", + payload={"delivery_mode": "session_queue"}, + ) + return payload, True + + def _recover_queued_turn_input(self, session_id: str, turn: dict[str, Any]) -> None: + """Finish interrupted input IO under the caller's Session fence. + + The existing queued event follows the transcript append. It is the + publication marker, so healthy claims need no full transcript scan. + Recovery uses the saved request, never a retry's replacement text. + """ + from .chat_store import _read_jsonl + + if "_acceptance" in turn: + raise ValueError("managed Turn acceptance must finish before queue dispatch") + turn_id = str(turn["turn_id"]) + event_path = self._event_path(session_id, turn_id) + try: + with event_path.open("rb") as stream: + first = stream.readline() + except FileNotFoundError: + first = b"" + if first.strip(): + import json + + event = json.loads(first) + if isinstance(event, dict) and event.get("kind") == "turn.queued": + return + self.flush_events(session_id, turn_id) + events = _read_jsonl(event_path, raise_on_error=True, strict=True) + if any(row["kind"] == "turn.queued" for row in events): + return + inputs = [row for row in _read_jsonl(self._session_dir(session_id) / "messages.jsonl", raise_on_error=True, strict=True) + if row.get("turn_id") == turn_id and row.get("role") == "user"] + expected = {"message_id": "queue." + turn_id, "role": "user", "turn_id": turn_id, + "text": turn["message"], "origin": turn["origin"], + "attachments": turn.get("attachments")} + if inputs: + if len(inputs) != 1: + raise ValueError("queued Turn has ambiguous input messages") + require_matching_message_replay(inputs[0], expected) + else: + self.append_message(session_id, **expected) + self.append_event(session_id, turn_id, kind="turn.queued", payload={"delivery_mode": "session_queue"}) + def create_ingress_receipt( self, session_id: str, diff --git a/loopx/chat_store.py b/loopx/chat_store.py index fe036313e4..a3cb2c109f 100644 --- a/loopx/chat_store.py +++ b/loopx/chat_store.py @@ -16,7 +16,6 @@ from .chat import ( require_matching_message_replay, - require_matching_replay, resolve_attached_completion_replay, ) from .capabilities.steward_executor.allocation import ( @@ -26,7 +25,11 @@ MODEL_SELECTION_UNSET as _MODEL_SELECTION_UNSET, normalize_agent_model_id, normalize_session_model_settings, ) from .chat_event_cache import ChatEventCache -from .chat_ingress import ChatIngressStore +from .chat_ingress import ( + ChatIngressStore, + SESSION_QUEUE_MAX_PENDING as SESSION_QUEUE_MAX_PENDING, + SESSION_QUEUE_TTL_SECONDS as SESSION_QUEUE_TTL_SECONDS, +) from .control_plane.chat_turn_acceptance import ( CHAT_TURN_ACCEPTANCE_CAPSULE_SCHEMA, AcceptedManagedTurn, @@ -47,8 +50,6 @@ TERMINAL_TURN_STATES = {"completed", "interrupted", "timed_out", "failed"} TERMINAL_EVENT_KINDS = {"turn.completed", "turn.failed", "turn.interrupted"} REPLAY_ONLY_EVENT_KINDS = {"answer.delta", "assistant.delta", "agent.phase", "turn.activity"} -SESSION_QUEUE_MAX_PENDING = 20 -SESSION_QUEUE_TTL_SECONDS = 3600 _OPAQUE_ID = re.compile(r"^[A-Za-z0-9._-]{1,160}$") @@ -87,7 +88,7 @@ def _read_json(path: Path) -> dict[str, Any]: def _read_jsonl( - path: Path, *, raise_on_error: bool = False, + path: Path, *, raise_on_error: bool = False, strict: bool = False, ) -> list[dict[str, Any]]: try: lines = path.read_bytes().split(b"\n") @@ -99,12 +100,18 @@ def _read_jsonl( return [] rows: list[dict[str, Any]] = [] for line in lines: + if not line.strip(): + continue try: item = json.loads(line) - except (UnicodeDecodeError, json.JSONDecodeError): + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + if strict: + raise ValueError("invalid Chat journal record") from exc continue if isinstance(item, dict): rows.append(item) + elif strict: + raise ValueError("invalid Chat journal record") return rows @@ -1118,119 +1125,6 @@ def accept_managed_turn( dispatch_reason=plan.dispatch_reason, ) - def create_queued_turn( - self, - session_id: str, - *, - client_turn_id: str, - message: str, - attachments: list[dict[str, Any]] | None = None, - goal_instance_id: str | None = None, - ttl_seconds: int = SESSION_QUEUE_TTL_SECONDS, - origin: str = "external", - external_agent_target: dict[str, Any] | None = None, - ) -> tuple[dict[str, Any], bool]: - """Persist one bounded follow-up without replacing the active Turn.""" - - client_id = _opaque_id(client_turn_id, field="client_turn_id") - from .chat_attachments import normalize_chat_image_attachments, validate_chat_turn_envelope - normalized_attachments = normalize_chat_image_attachments(attachments) or None - if normalized_attachments: - validate_chat_turn_envelope({"message": message, "attachments": normalized_attachments}) - session_path = self._session_path(session_id) - with self._session_lock(session_id): - with exclusive_file_lock( - session_path, - agent_id="loopx-chat", - operation="create_chat_queued_turn", - ): - existing = self.turn_for_client(session_id, client_id) - if existing is not None: - require_matching_replay( - existing, - identity="client_turn_id", - request={ - "message": str(message), - "attachments": normalized_attachments, - "origin": _opaque_id(origin, field="origin"), - "external_agent_target": external_agent_target, - }, - ) - return existing, False - session = self.load_session(session_id) - if session is None or session.get("status") == "closed": - raise KeyError("chat session was not found") - now = datetime.now(timezone.utc) - queued = [ - turn - for turn in self.queued_turns(session_id) - if (_instant(turn.get("expires_at")) or now + timedelta(seconds=1)) > now - ] - if len(queued) >= SESSION_QUEUE_MAX_PENDING: - raise RuntimeError("session_queue_full") - turn_id = uuid.uuid4().hex - now_text = now.isoformat().replace("+00:00", "Z") - payload = { - "schema_version": CHAT_TURN_SCHEMA_VERSION, - "turn_id": turn_id, - "session_id": session_id, - **( - { - "goal_instance_id": _opaque_id( - goal_instance_id, - field="goal_instance_id", - ) - } - if goal_instance_id is not None - else {} - ), - "client_turn_id": client_id, - "status": "queued", - **({"external_agent_target": external_agent_target} if external_agent_target is not None else {}), - "message": str(message), - **({"attachments": normalized_attachments} if normalized_attachments else {}), - "origin": _opaque_id(origin, field="origin"), - "upstream_turn_id": None, - "response": None, - "error_code": None, - "error": None, - "created_at": now_text, - "expires_at": (now + timedelta(seconds=max(1, ttl_seconds))) - .isoformat() - .replace("+00:00", "Z"), - "started_at": None, - "first_event_at": None, - "completed_at": None, - "last_activity_at": now_text, - "delta_count": 0, - "sse_reconnect_count": 0, - } - path = self._turn_path(session_id, turn_id) - _atomic_write_json(path, payload) - os.chmod(path, 0o600) - session.update( - { - "last_activity_at": now_text, - "updated_at": utc_now(), - } - ) - _atomic_write_json(session_path, session, preserve_mode=True) - self.append_message( - session_id, - role="user", - text=message, - turn_id=turn_id, - origin=origin, - attachments=normalized_attachments, - ) - self.append_event( - session_id, - turn_id, - kind="turn.queued", - payload={"delivery_mode": "session_queue"}, - ) - return payload, True - def queued_turns(self, session_id: str) -> list[dict[str, Any]]: session = self.load_session(session_id) if session is None: @@ -1358,6 +1252,7 @@ def claim_next_queued_turn( raise ValueError( "queued Turn Goal instance does not match its Session" ) + self._recover_queued_turn_input(session_id, turn) if host_claim_id: now_text = utc_now() turn.update( diff --git a/tests/test_chat_conversation_bindings.py b/tests/test_chat_conversation_bindings.py index 551bc8cca2..a55c4bbf48 100644 --- a/tests/test_chat_conversation_bindings.py +++ b/tests/test_chat_conversation_bindings.py @@ -438,6 +438,123 @@ def _bound_project(fixture): return bindings, args, binding, session, turn +@pytest.mark.parametrize("committed", [False, True]) +def test_http_queue_retry_retains_original_managed_acceptance(ordinary, monkeypatch, committed): # noqa: F811 + store, runtime, contexts, request, capture, _fake, _workspace = ordinary + status, session = request("/api/chat/sessions", { + "context_kind": "project", "project_ref": contexts.available()[0]["project_ref"]}) + assert status == 201 + sid = session["session_id"] + thread = store.load_session(sid)["upstream_thread_id"] + body = {"message": "Continue the original request 原文", "client_turn_id": "original-managed-input"} + append = store.append_message + + def interrupted(*args, **kwargs): + if kwargs.get("role") == "user": + if committed: + append(*args, **kwargs) + raise OSError("managed input publication interrupted") + return append(*args, **kwargs) + + try: + with monkeypatch.context() as patch: + patch.setattr(store, "append_message", interrupted) + status, failed = request(f"/api/chat/sessions/{sid}/turns", body) + assert status == 503 and failed["error_code"] == "chat_turn_acceptance_unavailable" + prepared = store.turn_for_client(sid, body["client_turn_id"]) + message_id = prepared["_acceptance"]["message_id"] + inputs = store.messages(sid) + for _ in range(2): + status, replay = request(f"/api/chat/sessions/{sid}/turns", {**body, "delivery": "queue"}) + assert status == 202 and not replay["created"] and replay["turn_id"] == prepared["turn_id"] + assert store.messages(sid) == inputs + assert store.events_after(sid, prepared["turn_id"], None) == [] + assert request(f"/api/chat/sessions/{sid}/turns", {**body, "message": "Changed", "delivery": "queue"})[0] == 400 + status, accepted = request(f"/api/chat/sessions/{sid}/turns", body) + assert status == 202 and not accepted["created"] and accepted["turn_id"] == prepared["turn_id"] + result = runtime.wait_for_turn(session_id=sid, turn_id=prepared["turn_id"], timeout_sec=10) + assert result["status"] == "completed" and "_acceptance" not in result + assert store.load_session(sid)["upstream_thread_id"] == thread + assert store.load_session(sid)["goal_id"] is None + inputs = [row for row in store.messages(sid) if row["role"] == "user"] + assert len(inputs) == 1 and inputs[0]["message_id"] == message_id and inputs[0]["text"] == body["message"] + calls = [json.loads(line) for line in capture.read_text().splitlines()] + assert sum(row.get("method") == "turn/start" for row in calls) == 1 + finally: + runtime.close() + + +@pytest.mark.parametrize("committed", [False, True]) +def test_bound_http_retry_repairs_interrupted_queue_input_once(ordinary, monkeypatch, committed): # noqa: F811 + store, runtime, _contexts, request, capture, _fake, workspace = ordinary + _bindings, _args, _binding, session, active = _bound_project(ordinary) + sid = session["session_id"] + original_thread = store.load_session(sid)["upstream_thread_id"] + message = "Continue with the complete original input: 原文\nsecond line" + append = store.append_message + + def interrupted(*args, **kwargs): + if kwargs.get("role") == "user" and kwargs.get("text") == message: + if committed: + append(*args, **kwargs) + raise OSError("queued input publication interrupted") + return append(*args, **kwargs) + + with monkeypatch.context() as patch: + patch.setattr(store, "append_message", interrupted) + with pytest.raises(OSError, match="queued input publication interrupted"): + runtime.enqueue_turn(session_id=sid, client_turn_id="original-input", message=message, + work_dir=workspace, objective="", origin="web") + original = store.turn_for_client(sid, "original-input") + assert original["status"] == "queued" + assert store.load_session(sid)["active_turn_id"] == active["turn_id"] + body = {"message": message, "client_turn_id": "original-input", "delivery": "queue"} + for _ in range(2): + status, recovered = request(f"/api/chat/sessions/{sid}/turns", body) + assert status == 202 and not recovered["created"] + assert recovered["turn_id"] == original["turn_id"] + inputs = [row for row in store.messages(sid) if row["turn_id"] == original["turn_id"] and row["role"] == "user"] + assert len(inputs) == 1 and inputs[0]["text"] == message and inputs[0]["origin"] == "web" + assert sum(row["kind"] == "turn.queued" for row in store.events_after(sid, original["turn_id"], None)) == 1 + assert store.load_session(sid)["active_turn_id"] == active["turn_id"] + runtime.interrupt_turn(session_id=sid, turn_id=active["turn_id"]) + assert runtime.wait_for_turn(session_id=sid, turn_id=original["turn_id"], timeout_sec=10)["status"] == "completed" + assert store.load_session(sid)["upstream_thread_id"] == original_thread + assert store.load_session(sid)["goal_id"] is None + calls = [json.loads(line) for line in capture.read_text().splitlines()] + assert sum(row.get("method") == "turn/start" for row in calls) == 2 + status, replay = request(f"/api/chat/sessions/{sid}/turns", body) + assert status == 202 and not replay["created"] and replay["turn_id"] == original["turn_id"] + assert [row for row in store.messages(sid) if row["turn_id"] == original["turn_id"] and row["role"] == "user"] == inputs + + +def test_bound_queue_recovery_does_not_bypass_revoked_source(ordinary, monkeypatch): # noqa: F811 + store, runtime, _contexts, request, _capture, _fake, workspace = ordinary + bindings, _args, binding, session, _active = _bound_project(ordinary) + sid = session["session_id"] + append = store.append_message + + def interrupted(*args, **kwargs): + if kwargs.get("role") == "user" and kwargs.get("text") == "Original queued request": + raise OSError("queued input publication interrupted") + return append(*args, **kwargs) + + with monkeypatch.context() as patch: + patch.setattr(store, "append_message", interrupted) + with pytest.raises(OSError): + runtime.enqueue_turn(session_id=sid, client_turn_id="revoked-input", message="Original queued request", + work_dir=workspace, objective="", origin="web") + original = store.turn_for_client(sid, "revoked-input") + bindings.disconnect(binding["binding_id"], expected_revision=bindings.read()["revision"]) + before = store.messages(sid) + status, _ = request(f"/api/chat/sessions/{sid}/turns", { + "message": "Original queued request", "client_turn_id": "revoked-input", "delivery": "queue"}) + assert status == 400 + assert store.messages(sid) == before + assert store.load_turn(sid, original["turn_id"])["status"] == "interrupted" + assert not store.load_turn(sid, original["turn_id"]).get("upstream_turn_id") + + @pytest.mark.parametrize("entrypoint", ["http", "runtime_context"]) def test_bound_session_identity_is_checked_before_input_or_runtime_resolution(ordinary, monkeypatch, entrypoint): # noqa: F811 from loopx.capabilities.native_chat.external_conversations import ChatExternalConversations diff --git a/tests/test_chat_queue_preparation.py b/tests/test_chat_queue_preparation.py index 7919cdad32..b1bca36c75 100644 --- a/tests/test_chat_queue_preparation.py +++ b/tests/test_chat_queue_preparation.py @@ -3,6 +3,7 @@ import json from pathlib import Path import runpy +import subprocess import sys import threading import time @@ -37,6 +38,252 @@ def session_runtime(tmp_path, *, channel="goal.public-research"): ) +@pytest.mark.parametrize("failure", ["before_message", "after_message", "before_event"]) +@pytest.mark.parametrize("display", ["Execution request", "Original visible request 原文"]) +def test_queue_replay_leaves_prepared_managed_input_with_its_owner(tmp_path, monkeypatch, failure, display): + """A queue-mode retry cannot replace the managed owner's display or identity.""" + store, sid, _ = session_runtime(tmp_path) + with monkeypatch.context() as fault: + append_message = store.append_message + + def interrupted_message(*args, **kwargs): + if failure == "before_message": + raise OSError("managed acceptance interrupted") + row = append_message(*args, **kwargs) + if failure == "after_message": + raise OSError("managed acceptance interrupted") + return row + + fault.setattr(store, "append_message", interrupted_message) + if failure == "before_event": + fault.setattr(store, "append_event", lambda *_, **__: (_ for _ in ()).throw(OSError("managed acceptance interrupted"))) + with pytest.raises(OSError, match="managed acceptance interrupted"): + store.create_turn(sid, client_turn_id="original", message="Execution request", + display_message=display, origin="web") + + prepared = store.turn_for_client(sid, "original") + assert prepared and "_acceptance" in prepared + # Reopen after the crash; the durable capsule belongs to managed acceptance. + restarted = ChatSessionStore(tmp_path) + before_messages = restarted.messages(sid) + before_events = restarted.events_after(sid, prepared["turn_id"], None) + replay, created = restarted.create_queued_turn( + sid, client_turn_id="original", message="Execution request", origin="web") + assert not created and replay == prepared + assert restarted.messages(sid) == before_messages + assert restarted.events_after(sid, prepared["turn_id"], None) == before_events + accepted = restarted.accept_managed_turn(sid, client_turn_id="original", message="Execution request", + display_message=display, origin="web") + assert accepted.turn["turn_id"] == prepared["turn_id"] and "_acceptance" not in accepted.turn + assert accepted.dispatch_required + inputs = restarted.messages(sid) + assert len(inputs) == 1 and inputs[0]["text"] == display + assert inputs[0]["message_id"] == prepared["_acceptance"]["message_id"] + + +def test_queue_claim_cannot_dispatch_prepared_managed_acceptance(tmp_path, monkeypatch): + """A crash before activation still requires the original acceptance owner.""" + import loopx.chat_store as chat_store + + store, sid, _ = session_runtime(tmp_path) + write = chat_store._atomic_write_json + + def interrupted(path, value, **kwargs): + if path == store._session_path(sid): + raise OSError("activation interrupted") + return write(path, value, **kwargs) + + with monkeypatch.context() as fault: + fault.setattr(chat_store, "_atomic_write_json", interrupted) + with pytest.raises(OSError, match="activation interrupted"): + store.create_turn(sid, client_turn_id="original", message="Execution request", + display_message="Original visible request", origin="web") + prepared = store.turn_for_client(sid, "original") + restarted = ChatSessionStore(tmp_path) + assert restarted.load_session(sid)["active_turn_id"] is None + with pytest.raises(ValueError, match="managed.*acceptance"): + restarted.claim_next_queued_turn(sid, host_claim_id="queue-host") + assert restarted.turn_for_client(sid, "original") == prepared + assert restarted.messages(sid) == [] + assert restarted.events_after(sid, prepared["turn_id"], None) == [] + assert restarted.load_session(sid)["active_turn_id"] is None + accepted = restarted.accept_managed_turn(sid, client_turn_id="original", message="Execution request", + display_message="Original visible request", origin="web") + assert accepted.dispatch_required and accepted.turn["turn_id"] == prepared["turn_id"] + assert restarted.messages(sid)[0]["text"] == "Original visible request" + + +@pytest.mark.parametrize("failure", ["before_message", "after_message", "before_event", "after_event"]) +@pytest.mark.parametrize("recovery", ["retry", "claim"]) +def test_queued_input_publication_recovers_after_restart(tmp_path, monkeypatch, failure, recovery): + """A durable queue entry must supply its original input exactly once.""" + store, sid, _ = session_runtime(tmp_path) + message = "Keep the complete request: 原文\nsecond line" + attachments = [{"id": "fixture", "name": "pixel.png", "data_url": + "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+a1ioAAAAASUVORK5CYII="}] + append_message, append_event = store.append_message, store.append_event + + def interrupted_message(*args, **kwargs): + if failure == "before_message": + raise OSError("input publication interrupted") + value = append_message(*args, **kwargs) + if failure == "after_message": + raise OSError("input publication interrupted") + return value + + def interrupted_event(*args, **kwargs): + if failure == "before_event": + raise OSError("input publication interrupted") + value = append_event(*args, **kwargs) + if failure == "after_event": + raise OSError("input publication interrupted") + return value + + monkeypatch.setattr(store, "append_message", interrupted_message) + monkeypatch.setattr(store, "append_event", interrupted_event) + with pytest.raises(OSError, match="input publication interrupted"): + store.create_queued_turn(sid, client_turn_id="original", message=message, origin="external", attachments=attachments) + original = store.turn_for_client(sid, "original") + assert original and original["status"] == "queued" + + # Reopen the real store, discarding any process-local event buffer/locks. + restarted = ChatSessionStore(tmp_path) + if recovery == "retry": + for _ in range(2): + replay, created = restarted.create_queued_turn( + sid, client_turn_id="original", message=message, origin="external", attachments=attachments) + assert not created and replay == original + claimed = restarted.claim_next_queued_turn(sid, host_claim_id="original-host") + else: + claimed = restarted.claim_next_queued_turn(sid, host_claim_id="original-host") + assert claimed["turn_id"] == original["turn_id"] + assert restarted.load_session(sid)["upstream_thread_id"] == "public-thread" + assert [row["turn_id"] for row in restarted.queued_turns(sid)] == [] + inputs = [row for row in restarted.messages(sid) if row["role"] == "user"] + assert len(inputs) == 1 + assert (inputs[0]["turn_id"], inputs[0]["text"], inputs[0]["origin"]) == ( + original["turn_id"], message, "external") + assert inputs[0]["attachments"] == original["attachments"] + events = restarted.events_after(sid, original["turn_id"], None) + assert sum(row["kind"] == "turn.queued" for row in events) == 1 + assert restarted.claim_next_queued_turn(sid, host_claim_id="original-host") == claimed + + +def test_queued_input_recovery_refuses_changed_retry_before_repair(tmp_path, monkeypatch): + store, sid, _ = session_runtime(tmp_path) + monkeypatch.setattr(store, "append_message", lambda *_, **__: (_ for _ in ()).throw(OSError("interrupted"))) + with pytest.raises(OSError): + store.create_queued_turn(sid, client_turn_id="original", message="Original", origin="external") + restarted = ChatSessionStore(tmp_path) + original = restarted.turn_for_client(sid, "original") + for change in [{"message": "Changed"}, {"origin": "web"}]: + with pytest.raises(ValueError, match="different request"): + restarted.create_queued_turn(sid, **{"client_turn_id": "original", "message": "Original", + "origin": "external", **change}) + assert restarted.messages(sid) == [] + assert restarted.turn_for_client(sid, "original") == original + assert restarted.load_session(sid)["active_turn_id"] is None + + +def test_queued_input_recovery_rejects_conflicting_transcript_before_claim(tmp_path, monkeypatch): + from loopx.chat import ChatMessagePayloadConflictError + + store, sid, _ = session_runtime(tmp_path) + append = store.append_message + monkeypatch.setattr(store, "append_message", lambda *_, **__: (_ for _ in ()).throw(OSError("interrupted"))) + with pytest.raises(OSError): + store.create_queued_turn(sid, client_turn_id="original", message="Original", origin="external") + original = store.turn_for_client(sid, "original") + append(sid, role="user", text="Conflicting input", turn_id=original["turn_id"], origin="external") + restarted = ChatSessionStore(tmp_path) + before = restarted.messages(sid) + with pytest.raises(ChatMessagePayloadConflictError): + restarted.claim_next_queued_turn(sid, host_claim_id="original-host") + assert restarted.messages(sid) == before + assert restarted.turn_for_client(sid, "original") == original + assert restarted.load_session(sid)["active_turn_id"] is None + + +@pytest.mark.parametrize("lane", ["message", "event"]) +def test_queued_input_recovery_keeps_unreadable_publication_unresolved(tmp_path, monkeypatch, lane): + store, sid, _ = session_runtime(tmp_path) + monkeypatch.setattr(store, "append_message", lambda *_, **__: (_ for _ in ()).throw(OSError("interrupted"))) + with pytest.raises(OSError): + store.create_queued_turn(sid, client_turn_id="original", message="Original", origin="external") + original = store.turn_for_client(sid, "original") + path = (store.root / "sessions" / sid / "messages.jsonl" if lane == "message" + else store._event_path(sid, original["turn_id"])) + path.write_text('{"incomplete":') + restarted = ChatSessionStore(tmp_path) + with pytest.raises(ValueError): + restarted.claim_next_queued_turn(sid, host_claim_id="original-host") + assert path.read_text() == '{"incomplete":' + assert restarted.turn_for_client(sid, "original") == original + assert restarted.load_session(sid)["active_turn_id"] is None + + +def test_queued_input_replay_does_not_reinterpret_later_execution_events(tmp_path): + store, sid, _ = session_runtime(tmp_path) + original, _ = store.create_queued_turn(sid, client_turn_id="original", message="Original", origin="external") + inputs = store.messages(sid) + path = store._event_path(sid, original["turn_id"]) + # Input publication completed before this unrelated interrupted output. + with path.open("ab") as stream: + stream.write(b'{"partial_execution_event":') + events = path.read_bytes() + replay, created = ChatSessionStore(tmp_path).create_queued_turn( + sid, client_turn_id="original", message="Original", origin="external") + assert not created and replay == original + assert store.messages(sid) == inputs and path.read_bytes() == events + + +def test_queue_claim_cannot_overtake_input_publication(tmp_path, monkeypatch): + from concurrent.futures import ThreadPoolExecutor + import subprocess + + store, sid, _ = session_runtime(tmp_path) + entered, release = threading.Event(), threading.Event() + append = store.append_message + + def delayed(*args, **kwargs): + entered.set() + assert release.wait(10) + return append(*args, **kwargs) + + monkeypatch.setattr(store, "append_message", delayed) + child = None + with ThreadPoolExecutor(max_workers=1) as workers: + created = workers.submit(store.create_queued_turn, sid, client_turn_id="original", message="Original", origin="external") + try: + assert entered.wait(5) + # A second process has no shared Python locks or event cache. + script = '''import json,sys +from pathlib import Path +from loopx.chat_store import ChatSessionStore +store=ChatSessionStore(Path(sys.argv[1])); sid=sys.argv[2] +print("claiming",flush=True) +turn=store.claim_next_queued_turn(sid,host_claim_id="original-host") +inputs=[row for row in store.messages(sid) if row["turn_id"]==turn["turn_id"] and row["role"]=="user"] +assert len(inputs)==1 and inputs[0]["text"]=="Original", inputs +print(json.dumps({"turn_id":turn["turn_id"]}),flush=True) +''' + child = subprocess.Popen([sys.executable, "-c", script, str(tmp_path), sid], cwd=tmp_path, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + assert child.stdout.readline().strip() == "claiming" + with pytest.raises(subprocess.TimeoutExpired): + child.wait(timeout=.2) + release.set() + turn, was_created = created.result(timeout=5) + output, errors = child.communicate(timeout=5) + assert child.returncode == 0, errors + assert was_created and json.loads(output)["turn_id"] == turn["turn_id"] + finally: + release.set() + if child is not None and child.poll() is None: + child.kill() + child.communicate(timeout=5) + + def drain(runtime, session_id, tmp_path): runtime.resume_session_queue(session_id=session_id, work_dir=tmp_path, objective="Research") # A worker that already retired has completed its queue cleanup. @@ -48,12 +295,27 @@ def drain(runtime, session_id, tmp_path): @pytest.fixture -def codex_startup(tmp_path): +def codex_startup(tmp_path, monkeypatch): """Exercise the real stdio handshake and cleanup, without model execution.""" capture = tmp_path / "requests.jsonl" attempts = tmp_path / "attempts" entered, release = tmp_path / "entered", tmp_path / "release" source = runpy.run_path(str(Path(__file__).parents[1] / "examples/loopx-chat-runtime-smoke.py"))["FAKE_CODEX"] + native_processes = [] + popen = subprocess.Popen + + def start_process(command, *args, **kwargs): + owned = isinstance(command, list) and command and command[0] == str(tmp_path / "codex") + if owned and native_processes: + # Check the actual owned handle before another process starts. + # The child must not infer cleanup from a potentially reused PID. + assert native_processes[-1].returncode is not None, "previous startup process must exit and be reaped before retry" + process = popen(command, *args, **kwargs) + if owned: + native_processes.append(process) + return process + + monkeypatch.setattr(subprocess, "Popen", start_process) def start(mode, *, stage="thread/resume"): script = source.replace("#!/usr/bin/env python3", f"#!{sys.executable}") @@ -62,13 +324,6 @@ def start(mode, *, stage="thread/resume"): from pathlib import Path attempts = Path({str(attempts)!r}) previous = json.loads(attempts.read_text()) if attempts.exists() else [] -if previous: - try: - os.kill(previous[-1], 0) - except ProcessLookupError: - pass - else: - raise AssertionError("previous startup process must be closed") attempt = len(previous) + 1 attempts.write_text(json.dumps([*previous, os.getpid()]))''') script = script.replace(' method = request.get("method")', f'''