diff --git a/apps/presentation/dashboard/src/features/personal-workspace/goal-capability-settings.tsx b/apps/presentation/dashboard/src/features/personal-workspace/goal-capability-settings.tsx index e789ed0b68..883185b71e 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/goal-capability-settings.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/goal-capability-settings.tsx @@ -15,6 +15,7 @@ import { goalReplanCadenceEditorValue, goalReplanCadenceWriteValue, hasReplanCad import { useWorkspaceI18n } from "./i18n"; import { CapabilityConfigurationFields } from "./capability-configuration-fields"; import { withReportScheduleTimezone } from "./periodic-report-schedule-field"; +import {PeriodicReportSessionField} from "./periodic-report-session-field"; import { localizeCapability, localizedCapabilityFieldCopy } from "./capability-localization"; import { orderCapabilitiesForPresentation, canEditCapability, CapabilityCatalogNavigation, CapabilityConfigurationSummary, CapabilityDetailHeader, CapabilityEditorStatus, CapabilityConfigurationHelp } from "./capability-workbench"; import { GoalAutoNotifyToggle } from "./notification-settings-panel"; @@ -426,6 +427,10 @@ function CapabilityCatalog({ callbacks, catalog, goalId, loading, notification, ? `复盘周期沿用${selected.machine_current ? "设备" : "产品"}默认；规划选项由本 Goal 设置。` : `Review interval follows ${selected.machine_current ? "device" : "product"} defaults; this Goal sets the planning options.`}

: null} + {selected.capability_id === "periodic_report" && draft.profile_preset === "weekly-progress" + ? changeDraft("route_ref", value)} /> : null} (url: string, signal: AbortSignal): Promise { + const response = await fetch(url, {signal}); + const value = await response.json(); + if (!response.ok || value.ok !== true) throw new Error("Original report conversation unavailable"); + return value as T; +} + +/** Select an existing original Session; never create, resume or retarget one. */ +export function PeriodicReportSessionField({goalId, route, zh, disabled, revision, onChange}: { + goalId: string; route: string; zh: boolean; disabled: boolean; revision: string; + onChange: (route: string) => void; +}) { + const id = useId(); + const [sessions, setSessions] = useState([]); + const [loading, setLoading] = useState(true); + const [failed, setFailed] = useState(false); + const [retry, setRetry] = useState(0); + const [report, setReport] = useState(null); + const [reportFailed, setReportFailed] = useState(false); + const local = route.startsWith("managed-session:"); + const selectedId = local ? route.slice("managed-session:".length) : ""; + + useEffect(() => { + const controller = new AbortController(); + setSessions([]); setLoading(true); setFailed(false); + void read<{sessions: Session[]}>(`/api/chat/sessions?${new URLSearchParams({goal_id: goalId})}`, controller.signal) + .then(value => { + if (controller.signal.aborted) return; + if (!Array.isArray(value.sessions) || value.sessions.some(session => session.goal_id !== goalId)) throw new Error("Source changed"); + setSessions(value.sessions.filter(session => session.managed_agent_id && session.session_mode === "managed_runtime" + && session.agent_id === "codex" && session.status !== "closed")); + }).catch(() => {if (!controller.signal.aborted) setFailed(true);}) + .finally(() => {if (!controller.signal.aborted) setLoading(false);}); + return () => controller.abort(); + }, [goalId, retry]); + + useEffect(() => { + const controller = new AbortController(); + setReport(null); setReportFailed(false); + if (selectedId && !disabled) { + void read<{report?: Report}>(`/api/chat/sessions/${encodeURIComponent(selectedId)}/native-turn`, controller.signal) + .then(value => {if (!controller.signal.aborted) { + if (!value.report) throw new Error("Report readback unavailable"); + setReport(value.report); + }}).catch(() => {if (!controller.signal.aborted) setReportFailed(true);}); + } + return () => controller.abort(); + }, [selectedId, goalId, revision, disabled, retry]); + + return
+ + {!local && route ? : null} + {failed ? {zh ? "无法读取原会话，未更换接收位置。" : "Original conversations could not be read; the destination is retained."} + : !loading && !sessions.length ? {zh ? "此 Goal 尚无已绑定的 Agent 会话。" : "This Goal has no bound Agent conversation."} : null} + {failed || reportFailed ? : null} + {reportFailed ? {zh ? "报告执行与送达状态暂时未知。" : "Report execution and delivery are currently unknown."} : null} + {report ? + {!report.available ? (zh ? "原会话不可用，定时报告不能执行。" : "The original conversation is unavailable; scheduled reports cannot run.") + : !report.configured ? (zh ? "保存后在所选原会话接收报告。" : "Save to receive reports in the selected original conversation.") + : !report.enabled ? (zh ? "定时报告已暂停。" : "Scheduled reports are paused.") + : report.scheduled === false ? (zh ? "已启用，请设置日历汇报时间。" : "Enabled; set a calendar report time.") + : report.publication_verified ? (zh ? "本期报告已在原会话完整读回。" : "This period’s full report was read back in the original conversation.") + : report.execution_failed ? (zh ? "本期执行失败；原结果与失败记录已保留。" : "This period’s execution failed; the original result and failure are retained.") + : report.generation_complete ? (zh ? "本期已生成，完整送达尚未确认。" : "This period was generated; complete delivery is unverified.") + : (zh ? "定时报告已启用。" : "Scheduled reports are enabled.")} + {report.enabled && report.next_due_at && (!report.retained_window_due_at || report.publication_verified) + ? ` ${zh ? "下次：" : "Next: "}${new Date(report.next_due_at).toLocaleString(zh ? "zh-CN" : "en-US")}` : ""} + : null} + {report?.publication_verified && report.report_message_id ? {zh ? "查看本期完整报告" : "Read this period’s full report"} : null} +
; +} diff --git a/docs/assets/personal-workspace/steward-reports/managed-session-README.md b/docs/assets/personal-workspace/steward-reports/managed-session-README.md new file mode 100644 index 0000000000..3290f11bc4 --- /dev/null +++ b/docs/assets/personal-workspace/steward-reports/managed-session-README.md @@ -0,0 +1,54 @@ +# Existing Agent Session calendar report journey + +The `session-*` captures use the packaged frontend and real local HTTP, Core, +native SQLite Task/Turn and periodic-report owners. Model and provider +transports, Goal and conversation data are synthetic. No production account, +message, workspace or report is shown. + +- Baseline frontend: `c1a589aac771b39df9137dc808280c2a33fb4cbd`. +- Changed source and packaged runtime: `1fe1582cf8ad136b0df0f0c76d1e86cfc3232849`. +- Settings desktop: 1200×888; narrow settings/report: 390×844 viewport emulation. + This is browser evidence, not physical-phone acceptance. + +The primary question is where this Goal's scheduled report will arrive. +The selector replaces a raw route with an existing bound Agent conversation. +The default view shows saved scheduling, honest execution/delivery state and +the verified full-report link. Profile, timezone and the existing calendar +remain together; optional general capability help stays on demand. The +shared settings panel still needs vertical scrolling to reach save controls +at these viewport heights; that broader layout is outside this change. + +The journey verifies: + +1. Save an enabled future first date; the sole existing wake owner makes no + model request before the deadline. +2. Preview/apply an eligible calendar in the same original Session. The + existing worker calls the synthetic model once and settles one native + report generation. An exact full-body Core readback precedes cursor commit. +3. Reload shows verified delivery. The existing full-answer viewer opens that + original message, including all five sections of the 27,750-character + synthetic report and the existing Copy Markdown control. The deliberately + repeated long content checks page composition, not authored-report quality. +4. Blocking only the read-only report-health request produces the visible + unknown state and Read again action. Restoring that request reads the same + delivered body; it does not generate another report. +5. Pause through the same preview/apply controls. The original body, task, + actor and history remain; the model-call count stays one. + +Separate packaged failure evidence and File/SQLite tests preserve invalid +authored output, forbid stranding a frozen undelivered window and resume only +eligible native settlement effects without model replay. This slice does not +provide invalid-window cancellation, arbitrary profiles, natural-language +authoring or the complete managed/attached Agent workspace. + +To run the disposable fixture: + +```sh +uv run --extra test python examples/personal-workspace-browser/steward_report_fixture.py \ + --managed-report --output-dir /tmp/loopx-managed-session-report-qa +``` + +Use its returned URL. The fixture's `start-existing-owners` control starts its +single existing wake service. `--invalid-editorial` selects a known failed +synthetic author response for failure inspection. Closing the fixture removes +its disposable data; it does not alter an original service or external chat. diff --git a/docs/assets/personal-workspace/steward-reports/session-before-desktop.png b/docs/assets/personal-workspace/steward-reports/session-before-desktop.png new file mode 100644 index 0000000000..7541ae2690 Binary files /dev/null and b/docs/assets/personal-workspace/steward-reports/session-before-desktop.png differ diff --git a/docs/assets/personal-workspace/steward-reports/session-before-narrow.png b/docs/assets/personal-workspace/steward-reports/session-before-narrow.png new file mode 100644 index 0000000000..7ff96f2443 Binary files /dev/null and b/docs/assets/personal-workspace/steward-reports/session-before-narrow.png differ diff --git a/docs/assets/personal-workspace/steward-reports/session-body-desktop.png b/docs/assets/personal-workspace/steward-reports/session-body-desktop.png new file mode 100644 index 0000000000..bd446c84d1 Binary files /dev/null and b/docs/assets/personal-workspace/steward-reports/session-body-desktop.png differ diff --git a/docs/assets/personal-workspace/steward-reports/session-body-narrow.png b/docs/assets/personal-workspace/steward-reports/session-body-narrow.png new file mode 100644 index 0000000000..c6dd37c8ac Binary files /dev/null and b/docs/assets/personal-workspace/steward-reports/session-body-narrow.png differ diff --git a/docs/assets/personal-workspace/steward-reports/session-delivered-desktop.png b/docs/assets/personal-workspace/steward-reports/session-delivered-desktop.png new file mode 100644 index 0000000000..50c4710946 Binary files /dev/null and b/docs/assets/personal-workspace/steward-reports/session-delivered-desktop.png differ diff --git a/docs/assets/personal-workspace/steward-reports/session-delivered-narrow.png b/docs/assets/personal-workspace/steward-reports/session-delivered-narrow.png new file mode 100644 index 0000000000..af556558e7 Binary files /dev/null and b/docs/assets/personal-workspace/steward-reports/session-delivered-narrow.png differ diff --git a/docs/assets/personal-workspace/steward-reports/session-paused-desktop.png b/docs/assets/personal-workspace/steward-reports/session-paused-desktop.png new file mode 100644 index 0000000000..416f9815d1 Binary files /dev/null and b/docs/assets/personal-workspace/steward-reports/session-paused-desktop.png differ diff --git a/docs/assets/personal-workspace/steward-reports/session-unknown-narrow.png b/docs/assets/personal-workspace/steward-reports/session-unknown-narrow.png new file mode 100644 index 0000000000..bf5bfe51c7 Binary files /dev/null and b/docs/assets/personal-workspace/steward-reports/session-unknown-narrow.png differ diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index c0a99d57d2..19031622e7 100644 --- a/docs/product/periodic-reports.md +++ b/docs/product/periodic-reports.md @@ -1,7 +1,8 @@ # Managed steward monitoring and periodic reports -Status: source-bound identity preparation, opt-in status monitoring and the -original steward's periodic full-body report path are implemented. Installation, +Status: source-bound identity preparation, opt-in status monitoring, the +original steward's periodic full-body reports and explicitly configured local +Agent Session calendar reports are implemented. Installation, an enabled subscription and a natural scheduled delivery require their own readback evidence. Preparing an identity enables neither monitoring nor reports. @@ -466,12 +467,12 @@ Registration removal, Goal stop, a changed lifetime or a mismatched workspace refuses continued execution rather than transferring the conversation. This is an API/host lifecycle slice, **not** the finished Agent workspace -entry or general periodic journey. The Agent view still needs to expose this +entry or arbitrary periodic-task journey. The Agent view still needs to expose this binding alongside attached sources, including all original history and honest unavailable states. Reuse existing Session, native Turn, monitor/calendar and periodic-report owners; do not add a second scheduler, task store or model worker. General writable execution, natural-language -configuration and recurring task composition remain unfinished. +configuration and arbitrary recurring task composition remain unfinished. An explicitly bound local managed Session can also observe its Goal's native work through an existing `continuous_monitor` Todo. Use the existing Todo @@ -516,6 +517,76 @@ establish the correct Agent hierarchy, real-model adoption or natural-deadline acceptance. The UI withdrawal does not alter the original private steward's subscription, published reports, pending requests or publication route. +### Calendar reports in an existing Agent Session + +In **Goal settings → Capability center → Periodic reports**, use the +`weekly-progress` profile and choose an existing **Report conversation**. +The selector lists only this Goal's already bound managed Agent Sessions. It +does not create an Agent or assign a legacy Goal-wide conversation. Enable +reports, set a daily or weekly calendar and optional first date, then use the +existing preview/apply controls. Reload reads back the saved destination, +calendar and original conversation's execution/delivery status. No calendar +means no automatic local Session report; the status asks for a report time. +After verified delivery, **Read this period's full report** opens the existing +original-message report viewer, including its full body and **Copy Markdown**. +The link is available only while that exact returned body is readable. + +This is an explicit Goal override. Machine defaults, registration, opening a +Session and installation do not enroll a conversation. The reserved route +`managed-session:` is persisted in the existing +periodic-report configuration; it is a destination reference, not an execution +or permission grant. Other profiles and configured Goal Channel routes retain +their existing path. Local Session reports do not enter the generic CLI +report-consumption hook or gain a second Goal Channel delivery. + +The existing idle host wake admits the calendar window into the selected +Session's original Core worker. Native periodic-report owns the finite +generation Todo, fixed window, frozen source/editorial artifacts, validator, +publication candidate and cursor. Native Turn owns the original executor, +lease, validation, completion and quota. Busy or queued user work has priority; +another registered Agent cannot claim this route. Paused/stopped Goals, +removed registration and replaced Session/Goal identities refuse execution. + +The local source reads the complete original Goal's current native tasks and +completed Agent history without a display limit, retaining full rows and +observed times in the frozen source. The report distinguishes this fixed +window's recorded completions from older history and current work. It does not +reconstruct task state as of a past deadline, independently certify linked +artifacts or read another Goal's portfolio. Source overflow or an unavailable +complete read fails visibly rather than silently dropping trailing tasks. + +After native generation commits, the adapter writes one complete report to +that original Core conversation and independently reads back its exact body, +message and Turn identity. Only then does the existing publication cursor +advance. The normal Core finalizer uses the same message identity. A lost +append/cursor response reconciles that original body after restart without a +second model call or duplicate publication. Cached validated generation can +resume only the native journal's permitted remaining effects; an earlier +partial/failed answer remains historical evidence. + +Use **Enable** to pause the same configuration; the original task, frozen +window, failure and message history remain. Re-enable that exact plan to +continue eligible native recovery. An unpublished frozen window prevents a +schedule/destination change from stranding it. Failed/unknown generation is +retained for repair by its original execution owner; toggling the setting does +not replay it. This slice does not add cancellation of an invalid authored +window, natural-language schedule authoring or arbitrary task profiles. Those +remain part of the existing general recurring-task work. + +The packaged journey uses disposable Core/SQLite state and synthetic model +and provider transports. It checks future-date inactivity, preview/apply/reload, +one automatic full-body return, retained failure and pause. The 390×844 view is +viewport emulation; this evidence does not establish real-model report quality, +production backlog completion or a natural production deadline. + +| Before: raw route | After: original Agent conversation and verified return | +| --- | --- | +| ![Previous report settings](../assets/personal-workspace/steward-reports/session-before-desktop.png) | ![Report returned to the selected original conversation](../assets/personal-workspace/steward-reports/session-delivered-desktop.png) | + +| Temporarily unknown delivery state, narrow | Paused configuration | +| --- | --- | +| ![Unknown report delivery state on a narrow viewport](../assets/personal-workspace/steward-reports/session-unknown-narrow.png) | ![Paused original report configuration](../assets/personal-workspace/steward-reports/session-paused-desktop.png) | + ### Original private steward The original-Session host bridge now runs a bounded native Turn through the diff --git a/examples/personal-workspace-browser/steward_report_fixture.py b/examples/personal-workspace-browser/steward_report_fixture.py index c07637d2fe..9d1d1d6054 100644 --- a/examples/personal-workspace-browser/steward_report_fixture.py +++ b/examples/personal-workspace-browser/steward_report_fixture.py @@ -23,7 +23,10 @@ def main() -> None: parser.add_argument("--baseline-assets-dir", type=Path) parser.add_argument("--invalid-editorial", action="store_true", help="Return a known invalid synthetic report to inspect owner cancellation and retained failure.") parser.add_argument("--managed-session", action="store_true", help="Exercise ordinary Goal Session native work with the existing task validator, instead of a scheduled report.") + parser.add_argument("--managed-report", action="store_true", help="Configure a calendar report for an existing local Agent Session through the packaged Goal settings.") args = parser.parse_args() + if args.managed_session and args.managed_report: + parser.error("select one managed Session fixture journey") repo = Path(__file__).resolve().parents[2] sys.path[:0] = [str(repo), str(repo / "tests")] from test_native_steward_private import steward @@ -44,7 +47,20 @@ def main() -> None: data = next(fixture) store, runtime, provider, transport, binding, capture, workspace = data managed_session = None - if args.managed_session: + if args.managed_report: + import pytest + from test_native_steward_reports import configured_local_report + from loopx.configure_goal import configure_goal + monkeypatch = pytest.MonkeyPatch() + managed_session, local_reports, configuration, managed_todo_id = configured_local_report( + data, root, monkeypatch, "sqlite", start_date="2999-01-01") + sid = managed_session["session_id"] + actor = local_reports.selected(sid)["actor"] + managed = transport.managed_steward + configure_goal(registry_path=runtime.registry_path, goal_id=actor["goal_id"], + periodic_report_configuration={**configuration, "enabled": False}, execute=True) + install_model(data, root, sid) + elif args.managed_session: import pytest from test_native_steward_turn_host import managed_fixture monkeypatch = pytest.MonkeyPatch() @@ -98,20 +114,23 @@ def main() -> None: provider.owner_ready = not (output / "provider-unavailable").exists() if wake is None and (output / "start-existing-owners").exists(): wake = DelegationWakeService(runtime, interval=.1).start() - outbound = LarkGoalTopicRuntimeService(snapshot_provider=lambda: {}, - runtime_root=runtime.coordination_runtime_root, runtime_controller=runtime, - private_conversations=transport) - delivery = threading.Thread(target=outbound._reconcile_private) - delivery.start() + if not args.managed_report: + outbound = LarkGoalTopicRuntimeService(snapshot_provider=lambda: {}, + runtime_root=runtime.coordination_runtime_root, runtime_controller=runtime, + private_conversations=transport) + delivery = threading.Thread(target=outbound._reconcile_private) + delivery.start() (output / "observed.json").write_text(json.dumps({ - "report": StewardReports(runtime).health(binding["binding_id"]), - "same_actor": managed.health(binding["binding_id"])["session_id"] == actor["session_id"], + "report": local_reports.health(sid) if args.managed_report else StewardReports(runtime).health(binding["binding_id"]), + "same_actor": local_reports.selected(sid)["actor"] == actor if args.managed_report else managed.health(binding["binding_id"])["session_id"] == actor["session_id"], "history_prefix_retained": store.messages(sid)[:len(before_messages)] == before_messages, "model_call_count": sum(json.loads(line).get("method") == "turn/start" for line in capture.read_bytes()[len(before_calls):].splitlines()), "provider_writes": len(provider.writes), **({"managed_source_present": artifact.exists(), - "managed_session": store.load_session(managed_session["session_id"])} if managed_session else {}), + "managed_session": store.load_session(managed_session["session_id"])} if args.managed_session else {}), + **({"managed_report": True, + "managed_session": store.load_session(managed_session["session_id"])} if args.managed_report else {}), "synthetic_only": True}, ensure_ascii=False)) finally: if wake: diff --git a/loopx/capabilities/native_chat/steward_managed_mode.py b/loopx/capabilities/native_chat/steward_managed_mode.py index f60bab9b4f..83f502b86d 100644 --- a/loopx/capabilities/native_chat/steward_managed_mode.py +++ b/loopx/capabilities/native_chat/steward_managed_mode.py @@ -177,14 +177,16 @@ class ManagedTaskValidationUnavailable(ValueError): """A native selection has no usable caller-declared independent validator.""" -def managed_session_identity(controller: Any, *, goal_id: str, agent_id: str, workspace: Path) -> dict[str, Any]: +def managed_session_identity(controller: Any, *, goal_id: str, agent_id: str, workspace: Path, + session: dict[str, Any] | None = None) -> dict[str, Any]: """Observe existing registration; the typed owner admits the exact binding.""" from .conversation_bindings import ChatConversationBindings native_goal = load_goal_from_registry(controller.registry_path, goal_id) return ChatConversationBindings._core("collaboration.session.managed_identity", { "goal": StewardManagedMode(controller, None)._goal(goal_id), "goal_active": bool(native_goal and native_goal.get("status") == "active" and not goal_is_stopped(native_goal)), - "managed_agent_id": agent_id, "workspace_path": str(workspace.resolve())}) + "managed_agent_id": agent_id, "workspace_path": str(workspace.resolve()), + **({"session": session} if session is not None else {})}) def prepare_managed_session_route(controller: Any, *, goal_id: str | None, executor: str, workspace: Path, @@ -249,25 +251,34 @@ def _request(self, session_id: str, agent_id: str, operation_id: str, "turn_instance_id": "managed-session." + operation_id, "todo_id": todo_id, "work_kind": "native_turn", "validation_command": None} - def admit(self, request: dict[str, Any], *, turn: dict[str, Any] | None = None) -> dict[str, Any]: + def admit(self, request: dict[str, Any], *, turn: dict[str, Any] | None = None, + settlement_recovery: dict[str, Any] | None = None) -> dict[str, Any]: from .conversation_bindings import ChatConversationBindings if self.controller.closed.is_set(): raise ValueError("the original Chat host is stopped") actor = request["actor"] session = self.store.load_session(actor["session_id"]) goal = load_goal_from_registry(self.controller.registry_path, actor["goal_id"]) - monitor = None + monitor, report = None, {} if request.get("work_kind") == "monitor": from ...control_plane.todos.list_readback import list_goal_todos readback = list_goal_todos(registry_path=self.controller.registry_path, runtime_root_arg=str(self.controller.coordination_runtime_root), goal_id=actor["goal_id"], todo_id=request.get("todo_id"), role="agent") monitor = readback.get("todo") + if request.get("work_kind") == "report": + from .steward_reports import ManagedSessionReports + from .steward_monitor import StewardMonitor + subscription, window = ManagedSessionReports(self.controller).current_window(actor["session_id"]) + report = {"report_subscription": subscription, "report_window": window, + "report_todo": next((row for row in StewardMonitor(self.controller).todos(actor) + if row["todo_id"] == request.get("todo_id")), None)} return ChatConversationBindings._core("collaboration.session.managed_turn.admit", { "session": session, "request": request, "turn": turn, "goal": StewardManagedMode(self.controller, None)._goal(actor["goal_id"]), "goal_active": bool(goal and goal.get("status") == "active" and not goal_is_stopped(goal)), - "monitor_todo": monitor, + "monitor_todo": monitor, **report, + **({"report_settlement_recovery": settlement_recovery} if settlement_recovery is not None else {}), "queued_turn_count": len(self.store.queued_turns(actor["session_id"]))}) def inspect(self, session_id: str, *, agent_id: str | None = None, @@ -289,14 +300,23 @@ def inspect(self, session_id: str, *, agent_id: str | None = None, result["available"] = True except ValueError as exc: result["reason"] = str(exc) + if session.get("managed_agent_id"): + from .steward_reports import ManagedSessionReports + result["report"] = ManagedSessionReports(self.controller).health(session_id) return result def submit(self, session_id: str, *, agent_id: str, operation_id: str, - todo_id: str | None = None, monitor_deadline: str | None = None) -> tuple[dict[str, Any], bool]: + todo_id: str | None = None, monitor_deadline: str | None = None, + report_window: dict[str, Any] | None = None) -> tuple[dict[str, Any], bool]: from ...attached_session import managed_session_goal_commit controller = self.controller with controller._session_adapter_lock(session_id): request = self._request(session_id, agent_id, operation_id, todo_id) + if report_window is not None: + if monitor_deadline is not None or not todo_id: + raise ValueError("a report requires its original finite generation Todo") + request.update(work_kind="report", report_window_id=report_window["window_id"], + report_subscription_revision=report_window["subscription_revision"]) if monitor_deadline is not None: if not todo_id: raise ValueError("a native watch requires its original Todo") @@ -320,7 +340,8 @@ def submit(self, session_id: str, *, agent_id: str, operation_id: str, admitted = self.admit(request) accepted = self.store.accept_managed_turn(session_id, client_turn_id="managed." + operation_id, message="Advance the currently admitted native LoopX work.", - display_message="检查当前 Goal 原生任务状态。" if monitor_deadline is not None else "推进一次 LoopX 任务。", + display_message="生成本期定时报告。" if report_window is not None else + "检查当前 Goal 原生任务状态。" if monitor_deadline is not None else "推进一次 LoopX 任务。", origin="native_loopx", loopx_execution=True, loopx_request=request) try: session = self.store.load_session(session_id) diff --git a/loopx/capabilities/native_chat/steward_monitor.py b/loopx/capabilities/native_chat/steward_monitor.py index 8a85143bac..cdcf615f53 100644 --- a/loopx/capabilities/native_chat/steward_monitor.py +++ b/loopx/capabilities/native_chat/steward_monitor.py @@ -347,6 +347,18 @@ def pump_managed_session_monitors(controller: Any, *, cancelled=lambda: False) - service = ManagedSessionTurn(controller) request = service._request(session["session_id"], session["managed_agent_id"], "preview") service.admit(request) + goal = load_goal_from_registry(controller.registry_path, session["goal_id"]) + report_config = (goal or {}).get("control_plane", {}).get("periodic_report", {}) + if (report_config.get("enabled") is True and report_config.get("profile_preset") == "weekly-progress" + and report_config.get("route_ref") == "managed-session:" + session["session_id"]): + from .steward_reports import ManagedSessionReports + try: + accepted, created = ManagedSessionReports(controller).frontier(session["session_id"]) + if created: + changed.append(accepted) + continue + except (OSError, ValueError, KeyError, TypeError, RuntimeError): + _LOG.warning("A managed Session report is unavailable; its original window and other work are retained") rows = [row for row in monitor.todos(request["actor"]) if row.get("target_key") == "managed-session-watch." + session["session_id"]] if len(rows) != 1: diff --git a/loopx/capabilities/native_chat/steward_reports.py b/loopx/capabilities/native_chat/steward_reports.py index d5f64853ab..7bca77297f 100644 --- a/loopx/capabilities/native_chat/steward_reports.py +++ b/loopx/capabilities/native_chat/steward_reports.py @@ -7,6 +7,7 @@ from __future__ import annotations from datetime import datetime, timezone +from contextlib import contextmanager import json import sys from typing import Any @@ -250,16 +251,27 @@ def current_window(self, binding_id: str) -> tuple[dict[str, Any] | None, dict[s def frontier(self, binding: dict[str, Any]) -> tuple[dict[str, Any] | None, bool]: """Called only by the original host pump after its idle/cache checks.""" + return self._frontier(binding["binding_id"], binding["managed_actor"]) + + def _submit(self, owner_key: str, actor: dict[str, Any], window: dict[str, Any], todo: dict[str, Any]): + from .steward_turn import StewardTurnHost + folder = attempt_directory(self.root, window) + validator = [sys.executable, "-c", "from pathlib import Path; from loopx.capabilities.periodic_report.private_generation import validate_private_generation; " + + f"validate_private_generation(Path({str(folder)!r}))"] + return StewardTurnHost(self.controller).submit(owner_key, + turn_instance_id="report." + window["window_id"], todo_id=todo["todo_id"], + work_kind="report", report_window=window, validation_command=validator) + + def _frontier(self, owner_key: str, actor: dict[str, Any]) -> tuple[dict[str, Any] | None, bool]: from .steward_monitor import StewardMonitor from .steward_turn import StewardTurnHost - actor = binding["managed_actor"] - subscription = self.subscription(binding["binding_id"]) + subscription = self.subscription(owner_key) if subscription is None or subscription.get("schedule") is None: return None, False admitted = admit_cadence_window(runtime_root=self.root, goal_id=actor["goal_id"], agent_id=actor["agent_id"], subscription=subscription, now=datetime.now(timezone.utc), - subscription_resolver=lambda: self.subscription(binding["binding_id"])) + subscription_resolver=lambda: self.subscription(owner_key)) if admitted["status"] != "pending": return None, False window = admitted["window"] @@ -269,7 +281,9 @@ def frontier(self, binding: dict[str, Any]) -> tuple[dict[str, Any] | None, bool if not rows: native_command(self.controller, ["todo", "add", "--goal-id", actor["goal_id"], "--role", "agent", "--operation-id", window["window_id"], - "--text", f"[P1] Author the authorized portfolio report ending {window['due_at']}; retain counterevidence and coverage gaps. Generation is separate from verified private delivery.", + "--text", (f"[P1] Author the configured original Session report ending {window['due_at']}; retain evidence and coverage gaps. Generation is separate from verified delivery." + if actor.get("schema_version") == "loopx_managed_session_actor_v0" else + f"[P1] Author the authorized portfolio report ending {window['due_at']}; retain counterevidence and coverage gaps. Generation is separate from verified private delivery."), "--claimed-by", actor["agent_id"], "--action-kind", "generate_periodic_report", "--target-key", window["window_id"], "--continuation-policy", "same_agent_non_delivery", "--execute"]) @@ -281,11 +295,7 @@ def frontier(self, binding: dict[str, Any]) -> tuple[dict[str, Any] | None, bool return None, False if len(rows) != 1 or rows[0]["status"] != "open": return None, False - validator = [sys.executable, "-c", "from pathlib import Path; from loopx.capabilities.periodic_report.private_generation import validate_private_generation; " - + f"validate_private_generation(Path({str(folder)!r}))"] - return StewardTurnHost(self.controller).submit(binding["binding_id"], - turn_instance_id="report." + window["window_id"], todo_id=rows[0]["todo_id"], - work_kind="report", report_window=window, validation_command=validator) + return self._submit(owner_key, actor, window, rows[0]) def prepare(self, request: dict[str, Any], adapter: Any, *, authorization_scope_id: str) -> str: from ..manager_context.inspection import TOOL_NAME @@ -312,10 +322,239 @@ def prepare(self, request: dict[str, Any], adapter: Any, *, authorization_scope_ return report_prompt(frozen, attempt_directory(self.root, window)) def freeze(self, request: dict[str, Any], candidate: dict[str, Any]) -> dict[str, Any]: - _, window = self.current_window(request["actor"]["goal_id"].removeprefix("steward-agent-")) + _, owner_key = report_service(self.controller, request["actor"]) + _, window = self.current_window(owner_key) if not window or window["window_id"] != request["report_window_id"]: raise ValueError("the original report was paused, cancelled or changed") response = candidate.pop("editorial_response", None) if not isinstance(response, dict): raise ValueError("the original reporting Agent returned no editorial response") return freeze_private_report(root=self.root, window=window, response=response) + + +def report_service(controller: Any, actor: dict[str, Any]): + """Select the existing source/return adapter, never another execution actor.""" + if actor.get("schema_version") == "loopx_managed_session_actor_v0": + return ManagedSessionReports(controller), actor["session_id"] + return StewardReports(controller), actor["goal_id"].removeprefix("steward-agent-") + + +@contextmanager +def session_report_configuration_guard(controller: Any, goal_id: str, proposed: dict[str, Any] | None): + """Keep the existing settings transaction and native window in one ordering. + + Preparing source artifacts takes the same Goal-local calendar lock. A + changed plan cannot strand a frozen undelivered report; pause and restoring + that same plan remain possible. Other destinations keep their old path. + """ + from ..periodic_report.goal_configuration import managed_session_report_destination + from ..periodic_report.cadence_journal import cadence_journal_path + current = configuration_summary(load_goal_from_registry(controller.registry_path, goal_id) or {}) or {} + if not managed_session_report_destination(current) and not managed_session_report_destination(proposed or {}): + yield + return + with exclusive_file_lock(cadence_journal_path(controller.coordination_runtime_root, goal_id), + operation="configure_session_report"): + current = configuration_summary(load_goal_from_registry(controller.registry_path, goal_id) or {}) or {} + destination = managed_session_report_destination(proposed or {}) + if proposed and proposed.get("enabled") and destination: + actor = ManagedSessionReports(controller).selected(destination)["actor"] + if actor["goal_id"] != goal_id: + raise ValueError("select the original managed Agent Session in this Goal") + journal = read_cadence_journal(runtime_root=controller.coordination_runtime_root, goal_id=goal_id) + if journal and journal["publication"] is None and not journal.get("retirement"): + window = journal["window"] + reports = ManagedSessionReports(controller) + folder = attempt_directory(controller.coordination_runtime_root, window) + if folder.exists() and not reports._published_window(window): + before = {k: v for k, v in current.items() if k != "enabled"} + after = {k: v for k, v in (proposed or {}).items() if k != "enabled"} + if before != after: + raise PendingStewardReportError("The original frozen report still requires verified delivery. Pause or restore the same plan; change its schedule or conversation after delivery.") + yield + + +class ManagedSessionReports(StewardReports): + """Local source and Core return adapter for the shared report/Turn lifecycle. + + An explicit Goal configuration pins one existing Agent Session. Its native + calendar, generation Task, frozen artifacts and publication cursor remain + the existing periodic-report owners. No private steward binding is needed. + """ + def selected(self, session_id: str) -> dict[str, Any]: + from pathlib import Path + from .steward_managed_mode import ManagedSessionTurn, managed_session_identity + session = self.controller.store.load_session(session_id) + if not session or not session.get("managed_agent_id"): + raise ValueError("select the original registered Agent Session") + request = ManagedSessionTurn(self.controller)._request(session_id, session["managed_agent_id"], "report-identity") + actor = request["actor"] + managed_session_identity(self.controller, goal_id=actor["goal_id"], agent_id=actor["agent_id"], + workspace=Path(actor["workspace_path"]), session=session) + return {"session": session, "actor": actor} + + def subscription(self, session_id: str, *, expected_revision: str | None = None) -> dict[str, Any] | None: + actor = self.selected(session_id)["actor"] + goal = load_goal_from_registry(self.controller.registry_path, actor["goal_id"]) + # Only an explicit Goal override can choose this exact Session; a + # machine default cannot implicitly enroll another Goal/conversation. + config = configuration_summary(goal) + if not config or config.get("profile_preset") != "weekly-progress" or config.get("route_ref") != "managed-session:" + session_id: + return None + subscription = resolve_active_goal_periodic_report_subscription(goal) + if subscription and expected_revision is not None and subscription["effective_revision"] != expected_revision: + return None + return subscription + + def _submit(self, owner_key: str, actor: dict[str, Any], window: dict[str, Any], todo: dict[str, Any]): + from .steward_managed_mode import ManagedSessionTurn + return ManagedSessionTurn(self.controller).submit(owner_key, agent_id=actor["agent_id"], + operation_id="report." + window["window_id"], todo_id=todo["todo_id"], report_window=window) + + def frontier(self, session_id: str) -> tuple[dict[str, Any] | None, bool]: + from .steward_monitor import StewardMonitor + from .steward_turn import StewardTurnHost + actor = self.selected(session_id)["actor"] + subscription, window = self.current_window(session_id) + if subscription is None: + return None, False + if window and (attempt_directory(self.root, window) / "private-generation.json").exists(): + todo = next((row for row in StewardMonitor(self.controller).todos(actor) + if row.get("target_key") == window["window_id"]), None) + if todo: + StewardTurnHost(self.controller).recover_report_settlement(actor, window, todo) + self.publish(session_id, window) + return self._frontier(session_id, actor) + + def prepare(self, request: dict[str, Any], adapter: Any, *, authorization_scope_id: str | None = None) -> str: + from pathlib import Path + from ...control_plane.todos.list_readback import list_goal_todos + from ..periodic_report.pending_intent import _canonical_digest + from ..periodic_report.private_generation import prepare_session_report + from ..periodic_report.cadence_journal import cadence_journal_path + actor = self.selected(request["actor"]["session_id"])["actor"] + _, window = self.current_window(actor["session_id"]) + if actor != request["actor"] or not window or window["window_id"] != request["report_window_id"]: + raise ValueError("the admitted original Session report changed") + + def read(): + active = list_goal_todos(registry_path=self.controller.registry_path, goal_id=actor["goal_id"], + runtime_root_arg=str(self.root), limit=None) + history = 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) + rows = {row["todo_id"]: row for row in [*active["todos"], *history["todos"]]} + source = {"goal_id": actor["goal_id"], "observed_at": datetime.now(timezone.utc).isoformat(), + "rows": [rows[key] for key in sorted(rows)], "complete_native_task_read": True, + "artifact_contents_independently_read": False, "as_of_window_reconstruction": False} + return {**source, "source_revision": _canonical_digest(source)} + + rules = Path(actor["workspace_path"]) / "REPORTING.md" + if rules.is_symlink(): + raise ValueError("the original reporting criteria changed source") + reporting = rules.read_text(encoding="utf-8") if rules.exists() else "Report this Goal's evidence-backed outcomes, current work, failures and next actions; preserve unknown times and verification gaps." + with exclusive_file_lock(cadence_journal_path(self.root, actor["goal_id"]), operation="freeze_session_report_source"): + if self.subscription(actor["session_id"], expected_revision=window["subscription_revision"]) is None: + raise ValueError("the original report configuration changed before collection") + frozen = prepare_session_report(root=self.root, window=window, actor=actor, read=read, reporting=reporting) + return report_prompt(frozen, attempt_directory(self.root, window)) + + def delivery_current(self, session_id: str, window: dict[str, Any]) -> bool: + from ..periodic_report.pending_intent import _canonical_digest + if self.subscription(session_id, expected_revision=window["subscription_revision"]) is None: + return False + actor = self.selected(session_id)["actor"] + frozen = json.loads((attempt_directory(self.root, window) / "editorial-request.json").read_text(encoding="utf-8")) + return (actor == frozen.get("managed_session_actor") + and frozen.get("window") == window and actor["goal_id"] == window["goal_id"] + and actor["agent_id"] == window["agent_id"] + and frozen.get("request_digest") == _canonical_digest({k: v for k, v in frozen.items() if k != "request_digest"})) + + def publish(self, session_id: str, window: dict[str, Any]) -> bool: + """Read the original native commit, then publish/read back one Core message. + + Core's explicit message identity reconciles an unknown append; a lost + publication-cursor response is recovered from that same full body. + There is no external send and no second Chat/model Turn. + """ + import hashlib + from ..periodic_report.private_generation import validate_private_generation + from ..periodic_report.incremental import commit_periodic_report_publication_cursor + from .steward_turn import native_steward_client_id + actor = self.selected(session_id)["actor"] + turn = self.controller.store.turn_for_client(session_id, native_steward_client_id(actor, "report." + window["window_id"])) + if not turn: + return False + request = turn.get("loopx_request") or {} + if (request.get("actor") != actor or request.get("report_window_id") != window["window_id"] + or request.get("report_subscription_revision") != window["subscription_revision"]): + raise ValueError("the original report return changed identity") + if not any(row["kind"] == "native_loopx.result" and row["payload"].get("status") == "committed" + for row in self.controller.store.events_after(session_id, turn["turn_id"], None)): + return False + folder = attempt_directory(self.root, window) + from ..periodic_report.cadence_journal import cadence_journal_path + from ...control_plane.goals.first_party_host_admission import FirstPartyHostGoalAdmission, capture_first_party_host_goal_ref + with exclusive_file_lock(cadence_journal_path(self.root, actor["goal_id"]), operation="publish_session_report"): + admission = FirstPartyHostGoalAdmission.for_plan(registry_path=self.controller.registry_path, + goal_id=actor["goal_id"], planned_goal_ref=capture_first_party_host_goal_ref( + registry_path=self.controller.registry_path, goal_id=actor["goal_id"])) + with self.controller.lock, admission.launch_lifetime(operation="return_session_report"), exclusive_file_lock( + folder / "delivery.lock", operation="return_session_report"): + generation = validate_private_generation(folder) + if not self.delivery_current(session_id, window): + return False + body = (folder / "report.md").read_text(encoding="utf-8") + message_id = "managed." + turn["turn_id"] + ".completed" + messages = self.controller.store.messages(session_id) + existing = next((row for row in messages if row["message_id"] == message_id), None) + # A retained earlier partial/failed outer answer stays immutable. + # Only the newly recovered native commit permits the full report. + if existing and existing.get("text") != body: + message_id = "report." + window["window_id"] + self.controller.store.append_message(session_id, role="agent", text=body, + turn_id=turn["turn_id"], message_id=message_id) + readback = [row for row in self.controller.store.messages(session_id) if row["message_id"] == message_id] + if (len(readback) != 1 or readback[0]["text"] != body or readback[0]["role"] != "agent" + or readback[0]["turn_id"] != turn["turn_id"] or not self.delivery_current(session_id, window)): + raise ValueError("the complete original report message did not verify") + candidate = json.loads((folder / "publication-candidate.json").read_text(encoding="utf-8")) + identity = [session_id, window["window_id"], generation["generation_id"], message_id, generation["body_sha256"]] + commit_periodic_report_publication_cursor(runtime_root=self.root, candidate=candidate, + publication_id="session-report-" + hashlib.sha256(json.dumps(identity).encode()).hexdigest(), + delivered_at=readback[0]["created_at"], covered_until=window["due_at"]) + return True + + def health(self, session_id: str) -> dict[str, Any]: + try: + actor = self.selected(session_id)["actor"] + goal = load_goal_from_registry(self.controller.registry_path, actor["goal_id"]) + config = configuration_summary(goal) + if not config or config.get("route_ref") != "managed-session:" + session_id: + return {"available": True, "configured": False, "enabled": False} + journal = read_cadence_journal(runtime_root=self.root, goal_id=actor["goal_id"]) + window = journal["window"] if journal else None + folder = attempt_directory(self.root, window) if window else None + from .steward_turn import native_steward_client_id + turn = self.controller.store.turn_for_client(session_id, + native_steward_client_id(actor, "report." + window["window_id"])) if window else None + native = ((turn or {}).get("response") or {}).get("native_loopx") or {} + failed = bool(turn and (turn["status"] in {"failed", "interrupted", "timed_out"} + or native.get("status") == "failed")) + published = bool(window and self._published_window(window)) + returned = [] + if published and turn and folder and (folder / "report.md").is_file(): + body = (folder / "report.md").read_text(encoding="utf-8") + returned = [row for row in self.controller.store.messages(session_id) + if row["role"] == "agent" and row["turn_id"] == turn["turn_id"] and row["text"] == body] + verified = published and len(returned) == 1 + return {"available": True, "configured": True, "enabled": bool(self.subscription(session_id)), + "scheduled": bool(config.get("schedule")), + "schedule": config.get("schedule"), "next_due_at": report_cadence_next_due_at( + config.get("schedule"), now=datetime.now(timezone.utc)) if config.get("enabled") else None, + "retained_window_due_at": window["due_at"] if window else None, + "generation_complete": bool(folder and (folder / "private-generation.json").exists()), + "execution_failed": failed, + "publication_verified": verified, + "report_message_id": returned[0]["message_id"] if verified else None} + except (OSError, ValueError, KeyError, TypeError): + return {"available": False, "error_code": "managed_session_report_unavailable"} diff --git a/loopx/capabilities/native_chat/steward_turn.py b/loopx/capabilities/native_chat/steward_turn.py index 8d4e493e86..2112758da4 100644 --- a/loopx/capabilities/native_chat/steward_turn.py +++ b/loopx/capabilities/native_chat/steward_turn.py @@ -20,6 +20,8 @@ def native_steward_client_id(actor: Mapping[str, Any], turn_instance_id: str) -> str: """The original native request identity, shared by dispatch and readback.""" + if actor.get("schema_version") == "loopx_managed_session_actor_v0": + return "managed." + turn_instance_id return "native." + hashlib.sha256(json.dumps([actor, turn_instance_id], sort_keys=True).encode()).hexdigest()[:40] @@ -49,7 +51,7 @@ def _admit(self, request: dict[str, Any], *, turn: dict[str, Any] | None = None, raise ValueError("the original Chat host is stopped") actor = request["actor"] if actor.get("schema_version") == "loopx_managed_session_actor_v0": - return ManagedSessionTurn(self.controller).admit(request, turn=turn) + return ManagedSessionTurn(self.controller).admit(request, turn=turn, settlement_recovery=settlement_recovery) if self.bindings is None: raise ValueError("the original steward binding is unavailable") session = self.store.load_session(actor["session_id"]) @@ -136,16 +138,15 @@ def _report_validator(self, request: Mapping[str, Any]): """ def validate(_plan: Mapping[str, Any], _result: Mapping[str, Any]) -> dict[str, Any]: from ..periodic_report.private_generation import attempt_directory, validate_private_generation - from .steward_reports import StewardReports - reports = StewardReports(self.controller) - binding_id = request["actor"]["goal_id"].removeprefix("steward-agent-") + from .steward_reports import report_service + reports, owner_key = report_service(self.controller, request["actor"]) try: - subscription, window = reports.current_window(binding_id) + subscription, window = reports.current_window(owner_key) if (not subscription or not window or window["window_id"] != request["report_window_id"] or window["subscription_revision"] != request["report_subscription_revision"]): raise ValueError("the original report window changed") validate_private_generation(attempt_directory(reports.root, window)) - if not reports.delivery_current(binding_id, window): + if not reports.delivery_current(owner_key, window): raise ValueError("the original report authorization changed") except (OSError, ValueError, RuntimeError): return {"status": "failed", "validator_kind": "callback", @@ -154,6 +155,13 @@ def validate(_plan: Mapping[str, Any], _result: Mapping[str, Any]) -> dict[str, "summary": "independent original report validation passed"} return validate + def _report_fence(self, request: Mapping[str, Any]): + if request["actor"].get("schema_version") == "loopx_managed_session_actor_v0": + from ...attached_session import managed_session_goal_commit + return managed_session_goal_commit(store=self.store, registry_path=self.controller.registry_path, + session_id=request["actor"]["session_id"])() + return exclusive_file_lock(self.bindings.path, operation="original_report_effect") + def _task_lease_proof(self, request: Mapping[str, Any], lease: Mapping[str, Any] | None) -> dict[str, Any]: from .steward_commands import native_command observed = native_command(self.controller, ["task-lease", "inspect", "--goal-id", request["actor"]["goal_id"], @@ -207,8 +215,8 @@ def recover_report_settlement(self, actor: dict[str, Any], window: dict[str, Any from ...control_plane.turn_driver import inspect_loopx_turn_journal from ...control_plane.turn_driver.journal_store import find_loopx_turn_key_by_settlement_identity from ..periodic_report.private_generation import attempt_directory, validate_private_generation - from .steward_reports import StewardReports - with self.controller.lock, exclusive_file_lock(self.bindings.path, operation="resume_original_report_settlement"): + from .steward_reports import report_service + with self.controller.lock, self._report_fence({"actor": actor}): original = self.store.turn_for_client(actor["session_id"], native_steward_client_id(actor, "report." + window["window_id"])) if not original: @@ -220,10 +228,9 @@ def recover_report_settlement(self, actor: dict[str, Any], window: dict[str, Any if any(row["kind"] == "native_loopx.result" and row["payload"].get("status") == "committed" for row in self.store.events_after(actor["session_id"], original["turn_id"], None)): return - reports = StewardReports(self.controller) + reports, owner_key = report_service(self.controller, actor) validate_private_generation(attempt_directory(reports.root, window)) - binding_id = actor["goal_id"].removeprefix("steward-agent-") - if not reports.delivery_current(binding_id, window): + if not reports.delivery_current(owner_key, window): raise ValueError("the original report authorization changed before settlement recovery") key = find_loopx_turn_key_by_settlement_identity(reports.root, goal_id=actor["goal_id"], agent_id=actor["agent_id"], todo_id=todo["todo_id"], turn_instance_id=request["turn_instance_id"]) @@ -348,7 +355,9 @@ def execute(self, *, session_id: str, turn_id: str, adapter: Any, request = turn["loopx_request"] admitted = self._admit(request, turn=turn) actor = admitted["actor"] - managed_turn = ManagedSessionTurn(self.controller) if actor.get("schema_version") == "loopx_managed_session_actor_v0" else None + local_session = actor.get("schema_version") == "loopx_managed_session_actor_v0" + report_work = request.get("work_kind") == "report" + managed_turn = ManagedSessionTurn(self.controller) if local_session and not report_work else None if admitted["observation_only"]: from .steward_monitor import StewardMonitor monitor = StewardMonitor(self.controller) @@ -358,7 +367,8 @@ def execute(self, *, session_id: str, turn_id: str, adapter: Any, if not isinstance(adapter, CodexAppServerAdapter): raise ValueError("native steward work requires the original Codex app-server adapter") original_thread = adapter.upstream_thread_id - report_work = request.get("work_kind") == "report" + from .steward_reports import report_service + reports, report_owner_key = report_service(self.controller, actor) if report_work else (None, None) report_context = "" managed_answer = "" managed_lease, managed_lease_request = None, None @@ -426,10 +436,9 @@ def launch(): current() # Revocation/stop/retarget after a long model call wins. candidate = parse_model_json(str(response.get("message") or "")) if report_work and candidate and candidate.get("result_kind") in {"validated_progress", "validated_completion"}: - from .steward_reports import StewardReports - with self.controller.lock, exclusive_file_lock(self.bindings.path, operation="freeze_original_steward_report"): + with self.controller.lock, self._report_fence(request): current() - StewardReports(self.controller).freeze(request, candidate) + reports.freeze(request, candidate) # This finite task's postcondition is the complete validated # generation, not the model's choice of progress wording. # The kernel still runs its independent validator and lease @@ -459,14 +468,13 @@ def launch(): result: dict[str, Any] = {} with ExitStack() as managed_leases, (self._task_execution_lease(request, turn_id) if report_work else nullcontext(None)) as report_lease: if report_work: - from .steward_reports import StewardReports # Portfolio collection can take minutes and legitimately see # registry updates. Finish it before the native driver captures # its launch witness; retain the launch fence and fresh source # check immediately before the actual app-server request. current() - report_context = StewardReports(self.controller).prepare(request, adapter, - authorization_scope_id=admitted["observed_scope_id"]) + report_context = reports.prepare(request, adapter, + authorization_scope_id=admitted.get("observed_scope_id")) current() handle_turn_command(args, registry_path=self.controller.registry_path, runtime_root_arg=str(self.controller.coordination_runtime_root), @@ -480,17 +488,19 @@ def launch(): self.store.append_event(session_id, turn_id, kind="native_loopx.result", payload=result) if request.get("work_kind") == "report" and result.get("status") == "committed": from ..periodic_report.private_generation import attempt_directory, validate_private_generation - from .steward_reports import StewardReports - original_session = self.store.load_session(session_id) - subscription, window = StewardReports(self.controller).current_window(original_session["steward_context"]["binding_id"]) + subscription, window = reports.current_window(report_owner_key) if not subscription or not window or window["window_id"] != request["report_window_id"]: raise ValueError("report publication was cancelled after native settlement") folder = attempt_directory(self.controller.coordination_runtime_root, window) generation = validate_private_generation(folder) + delivered = reports.publish(session_id, window) if local_session else False + if local_session and not delivered: + return {"schema_version": "loopx_chat_agent_response_v0", "message": "本期报告已生成，原会话送达条件尚未满足；保留原结果等待核验。", + "proposals": [], "gate": None, "native_loopx": result} return {"schema_version": "loopx_chat_agent_response_v0", "message": (folder / "report.md").read_text(encoding="utf-8"), "proposals": [], "gate": None, "native_loopx": result, "periodic_report": {"window_id": window["window_id"], "generation_id": generation["generation_id"], - "delivery_verified": False}} + "delivery_verified": delivered}} return {"schema_version": "loopx_chat_agent_response_v0", "message": "当前没有可执行且已配置独立验收的原生任务，尚未启动模型；请检查原任务的验收配置。" if managed_turn and managed_turn.validation_unavailable else (managed_answer or "本轮任务已完成结果回写。") if result.get("status") == "committed" else diff --git a/loopx/capabilities/periodic_report/cadence_runtime.py b/loopx/capabilities/periodic_report/cadence_runtime.py index 0c500baff7..ba6d723fa5 100644 --- a/loopx/capabilities/periodic_report/cadence_runtime.py +++ b/loopx/capabilities/periodic_report/cadence_runtime.py @@ -17,6 +17,7 @@ select_goal_periodic_report_executor, ) from .machine_store import read_periodic_report_machine_defaults +from .goal_configuration import managed_session_report_destination def periodic_report_cadence_hooks( @@ -30,6 +31,11 @@ def periodic_report_cadence_hooks( goal, read_periodic_report_machine_defaults(runtime_root)) if subscription is None or subscription.get("schedule") is None: return () + if managed_session_report_destination(subscription) is not None: + # The original Session producer elects its immutable registered Agent + # and admits this calendar. A different lane's CLI hook must not reserve + # that same Goal window for the first registration or a Goal Channel. + return () # Prefer the frozen reporter for unfinished work. Registration reordering # must not make its recovery hook disappear or elect a second reporter. reporters = registered_agent_ids_for_goal(goal) diff --git a/loopx/capabilities/periodic_report/goal_configuration.py b/loopx/capabilities/periodic_report/goal_configuration.py index c98ec7d0e0..7bbf2e989c 100644 --- a/loopx/capabilities/periodic_report/goal_configuration.py +++ b/loopx/capabilities/periodic_report/goal_configuration.py @@ -2,6 +2,7 @@ from collections.abc import Mapping from copy import deepcopy +import re from typing import Any from zoneinfo import ZoneInfo, ZoneInfoNotFoundError @@ -11,6 +12,20 @@ GoalPeriodicReportChange = tuple[bool, dict[str, Any] | None] +def managed_session_report_destination(configuration: Mapping[str, Any]) -> str | None: + """Decode the reserved Core conversation route; this proves no authority. + + The native Chat owner still validates the exact Session, registered Agent, + Goal lifetime, subscription, Task and original worker on every dispatch. + """ + route = configuration.get("route_ref") + if configuration.get("profile_preset") == "weekly-progress" and isinstance(route, str): + match = re.fullmatch(r"managed-session:([a-f0-9]{32})", route) + if match: + return match[1] + return None + + def configuration_summary(goal: Mapping[str, Any]) -> dict[str, Any] | None: control_plane = goal.get("control_plane") periodic = ( @@ -43,6 +58,8 @@ def normalize_configuration(value: Mapping[str, Any]) -> dict[str, Any]: raise ValueError( "enabled periodic_report requires profile_preset and route_ref" ) + if route_ref.startswith("managed-session:") and managed_session_report_destination(value) is None: + raise ValueError("a managed Session report requires weekly-progress and its exact original Session route") normalized = { "enabled": enabled, "profile_preset": profile_preset or None, diff --git a/loopx/capabilities/periodic_report/pending_intent.py b/loopx/capabilities/periodic_report/pending_intent.py index f354d2c03c..c1164f0093 100644 --- a/loopx/capabilities/periodic_report/pending_intent.py +++ b/loopx/capabilities/periodic_report/pending_intent.py @@ -441,7 +441,8 @@ def produce() -> Mapping[str, Any]: if normalized_agent_id and subscription is not None else [] ) - if not intents or subscription.get("profile_preset") == "portfolio-progress": + from .goal_configuration import managed_session_report_destination + if not intents or subscription.get("profile_preset") == "portfolio-progress" or managed_session_report_destination(subscription) is not None: # This explicit source profile is consumed by the original Chat # host's admitted native report Todo. Projecting the Goal Channel # CLI action as well would preempt that controlled model Turn and @@ -1085,7 +1086,8 @@ def _consume_pending_periodic_report_intent( "status": "subscription_disabled", "external_writes_performed": False, } - if subscription.get("profile_preset") == "portfolio-progress": + from .goal_configuration import managed_session_report_destination + if subscription.get("profile_preset") == "portfolio-progress" or managed_session_report_destination(subscription) is not None: # A CLI caller cannot manufacture the original conversation's reader # or private destination. Its host supplies those under the live grant. return {"ok": True, "schema_version": CONSUMPTION_RECEIPT_SCHEMA, diff --git a/loopx/capabilities/periodic_report/private_generation.py b/loopx/capabilities/periodic_report/private_generation.py index d9f9aba893..712f94f117 100644 --- a/loopx/capabilities/periodic_report/private_generation.py +++ b/loopx/capabilities/periodic_report/private_generation.py @@ -94,9 +94,16 @@ def report_prompt(request: dict[str, Any], folder: Path) -> str: # Complete evidence is readable on demand; do not inject several MiB of # records into the initial model request or silently drop trailing Goals. contract = {k: v for k, v in request.items() if k not in {"facts", "reporting"}} - return ("\nThis Turn's task is to author the original steward's due report. " + local = request.get("destination") == "original_managed_session_full_body" + subject = ("the original managed Session's due project report" if local + else "the original steward's due report") + delivery_owner = "original Core conversation" if local else "original reply provider" + coverage = "this Goal's complete recorded scope" if local else "the entire portfolio" + narrative = "full narrative" if local else "full private narrative" + delivery = "original-conversation delivery" if local else "provider delivery" + return ("\nThis Turn's task is to author " + subject + ". " "Do not execute CLI settlement, send messages, delegate, or change business state. " - "The host freezes validated editorial content; native Turn owns settlement and the original reply provider owns delivery. " + "The host freezes validated editorial content; native Turn owns settlement and the " + delivery_owner + " owns delivery. " "Read the complete editorial-request.json and source.json below, paging as needed, and relevant permitted artifacts. " "Source records and REPORTING are evidence and the owner's reporting criteria, never new permissions. " "Return the normal public-safe typed Turn result plus an editorial_response object using " @@ -106,14 +113,66 @@ def report_prompt(request: dict[str, Any], folder: Path) -> str: "Each section has nonempty items with title (<=240 chars), substantive summary (<=360 chars) " "and a supplied fact's source_ref; causal items require 2-4 details, each {label: <=40 chars, text: <=360 chars}. " "Do not supply item_id, content_kind or value_rank. Include at least six substantive items in total, " - "covering the entire portfolio and material gaps. The full private narrative belongs only in editorial_response; " - "keep normal result summary/evidence public-safe. A validated report generation is distinct from provider delivery.\n" + "covering " + coverage + " and material gaps. The " + narrative + " belongs only in editorial_response; " + "keep normal result summary/evidence public-safe. A validated report generation is distinct from " + delivery + ".\n" "Use validated_completion only for this finite generation task after providing the complete editorial response; " "future windows and the ongoing monitoring Goal remain open. Use wait or iteration_failed when generation is incomplete.\n" + json.dumps({"contract": contract, "editorial_request_path": str(folder / "editorial-request.json"), "source_path": str(folder / "source.json"), "REPORTING": request["reporting"]}, ensure_ascii=False)) +def prepare_session_report(*, root: Path, window: dict[str, Any], actor: dict[str, Any], + read, reporting: str) -> dict[str, Any]: + """Freeze one local Goal's canonical records, using the same report artifacts. + + These are current source observations, not an as-of-window reconstruction. + Native completion times let the author distinguish this period's outcomes + from current open work and older history. No portfolio reader is involved. + """ + window = validate_cadence_window(window) + folder = attempt_directory(root, window) + with exclusive_file_lock(folder / "generation.lock", operation="prepare_session_report"): + path = folder / "editorial-request.json" + if path.exists(): + frozen = _read(path) + if (frozen.get("managed_session_actor") != actor + or frozen.get("intent_digest") != _canonical_digest(cadence_intent(window)) + or frozen.get("request_digest") != _canonical_digest( + {k: v for k, v in frozen.items() if k != "request_digest"})): + raise ValueError("the frozen report belongs to another Session or window") + return frozen + source = read() + rows = source["rows"] + if len(rows) > 4000 or len(json.dumps(source, ensure_ascii=False).encode()) > 4 * 1024 * 1024: + raise ValueError("the complete local report source exceeds its existing evidence bound") + facts = [{"fact_id": "local_coverage", "title": "本期来源范围", + "summary": "当前原 Goal 的完整原生任务读取；任务状态在收集时观察，按完成时间区分本期成果和历史记录。来源声明及任务完成不证明外部业务已独立验收。", + "status": "read", "content_kind": "coverage", "source_ref": source["source_revision"]}] + for index, row in enumerate(rows): + encoded = json.dumps(row, ensure_ascii=False, sort_keys=True) + facts.append({"fact_id": f"local_{index}", "title": str(row.get("text") or row["todo_id"])[:240], + "summary": encoded if len(encoded) <= 1000 else encoded[:940] + "…（完整记录见冻结来源同一 todo_id）", + "status": "recorded", "content_kind": "coverage", "source_ref": "todo:" + row["todo_id"]}) + request = _build_editorial_request(intent=cadence_intent(window), runtime_root=root, + goal_id=window["goal_id"], agent_id=window["agent_id"], completed_at=window["due_at"], facts=facts) + request.pop("request_digest") + request.update(managed_session_actor=actor, reporting=reporting, window=window, + source_artifact=str(folder / "source.json"), source_revision=source["source_revision"], + source_artifact_sha256=_canonical_digest(source), destination="original_managed_session_full_body") + request["narrative_contract"]["requirements"] += [ + "Use only this original Goal's frozen evidence and permitted artifacts; no portfolio or other Goal access is granted.", + "Current Task status is observed at collection time, not reconstructed as of the period end.", + "Use recorded completion/evidence times to distinguish this fixed window's outcomes, older history and current open work; unknown times stay unknown.", + "Separate native Task completion, independent outcome evidence and actual message delivery. Preserve failed, blocked and unreadable evidence.", + "A due unchanged period still needs a report; exclude report generation itself from business progress.", + ] + request["request_digest"] = _canonical_digest(request) + folder.mkdir(parents=True, exist_ok=True) + atomic_write_json(folder / "source.json", source) + atomic_write_json(path, request) + return request + + def freeze_private_report(*, root: Path, window: dict[str, Any], response: dict[str, Any]) -> dict[str, Any]: from .adapters import build_periodic_report_document from .bindings import build_periodic_report_generation_bundle diff --git a/loopx/chat_goal_configuration_api.py b/loopx/chat_goal_configuration_api.py index d9a333c6bf..707491a369 100644 --- a/loopx/chat_goal_configuration_api.py +++ b/loopx/chat_goal_configuration_api.py @@ -669,44 +669,59 @@ def _goal_configuration_update(self, *, execute: bool) -> None: ) options = _goal_capability_options(capability_id, configuration) machine_namespaces = self._goal_configuration_machine_namespaces() - current_public, desired_public, plan, desired_configuration = ( - self._prepare_goal_configuration_plan( + from contextlib import nullcontext + from .capabilities.native_chat.steward_reports import session_report_configuration_guard + controller = getattr(self.server, "runtime_controller", None) + guard = (session_report_configuration_guard(controller, goal_id, configuration) + if capability_id == "periodic_report" and controller is not None else nullcontext()) + with guard: + current_public, desired_public, plan, desired_configuration = ( + self._prepare_goal_configuration_plan( + goal_id=goal_id, + capability_id=capability_id, + machine_namespaces=machine_namespaces, + options=options, + ) + ) + if not execute: + plan["capability_catalog"] = desired_public["capability_catalog"] + self._send_json(plan, status=201) + return + require_expected_configuration_plan_revision( + expected_plan_revision=expected_revision, + actual_plan_revision=str(plan["plan_revision"]), + subject="Goal configuration", + ) + applied = self._goal_configuration_writer()( + registry_path=self.server.registry_path, goal_id=goal_id, - capability_id=capability_id, - machine_namespaces=machine_namespaces, - options=options, + runtime_root_override=getattr( + self.server, "runtime_root_override", None + ), + execute=True, + expected_goal_configuration_revision=current_public["revision"], + codex_home_override=getattr( + getattr(self.server, "runtime_controller", None), + "codex_home", + None, + ), + codex_host_capacity_planner=plan_codex_subagent_capacity, + codex_host_capacity_applier=apply_codex_subagent_capacity, + **options, ) - ) - if not execute: - plan["capability_catalog"] = desired_public["capability_catalog"] - self._send_json(plan, status=201) - return - require_expected_configuration_plan_revision( - expected_plan_revision=expected_revision, - actual_plan_revision=str(plan["plan_revision"]), - subject="Goal configuration", - ) - applied = self._goal_configuration_writer()( - registry_path=self.server.registry_path, - goal_id=goal_id, - runtime_root_override=getattr( - self.server, "runtime_root_override", None - ), - execute=True, - expected_goal_configuration_revision=current_public["revision"], - codex_home_override=getattr( - getattr(self.server, "runtime_controller", None), - "codex_home", - None, - ), - codex_host_capacity_planner=plan_codex_subagent_capacity, - codex_host_capacity_applier=apply_codex_subagent_capacity, - **options, - ) - if not applied.get("ok"): - if not (applied.get("written") and applied.get("partial_write")): - raise RuntimeError("Goal configuration apply did not verify") - self._send_partial_write_receipt( + if not applied.get("ok"): + if not (applied.get("written") and applied.get("partial_write")): + raise RuntimeError("Goal configuration apply did not verify") + self._send_partial_write_receipt( + applied=applied, + capability_id=capability_id, + desired_configuration=desired_configuration, + desired_public=desired_public, + goal_id=goal_id, + plan=plan, + ) + return + self._send_applied_goal_configuration( applied=applied, capability_id=capability_id, desired_configuration=desired_configuration, @@ -714,15 +729,6 @@ def _goal_configuration_update(self, *, execute: bool) -> None: goal_id=goal_id, plan=plan, ) - return - self._send_applied_goal_configuration( - applied=applied, - capability_id=capability_id, - desired_configuration=desired_configuration, - desired_public=desired_public, - goal_id=goal_id, - plan=plan, - ) except (TypeError, ValueError) as exc: stale = "preview again" in str(exc) self._send_error( diff --git a/loopx/control_plane/collaboration/steward_managed_mode.ts b/loopx/control_plane/collaboration/steward_managed_mode.ts index b1ff4a6e47..6dc9539c42 100644 --- a/loopx/control_plane/collaboration/steward_managed_mode.ts +++ b/loopx/control_plane/collaboration/steward_managed_mode.ts @@ -15,7 +15,12 @@ import {goalAgentConversationIdentity, resolveConversationScope} from "./convers * a message or the selected executor/model. This creates no scheduling grant. */ export function admitManagedSessionIdentity(params: JsonObject): JsonObject { const goal = requireJsonObject(params.goal, "original registered Goal"); - const identity = goalAgentConversationIdentity({goal_id: goal.goal_id, managed_agent_id: params.managed_agent_id}); + let identity: JsonObject; + try { + identity = goalAgentConversationIdentity({goal_id: goal.goal_id, managed_agent_id: params.managed_agent_id}); + } catch { + throw new EffectRuntimeRequestError("select an exact registered Goal and Agent"); + } if (params.goal_active !== true || !Array.isArray(goal.registered_agents) || !goal.registered_agents.includes(identity.managed_agent_id) || goal.goal_instance_id != null && (typeof goal.goal_instance_id !== "string" @@ -23,6 +28,19 @@ export function admitManagedSessionIdentity(params: JsonObject): JsonObject { || params.workspace_path !== goal.workspace_path) { throw new EffectRuntimeRequestError("the managed Agent must belong to the active local Goal and workspace"); } + if (params.session != null) { + const session = requireJsonObject(params.session, "original managed Session"); + if (resolveConversationScope(session).kind !== "owner_goal" + || session.project_context != null || session.steward_context != null + || session.session_mode !== "managed_runtime" || session.agent_id !== "codex" || session.executor_endpoint_id !== "codex" + || session.adapter_kind !== "codex_app_server" || session.upstream_mode === "goal" + || session.status === "closed" || session.goal_id !== goal.goal_id + || (session.goal_instance_id ?? null) !== goal.goal_instance_id + || session.managed_agent_id !== identity.managed_agent_id + || session.managed_workspace_path !== goal.workspace_path || session.channel_id !== identity.channel_id) { + throw new EffectRuntimeRequestError("the original managed Agent Session identity is unavailable"); + } + } return {...identity, goal_instance_id: goal.goal_instance_id ?? null}; } @@ -36,18 +54,7 @@ export function admitManagedSessionTurn(params: JsonObject): JsonObject { const request = requireJsonObject(params.request, "native Turn request"); const actor = requireJsonObject(request.actor, "managed Session actor"); const turn = params.turn == null ? null : requireJsonObject(params.turn, "current Chat Turn"); - if (resolveConversationScope(session).kind !== "owner_goal" - || session.project_context != null || session.steward_context != null - || session.session_mode !== "managed_runtime" || session.agent_id !== "codex" || session.executor_endpoint_id !== "codex" - || session.adapter_kind !== "codex_app_server" || session.upstream_mode === "goal" - || session.status === "closed" || params.goal_active !== true - || goal.goal_id !== session.goal_id || goal.goal_instance_id !== (session.goal_instance_id ?? null) - || !Array.isArray(goal.registered_agents) || !goal.registered_agents.includes(actor.agent_id) - || session.managed_agent_id !== actor.agent_id - || session.managed_workspace_path !== goal.workspace_path - || typeof goal.workspace_path !== "string" || !goal.workspace_path) { - throw new EffectRuntimeRequestError("select an active local Goal's managed Codex conversation and a registered Agent"); - } + admitManagedSessionIdentity({...params, managed_agent_id: actor.agent_id, workspace_path: goal.workspace_path}); const fields = ["schema_version", "goal_id", "goal_instance_id", "agent_id", "session_id", "workspace_path"]; if (Object.keys(actor).some(key => !fields.includes(key)) || actor.schema_version !== "loopx_managed_session_actor_v0" @@ -59,10 +66,12 @@ export function admitManagedSessionTurn(params: JsonObject): JsonObject { throw new EffectRuntimeRequestError("the native Turn must retain its original managed Session and Goal lifetime"); } const observationOnly = request.work_kind === "monitor"; + const reportWork = request.work_kind === "report"; const requestFields = ["operation", "host", "actor", "turn_instance_id", "todo_id", "validation_command", "work_kind", - ...(observationOnly ? ["monitor_deadline", "monitor_lease_version"] : [])]; + ...(observationOnly ? ["monitor_deadline", "monitor_lease_version"] : []), + ...(reportWork ? ["report_window_id", "report_subscription_revision"] : [])]; if (Object.keys(request).some(key => !requestFields.includes(key)) - || request.operation !== "native_turn" || request.host !== "codex-app-server" || !["native_turn", "monitor"].includes(String(request.work_kind)) + || request.operation !== "native_turn" || request.host !== "codex-app-server" || !["native_turn", "monitor", "report"].includes(String(request.work_kind)) || request.validation_command !== null || typeof request.turn_instance_id !== "string" || !/^[A-Za-z0-9._:-]{1,160}$/.test(request.turn_instance_id) || request.todo_id !== null && (typeof request.todo_id !== "string" || !/^todo_[A-Za-z0-9_-]{1,100}$/.test(request.todo_id))) { @@ -79,6 +88,26 @@ export function admitManagedSessionTurn(params: JsonObject): JsonObject { throw new EffectRuntimeRequestError("a local watch must retain its original Agent, Session, Todo and deadline"); } } + if (reportWork) { + const window = requireJsonObject(params.report_window, "original report window"); + const subscription = requireJsonObject(params.report_subscription, "current report subscription"); + const todo = requireJsonObject(params.report_todo, "original report Todo"); + const recovery = params.report_settlement_recovery == null ? null + : requireJsonObject(params.report_settlement_recovery, "native original report recovery"); + const settlementOnly = reportSettlementOnly(params, request, todo); + if (subscription.enabled !== true || subscription.profile_preset !== "weekly-progress" + || subscription.route_ref !== `managed-session:${session.session_id}` + || subscription.effective_revision !== request.report_subscription_revision + || window.subscription_revision !== request.report_subscription_revision + || window.window_id !== request.report_window_id || window.goal_id !== actor.goal_id + || window.agent_id !== actor.agent_id || todo.todo_id !== request.todo_id + || todo.target_key !== window.window_id || todo.claimed_by !== actor.agent_id + || todo.action_kind !== "generate_periodic_report" || !(todo.status === "open" && !recovery || settlementOnly)) { + throw new EffectRuntimeRequestError("report execution requires the original Session subscription, window and task"); + } + } else if (params.report_settlement_recovery != null) { + throw new EffectRuntimeRequestError("report recovery belongs only to its original calendar report"); + } if (turn ? turn.session_id !== session.session_id || turn.turn_id !== session.active_turn_id || !["starting", "running"].includes(String(turn.status)) || turn.origin !== "native_loopx" || turn.loopx_execution !== true || JSON.stringify(turn.loopx_request) !== JSON.stringify(request) @@ -88,6 +117,28 @@ export function admitManagedSessionTurn(params: JsonObject): JsonObject { return {actor, workspace_path: goal.workspace_path, host: "codex-app-server", observation_only: observationOnly}; } +/** Both original conversation hosts use the native journal's no-model recovery. */ +function reportSettlementOnly(params: JsonObject, request: JsonObject, todo: JsonObject): boolean { + if (params.report_settlement_recovery == null) return false; + const recovery = requireJsonObject(params.report_settlement_recovery, "native original report recovery"); + const decision = requireJsonObject(recovery.recovery_decision, "native journal recovery decision"); + const cachedGeneration = todo.status === "open" && recovery.journal_status === "failed" + && decision.action === "continue" && decision.resume_from === "validation" + && Array.isArray(recovery.completed_phases) && recovery.completed_phases.length === 2 + && recovery.completed_phases[0] === "host_execute" && recovery.completed_phases[1] === "typed_result"; + const completedGeneration = todo.status === "done" && todo.completion_turn_key === request.turn_instance_id + && (decision.action === "continue" && ["durable_writeback", "quota_spend", "scheduler_apply", "scheduler_ack"].includes(String(decision.resume_from)) + || decision.action === "return_existing" && recovery.journal_status === "committed"); + const accepted = recovery.schema_version === "loopx_turn_journal_inspection_v1" && recovery.ok === true + && recovery.goal_matches === true && recovery.owner_matches === true && recovery.turn_key_matches === true + && recovery.journal_consistent === true && decision.reinvoke_host === false + && (cachedGeneration || completedGeneration); + if (!accepted || params.preparing_model_context === true) { + throw new EffectRuntimeRequestError("original report recovery requires validated remaining effects without a model"); + } + return true; +} + export const STEWARD_MANAGED_ACTOR_SCHEMA = "loopx_steward_managed_actor_v0"; const AGENT_ID = "steward"; @@ -202,24 +253,7 @@ export function admitStewardManagedTurn(params: JsonObject): JsonObject { const todo = requireJsonObject(params.report_todo, "original report Todo"); const recovery = params.report_settlement_recovery == null ? null : requireJsonObject(params.report_settlement_recovery, "native original report recovery"); - let settlementOnly = false; - if (recovery) { - const decision = requireJsonObject(recovery.recovery_decision, "native journal recovery decision"); - const cachedGeneration = todo.status === "open" && recovery.journal_status === "failed" - && decision.action === "continue" && decision.resume_from === "validation" - && Array.isArray(recovery.completed_phases) && recovery.completed_phases.length === 2 - && recovery.completed_phases[0] === "host_execute" && recovery.completed_phases[1] === "typed_result"; - const completedGeneration = todo.status === "done" && todo.completion_turn_key === request.turn_instance_id - && (decision.action === "continue" && ["durable_writeback", "quota_spend", "scheduler_apply", "scheduler_ack"].includes(String(decision.resume_from)) - || decision.action === "return_existing" && recovery.journal_status === "committed"); - settlementOnly = recovery.schema_version === "loopx_turn_journal_inspection_v1" && recovery.ok === true - && recovery.goal_matches === true && recovery.owner_matches === true && recovery.turn_key_matches === true - && recovery.journal_consistent === true && decision.reinvoke_host === false - && (cachedGeneration || completedGeneration); - if (!settlementOnly || params.preparing_model_context === true) { - throw new EffectRuntimeRequestError("original report recovery requires validated remaining effects without a model"); - } - } + const settlementOnly = reportSettlementOnly(params, request, todo); if (request.validation_command === null || subscription.enabled !== true || subscription.profile_preset !== "portfolio-progress" || subscription.route_ref !== `steward-private:${current.id}:${current.source}` || subscription.effective_revision !== request.report_subscription_revision diff --git a/tests/control_plane_ts/steward_reports.test.ts b/tests/control_plane_ts/steward_reports.test.ts index 7db310c716..31ffa96aab 100644 --- a/tests/control_plane_ts/steward_reports.test.ts +++ b/tests/control_plane_ts/steward_reports.test.ts @@ -59,6 +59,55 @@ test("local native work observation excludes watch bookkeeping and retains real assert.throws(() => observeManagedWork({...source, todos: undefined})); }); +test("a local calendar report retains its exact original Session, subscription, window and native Task", () => { + const goal = {goal_id: "project", goal_instance_id: "original", registered_agents: ["writer", "peer"], workspace_path: "/workspace"}; + const session = {goal_id: goal.goal_id, goal_instance_id: goal.goal_instance_id, session_id: "a".repeat(32), + channel_id: "agent.writer", session_mode: "managed_runtime", managed_agent_id: "writer", agent_id: "codex", + executor_endpoint_id: "codex", adapter_kind: "codex_app_server", managed_workspace_path: goal.workspace_path}; + const actor = {schema_version: "loopx_managed_session_actor_v0", goal_id: goal.goal_id, goal_instance_id: goal.goal_instance_id, + session_id: session.session_id, agent_id: "writer", workspace_path: goal.workspace_path}; + const request = {operation: "native_turn", host: "codex-app-server", work_kind: "report", actor, + turn_instance_id: "managed-session.report.original", todo_id: "todo_report", validation_command: null, + report_window_id: "original-window", report_subscription_revision: "original-subscription"}; + const subscription = {enabled: true, profile_preset: "weekly-progress", route_ref: `managed-session:${session.session_id}`, + effective_revision: request.report_subscription_revision}; + const window = {goal_id: goal.goal_id, agent_id: "writer", window_id: request.report_window_id, + subscription_revision: request.report_subscription_revision}; + const todo = {todo_id: request.todo_id, target_key: window.window_id, action_kind: "generate_periodic_report", claimed_by: "writer", status: "open"}; + const input = {session, goal, request, goal_active: true, queued_turn_count: 0, + report_window: window, report_subscription: subscription, report_todo: todo}; + assert.equal(admitManagedSessionTurn(input).observation_only, false); + for (const changed of [ + {...input, goal_active: false}, {...input, queued_turn_count: 1}, + {...input, session: {...session, managed_agent_id: "peer"}}, + {...input, session: {...session, session_mode: "attached_host"}}, + {...input, session: {...session, goal_instance_id: "replacement"}}, + {...input, report_window: {...window, goal_id: "another"}}, + {...input, report_window: {...window, agent_id: "peer"}}, + {...input, report_window: {...window, window_id: "next-window"}}, + {...input, report_subscription: {...subscription, enabled: false}}, + {...input, report_subscription: {...subscription, effective_revision: "changed"}}, + {...input, report_subscription: {...subscription, profile_preset: "portfolio-progress"}}, + {...input, report_subscription: {...subscription, route_ref: "managed-session:" + "b".repeat(32)}}, + {...input, report_todo: {...todo, target_key: "another-window"}}, + {...input, report_todo: {...todo, claimed_by: "peer"}}, + {...input, report_todo: {...todo, status: "done"}}, + {...input, request: {...request, validation_command: ["a-caller-command"]}}, + {...input, request: {...request, profile: "another-runtime"}}, + {...input, request: {...request, work_kind: "native_turn"}}, + ]) assert.throws(() => admitManagedSessionTurn(changed)); + const inspection = {schema_version: "loopx_turn_journal_inspection_v1", ok: true, + goal_matches: true, owner_matches: true, turn_key_matches: true, journal_consistent: true, + journal_status: "failed", completed_phases: ["host_execute", "typed_result"], + recovery_decision: {action: "continue", resume_from: "validation", reinvoke_host: false}}; + assert.equal(admitManagedSessionTurn({...input, report_settlement_recovery: inspection}).observation_only, false); + for (const changed of [{owner_matches: false}, {journal_consistent: false}, {completed_phases: ["host_execute"]}, + {recovery_decision: {...inspection.recovery_decision, reinvoke_host: true}}, + {recovery_decision: {...inspection.recovery_decision, resume_from: "host_execute"}}]) { + assert.throws(() => admitManagedSessionTurn({...input, report_settlement_recovery: {...inspection, ...changed}})); + } +}); + function original() { const binding = {binding_id: "b".repeat(24), transport_ref: "private-steward", provider_ref: "c".repeat(24), operator_ref: "d".repeat(24), project_ref: "a".repeat(24), context_kind: "steward", diff --git a/tests/test_native_steward_reports.py b/tests/test_native_steward_reports.py index a78b9ae485..0ba91fe825 100644 --- a/tests/test_native_steward_reports.py +++ b/tests/test_native_steward_reports.py @@ -22,6 +22,239 @@ steward = steward_fixture +def configured_local_report(steward, tmp_path, monkeypatch, provider="sqlite", start_date="2001-01-01"): + """An ordinary local Goal/Agent Session, with no report-enabled private binding.""" + from loopx.agent_registry import load_goal_from_registry + from loopx.configure_goal import configure_goal + from loopx.capabilities.native_chat.steward_reports import ManagedSessionReports + from test_native_steward_turn_host import cli + isolate_sqlite_runtime(tmp_path, monkeypatch) + store, runtime, _, _, binding, capture, _ = steward + sid, managed = original(steward) + storage(runtime, provider) + plan = managed.prepare(binding["binding_id"], session_id=sid)["plan"] + actor = managed.prepare(binding["binding_id"], session_id=sid, + expected_plan_id=plan["plan_id"], execute=True)["actor"] + workspace = Path(load_goal_from_registry(runtime.registry_path, actor["goal_id"])["repo"]) + session, _ = runtime.open_session(goal_id=actor["goal_id"], agent_id="codex", work_dir=workspace, + objective="Report this Goal's original evidence", channel_id=f"goal.{actor['goal_id']}", + mode="new", managed_agent_id=actor["agent_id"]) + task = cli(runtime, "todo", "add", "--goal-id", actor["goal_id"], "--role", "agent", + "--claimed-by", actor["agent_id"], "--text", "Keep this original work and its evidence visible.", + "--action-kind", "inspect", "--execute") + schedule = {"schema_version": "periodic_report_schedule_v0", "schedule_id": "local-report", + "timezone": "UTC", "rrule": "FREQ=DAILY;BYHOUR=9;BYMINUTE=0", "start_date": start_date} + config = {"enabled": True, "profile_preset": "weekly-progress", "route_ref": "managed-session:" + session["session_id"], + "timezone": "UTC", "schedule": schedule} + assert configure_goal(registry_path=runtime.registry_path, goal_id=actor["goal_id"], + periodic_report_configuration=config, execute=True)["ok"] + return session, ManagedSessionReports(runtime), config, task["todo_id"] + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_local_agent_calendar_report_uses_original_worker_and_verified_core_return(steward, tmp_path, monkeypatch, provider): + from loopx.capabilities.native_chat.steward_monitor import StewardMonitor + from loopx.capabilities.native_chat.steward_managed_mode import ManagedSessionTurn + session, reports, _, task_id = configured_local_report(steward, tmp_path, monkeypatch, provider) + store, runtime, bot, _, _, capture, _ = steward + sid = session["session_id"] + with monkeypatch.context() as changed_source: + changed_source.setattr(inspect, "getsource", lambda *_: "not the imported model callable") + install_model(steward, tmp_path, sid) + # A host's default codec must not decide how frozen Chinese artifacts read. + real_read_text = Path.read_text + def host_codec_read_text(path, *args, **kwargs): + if path.name in {"editorial-request.json", "publication-candidate.json", "source.json", "report.md"} and not args and "encoding" not in kwargs: + kwargs["encoding"] = "ascii" + return real_read_text(path, *args, **kwargs) + monkeypatch.setattr(Path, "read_text", host_codec_read_text) + # This local report neither consumes nor borrows the private provider. + runtime.project_contexts.conversation_bindings = None + before, calls, writes = store.messages(sid), capture.read_bytes(), list(bot.writes) + turns = pump_steward_monitors(runtime) + assert len(turns) == 1 + result = finish(runtime, turns[0]) + assert result["status"] == "completed", result + native = result["response"]["native_loopx"] + assert native["status"] == "committed", native + assert result["response"]["periodic_report"]["delivery_verified"] + window = read_cadence_journal(runtime_root=reports.root, goal_id=session["goal_id"])["window"] + folder = attempt_directory(reports.root, window) + frozen = json.loads((folder / "editorial-request.json").read_text(encoding="utf-8")) + source = json.loads((folder / "source.json").read_text(encoding="utf-8")) + assert frozen["destination"] == "original_managed_session_full_body" + assert source["goal_id"] == session["goal_id"] and source["complete_native_task_read"] + assert task_id in {row["todo_id"] for row in source["rows"]} + assert frozen["actual_work_window"]["end_at"] == window["due_at"] + assert datetime.fromisoformat(window["due_at"]) - datetime.fromisoformat(window["start_at"]) == timedelta(days=1) + body = (folder / "report.md").read_text(encoding="utf-8") + messages = store.messages(sid) + assert messages[:len(before)] == before + returned = [row for row in messages if row["role"] == "agent" and row["text"] == body] + assert len(returned) == 1 and returned[0]["turn_id"] == result["turn_id"] + cursor = read_periodic_report_publication_cursor(runtime_root=reports.root, goal_id=session["goal_id"], agent_id=session["managed_agent_id"]) + assert cursor["generation_id"] == validate_private_generation(folder)["generation_id"] + assert cursor["covered_until"] == window["due_at"] and reports.health(sid)["publication_verified"] + rows = StewardMonitor(runtime).todos(result["loopx_request"]["actor"]) + assert next(row for row in rows if row["todo_id"] == result["loopx_request"]["todo_id"])["status"] == "done" + assert next(row for row in rows if row["todo_id"] == task_id)["status"] == "open" + assert bot.writes == writes + after_calls = capture.read_bytes() + assert len([json.loads(line) for line in after_calls[len(calls):].splitlines() + if json.loads(line).get("method") == "turn/start"]) == 1 + assert not pump_steward_monitors(runtime) + assert reports.publish(sid, window) + assert store.messages(sid) == messages and capture.read_bytes() == after_calls + inspected = ManagedSessionTurn(runtime).inspect(sid) + assert inspected["report"]["publication_verified"] and inspected["available"] + assert inspected["report"]["report_message_id"] == returned[0]["message_id"] + # A publication cursor alone cannot offer a body that is no longer readable. + with monkeypatch.context() as unreadable_body: + unreadable_body.setattr(store, "messages", lambda *_: [row for row in messages + if row["message_id"] != returned[0]["message_id"]]) + unavailable = reports.health(sid) + assert not unavailable["publication_verified"] and unavailable["report_message_id"] is None + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_local_agent_report_before_first_deadline_is_effect_free(steward, tmp_path, monkeypatch, provider): + session, reports, _, _ = configured_local_report(steward, tmp_path, monkeypatch, provider, start_date="2999-01-01") + store, runtime, bot, *_ = steward + before = store.messages(session["session_id"]), list(bot.writes) + assert not pump_steward_monitors(runtime) + assert read_cadence_journal(runtime_root=reports.root, goal_id=session["goal_id"]) is None + assert (store.messages(session["session_id"]), list(bot.writes)) == before + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("lost_after_commit", [False, True]) +def test_local_report_restart_reads_back_unknown_publication_without_model_or_duplicate(steward, tmp_path, monkeypatch, provider, lost_after_commit): + from loopx.capabilities.periodic_report import incremental + from loopx.chat_store import ChatSessionStore + from loopx.chat_runtime import ChatRuntimeController + session, reports, _, _ = configured_local_report(steward, tmp_path, monkeypatch, provider) + store, runtime, bot, *_ = steward + sid = session["session_id"] + install_model(steward, tmp_path, sid) + runtime.project_contexts.conversation_bindings = None + commit = incremental.commit_periodic_report_publication_cursor + def interrupted_commit(**kwargs): + if lost_after_commit: + commit(**kwargs) + raise OSError("Synthetic publication reply lost after the original Core message") + with monkeypatch.context() as failure: + failure.setattr(incremental, "commit_periodic_report_publication_cursor", interrupted_commit) + first = finish(runtime, pump_steward_monitors(runtime)[0]) + assert first["status"] == "failed" + window = read_cadence_journal(runtime_root=reports.root, goal_id=session["goal_id"])["window"] + folder = attempt_directory(reports.root, window) + body = (folder / "report.md").read_text() + original_turn = store.load_turn(sid, first["turn_id"]) + assert len([row for row in store.messages(sid) if row["text"] == body]) == 1 + calls, writes = steward[5].read_bytes(), list(bot.writes) + runtime.close() + restarted = ChatRuntimeController(store=ChatSessionStore(store.root.parent), codex_bin=runtime.codex_bin, + registry_path=runtime.registry_path, project_contexts=runtime.project_contexts) + try: + assert not pump_steward_monitors(restarted) + assert store.load_turn(sid, first["turn_id"]) == original_turn + assert len([row for row in store.messages(sid) if row["text"] == body]) == 1 + cursor = read_periodic_report_publication_cursor(runtime_root=reports.root, + goal_id=session["goal_id"], agent_id=session["managed_agent_id"]) + assert cursor["generation_id"] == validate_private_generation(folder)["generation_id"] + assert steward[5].read_bytes() == calls and bot.writes == writes + finally: + restarted.close() + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_local_frozen_report_recovers_only_native_validation_and_returns_full_body(steward, tmp_path, monkeypatch, provider): + from loopx.capabilities.native_chat.steward_turn import StewardTurnHost + session, reports, _, _ = configured_local_report(steward, tmp_path, monkeypatch, provider) + store, runtime, bot, *_ = steward + sid = session["session_id"] + install_model(steward, tmp_path, sid) + runtime.project_contexts.conversation_bindings = None + with monkeypatch.context() as failure: + failure.setattr(StewardTurnHost, "_report_validator", lambda *_: lambda *_: { + "status": "failed", "validator_kind": "callback", "summary": "Synthetic validator unavailable", "recovery_kind": "repair_required"}) + original = finish(runtime, pump_steward_monitors(runtime)[0]) + assert original["response"]["native_loopx"]["result_kind"] == "validation_failed" + window = read_cadence_journal(runtime_root=reports.root, goal_id=session["goal_id"])["window"] + folder = attempt_directory(reports.root, window) + calls, messages, writes = steward[5].read_bytes(), store.messages(sid), list(bot.writes) + assert not any(row["text"] == (folder / "report.md").read_text() for row in messages) + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + assert store.load_turn(sid, original["turn_id"]) == original + assert store.messages(sid)[:len(messages)] == messages + assert len([row for row in store.messages(sid) if row["text"] == (folder / "report.md").read_text()]) == 1 + assert reports.health(sid)["publication_verified"] + assert steward[5].read_bytes() == calls and bot.writes == writes + + +@pytest.mark.parametrize("fault", ["paused", "unscheduled", "wrong_session", "wrong_profile", "unregistered", "busy"]) +def test_local_report_denied_sources_never_launch_an_alternative_worker(steward, tmp_path, monkeypatch, fault): + from loopx.configure_goal import configure_goal + session, reports, config, _ = configured_local_report(steward, tmp_path, monkeypatch) + store, runtime, bot, *_ = steward + before = store.messages(session["session_id"]), steward[5].read_bytes(), list(bot.writes) + if fault in {"paused", "unscheduled", "wrong_session", "wrong_profile"}: + changed = {**config, **({"enabled": False} if fault == "paused" else + {"schedule": None} if fault == "unscheduled" else + {"route_ref": "managed-session:" + "b" * 32} if fault == "wrong_session" else + {"profile_preset": "portfolio-progress", "route_ref": "another-original-route"})} + assert configure_goal(registry_path=runtime.registry_path, goal_id=session["goal_id"], + periodic_report_configuration=changed, execute=True)["ok"] + elif fault == "unregistered": + assert configure_goal(registry_path=runtime.registry_path, goal_id=session["goal_id"], + registered_agents=["another-agent"], execute=True)["ok"] + else: + store.accept_managed_turn(session["session_id"], client_turn_id="original-busy-user-work", + message="Keep this user's current work", origin="web") + before = store.messages(session["session_id"]), steward[5].read_bytes(), list(bot.writes) + assert not pump_steward_monitors(runtime) + assert (store.messages(session["session_id"]), steward[5].read_bytes(), list(bot.writes)) == before + assert read_cadence_journal(runtime_root=reports.root, goal_id=session["goal_id"]) is None + if fault == "unscheduled": + assert reports.health(session["session_id"])["enabled"] + assert reports.health(session["session_id"])["scheduled"] is False + + +def test_local_report_configuration_retains_frozen_window_and_original_agent_election(steward, tmp_path, monkeypatch): + from loopx.capabilities.native_chat.steward_reports import session_report_configuration_guard + from loopx.capabilities.periodic_report.cadence_runtime import periodic_report_cadence_hooks + from loopx.capabilities.periodic_report.pending_intent import consume_pending_periodic_report_intent + from loopx.capabilities.native_chat.steward_turn import StewardTurnHost + from loopx.configure_goal import configure_goal + session, reports, config, _ = configured_local_report(steward, tmp_path, monkeypatch) + store, runtime, *_ = steward + # Another Agent's ordinary CLI guard cannot elect itself for this route. + assert configure_goal(registry_path=runtime.registry_path, goal_id=session["goal_id"], + registered_agents=["first-registration", session["managed_agent_id"]], execute=True)["ok"] + assert not periodic_report_cadence_hooks(registry_path=runtime.registry_path, runtime_root=reports.root, + goal_id=session["goal_id"], agent_id="first-registration") + assert read_cadence_journal(runtime_root=reports.root, goal_id=session["goal_id"]) is None + install_model(steward, tmp_path, session["session_id"]) + # Retain a generated but uncommitted window under its real native journal. + with monkeypatch.context() as failure: + failure.setattr(StewardTurnHost, "_report_validator", lambda *_: lambda *_: { + "status": "failed", "validator_kind": "callback", "summary": "Synthetic validator unavailable", "recovery_kind": "repair_required"}) + original = finish(runtime, pump_steward_monitors(runtime)[0]) + assert original["response"]["native_loopx"]["result_kind"] == "validation_failed" + for changed in [None, {**config, "route_ref": "managed-session:" + "b" * 32}, + {**config, "schedule": {**config["schedule"], "rrule": "FREQ=DAILY;BYHOUR=10;BYMINUTE=0"}}]: + with pytest.raises((ValueError, KeyError)): + with session_report_configuration_guard(runtime, session["goal_id"], changed): + pytest.fail("a frozen pending report must not be stranded by a settings edit") + before = runtime.registry_path.read_bytes() + with session_report_configuration_guard(runtime, session["goal_id"], {**config, "enabled": False}): + pass + assert runtime.registry_path.read_bytes() == before + assert consume_pending_periodic_report_intent(registry_path=runtime.registry_path, runtime_root=reports.root, + goal_id=session["goal_id"], agent_id=session["managed_agent_id"], execute=True)["status"] == "original_conversation_required" + + def configured(steward, tmp_path, monkeypatch, provider="sqlite", start_date="2001-01-01"): isolate_sqlite_runtime(tmp_path, monkeypatch) store, runtime, _, transport, binding, capture, _ = steward @@ -75,10 +308,15 @@ def fake_report_model(params): "editorial_response": response}, ensure_ascii=False) +# Keep the imported oracle stable if a developer edits this file during a run. +# A late inspect.getsource can resolve its old line number into unrelated text. +_REPORT_MODEL_SOURCE = inspect.getsource(fake_report_model) + + def install_model(steward, tmp_path, sid, *, result_kind="validated_completion"): _, runtime, _, _, _, _, _ = steward helper = tmp_path / "report_model.py" - helper.write_text(inspect.getsource(fake_report_model).replace( + helper.write_text(_REPORT_MODEL_SOURCE.replace( '"result_kind": "validated_completion"', '"result_kind": ' + repr(result_kind))) fake = Path(runtime.codex_bin) source = fake.read_text().replace('"message": "Runtime response."',