From 6c7938844a8204264f5ec886d6b10055a202dac8 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 29 Oct 2026 20:55:00 +0800 Subject: [PATCH 1/2] feat: discover repeatable native Tasks in their managed Session Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/product/periodic-reports.md | 23 +++++ .../native_chat/session_calendar_tasks.py | 45 ++++++++-- .../native_chat/steward_managed_mode.py | 7 +- loopx/chat_server.py | 15 +++- tests/test_managed_session_calendar_tasks.py | 89 +++++++++++++++++++ 5 files changed, 171 insertions(+), 8 deletions(-) diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index 9a0ba85b1a..972bdb2c06 100644 --- a/docs/product/periodic-reports.md +++ b/docs/product/periodic-reports.md @@ -1020,6 +1020,29 @@ renderer preset; it does not invoke the progress-report source or editorial format. Task descriptions, scripts and validation argv cannot be supplied in this configuration. +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 +history, including archived Tasks, and applies the same original Agent and +independent argv-declaration checks used when configuring execution. It returns +Task references, display titles, validation digests and archive state, without +installing commands or copying validators into the configuration. Open work, +other Agents' work, report generation and generated calendar periods are not +offered as repeating sources. A missing, changed or unreadable declaration +fails this read visibly; it is not represented as an empty valid list. +Ordinary inspection omits this optional catalog. Repeated discovery does not +configure a subscription, wake an Agent, create a Task, invoke a model, spend +quota or send a message. Configuration and every due execution still recheck +the original source; a displayed reference grants no launch authority. + +调用方可在原会话的 `native-turn` 读取中显式加入 +`calendar_task_sources=true`，发现同一 Agent 可重复执行的已完成 Task，包含已归档项。 +此只读视图复用原生已完成历史和执行入口的验收声明检查；返回原引用、显示标题、 +验收摘要和归档状态，不传入新命令。未完成任务、其它 Agent 的任务、报告生成和 +已生成的日历窗口不会成为候选。声明缺失、变化或不可读时明确失败，不伪装为空列表。 +普通状态读取不加载此目录；重复发现不会启用计划、唤醒 Agent、创建 Task、调用模型、 +扣额或发消息。保存配置和每次到期执行仍重新核验原来源。 + At the first due boundary, the original idle Chat pump freezes the source Task's declaration into the existing cadence attempt. It creates one finite native Task for that window, inheriting the source Task's scope, repository, diff --git a/loopx/capabilities/native_chat/session_calendar_tasks.py b/loopx/capabilities/native_chat/session_calendar_tasks.py index 78df73d1db..17c0b5a9d1 100644 --- a/loopx/capabilities/native_chat/session_calendar_tasks.py +++ b/loopx/capabilities/native_chat/session_calendar_tasks.py @@ -84,17 +84,50 @@ def source_task(self, actor, task): row = list_goal_todos(registry_path=self.controller.registry_path, goal_id=actor["goal_id"], runtime_root_arg=str(self.root), todo_id=task["source_todo_id"], role="agent", status="done", read_scope="completed_history").get("todo") - if (not row or row.get("status") != "done" or row.get("claimed_by") != actor["agent_id"] - or row.get("task_class") != "advancement_task" or row.get("completion_validation_required") is not True - or row.get("action_kind") == "generate_periodic_report" - or re.fullmatch(r"cadence_[a-f0-9]{64}", row.get("target_key") or "") - or actor["agent_id"] in row.get("excluded_agents", [])): + return row, self._source_validation(actor, row) + + @staticmethod + def _source_candidate(actor, row): + return bool(row and row.get("status") == "done" and row.get("claimed_by") == actor["agent_id"] + and row.get("task_class") == "advancement_task" and row.get("completion_validation_required") is True + and row.get("action_kind") != "generate_periodic_report" + and not re.fullmatch(r"cadence_[a-f0-9]{64}", row.get("target_key") or "") + and actor["agent_id"] not in row.get("excluded_agents", [])) + + def _source_validation(self, actor, row): + if not self._source_candidate(actor, row): raise ValueError("select this Agent's existing completed Task with independent argv validation") declaration = read_completion_validation_declaration(runtime_root=self.root, goal_id=actor["goal_id"], todo_id=row["todo_id"], expected_digest=row["completion_validation_sha256"]) if not declaration or not declaration.get("validation_command_argv"): raise ValueError("the existing source task has no independent argv validation") - return row, declaration + return declaration + + def source_options(self, session_id): + """Discover references through the same source admission, without enrolling work. + + Completed-history reads include active and archived native Tasks. The + canonical owner drains its snapshot pages before projecting this view; + no hot-list limit or private second index determines membership here. + """ + from ...control_plane.todos.list_readback import list_goal_todos + actor = self.selected(session_id)["actor"] + snapshot = list_goal_todos(registry_path=self.controller.registry_path, goal_id=actor["goal_id"], + runtime_root_arg=str(self.root), role="agent", status="done", read_scope="completed_history", limit=None) + sources = [] + for row in snapshot["todos"]: + if not self._source_candidate(actor, row): + continue + # An unreadable or changed declaration must fail the read instead + # of turning a source-coverage failure into an empty valid catalog. + validation = self._source_validation(actor, row) + sources.append({"source_todo_id": row["todo_id"], "title": row.get("title") or row["text"], + "completion_validation_sha256": row["completion_validation_sha256"], + "validation_label": validation.get("validation_label"), + "archived": row.get("archive_state") == "archive"}) + return {"schema_version": "managed_session_task_sources_v0", "session_id": session_id, + "goal_id": actor["goal_id"], "agent_id": actor["agent_id"], "read_only": True, + "sources": sources, "authority_read": snapshot["authority_read"]} def admission(self, request): actor = request["actor"] diff --git a/loopx/capabilities/native_chat/steward_managed_mode.py b/loopx/capabilities/native_chat/steward_managed_mode.py index aacdc29332..dbac78cd2e 100644 --- a/loopx/capabilities/native_chat/steward_managed_mode.py +++ b/loopx/capabilities/native_chat/steward_managed_mode.py @@ -290,7 +290,7 @@ def admit(self, request: dict[str, Any], *, turn: dict[str, Any] | None = None, "queued_turn_count": len(self.store.queued_turns(actor["session_id"]))}) def inspect(self, session_id: str, *, agent_id: str | None = None, - operation_id: str | None = None) -> dict[str, Any]: + operation_id: str | None = None, include_calendar_task_sources: bool = False) -> dict[str, Any]: session = self.store.load_session(session_id) if not session: raise KeyError("chat session was not found") @@ -315,6 +315,11 @@ def inspect(self, session_id: str, *, agent_id: str | None = None, task = SessionCalendarTasks(self.controller).health(session_id) if task.get("configured"): result["scheduled_task"] = task + if include_calendar_task_sources: + if agent_id is not None and agent_id != session.get("managed_agent_id"): + raise ValueError("select the original registered Agent for calendar Task sources") + from .session_calendar_tasks import SessionCalendarTasks + result["calendar_task_sources"] = SessionCalendarTasks(self.controller).source_options(session_id) return result def submit(self, session_id: str, *, agent_id: str, operation_id: str, diff --git a/loopx/chat_server.py b/loopx/chat_server.py index ccaf3052c1..bc10ac68c3 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -952,6 +952,7 @@ def _session_snapshot(self, session_id: str) -> None: def _managed_native_turn(self, session_id: str, *, execute: bool = False) -> None: from .capabilities.native_chat.steward_managed_mode import ManagedSessionTurn service = ManagedSessionTurn(self.server.runtime_controller) + include_sources = False try: if execute: body = self._read_json() @@ -960,12 +961,24 @@ def _managed_native_turn(self, session_id: str, *, execute: bool = False) -> Non turn, created = service.submit(session_id, **body) return self._send_json({"ok": True, "session_id": session_id, "turn": turn, "created": created}, status=202) query = parse_qs(urlparse(self.path).query, keep_blank_values=True) + sources = query.get("calendar_task_sources", ["false"]) + if sources not in (["true"], ["false"]): + raise ValueError("calendar_task_sources must be one true or false value") + include_sources = sources == ["true"] + if include_sources and (set(query) - {"calendar_task_sources", "agent_id", "operation_id"} + or any(len(values) != 1 for values in query.values())): + raise ValueError("select one original managed Session and Agent for Task sources") self._send_json(service.inspect(session_id, agent_id=query.get("agent_id", [None])[0], - operation_id=query.get("operation_id", [None])[0])) + operation_id=query.get("operation_id", [None])[0], include_calendar_task_sources=include_sources)) except KeyError: self._send_error("chat session was not found", status=404) except (TypeError, ValueError, RuntimeError) as exc: self._send_error(str(exc), status=409, error_code="managed_turn_unavailable") + except OSError: + if not include_sources: + raise + self._send_error("Original managed Task sources are temporarily unreadable", status=503, + error_code="managed_task_sources_unavailable") def _list_sessions(self) -> None: query = parse_qs(urlparse(self.path).query, keep_blank_values=True) diff --git a/tests/test_managed_session_calendar_tasks.py b/tests/test_managed_session_calendar_tasks.py index fc2edd6abc..0db25ce785 100644 --- a/tests/test_managed_session_calendar_tasks.py +++ b/tests/test_managed_session_calendar_tasks.py @@ -140,6 +140,95 @@ def now(cls, tz=None): return point +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_original_session_discovers_repeatable_tasks_without_enrolling_or_replaying(steward, tmp_path, monkeypatch, provider): + from loopx.capabilities.native_chat.steward_managed_mode import ManagedSessionTurn + from loopx.capabilities.native_chat.steward_monitor import StewardMonitor + session, actor, unrelated, capture, _, config = setup_task(steward, tmp_path, monkeypatch, provider) + store, runtime, bot, *_ = steward + sid = session["session_id"] + fixed_clock(monkeypatch) + calls, history, writes = capture.read_bytes(), store.messages(sid), list(bot.writes) + before = load_goal_from_registry(runtime.registry_path, actor["goal_id"]) + with settings_server(steward) as request: + url = f"/api/chat/sessions/{sid}/native-turn" + status, plain = request(url, None, method="GET") + assert status == 200 and "calendar_task_sources" not in plain + status, view = request(url + "?calendar_task_sources=true", None, method="GET") + assert status == 200, view + options = view["calendar_task_sources"] + assert options["session_id"] == sid and options["goal_id"] == actor["goal_id"] + assert options["agent_id"] == actor["agent_id"] and options["read_only"] + assert options["authority_read"]["source_authority"] == ("file_v0" if provider == "file" else "sqlite_v0") + assert [row["source_todo_id"] for row in options["sources"]] == [config["task"]["source_todo_id"]] + chosen = options["sources"][0] + assert chosen["title"] and len(chosen["completion_validation_sha256"]) == 64 + assert not chosen["archived"] and "validation_command_argv" not in json.dumps(options) + for selector in ("yes", "true&calendar_task_sources=false", f"true&agent_id={actor['agent_id']}-other", + f"true&agent_id={actor['agent_id']}&agent_id=other", "true&goal_id=other"): + status, rejected = request(url + "?calendar_task_sources=" + selector, None, method="GET") + assert status == 409 and rejected["ok"] is False + # Discovery is opt-in and has not changed the original Goal, Task, + # model calls, original messages, or private provider delivery. + assert load_goal_from_registry(runtime.registry_path, actor["goal_id"]) == before + assert capture.read_bytes() == calls and store.messages(sid) == history and bot.writes == writes + assert read_cadence_journal(runtime_root=runtime.coordination_runtime_root, goal_id=actor["goal_id"]) is None + apply_configuration(request, actor["goal_id"], {**config, "task": {"source_todo_id": chosen["source_todo_id"]}}) + completed = finish(runtime, pump_steward_monitors(runtime)[0]) + assert completed["response"]["native_loopx"]["status"] == "committed" + # A completed generated period cannot become another repeating source. + assert [row["source_todo_id"] for row in ManagedSessionTurn(runtime).inspect( + sid, include_calendar_task_sources=True)["calendar_task_sources"]["sources"]] == [chosen["source_todo_id"]] + cli(runtime, "todo", "archive-completed", "--goal-id", actor["goal_id"], "--role", "agent", + "--max-active-done", "0", "--execute") + calls = capture.read_bytes() + status, archived = request(url + "?calendar_task_sources=true", None, method="GET") + assert status == 200, archived + assert archived["calendar_task_sources"]["sources"] == [{**chosen, "archived": True}] + assert archived["scheduled_task"]["completion_verified"] and capture.read_bytes() == calls + assert next(row for row in StewardMonitor(runtime).todos(actor) if row["todo_id"] == unrelated)["status"] == "open" + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("fault", ["malformed", "directory"]) +def test_calendar_task_discovery_retains_unreadable_validator_and_recovers_without_work(steward, tmp_path, monkeypatch, provider, fault): + from loopx.control_plane.todos.completion_validation_store import completion_validation_declaration_path + session, actor, _, capture, _, config = setup_task(steward, tmp_path, monkeypatch, provider) + store, runtime, bot, *_ = steward + service = SessionCalendarTasks(runtime) + options = service.source_options(session["session_id"]) + source = options["sources"][0] + path = completion_validation_declaration_path(runtime_root=runtime.coordination_runtime_root, + goal_id=actor["goal_id"], todo_id=source["source_todo_id"]).parent / "blobs" / (source["completion_validation_sha256"] + ".json") + original = path.read_bytes() + calls, history, writes = capture.read_bytes(), store.messages(session["session_id"]), list(bot.writes) + path.unlink() + if fault == "directory": + path.mkdir() + else: + path.write_text("{}") + try: + with settings_server(steward) as request: + url = f"/api/chat/sessions/{session['session_id']}/native-turn?calendar_task_sources=true" + status, failed = request(url, None, method="GET") + assert status == (503 if fault == "directory" else 409), failed + assert failed["error_code"] == ("managed_task_sources_unavailable" if fault == "directory" else "managed_turn_unavailable") + assert "calendar_task_sources" not in failed + with pytest.raises(OSError if fault == "directory" else ValueError): + service.source_task(actor, config["task"]) + if path.is_dir(): + path.rmdir() + path.write_bytes(original) + status, recovered = request(url, None, method="GET") + assert status == 200 and recovered["calendar_task_sources"]["sources"] == options["sources"] + finally: + if path.is_dir(): + path.rmdir() + path.write_bytes(original) + assert capture.read_bytes() == calls and store.messages(session["session_id"]) == history and bot.writes == writes + assert read_cadence_journal(runtime_root=runtime.coordination_runtime_root, goal_id=actor["goal_id"]) is None + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_calendar_preserves_canonical_decision_scopes_through_original_return(steward, tmp_path, monkeypatch, provider): from loopx.control_plane.todos.contract import normalize_todo_required_decision_scopes From b4b6b836385f6db036b9589e131ad361e6d6330e Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 29 Oct 2026 20:57:44 +0800 Subject: [PATCH 2/2] refactor: keep calendar source read failures in their owner Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../native_chat/session_calendar_tasks.py | 37 ++++++++++--------- loopx/chat_server.py | 6 --- tests/test_managed_session_calendar_tasks.py | 3 +- 3 files changed, 21 insertions(+), 25 deletions(-) diff --git a/loopx/capabilities/native_chat/session_calendar_tasks.py b/loopx/capabilities/native_chat/session_calendar_tasks.py index 17c0b5a9d1..a552d9d07b 100644 --- a/loopx/capabilities/native_chat/session_calendar_tasks.py +++ b/loopx/capabilities/native_chat/session_calendar_tasks.py @@ -111,23 +111,26 @@ def source_options(self, session_id): no hot-list limit or private second index determines membership here. """ from ...control_plane.todos.list_readback import list_goal_todos - actor = self.selected(session_id)["actor"] - snapshot = list_goal_todos(registry_path=self.controller.registry_path, goal_id=actor["goal_id"], - runtime_root_arg=str(self.root), role="agent", status="done", read_scope="completed_history", limit=None) - sources = [] - for row in snapshot["todos"]: - if not self._source_candidate(actor, row): - continue - # An unreadable or changed declaration must fail the read instead - # of turning a source-coverage failure into an empty valid catalog. - validation = self._source_validation(actor, row) - sources.append({"source_todo_id": row["todo_id"], "title": row.get("title") or row["text"], - "completion_validation_sha256": row["completion_validation_sha256"], - "validation_label": validation.get("validation_label"), - "archived": row.get("archive_state") == "archive"}) - return {"schema_version": "managed_session_task_sources_v0", "session_id": session_id, - "goal_id": actor["goal_id"], "agent_id": actor["agent_id"], "read_only": True, - "sources": sources, "authority_read": snapshot["authority_read"]} + try: + actor = self.selected(session_id)["actor"] + snapshot = list_goal_todos(registry_path=self.controller.registry_path, goal_id=actor["goal_id"], + runtime_root_arg=str(self.root), role="agent", status="done", read_scope="completed_history", limit=None) + sources = [] + for row in snapshot["todos"]: + if not self._source_candidate(actor, row): + continue + # An unreadable or changed declaration must fail the read instead + # of turning a source-coverage failure into an empty valid catalog. + validation = self._source_validation(actor, row) + sources.append({"source_todo_id": row["todo_id"], "title": row.get("title") or row["text"], + "completion_validation_sha256": row["completion_validation_sha256"], + "validation_label": validation.get("validation_label"), + "archived": row.get("archive_state") == "archive"}) + return {"schema_version": "managed_session_task_sources_v0", "session_id": session_id, + "goal_id": actor["goal_id"], "agent_id": actor["agent_id"], "read_only": True, + "sources": sources, "authority_read": snapshot["authority_read"]} + except OSError as exc: + raise RuntimeError("Original managed Task sources are temporarily unreadable") from exc def admission(self, request): actor = request["actor"] diff --git a/loopx/chat_server.py b/loopx/chat_server.py index bc10ac68c3..58b2088fcf 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -952,7 +952,6 @@ def _session_snapshot(self, session_id: str) -> None: def _managed_native_turn(self, session_id: str, *, execute: bool = False) -> None: from .capabilities.native_chat.steward_managed_mode import ManagedSessionTurn service = ManagedSessionTurn(self.server.runtime_controller) - include_sources = False try: if execute: body = self._read_json() @@ -974,11 +973,6 @@ def _managed_native_turn(self, session_id: str, *, execute: bool = False) -> Non self._send_error("chat session was not found", status=404) except (TypeError, ValueError, RuntimeError) as exc: self._send_error(str(exc), status=409, error_code="managed_turn_unavailable") - except OSError: - if not include_sources: - raise - self._send_error("Original managed Task sources are temporarily unreadable", status=503, - error_code="managed_task_sources_unavailable") def _list_sessions(self) -> None: query = parse_qs(urlparse(self.path).query, keep_blank_values=True) diff --git a/tests/test_managed_session_calendar_tasks.py b/tests/test_managed_session_calendar_tasks.py index 0db25ce785..b2c722ced9 100644 --- a/tests/test_managed_session_calendar_tasks.py +++ b/tests/test_managed_session_calendar_tasks.py @@ -211,8 +211,7 @@ def test_calendar_task_discovery_retains_unreadable_validator_and_recovers_witho with settings_server(steward) as request: url = f"/api/chat/sessions/{session['session_id']}/native-turn?calendar_task_sources=true" status, failed = request(url, None, method="GET") - assert status == (503 if fault == "directory" else 409), failed - assert failed["error_code"] == ("managed_task_sources_unavailable" if fault == "directory" else "managed_turn_unavailable") + assert status == 409 and failed["error_code"] == "managed_turn_unavailable", failed assert "calendar_task_sources" not in failed with pytest.raises(OSError if fault == "directory" else ValueError): service.source_task(actor, config["task"])