From d6a2a426e4d533ec0d18a4661952c5a17fca8de7 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 28 Oct 2026 18:04:18 +0800 Subject: [PATCH] fix(chat): verify the complete original calendar task return Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/product/periodic-reports.md | 9 +++ .../native_chat/session_calendar_tasks.py | 9 ++- tests/test_managed_session_calendar_tasks.py | 75 +++++++++++++++++++ 3 files changed, 90 insertions(+), 3 deletions(-) diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index dfbc422c81..c7d30cc4db 100644 --- a/docs/product/periodic-reports.md +++ b/docs/product/periodic-reports.md @@ -979,6 +979,15 @@ completion receipt is distinct from report/provider publication; it never claims an external message was sent. Source and previous-period Tasks may be read from canonical completed history after archival. +Completion reconciliation requires one native result matching the complete +stored Turn result and a strictly readable original message journal. Duplicate +or conflicting result events, malformed JSON and non-object message records +leave completion unconfirmed. The calendar retains its window and configuration; +settings cannot clear that uncertainty as a successful or known failed task. +Once the original records are readable and unambiguous again, a restarted host +can acknowledge that same result without another model execution or answer, +then admit a later independent period. This path does not repair journal bytes. + Before the first deadline, while the Session has accepted user work, or while the schedule is disabled, this path starts no task model. Ordinary calendar and source reads still have local IO costs. Each actual execution uses native diff --git a/loopx/capabilities/native_chat/session_calendar_tasks.py b/loopx/capabilities/native_chat/session_calendar_tasks.py index 733d395a70..78df73d1db 100644 --- a/loopx/capabilities/native_chat/session_calendar_tasks.py +++ b/loopx/capabilities/native_chat/session_calendar_tasks.py @@ -135,11 +135,14 @@ def _completion(self, window): or todo.get("completion_turn_key") != request.get("turn_instance_id") or (response.get("native_loopx") or {}).get("status") != "committed"): return None - if not any(row["kind"] == "native_loopx.result" and row["payload"].get("status") == "committed" - for row in self.controller.store.events_after(actor["session_id"], turn["turn_id"], None)): + results = [row["payload"] for row in self.controller.store.events_after(actor["session_id"], turn["turn_id"], None) + if row["kind"] == "native_loopx.result"] + if results != [response["native_loopx"]]: return None body = response.get("message") - rows = [row for row in self.controller.store.messages(actor["session_id"]) + # Tolerant transcript rendering cannot establish a unique complete + # return: an unreadable record could contain a conflicting reply. + rows = [row for row in self.controller.store.messages(actor["session_id"], strict=True) if row["message_id"] == "managed." + turn["turn_id"] + ".completed"] if (not isinstance(body, str) or not body.strip() or len(rows) != 1 or rows[0]["role"] != "agent" or rows[0]["turn_id"] != turn["turn_id"] or rows[0]["text"] != body): diff --git a/tests/test_managed_session_calendar_tasks.py b/tests/test_managed_session_calendar_tasks.py index 8cfb69ef89..fc2edd6abc 100644 --- a/tests/test_managed_session_calendar_tasks.py +++ b/tests/test_managed_session_calendar_tasks.py @@ -434,6 +434,81 @@ def lost_response(**kwargs): recovered.close() +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("fault", ["malformed-message", "non-object-message", "conflicting-result", "duplicate-result", "none"]) +def test_calendar_completion_requires_complete_unambiguous_original_return( + steward, tmp_path, monkeypatch, provider, fault): + from loopx.chat_runtime import ChatRuntimeController + from loopx.chat_store import ChatSessionStore + + 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) + completed = finish(runtime, pump_steward_monitors(runtime)[0]) + assert completed["response"]["native_loopx"]["status"] == "committed" + before = read_cadence_journal(runtime_root=runtime.coordination_runtime_root, goal_id=actor["goal_id"]) + assert not before.get("task_completion") + messages = store._session_dir(sid) / "messages.jsonl" + events = store._event_path(sid, completed["turn_id"]) + originals = {path: path.read_bytes() for path in (messages, events)} + if fault.endswith("message"): + # The existing body remains readable, but a partial/corrupt journal + # cannot establish that it is the complete, unique original reply. + messages.write_bytes(originals[messages] + ( + b'{"message_id":' if fault == "malformed-message" else b'[]\n')) + elif fault in {"conflicting-result", "duplicate-result"}: + payload = dict(completed["response"]["native_loopx"]) + if fault == "conflicting-result": + payload["summary"] = "A conflicting native result, still labelled committed." + store.append_event(sid, completed["turn_id"], kind="native_loopx.result", payload=payload) + damaged = {path: path.read_bytes() for path in originals} + calls, writes = capture.read_bytes(), list(bot.writes) + service = SessionCalendarTasks(runtime) + health = service.health(sid) + if fault == "none": + assert health["completion_verified"] + else: + assert not health.get("completion_verified"), health + assert request("/api/chat/goal-configuration/preview", { + "goal_id": actor["goal_id"], "capability_id": "periodic_report", "configuration": None})[0] == 400 + point[0] += timedelta(days=1) + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + assert read_cadence_journal(runtime_root=runtime.coordination_runtime_root, goal_id=actor["goal_id"]) == before + assert load_goal_from_registry(runtime.registry_path, actor["goal_id"])["control_plane"]["periodic_report"] == config + assert {path: path.read_bytes() for path in originals} == damaged + point[0] -= timedelta(days=1) + assert capture.read_bytes() == calls and bot.writes == writes + assert store.load_turn(sid, completed["turn_id"]) == completed + # Restore only disposable fixture bytes, then restart the real adapter. + # Product code must neither repair these logs nor replay the old task. + for path, content in originals.items(): + path.write_bytes(content) + runtime.close() + recovered = ChatRuntimeController(store=ChatSessionStore(store.root.parent), codex_bin=runtime.codex_bin, + project_contexts=runtime.project_contexts, registry_path=runtime.registry_path) + try: + service = SessionCalendarTasks(recovered) + assert service.frontier(sid) == (None, False) + receipt = read_cadence_journal(runtime_root=recovered.coordination_runtime_root, + goal_id=actor["goal_id"])["task_completion"] + assert receipt["message_id"] == "managed." + completed["turn_id"] + ".completed" + assert capture.read_bytes() == calls and bot.writes == writes + assert recovered.store.load_turn(sid, completed["turn_id"]) == completed + (workspace / "total.json").write_text('{"total": -1}') + point[0] += timedelta(days=1) + turn, created = service.frontier(sid) + assert created and turn["turn_id"] != completed["turn_id"] + assert finish(recovered, turn)["response"]["native_loopx"]["status"] == "committed" + assert json.loads((workspace / "total.json").read_text()) == {"total": 42} + assert bot.writes == writes + finally: + recovered.close() + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_archived_source_and_completed_period_remain_usable_without_replay(steward, tmp_path, monkeypatch, provider): session, actor, _, capture, _, config = setup_task(steward, tmp_path, monkeypatch, provider)