Skip to content

fix(chat): recover interrupted queued input publication #6895

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
huangruiteng merged 4 commits into from
Oct 19, 2026
+565 −129
Merged

fix(chat): recover interrupted queued input publication #6895

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
4eb10ec
fix(chat): recover interrupted queued input publication
loopx-agent Oct 19, 2026
84a524a
fix(chat): preserve prepared managed acceptance on queue replay
loopx-agent Oct 19, 2026
d82ad56
Merge latest main recovery owners into queued-input fix
loopx-agent Oct 19, 2026
bd4891e
test(chat): check startup cleanup against owned process
loopx-agent Oct 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
173 changes: 171 additions & 2 deletions loopx/chat_ingress.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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,
Expand Down
133 changes: 14 additions & 119 deletions loopx/chat_store.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@

from .chat import (
require_matching_message_replay,
require_matching_replay,
resolve_attached_completion_replay,
)
from .capabilities.steward_executor.allocation import (
Expand All @@ -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,
Expand All @@ -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}$")

Expand Down Expand Up @@ -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")
Expand All @@ -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


Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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(
Expand Down
Loading
Loading