fix(chat): reject ambiguous calendar task completion readback

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

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):

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)