feat: discover repeatable Tasks in their original managed Session

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/<original-session-id>/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,

loopx/capabilities/native_chat/session_calendar_tasks.py

@@ -84,17 +84,53 @@ 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
+        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"]

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,

loopx/chat_server.py

@@ -960,8 +960,15 @@ 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:

tests/test_managed_session_calendar_tasks.py

@@ -140,6 +140,94 @@ 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 == 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"])
+            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