diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index 2fb3dc0ba2..35a7a8fe27 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/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}; } diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts index 15e4611c5c..1352baeb04 100644 --- a/tests/control_plane_ts/conversation_binding.test.ts +++ b/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"}; diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 2eceddc7f9..576e0615a4 100644 --- a/tests/test_lark_private_conversations.py +++ b/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