Skip to content

fix(chat): finish atomic stop and queued Turn delivery #7505

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
loopx-agent merged 9 commits into from
Oct 27, 2026
Merged

fix(chat): finish atomic stop and queued Turn delivery #7505

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
7836bee
fix(chat): keep a completed Turn's event history when a stop races it
JasonBuildAI Oct 27, 2026
43f7ddb
chore(canary): record the reviewed chat module ceilings
JasonBuildAI Oct 27, 2026
7ab114f
fix(chat): serialize a queued claim with the Turn interrupt transition
JasonBuildAI Oct 27, 2026
203590e
test: qualify claim-first and crash-safe interrupt transitions
loopx-agent Oct 27, 2026
01dc6b9
Merge remote-tracking branch 'refs/remotes/release7477-final/main' in…
loopx-agent Oct 27, 2026
7fdf3ca
test: expose queued expiry racing an accepted stop
loopx-agent Oct 27, 2026
318ccc1
fix(chat): serialize queued expiry and acceptance settlement with stops
loopx-agent Oct 27, 2026
b7bd894
test: preserve expired queue outcomes when stop arrives later
loopx-agent Oct 27, 2026
c184163
Merge commit 'b7bd894bf6772d492d486c641888e01b7608cff1' into codex/re…
loopx-agent Oct 27, 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
4 changes: 2 additions & 2 deletions loopx/canary/module_metric_baseline.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,15 +36,15 @@
"loopx/chat_runtime.py": {
"any_count": 37,
"dict_any_count": 0,
"lines": 2319
"lines": 2320
},
"loopx/chat_server.py": {
"any_count": 25,
"dict_any_count": 0,
"lines": 2020
},
"loopx/chat_store.py": {
"lines": 2003
"lines": 2066
},
"loopx/collaboration_mcp.py": {
"lines": 2058
Expand Down
9 changes: 5 additions & 4 deletions loopx/chat_runtime.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2101,13 +2101,14 @@ def interrupt_turn(self, *, session_id: str, turn_id: str) -> dict[str, Any]:
return turn
# Persist stop intent before changing status or contacting native
# IO, so a service crash cannot turn an interrupted stop into an
# automatic review of the work the user just cancelled.
self.store.append_event(session_id, turn_id, kind="turn.interrupt_requested", payload={})
interrupting = self.store.update_turn(
# automatic review of the work the user just cancelled. The store
# checks the status and writes the intent under the Turn lock, so a
# Turn that completed while this call was reading it keeps its
# terminal event history and is not relabelled as interrupted.
interrupting = self.store.begin_turn_interrupt(
session_id,
turn_id,
expected_statuses={"queued", "starting", "running", "interrupting"},
status="interrupting",
)
if interrupting is None:
current = self.store.load_turn(session_id, turn_id)
Expand Down
197 changes: 130 additions & 67 deletions loopx/chat_store.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1168,17 +1168,22 @@ def accept_managed_turn(
payload={},
)
if plan.writes.settle_turn:
settled = self.load_turn(session_token, plan.turn_id)
if settled is None:
raise ValueError(
"chat turn acceptance lost its prepared turn"
)
settled.pop("_acceptance", None)
_atomic_write_json(
with exclusive_file_lock(
self._turn_path(session_token, plan.turn_id),
settled,
preserve_mode=True,
)
agent_id="loopx-chat",
operation="settle_managed_chat_turn_acceptance",
):
settled = self.load_turn(session_token, plan.turn_id)
if settled is None:
raise ValueError(
"chat turn acceptance lost its prepared turn"
)
settled.pop("_acceptance", None)
_atomic_write_json(
self._turn_path(session_token, plan.turn_id),
settled,
preserve_mode=True,
)
payload = settled
else:
current = self.load_turn(session_token, plan.turn_id)
Expand Down Expand Up @@ -1229,32 +1234,35 @@ def _settle_expired_queued_turns(

live_turns: list[dict[str, Any]] = []
for turn in self.queued_turns(session_id):
expires_at = _instant(turn.get("expires_at"))
if expires_at is None or expires_at > now:
live_turns.append(turn)
continue
turn_id = str(turn["turn_id"])
completed = utc_now()
turn.update(
{
"status": "timed_out",
"error_code": "session_queue_expired",
"error": "Queued Agent ingress expired before dispatch.",
"completed_at": completed,
"last_activity_at": completed,
}
)
_atomic_write_json(
self._turn_path(session_id, turn_id),
turn,
preserve_mode=True,
)
self.append_event(
session_id,
turn_id,
kind="turn.failed",
payload={"error_code": "session_queue_expired"},
)
path = self._turn_path(session_id, turn_id)
with exclusive_file_lock(
path, agent_id="loopx-chat", operation="expire_queued_chat_turn",
):
# Queue enumeration is only a candidate; a stop or claim may
# already own the Turn by the time its write lock is acquired.
current = self.load_turn(session_id, turn_id)
if current is None or current.get("status") != "queued":
continue
expires_at = _instant(current.get("expires_at"))
if expires_at is None or expires_at > now:
live_turns.append(current)
continue
completed = utc_now()
current.update(
{
"status": "timed_out",
"error_code": "session_queue_expired",
"error": "Queued Agent ingress expired before dispatch.",
"completed_at": completed,
"last_activity_at": completed,
}
)
_atomic_write_json(path, current, preserve_mode=True)
self.append_event(
session_id, turn_id, kind="turn.failed",
payload={"error_code": "session_queue_expired"},
)
return live_turns

def claim_next_queued_turn(
Expand Down Expand Up @@ -1324,38 +1332,60 @@ def claim_next_queued_turn(
self._recover_queued_turn_input(session_id, turn)
if host_claim_id:
now_text = utc_now()
turn.update(
{
"status": "running",
"host_claim_id": _opaque_id(
host_claim_id,
field="host_claim_id",
),
**(
{
"admitted_goal_instance_id": _opaque_id(
admitted_goal_instance_id,
field="admitted_goal_instance_id",
)
}
if admitted_goal_instance_id is not None
else {}
),
"started_at": turn.get("started_at") or now_text,
"last_activity_at": now_text,
}
)
_atomic_write_json(
self._turn_path(session_id, turn_id),
turn,
preserve_mode=True,
)
self.append_event(
session_id,
turn_id,
kind="turn.claimed_by_attached_host",
payload={"host_claim_id": host_claim_id},
)
turn_path = self._turn_path(session_id, turn_id)
# Share the Turn write lock that `begin_turn_interrupt`
# takes: the claim is the other writer of this Turn's
# status and authority, so the queue scan above is only
# a candidate. Re-reading under the lock means an
# interrupt that already committed wins the transition
# instead of being overwritten by this write, and a
# claim that commits first is never rewritten from a
# snapshot that predates it.
with exclusive_file_lock(
turn_path,
agent_id="loopx-chat",
operation="claim_queued_chat_turn",
):
claimed = self.load_turn(session_id, turn_id)
if claimed is None or claimed.get("status") != "queued":
# Another writer owns this Turn now (an
# interrupt, or a claim from another surface).
# Starting a different queued Turn here would
# leave the queue head in an unclaimed state.
return None
claimed.update(
{
"status": "running",
"host_claim_id": _opaque_id(
host_claim_id,
field="host_claim_id",
),
**(
{
"admitted_goal_instance_id": _opaque_id(
admitted_goal_instance_id,
field="admitted_goal_instance_id",
)
}
if admitted_goal_instance_id is not None
else {}
),
"started_at": claimed.get("started_at") or now_text,
"last_activity_at": now_text,
}
)
_atomic_write_json(
turn_path,
claimed,
preserve_mode=True,
)
self.append_event(
session_id,
turn_id,
kind="turn.claimed_by_attached_host",
payload={"host_claim_id": host_claim_id},
)
turn = claimed
session.update(
{
"status": "busy",
Expand Down Expand Up @@ -1416,6 +1446,39 @@ def update_turn(
_atomic_write_json(path, payload, preserve_mode=True)
return payload

def begin_turn_interrupt(
self,
session_id: str,
turn_id: str,
*,
expected_statuses: set[str],
) -> dict[str, Any] | None:
"""Record stop intent and enter `interrupting` as one Turn transition.

Holding the Turn write lock across the status check, the durable intent
event and the status write closes the completion race: a Turn that
already reached a terminal state cannot gain a stop-intent event after
its terminal outcome, and a Turn this call does interrupt cannot lose
that intent to a later completion. The event still precedes the status
change, so a crash between them leaves the requested stop recorded
rather than silently turning it into an automatic review.
"""
path = self._turn_path(session_id, turn_id)
with exclusive_file_lock(
path,
agent_id="loopx-chat",
operation="begin_chat_turn_interrupt",
):
payload = self.load_turn(session_id, turn_id)
if payload is None:
raise KeyError("chat turn was not found")
if payload.get("status") not in expected_statuses:
return None
self.append_event(session_id, turn_id, kind="turn.interrupt_requested", payload={})
payload["status"] = "interrupting"
_atomic_write_json(path, payload, preserve_mode=True)
return payload

def append_completed_response_events(
self,
session_id: str,
Expand Down
Loading
Loading