fix(chat): retain timed-out requests after failure notices

loopx/control_plane/collaboration/conversation_binding.ts

@@ -444,6 +444,8 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject {
     const turn = observation.turn == null ? null : requireJsonObject(observation.turn, "observed Turn");
     // Pending work is not current retry eligibility: backoff, disabled bindings,
     // exhausted attempts and unavailable providers do not retire the request.
+    // A delivered timeout notice is not a business result. Its effects can
+    // remain unknown, so visibility must not grant another execution.
     // A delivered partial answer is not completion either. Keep explicit
     // unfinished work and a correlated quota continuation's unresolved gate
     // visible, without changing recovery eligibility or admitting execution.
@@ -463,7 +465,8 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject {
       && (unfinishedProjectWork(response) !== null || quotaGate);
     return {pending: row.command === null && row.status === "accepted"
       && turn?.session_id === row.session_id && turn?.turn_id === row.turn_id
-      && (unfinished || turn?.status === "failed" && turn.error_code === "usage_limit_exceeded")
+      && (unfinished || turn?.status === "failed"
+        && (turn.error_code === "usage_limit_exceeded" || turn.error_code === "response_timeout"))
       && !(current?.session_id === row.session_id && current?.status === "closed")
       && observation.stop_observed !== true};
   }

tests/control_plane_ts/conversation_binding.test.ts

@@ -764,6 +764,26 @@ test("quota backlog projection retires explicit lifecycle stops, not unavailable
   }
 });
 
+test("delivered timeout notices retain the exact failed request, including other executors", () => {
+  const request = {request_ref: "f".repeat(24), command: null, status: "accepted",
+    session_id: "original", turn_id: "timed-out", delivery_verified: true};
+  const turn = {session_id: "original", turn_id: "timed-out", status: "failed", error_code: "response_timeout"};
+  const packet = {request, current_session: {session_id: "original", status: "ready", agent_id: "antigravity"},
+    pending_observation: {turn, stop_observed: false}};
+  assert.deepEqual(planBoundConversationRequest(packet), {pending: true});
+  assert.equal(planBoundConversationRequest({...packet, current_session: null}).pending, true);
+  assert.equal(planBoundConversationRequest({...packet,
+    pending_observation: {turn, stop_observed: true}}).pending, false);
+  assert.equal(planBoundConversationRequest({...packet,
+    current_session: {...packet.current_session, status: "closed"}}).pending, false);
+  for (const changed of [{...turn, session_id: "foreign"}, {...turn, turn_id: "foreign"},
+    {...turn, status: "completed"}, {...turn, error_code: "policy"}]) {
+    assert.equal(planBoundConversationRequest({...packet,
+      pending_observation: {turn: changed, stop_observed: false}}).pending, false);
+  }
+  assert.equal(planBoundConversationRequest({...packet, request: {...request, command: "stop"}}).pending, false);
+});
+
 test("delivered partial results remain pending without authorizing another execution", () => {
   const original = {request_ref: "f".repeat(24), command: null, status: "accepted",
     session_id: "original", turn_id: "partial"};

tests/test_lark_private_conversations.py

@@ -275,6 +275,47 @@ def test_delivered_quota_failure_stays_pending_across_restart_without_replaying(
         runtime.close()
 
 
+@pytest.mark.parametrize("command", ["stop", "new"])
+def test_delivered_timeout_stays_pending_without_replaying_and_honors_stop(ordinary, monkeypatch, command):  # noqa: F811
+    from loopx.chat_agent import CodexChatAgentError
+    from loopx.capabilities.native_chat.codex_adapter import CodexAppServerAdapter
+
+    starts = []
+
+    def timeout_after_start(_adapter, message, event_sink):
+        starts.append(message)
+        event_sink("turn.started", {"upstream_turn_id": "synthetic-timeout"})
+        raise CodexChatAgentError("Synthetic upstream timeout", gate=None, error_code="response_timeout")
+
+    monkeypatch.setattr(CodexAppServerAdapter, "start_turn", timeout_after_start)
+    store, runtime, provider, transport = connect(ordinary)
+    try:
+        transport.admit("notes-app", provider.event("notes-app", "timeout-pending", "Complete this request."))
+        row = transport.core.pending()[0]
+        failed = runtime.wait_for_turn(session_id=row["session_id"], turn_id=row["turn_id"], timeout_sec=10)
+        assert failed["status"] == "failed" and failed["error_code"] == "response_timeout"
+        transport.reconcile()
+        saved = json.loads((transport.root / f"{row['request_ref']}.json").read_text())
+        assert saved["status"] == "delivered" and saved["deliveries"]["terminal"]["verified"]
+        restarted = LarkPrivateConversations(controller=runtime, runtime_root=transport.runtime_root,
+            runner=provider, cli_bin="lark-cli")
+        before = ordinary[4].read_bytes(), list(provider.writes), store.messages(row["session_id"])
+        with monkeypatch.context() as observation_only:
+            observation_only.setattr(runtime, "enqueue_turn", lambda **_: pytest.fail("health admitted work"))
+            observation_only.setattr(transport.core, "recover_failed_request", lambda *_a, **_kw: pytest.fail("health replayed work"))
+            assert transport.health()[row["binding_id"]] == {"pending_count": 1, "recovery_count": 0}
+            assert restarted.health()[row["binding_id"]] == {"pending_count": 1, "recovery_count": 0}
+        assert (ordinary[4].read_bytes(), provider.writes, store.messages(row["session_id"])) == before
+        assert store.load_turn(row["session_id"], row["turn_id"]) == failed
+        assert len(starts) == 1 and "Complete this request." in starts[0]
+        transport.admit("notes-app", provider.event("notes-app", "retire-timeout", f"/{command}"))
+        transport.reconcile()
+        assert transport.health()[row["binding_id"]]["pending_count"] == 0
+        assert len(starts) == 1
+    finally:
+        runtime.close()
+
+
 @pytest.mark.parametrize("command", ["stop", "new"])
 def test_delivered_unfinished_work_remains_visible_without_admitting_recovery(ordinary, monkeypatch, command):  # noqa: F811
     store, runtime, _, _, _, fake, _ = ordinary