From bd531b9fecc68647004404289e265f2b13309343 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 1 Nov 2026 12:55:24 +0800 Subject: [PATCH] test(calendar): retain managed tasks across process restarts Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- tests/test_managed_session_calendar_tasks.py | 105 +++++++++++++++++++ 1 file changed, 105 insertions(+) diff --git a/tests/test_managed_session_calendar_tasks.py b/tests/test_managed_session_calendar_tasks.py index f9c822b70a..6366d19d43 100644 --- a/tests/test_managed_session_calendar_tasks.py +++ b/tests/test_managed_session_calendar_tasks.py @@ -9,6 +9,7 @@ import http.client import json from pathlib import Path +import subprocess import sys import threading from types import SimpleNamespace @@ -657,6 +658,110 @@ def test_same_calendar_task_lifecycle_retains_results_and_rejects_late_dispatch( recovered.close() +def _resume_calendar_process(request): + """A fresh OS process opens only the supplied disposable physical store.""" + import os + from loopx.chat_runtime import ChatRuntimeController + from loopx.chat_store import ChatSessionStore + from loopx.capabilities.native_chat.project_context import ChatProjectContexts + import loopx.capabilities.native_chat.session_calendar_tasks as calendar + + point = datetime(2020, 1, request["day"], 9, 0, 1, tzinfo=timezone.utc) + class Clock(datetime): + @classmethod + def now(cls, tz=None): + return point.astimezone(tz) + calendar.datetime = Clock + runtime = ChatRuntimeController(store=ChatSessionStore(Path(request["runtime_root"])), + codex_bin=request["codex_bin"], registry_path=Path(request["registry_path"]), + project_contexts=ChatProjectContexts([Path(request["workspace"])])) + sid, actor = request["session_id"], request["actor"] + capture = Path(request["capture"]) + try: + service = SessionCalendarTasks(runtime) + assert service.selected(sid)["actor"] == actor + before = capture.read_bytes(), runtime.store.messages(sid) + # Neither a persisted pause nor cancellation can dispatch on startup. + assert not pump_steward_monitors(runtime) + assert (capture.read_bytes(), runtime.store.messages(sid)) == before + result = {"pid": os.getpid(), "session_id": sid, "actor": actor} + if request["continue"]: + # This local Session has no Lark transport or external recipient. + data = (runtime.store, runtime, None, None) + with settings_server(data) as call: + apply_configuration(call, actor["goal_id"], request["configuration"]) + getattr(runtime, "_steward_monitor_next_check", {}).clear() + turns = pump_steward_monitors(runtime) + assert len(turns) == 1 + completed = finish(runtime, turns[0]) + assert completed["response"]["native_loopx"]["status"] == "committed", completed + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + assert service.health(sid)["completion_verified"] + result["turn_id"] = completed["turn_id"] + apply_configuration(call, actor["goal_id"], None) + else: + assert not service.health(sid).get("enabled") + print(json.dumps(result)) + finally: + runtime.close() + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_calendar_physical_state_survives_independent_processes(steward, tmp_path, monkeypatch, provider): + import os + + session, actor, _, capture, _, config = setup_task(steward, tmp_path, monkeypatch, provider) + store, runtime, bot, _, _, _, original_workspace = steward + sid = session["session_id"] + actor = SessionCalendarTasks(runtime).selected(sid)["actor"] + fixed_clock(monkeypatch) + with settings_server(steward) as call: + apply_configuration(call, actor["goal_id"], config) + 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) + apply_configuration(call, actor["goal_id"], {**config, "enabled": False}) + first_bytes = store._turn_path(sid, first["turn_id"]).read_bytes() + history, calls, writes = store.messages(sid), capture.read_bytes(), list(bot.writes) + runtime.close() + assert runtime.closed.is_set() + + request = {"runtime_root": str(store.root.parent), "codex_bin": runtime.codex_bin, + "registry_path": str(runtime.registry_path), "workspace": str(original_workspace), + "session_id": sid, "actor": actor, "capture": str(capture), "configuration": config} + command = [sys.executable, "-c", + "import json,sys;sys.path.insert(0,'tests');" + "from test_managed_session_calendar_tasks import _resume_calendar_process;" + "_resume_calendar_process(json.load(sys.stdin))"] + def restart(*, continue_task, day=2): + completed = subprocess.run(command, input=json.dumps({**request, "continue": continue_task, "day": day}), + cwd=Path(__file__).resolve().parents[1], text=True, capture_output=True, timeout=90) + assert completed.returncode == 0, completed.stdout + completed.stderr + result = json.loads(completed.stdout) + assert result["pid"] != os.getpid() and result["session_id"] == sid and result["actor"] == actor + assert store._turn_path(sid, first["turn_id"]).read_bytes() == first_bytes + assert store.messages(sid)[:len(history)] == history and bot.writes == writes + return result + + restart(continue_task=False) + assert capture.read_bytes() == calls and store.messages(sid) == history + resumed = restart(continue_task=True) + second = store.load_turn(sid, resumed["turn_id"]) + assert second["turn_id"] != first["turn_id"] + assert second["loopx_request"]["schedule_window_id"] != first["loopx_request"]["schedule_window_id"] + requests = [json.loads(line) for line in capture.read_bytes()[len(calls):].splitlines()] + assert sum(row.get("method") == "turn/start" for row in requests) == 1 + for turn in (first, second): + replies = [row for row in store.messages(sid) if row["role"] == "agent" and row.get("turn_id") == turn["turn_id"]] + assert len(replies) == 1 and replies[0]["text"] == turn["response"]["message"] + final = capture.read_bytes(), store.messages(sid) + # A later due window would execute if cancellation were lost on restart. + restart(continue_task=False, day=3) + assert (capture.read_bytes(), store.messages(sid)) == final + + @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(