test: qualify managed calendar Task lifecycle across restarts

docs/product/periodic-reports.md

@@ -1052,6 +1052,24 @@ instance. `observed.json` records artifact, completion, retained history and
 actual post-setup model calls. This qualifies neither live-model understanding,
 complete Chat-process restart nor production delivery.
 
+For a controller-restart check, stop that wake owner, await accepted work, then
+write a unique operation token to `restart-runtime` in the same output directory.
+The fixture closes the old controller and reconstructs the normal controller,
+Chat store and transport over the original physical state, keeping its HTTP URL.
+`restart-readback.json` verifies the original Session, actor and message history
+and that restarting itself added no model or delivery effect. The wake owner
+remains stopped until `stop-existing-wake` is removed. Reusing a token within
+that fixture lifetime does not restart again. Missing stop or pending original
+work writes `restart-rejected.json` and leaves the controller running; that
+token cannot apply later after the condition changes. This is an isolated controller
+restart inside the fixture process, not an OS-process or installed-service test.
+
+`test_same_calendar_task_lifecycle_retains_results_and_rejects_late_dispatch`
+joins settings changes, pause/cancel, restart, original result readback,
+withdrawn Agent registration, late dispatch and three distinct calendar windows
+in one File/SQLite journey. Paused and withdrawn stages must add no model call
+or reply; each admitted window must have exactly one original Core result.
+
 Callers can discover eligible references on demand with
 `GET /api/chat/sessions/<original-session-id>/native-turn?calendar_task_sources=true`.
 The `calendar_task_sources` view reads the original Goal's canonical completed

examples/personal-workspace-browser/steward_report_fixture.py

@@ -151,6 +151,8 @@ def fail_report(_adapter, _prompt, event_sink):
                 "calendar_task_configuration": configuration} if args.managed_calendar_task else {})}))
         wake, outbound, delivery = None, None, None
         owner_start_count = 0
+        restart_tokens = set()
+        restart_refusals = set()
         try:
             while not stop.wait(.2):
                 server.assets_dir = current_assets if (output / "use-current-assets").exists() else baseline_assets
@@ -159,6 +161,58 @@ def fail_report(_adapter, _prompt, event_sink):
                 if owners_stopped and wake:
                     wake.close()
                     wake = None
+                restart_file = output / "restart-runtime"
+                restart_token = restart_file.read_text().strip() if restart_file.is_file() else None
+                if (args.managed_calendar_task and restart_token
+                        and restart_token not in restart_tokens | restart_refusals):
+                    with runtime.lock:
+                        restart_busy = (not owners_stopped or store.load_session(sid).get("active_turn_id")
+                            or any(not done.is_set() for done in runtime.turn_done_events.values()))
+                    if restart_busy:
+                        restart_refusals.add(restart_token)
+                        (output / "restart-rejected.json").write_text(json.dumps({
+                            "operation": restart_token, "reason": "stop_wake_and_await_original_work",
+                            "session_id": sid, "runtime_restarted": False, "synthetic_only": True}))
+                if (args.managed_calendar_task and restart_token
+                        and restart_token not in restart_tokens | restart_refusals):
+                    # Disposable QA only: reconstruct the normal controller over
+                    # its original physical state, never reseed the fixture or
+                    # retain a second wake/model owner across the restart.
+                    from loopx.chat_actions import ChatActionService
+                    from loopx.chat_action_store import ChatActionStore
+                    from loopx.chat_runtime import ChatRuntimeController
+                    from loopx.chat_store import ChatSessionStore
+                    from loopx.extensions.lark.private_conversations import LarkPrivateConversations
+                    history, prior_session = store.messages(sid), store.load_session(sid)
+                    prior_calls, prior_writes = capture.read_bytes(), list(provider.writes)
+                    server.shutdown()
+                    old_runtime = runtime
+                    old_runtime.close()
+                    store = ChatSessionStore(store.root.parent)
+                    runtime = ChatRuntimeController(store=store, codex_bin=old_runtime.codex_bin,
+                        project_contexts=old_runtime.project_contexts, registry_path=old_runtime.registry_path)
+                    transport = LarkPrivateConversations(controller=runtime, runtime_root=store.root.parent,
+                        runner=provider, cli_bin="lark-cli")
+                    runtime.manager_scope_resolver = transport.bindings.steward_scope
+                    transport.core.actions = ChatActionService(store=ChatActionStore(store.root / "actions"),
+                        registry_path=runtime.registry_path, chat_store=store, runtime_controller=runtime,
+                        workspace_roots=[workspace])
+                    local_tasks = SessionCalendarTasks(runtime)
+                    local_owner = local_tasks
+                    server.chat_store, server.runtime_controller = store, runtime
+                    server.lark_private_conversations = transport
+                    server.action_service, server.action_store = transport.core.actions, transport.core.actions.store
+                    threading.Thread(target=server.serve_forever, daemon=True).start()
+                    assert old_runtime.closed.is_set()
+                    assert store.load_session(sid) == prior_session and store.messages(sid) == history
+                    assert local_owner.selected(sid)["actor"] == actor
+                    assert capture.read_bytes() == prior_calls and provider.writes == prior_writes
+                    restart_tokens.add(restart_token)
+                    (output / "restart-readback.json").write_text(json.dumps({
+                        "operation": restart_token, "runtime_restarts": len(restart_tokens),
+                        "session_id": sid, "actor": actor, "message_count": len(history),
+                        "prior_runtime_closed": True, "history_and_identity_preserved": True,
+                        "model_or_delivery_replayed": False, "synthetic_only": True}))
                 if wake is None and not owners_stopped and (output / "start-existing-owners").exists():
                     wake = DelegationWakeService(runtime, interval=.1).start()
                     owner_start_count += 1
@@ -181,6 +235,8 @@ def fail_report(_adapter, _prompt, event_sink):
                     "provider_history_pages": sum("+chat-messages-list" in call for call in provider.calls),
                     "existing_wake_owner_running": wake is not None,
                     "existing_wake_owner_starts": owner_start_count,
+                    "runtime_restarts": len(restart_tokens),
+                    "runtime_restart_rejections": len(restart_refusals),
                     **({"managed_source_present": artifact.exists(),
                         "managed_session": store.load_session(managed_session["session_id"])} if args.managed_session else {}),
                     **({"managed_report": True,

tests/test_managed_session_calendar_tasks.py

@@ -547,6 +547,116 @@ def lost_response(**kwargs):
         recovered.close()
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_same_calendar_task_lifecycle_retains_results_and_rejects_late_dispatch(
+        steward, tmp_path, monkeypatch, provider):
+    from loopx.chat_runtime import ChatRuntimeController
+    from loopx.chat_store import ChatSessionStore
+    from loopx.capabilities.native_chat.steward_managed_mode import ManagedSessionTurn
+    from loopx.extensions.lark.private_conversations import LarkPrivateConversations
+
+    session, actor, _, capture, workspace, config = setup_task(steward, tmp_path, monkeypatch, provider)
+    store, runtime, bot, _, binding, _, original_workspace = steward
+    sid, goal_id = session["session_id"], actor["goal_id"]
+    actor = SessionCalendarTasks(runtime).selected(sid)["actor"]
+    point = fixed_clock(monkeypatch)
+    point[0] = datetime(2020, 1, 1, 8, 59, 59, tzinfo=timezone.utc)
+    initial = capture.read_bytes(), store.messages(sid), list(bot.writes)
+    later = {**config, "schedule": {**config["schedule"], "rrule": "FREQ=DAILY;BYHOUR=10;BYMINUTE=0"}}
+    with settings_server(steward) as request:
+        for proposed in (config, later, {**later, "enabled": False}, None, config):
+            apply_configuration(request, goal_id, proposed)
+            getattr(runtime, "_steward_monitor_next_check", {}).clear()
+            assert not pump_steward_monitors(runtime)
+        assert read_cadence_journal(runtime_root=runtime.coordination_runtime_root, goal_id=goal_id) is None
+        assert (capture.read_bytes(), store.messages(sid), list(bot.writes)) == initial
+        point[0] += timedelta(seconds=2)
+        runtime._steward_monitor_next_check.clear()
+        first = finish(runtime, pump_steward_monitors(runtime)[0])
+        assert first["response"]["native_loopx"]["status"] == "committed", first
+        runtime._steward_monitor_next_check.clear()
+        assert not pump_steward_monitors(runtime)
+        original_result = store.load_turn(sid, first["turn_id"])
+        completion = SessionCalendarTasks(runtime).health(sid)["completion"]
+        apply_configuration(request, goal_id, {**config, "enabled": False})
+
+    # A new runtime reads the same physical store, actor, paused plan and body.
+    runtime.close()
+    recovered = ChatRuntimeController(store=ChatSessionStore(store.root.parent), codex_bin=runtime.codex_bin,
+        project_contexts=runtime.project_contexts, registry_path=runtime.registry_path)
+    transport = LarkPrivateConversations(controller=recovered, runtime_root=store.root.parent,
+        runner=bot, cli_bin="lark-cli")
+    resumed = (recovered.store, recovered, bot, transport, binding, capture, original_workspace)
+    try:
+        service = SessionCalendarTasks(recovered)
+        assert service.selected(sid)["actor"] == actor
+        assert service.health(sid)["completion"] == completion and not service.health(sid)["enabled"]
+        paused = capture.read_bytes(), recovered.store.messages(sid), list(bot.writes)
+        point[0] += timedelta(days=1)
+        assert not pump_steward_monitors(recovered)
+        assert (capture.read_bytes(), recovered.store.messages(sid), list(bot.writes)) == paused
+        with settings_server(resumed) as request:
+            apply_configuration(request, goal_id, config)
+            recovered._steward_monitor_next_check.clear()
+            second = finish(recovered, pump_steward_monitors(recovered)[0])
+            assert second["response"]["native_loopx"]["status"] == "committed", second
+            assert second["turn_id"] != first["turn_id"]
+            assert recovered.store.load_turn(sid, first["turn_id"]) == original_result
+            recovered._steward_monitor_next_check.clear()
+            assert not pump_steward_monitors(recovered)
+            apply_configuration(request, goal_id, later)
+            point[0] = datetime(2020, 1, 2, 10, 0, 1, tzinfo=timezone.utc)
+            # Freeze through the real calendar owner, before any model dispatch.
+            with monkeypatch.context() as waiting:
+                waiting.setattr(ManagedSessionTurn, "submit", lambda *_a, **_kw: (None, False))
+                assert service.frontier(sid) == (None, False)
+            window = read_cadence_journal(runtime_root=recovered.coordination_runtime_root, goal_id=goal_id)["window"]
+            pending = service._todo(actor, window)
+            before = capture.read_bytes(), recovered.store.messages(sid), list(bot.writes)
+            registry_bytes = recovered.registry_path.read_bytes()
+            registry = json.loads(registry_bytes)
+            goal = next(row for row in registry["goals"] if row["id"] == goal_id)
+            goal["coordination"]["registered_agents"] = ["peer"]
+            recovered.registry_path.write_text(json.dumps(registry))
+            try:
+                assert not pump_steward_monitors(recovered)
+                with pytest.raises(ValueError):
+                    ManagedSessionTurn(recovered).submit(sid, agent_id=actor["agent_id"],
+                        operation_id="task." + window["window_id"], todo_id=pending["todo_id"], scheduled_window=window)
+            finally:
+                # Restore only this disposable counterfactual, never live grants.
+                recovered.registry_path.write_bytes(registry_bytes)
+            apply_configuration(request, goal_id, {**later, "enabled": False})
+            with pytest.raises(ValueError, match="subscription changed"):
+                ManagedSessionTurn(recovered).submit(sid, agent_id=actor["agent_id"],
+                    operation_id="task." + window["window_id"], todo_id=pending["todo_id"], scheduled_window=window)
+            assert (capture.read_bytes(), recovered.store.messages(sid), list(bot.writes)) == before
+            apply_configuration(request, goal_id, later)
+            recovered._steward_monitor_next_check.clear()
+            third = finish(recovered, pump_steward_monitors(recovered)[0])
+            assert third["response"]["native_loopx"]["status"] == "committed", third
+            assert third["loopx_request"]["todo_id"] == pending["todo_id"]
+            assert len({row["loopx_request"]["schedule_window_id"] for row in (first, second, third)}) == 3
+            assert json.loads((workspace / "total.json").read_text()) == {"total": 42}
+            final = capture.read_bytes(), recovered.store.messages(sid), list(bot.writes)
+            apply_configuration(request, goal_id, None)
+            point[0] += timedelta(days=1)
+            recovered._steward_monitor_next_check.clear()
+            assert not pump_steward_monitors(recovered)
+            assert (capture.read_bytes(), recovered.store.messages(sid), list(bot.writes)) == final
+            assert recovered.store.load_turn(sid, first["turn_id"]) == original_result
+            calls = [json.loads(line) for line in capture.read_bytes()[len(initial[0]):].splitlines()]
+            assert sum(row.get("method") == "turn/start" for row in calls) == 3
+            messages = recovered.store.messages(sid)
+            assert messages[:len(initial[1])] == initial[1]
+            for turn in (first, second, third):
+                replies = [row for row in messages if row["role"] == "agent" and row.get("turn_id") == turn["turn_id"]]
+                assert len(replies) == 1 and replies[0]["text"] == turn["response"]["message"]
+            assert bot.writes == initial[2]
+    finally:
+        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(