diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index 744ee06a4c..904d81102b 100644 --- a/docs/product/periodic-reports.md +++ b/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//native-turn?calendar_task_sources=true`. The `calendar_task_sources` view reads the original Goal's canonical completed diff --git a/examples/personal-workspace-browser/steward_report_fixture.py b/examples/personal-workspace-browser/steward_report_fixture.py index ae3cb2c37a..5e388514f8 100644 --- a/examples/personal-workspace-browser/steward_report_fixture.py +++ b/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, diff --git a/tests/test_managed_session_calendar_tasks.py b/tests/test_managed_session_calendar_tasks.py index 66447aaec7..f9c822b70a 100644 --- a/tests/test_managed_session_calendar_tasks.py +++ b/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(