From acf7df934c9c34449bda38988c1de4132a1e2700 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 27 Oct 2026 12:47:19 +0800 Subject: [PATCH] fix(chat): recover committed replies after lost event acknowledgments Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/product/periodic-reports.md | 6 ++ .../capabilities/native_chat/steward_turn.py | 14 ++- tests/test_managed_session_calendar_tasks.py | 101 +++++++++++++++++- 3 files changed, 115 insertions(+), 6 deletions(-) diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index ff16b2e654..aad46a8211 100644 --- a/docs/product/periodic-reports.md +++ b/docs/product/periodic-reports.md @@ -942,6 +942,12 @@ admission, a Task lease, independent validation and quota settlement. Token usage and fees depend on the configured host; the calendar makes no savings claim. Duplicate observations and a lost completion acknowledgement reconcile the existing Task and Core message without another model execution or answer. +If the native transaction has committed and the Chat result-event append loses +its acknowledgment, that same invocation may finish returning the original +answer after reading exactly one event with the complete matching result. +Missing, conflicting, duplicate or unreadable result events, and uncommitted +transactions, retain their failure. This readback neither repeats native +execution or settlement nor reopens historical failed Chat Turns. Set `enabled=false` to pause new work and delivery; retain the rest of the configuration to resume that plan. Clearing the override after a known failed diff --git a/loopx/capabilities/native_chat/steward_turn.py b/loopx/capabilities/native_chat/steward_turn.py index 4faeed3b11..0d59b7b1c1 100644 --- a/loopx/capabilities/native_chat/steward_turn.py +++ b/loopx/capabilities/native_chat/steward_turn.py @@ -736,7 +736,19 @@ def launch(): "native_task_validator": self._report_validator(request)} if report_work else {"native_task_validator": managed_turn.validator(request, turn_id=turn_id), "native_completion_lease_resolver": lambda: self._task_lease_proof(managed_lease_request, managed_lease)} if managed_turn else {})) - self.store.append_event(session_id, turn_id, kind="native_loopx.result", payload=result) + try: + self.store.append_event(session_id, turn_id, kind="native_loopx.result", payload=result) + except OSError: + # The native transaction already owns execution and settlement. + # A lost Chat append acknowledgment must not discard its answer + # when this exact committed event is independently readable. + # Reuse the store's ordered readback without re-entering native work. + if result.get("ok") is not True or result.get("status") != "committed": + raise + recorded = [row for row in self.store.events_after(session_id, turn_id, None) + if row["kind"] == "native_loopx.result"] + if len(recorded) != 1 or recorded[0]["payload"] != result: + raise if request.get("work_kind") == "report" and result.get("status") == "committed": from ..periodic_report.private_generation import attempt_directory, validate_private_generation subscription, window = reports.current_window(report_owner_key) diff --git a/tests/test_managed_session_calendar_tasks.py b/tests/test_managed_session_calendar_tasks.py index d36bde72e8..cf692ac085 100644 --- a/tests/test_managed_session_calendar_tasks.py +++ b/tests/test_managed_session_calendar_tasks.py @@ -32,19 +32,20 @@ def settings_server(steward): store, runtime, _, transport, *_ = steward server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler) server.chat_store, server.runtime_controller = store, runtime + server.runtime_root = runtime.coordination_runtime_root server.registry_path, server.runtime_root_override = runtime.registry_path, store.root.parent server.lark_private_conversations = transport server.lark_goal_topic_runtime = SimpleNamespace(health_snapshot=lambda: {}, close=lambda: None) server.verbose = False thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() - def request(path, body, origin=None): + def request(path, body, origin=None, *, method="POST"): connection = http.client.HTTPConnection(*server.server_address, timeout=20) try: headers = {"Content-Type": "application/json"} if origin: headers["Origin"] = origin - connection.request("POST", path, json.dumps(body), headers=headers) + connection.request(method, path, None if body is None else json.dumps(body), headers=headers) response = connection.getresponse() return response.status, json.loads(response.read()) finally: @@ -430,6 +431,58 @@ def test_frozen_task_cannot_change_work_or_scope_before_dispatch(steward, tmp_pa assert capture.read_bytes() == calls and bot.writes == writes +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_committed_calendar_result_survives_lost_event_ack_without_reexecution(steward, tmp_path, monkeypatch, provider): + session, actor, _, capture, workspace, config = setup_task(steward, tmp_path, monkeypatch, provider) + point = fixed_clock(monkeypatch) + store, runtime, bot, *_ = steward + sid = session["session_id"] + with settings_server(steward) as request: + apply_configuration(request, actor["goal_id"], config) + append, lost = store.append_event, [] + before = capture.read_bytes() + def lose_ack_after_write(*args, **kwargs): + result = append(*args, **kwargs) + if kwargs.get("kind") == "native_loopx.result" and not lost: + lost.append(True) + raise OSError("synthetic lost acknowledgment after the committed result event persisted") + return result + with monkeypatch.context() as lost_ack: + lost_ack.setattr(store, "append_event", lose_ack_after_write) + completed = finish(runtime, pump_steward_monitors(runtime)[0]) + assert lost and completed["status"] == "completed", completed + assert completed["response"]["native_loopx"]["status"] == "committed" + assert completed["response"]["native_loopx"]["effects"]["quota_spent"] is True + result_events = [row for row in store.events_after(sid, completed["turn_id"], None) + if row["kind"] == "native_loopx.result"] + assert len(result_events) == 1 + assert result_events[0]["payload"] == completed["response"]["native_loopx"] + messages = [row for row in store.messages(sid) + if row.get("turn_id") == completed["turn_id"] and row["role"] == "agent"] + assert len(messages) == 1 and messages[0]["text"] == "The independently validated total is 42." + status, conversation = request("/api/chat/sessions/" + sid, None, method="GET") + assert status == 200 and conversation["active_turn"] is None + assert messages[0] in conversation["messages"] + calls = [json.loads(line) for line in capture.read_bytes()[len(before):].decode().splitlines()] + assert len([row for row in calls if row.get("method") == "turn/start"]) == 1 + after = capture.read_bytes(), store.messages(sid), list(bot.writes) + for _ in range(2): + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + journal = read_cadence_journal(runtime_root=runtime.coordination_runtime_root, goal_id=actor["goal_id"]) + assert journal["task_completion"]["message_id"] == messages[0]["message_id"] + assert (capture.read_bytes(), store.messages(sid), list(bot.writes)) == after + (workspace / "total.json").write_text('{"total": -1}') + point[0] += timedelta(days=1) + runtime._steward_monitor_next_check.clear() + second = finish(runtime, pump_steward_monitors(runtime)[0]) + assert second["turn_id"] != completed["turn_id"] + assert second["response"]["native_loopx"]["status"] == "committed" + assert json.loads((workspace / "total.json").read_text()) == {"total": 42} + calls = [json.loads(line) for line in capture.read_bytes()[len(before):].decode().splitlines()] + assert len([row for row in calls if row.get("method") == "turn/start"]) == 2 + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_unknown_postcommit_return_never_becomes_cancelled_failure_or_model_replay(steward, tmp_path, monkeypatch, provider): session, actor, _, capture, _, config = setup_task(steward, tmp_path, monkeypatch, provider) @@ -440,11 +493,10 @@ def test_unknown_postcommit_return_never_becomes_cancelled_failure_or_model_repl append = store.append_event lost = [] def lose_native_return(*args, **kwargs): - result = append(*args, **kwargs) if kwargs.get("kind") == "native_loopx.result" and not lost: lost.append(True) - raise OSError("synthetic lost return after the original native commit") - return result + raise OSError("synthetic missing Chat result event after the original native commit") + return append(*args, **kwargs) with monkeypatch.context() as lost_return: lost_return.setattr(store, "append_event", lose_native_return) failed = finish(runtime, pump_steward_monitors(runtime)[0]) @@ -461,6 +513,45 @@ def lose_native_return(*args, **kwargs): assert store.load_turn(session["session_id"], failed["turn_id"]) == failed +@pytest.mark.parametrize("fault", ["conflicting", "duplicate", "unreadable", "uncommitted"]) +def test_lost_result_ack_requires_one_exact_readable_commit(steward, tmp_path, monkeypatch, fault): + session, actor, _, capture, _, config = setup_task(steward, tmp_path, monkeypatch, "sqlite") + fixed_clock(monkeypatch) + store, runtime, bot, *_ = steward + if fault == "uncommitted": + script = tmp_path / "calendar-model.py" + script.write_text(script.read_text().replace('"result_kind": "validated_completion"', + '"result_kind": "validated_progress"')) + with settings_server(steward) as request: + apply_configuration(request, actor["goal_id"], config) + append, read, lost = store.append_event, store.events_after, [] + def lose_ack(*args, **kwargs): + if kwargs.get("kind") != "native_loopx.result" or lost: + return append(*args, **kwargs) + lost.append(args[1]) + observed = {**kwargs, "payload": {**kwargs["payload"], "ok": False}} if fault == "conflicting" else kwargs + append(*args, **observed) + if fault == "duplicate": + append(*args, **kwargs) + raise OSError("synthetic result acknowledgment unavailable") + def unavailable_read(session_id, turn_id, event_id): + if fault == "unreadable" and turn_id in lost: + raise OSError("synthetic result readback unavailable") + return read(session_id, turn_id, event_id) + with monkeypatch.context() as lost_ack: + lost_ack.setattr(store, "append_event", lose_ack) + lost_ack.setattr(store, "events_after", unavailable_read) + failed = finish(runtime, pump_steward_monitors(runtime)[0]) + assert lost and failed["status"] == "failed", failed + before = capture.read_bytes(), store.messages(session["session_id"]), list(bot.writes) + for _ in range(2): + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + assert not SessionCalendarTasks(runtime).health(session["session_id"])["completion_verified"] + assert (capture.read_bytes(), store.messages(session["session_id"]), list(bot.writes)) == before + assert store.load_turn(session["session_id"], failed["turn_id"]) == failed + + def test_partial_progress_keeps_period_open_without_an_automatic_retry(steward, tmp_path, monkeypatch): session, actor, _, capture, _, config = setup_task(steward, tmp_path, monkeypatch, "sqlite") fixed_clock(monkeypatch)