diff --git a/apps/presentation/dashboard/src/data/chat.ts b/apps/presentation/dashboard/src/data/chat.ts index 6f37e449e8..bab7ca99ed 100644 --- a/apps/presentation/dashboard/src/data/chat.ts +++ b/apps/presentation/dashboard/src/data/chat.ts @@ -2622,6 +2622,11 @@ const privateConversationSchema = z.object({ pending_count: z.number().int(), unconfirmed_count: z.number().int(), failed_count: z.number().int()}), z.object({available: z.literal(false), error_code: z.string()}), ]).optional(), + managed_steward: z.discriminatedUnion("available", [ + z.object({available: z.literal(true), prepared: z.boolean(), scheduling_enabled: z.literal(false), + goal_id: z.string().optional(), agent_id: z.string().optional(), session_id: z.string().optional()}), + z.object({available: z.literal(false), error_code: z.string()}), + ]).optional(), agent_candidates: z.array(privateAgentSessionSchema).default([]), agent_targets: z.array(privateAgentTargetSchema).default([]), }); const privateConversationStopRecoverySchema = z.object({binding_id: z.string(), project_ref: z.string(), @@ -2647,6 +2652,23 @@ export async function disconnectPrivateConversation(bindingId: string, revision: })); } +const stewardPreparationSchema = z.object({ok: z.literal(true), applied: z.boolean(), plan: z.object({ + schema_version: z.literal("loopx_steward_managed_mode_plan_v0"), plan_id: z.string().regex(/^[a-f0-9]{64}$/), + session_id: z.string(), goal_id: z.string(), already_prepared: z.boolean(), + scheduling_enabled: z.literal(false), model_invoked: z.literal(false), +})}); +export async function prepareManagedSteward(bindingId: string) { + const path = "/api/chat/lark/private-conversations/managed-steward"; + const send = (body: unknown) => requestJson(path, {method: "POST", + headers: {"Content-Type": "application/json"}, body: JSON.stringify(body)}); + const preview = stewardPreparationSchema.parse(await send({binding_id: bindingId, execute: false})); + if (preview.plan.already_prepared) return preview; + const applied = stewardPreparationSchema.parse(await send({binding_id: bindingId, execute: true, + session_id: preview.plan.session_id, plan_id: preview.plan.plan_id})); + if (!applied.applied) throw new Error("The original steward identity has not been prepared."); + return applied; +} + export async function changePrivateAgentTarget(bindingId: string, revision: number, target: {session_id: string} | {target_ref: string}) { return privateConversationsSchema.parse(await requestJson("/api/chat/lark/private-conversations/agent-targets", { method: "POST", headers: {"Content-Type": "application/json"}, body: JSON.stringify({binding_id: bindingId, revision, ...target}), diff --git a/apps/presentation/dashboard/src/features/personal-workspace/private-conversation-panel.tsx b/apps/presentation/dashboard/src/features/personal-workspace/private-conversation-panel.tsx index 11a6bef9d8..61e0453d75 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/private-conversation-panel.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/private-conversation-panel.tsx @@ -1,9 +1,10 @@ import {useEffect, useRef, useState} from "react"; import {connectPrivateConversation, disconnectPrivateConversation, fetchPrivateConversations, - changePrivateAgentTarget, changeOwnerNotifications, fetchChatProjects, fetchChatCapabilities, fetchLarkGroupChats, type PrivateConversation, + changePrivateAgentTarget, changeOwnerNotifications, prepareManagedSteward, fetchChatProjects, fetchChatCapabilities, fetchLarkGroupChats, type PrivateConversation, discoverBoundChatSession, type BoundConversationTarget, type ChatProject, type LarkApp, type PrivateConversationStopRecovery} from "../../data/chat"; import {useWorkspaceI18n} from "./i18n"; +import {StewardManagedMode} from "./steward-managed-mode"; import "./private-conversation.css"; export function PrivateConversationPanel({apps, onOpenOriginal}: {apps: LarkApp[]; onOpenOriginal?: (target: BoundConversationTarget) => void}) { @@ -156,6 +157,9 @@ export function PrivateConversationPanel({apps, onOpenOriginal}: {apps: LarkApp[ {row.recovery_count > 0 ?

{zh ? "存在尚未确认的发送回执。服务会读取原回执恢复;不要重新发送同一任务。检查 App 登录、权限和原会话后刷新状态。" : "A send receipt is unconfirmed. The service reads the original receipt to recover; avoid resending the same task. Check this App login, permissions and original Session, then refresh status."}

: null} {!row.context_available ?

{zh ? "工作区授权已失效;请恢复原工作区或重新选择。旧会话不会移到其它工作区。" : "The workspace grant is unavailable. Restore the original workspace or select a new one; the old Session will not move."}

: null} {row.context_kind === "project" && row.audience !== "group" ? : null} + {row.context_kind === "steward" && row.audience === "owner" && row.executor_endpoint_id === "codex" + ? void act(() => prepareManagedSteward(row.binding_id))}/> : null} {row.context_kind === "steward" && row.goal_scope !== "all_registered" ? : null} )} diff --git a/apps/presentation/dashboard/src/features/personal-workspace/steward-managed-mode.tsx b/apps/presentation/dashboard/src/features/personal-workspace/steward-managed-mode.tsx new file mode 100644 index 0000000000..a1c504bf86 --- /dev/null +++ b/apps/presentation/dashboard/src/features/personal-workspace/steward-managed-mode.tsx @@ -0,0 +1,20 @@ +import type {PrivateConversation} from "../../data/chat"; + +/** Identity preparation is deliberately distinct from enabling scheduled work. */ +export function StewardManagedMode({state, zh, busy, available, onPrepare}: { + state: PrivateConversation["managed_steward"]; zh: boolean; busy: boolean; + available: boolean; onPrepare: () => void; +}) { + if (state === undefined) return null; + return
+ {zh ? "原生周期监控" : "Native periodic monitoring"} + {!state.available ?

{zh ? "管家工作身份暂不可用;请检查原会话和连接后重试。" : "The steward identity is unavailable. Check the original conversation and connection, then retry."}

+ : state.prepared ?

{zh ? "管家身份已准备。定时任务尚未启用。" : "The steward identity is prepared. Scheduled tasks are not enabled."}

+ : <> +

{zh ? "为当前管家准备一个可见的 LoopX 工作身份,保留原对话和权限。此步骤不会启动定时任务或模型。" : "Prepare one visible LoopX identity for this steward, retaining its original conversation and permissions. This step starts neither scheduled tasks nor a model."}

+ + } +
; +} diff --git a/docs/assets/personal-workspace/steward-native-identity-before.png b/docs/assets/personal-workspace/steward-native-identity-before.png new file mode 100644 index 0000000000..f46a4da95e Binary files /dev/null and b/docs/assets/personal-workspace/steward-native-identity-before.png differ diff --git a/docs/assets/personal-workspace/steward-native-identity-narrow.png b/docs/assets/personal-workspace/steward-native-identity-narrow.png new file mode 100644 index 0000000000..1ab23b516f Binary files /dev/null and b/docs/assets/personal-workspace/steward-native-identity-narrow.png differ diff --git a/docs/assets/personal-workspace/steward-native-identity-prepared.png b/docs/assets/personal-workspace/steward-native-identity-prepared.png new file mode 100644 index 0000000000..af0dde8e58 Binary files /dev/null and b/docs/assets/personal-workspace/steward-native-identity-prepared.png differ diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md new file mode 100644 index 0000000000..e06a0c6411 --- /dev/null +++ b/docs/product/periodic-reports.md @@ -0,0 +1,175 @@ +# Managed steward monitoring and periodic reports + +Status: native identity preparation is implemented. Scheduled execution and +daily portfolio reports remain the target described below; they are not enabled +by preparing an identity. + +The steward should become a persistent LoopX-managed Agent. Its original private +conversation remains the user's entry point. Native monitor Todos, Turn +admission, scheduling and periodic-report publication own recurring work. + +## What exists and what is missing + +The current components already provide most of the lifecycle: + +- `control_plane/scheduler/monitor_todo.py` and the typed scheduler project due + monitor work and the next deadline. +- `control_plane/turn_driver/` provides registered Goal/Agent admission, single + executor ownership, recovery, settlement and quota lineage. +- `capabilities/periodic_report/cadence.py`, `cadence_journal.py` and + `cadence_runtime.py` evaluate calendar windows, retain a pending reporter and + expose due report intents at Turn start. +- Native external-conversation admission and the existing Lark reply journal + retain original requests, audience identity and delivery evidence. + +The private steward currently uses the special `loopx-manager` conversation +identity. It does not have its own registered execution Goal/Agent. The existing +Chat LoopX-mode endpoint accepts a project Goal Chat, and rejects this steward. +Also, a report Turn-start hook does not independently wake a sleeping actor. +These are integration gaps; calendar calculation alone does not enable a report. + +Chat `/goal` transport supervision and a LoopX-managed Turn are distinct. Merely +relaxing `validate_goal_chat`, changing a Session field, adding a prompt, or +starting a Codex native Goal does not establish the LoopX lifecycle below. + +## One persistent steward actor + +Extend the existing source-bound upgrade/activation path. A typed preview must +identify the original private binding, its conversation, the steward workspace, +the persistent execution identity and the current host/profile. Apply is an +idempotent native transaction using the owner's existing authorization. + +Register one visible steward Goal/Agent, rather than creating a hidden Goal for +each monitor or borrowing a project worker's identity. The Goal owns the +steward's durable monitoring and reporting lifecycle. Its execution authority +does not include performing other projects' business work. + +The upgrade retains private chat identity, visible history, queued and accepted +requests, unfinished delegations, original delivery routes and existing grants. +In-flight work stays with its original accepted Turn; the upgrade waits for an +eligible idle boundary. If host tools require a replacement upstream thread, +preserve the visible conversation and history, and record that transport change. +Do not claim to preserve a physical thread that was actually replaced. + +Keep the steward's existing portfolio-read authority separate from its execution +actor. Resolve both through their native owners on every admission. Registering +the actor does not grant new project writes or broaden the original audience. +Keep ordinary project assistants unchanged. + +### Identity preparation available now + +In the private steward's connection settings, expand **Native periodic +monitoring** and choose **Prepare steward identity**. The same gesture obtains +a current preview and applies it. A stale preview, unverified provider, changed +grant, active Turn or queued user request refuses preparation. Resolve that +condition and retry in the original conversation. + +The existing local-owner endpoint is +`POST /api/chat/lark/private-conversations/managed-steward`. A preview accepts +`binding_id` and optional `session_id`. Apply additionally requires +`execute: true` and the preview's `plan_id`. Callers cannot select a Goal, Agent, +workspace, provider account or new grant. Existing loopback/origin checks apply. + +The typed conversation owner names one Goal and registered `steward` Agent from +the original binding. Existing native Goal creation and configuration owners +initialize File or SQLite storage according to machine settings. The existing +conversation binding stores only the resulting identity reference, including +the original Session and native creation operation. Source-session registries +retain their lifetime token; ordinary registries retain the existing nullable +lifetime contract. Preparation does not strengthen that legacy contract. + +A lost registration or binding response can resume the same verified creation; +it cannot adopt an unrelated Goal or recreate a missing bound Goal. Repeating +an applied request verifies the same registration and returns the same actor. +The original Session, history, upstream thread, accepted requests and portfolio +grant are retained. No model, native monitor, scheduler host or extra message +consumer starts. The UI explicitly shows **Scheduled tasks are not enabled**. + +This is a separately reviewable preparation boundary. There is no active host +to stop or migrate back. Disconnecting the conversation uses its existing owner +and leaves the registered, unscheduled Goal visible; it does not delete history +or accepted work. Removing that Goal remains an explicit native Goal lifecycle +operation. Activation and its pause/rollback controls are the next integration +stage, not an implied consequence of this preparation. + +## Native scheduling and execution + +The intended flow is: + +```text +original private ingress -> original steward conversation + | + steward Goal / Agent + | + native monitor/report deadline -> scheduler admission + | + one managed LoopX Turn + | + periodic-report generation/publication + | + original private reply + provider readback +``` + +Represent monitors as native `continuous_monitor` Todos under this actor. Reuse +their due times, observations and lifecycle; do not create another task store, +cron, model loop or message consumer. Expose the report calendar's next boundary +to that same scheduler frontier. The periodic-report calendar/journal remains +the owner of report windows; a monitor deadline is its scheduling projection, +not a second publication cursor. + +The existing runtime host checks the frontier without invoking a model. When +nothing is due, it sleeps until the next eligible native deadline. Only an +admitted Turn can invoke the model. A fixed daily report remains due even if +project state has not changed. Preserve the owner's minimum scheduling cadence; +align eligible calendar wakes to the requested wall time and disclose lateness +when the machine, host or actor cannot run then. + +User messages, due monitors, outstanding original replies and reports use the +same actor's queue and admission. A busy actor retains due work and rechecks it +at the normal terminal boundary. It does not create another executor. Existing +priority, replan, budget, stop and settlement policies remain effective. Pausing +the steward prevents new scheduled model Turns without discarding accepted work +or historical results. + +## Daily portfolio report + +Configure the original steward's daily report for 09:00 `Asia/Shanghai`. At +scheduled boundary T, freeze the reporting interval as `[T-24h,T)`. Keep scheduled +boundary, actual generation time and actual delivery time separate. + +Read all locally authorized registered Goals, including stopped Goals, through +their current native read surfaces. Do not reactivate them. Collect window +progress, relevant active/completed Todos, actual artifacts and validation, +unresolved requests/returns and concrete blockers. Distinguish reported claims +from independent verification, and report incomplete coverage explicitly. +Reconcile current approvals and dependencies before proposing a new owner ask. + +The steward produces the full Chinese body and returns it to the original +private chat. A summary, queued outbox row, attachment alone or assistant-side +completion flag is insufficient evidence that the user received the report. + +Use a stable identity for each scheduled window. Retain generation and delivery +progress separately. A restart resumes that window; confirmed delivery is not +sent again, and uncertain sends are read back before retry. After extended +offline time, produce the most recent complete window and disclose missing +windows, rather than starting an unbounded series of model calls. + +## Implementation order and acceptance + +1. Qualify the native steward upgrade and rollback at an idle boundary. Verify + one registered actor, preserved original history/requests/grants, and no + change to ordinary assistants or other project Agents. +2. Connect a real monitor deadline to that actor's native host and Turn driver. + Verify not-due checks invoke no model; due, busy, pause and restart cases + preserve one executor and native settlement. +3. Add the authorized portfolio report source to periodic-report, project its + next calendar deadline and configure the original private delivery route. + Reuse the existing report settings and monitor views for edit/pause/cancel, + next due time, last result, coverage and actual delivery status. +4. Run semantic File/SQLite and host tests for fixed windows, unchanged-but-due + reports, offline recovery, original audience, revoked grants, duplicate or + unknown sends, partial delivery and late callbacks after cancellation. +5. Qualify the exact final diff, publish and merge under the existing repository + policy, install it, then independently verify original-service adoption, + actual configuration and an end-to-end report. The first natural scheduled + report remains a separate acceptance event. diff --git a/loopx/capabilities/native_chat/conversation_bindings.py b/loopx/capabilities/native_chat/conversation_bindings.py index 537e79c284..e2d20562e0 100644 --- a/loopx/capabilities/native_chat/conversation_bindings.py +++ b/loopx/capabilities/native_chat/conversation_bindings.py @@ -104,6 +104,8 @@ def configure(self, *, transport_ref: str, project_ref: str, if "goal_scope" in previous: candidate["goal_scope"] = previous["goal_scope"] candidate["binding_id"] = previous["binding_id"] + if previous.get("managed_actor") is not None: + candidate["managed_actor"] = previous["managed_actor"] result = self._core("collaboration.conversation.binding", { "current": current, "expected_revision": current.get("revision"), "operation": "configure", "binding": candidate, "observation": observation, "available_projects": self.projects.available(), diff --git a/loopx/capabilities/native_chat/steward_managed_mode.py b/loopx/capabilities/native_chat/steward_managed_mode.py new file mode 100644 index 0000000000..2276fe3e40 --- /dev/null +++ b/loopx/capabilities/native_chat/steward_managed_mode.py @@ -0,0 +1,157 @@ +"""Native Goal/Agent preparation for the original private steward. + +The typed conversation owner fixes identity and consent. Existing Goal creation +and configuration perform durable effects. No model, monitor, scheduler or +second Chat Session is started here; preparation is distinct from activation. +""" +from __future__ import annotations + +import hashlib +from pathlib import Path +from typing import Any + +from ...agent_registry import load_goal_from_registry, registered_agent_ids_for_goal +from ...chat_store import _atomic_write_json +from ...file_lock import exclusive_file_lock + + +class StewardManagedMode: + def __init__(self, controller: Any, bindings: Any) -> None: + self.controller, self.bindings = controller, bindings + self.store = controller.store + + def _session(self, binding_id: str, session_id: str | None) -> dict[str, Any]: + if session_id is not None: + session = self.store.load_session(session_id) + else: + binding = next((row for row in self.bindings.read()["bindings"] + if row["binding_id"] == binding_id), {}) + actor = binding.get("managed_actor") or {} + if actor.get("session_id"): + session = self.store.load_session(actor["session_id"]) + else: + candidates = self.bindings.local_sessions(binding_id) + candidates = [row for row in candidates if row.get("status") != "closed"] + channels = {row.get("channel_id") for row in candidates} + if len(channels) != 1 or not candidates: + raise ValueError("select the original private steward conversation first") + latest = max(candidates, key=lambda row: str(row.get("updated_at") or "")) + session = self.store.load_session(latest["session_id"]) + if not session or (session.get("steward_context") or {}).get("binding_id") != binding_id: + raise ValueError("the original steward Session is unavailable") + return session + + def _workspace(self, session: dict[str, Any]) -> Path: + # Inspect the already established manager workspace. Calling its + # construction helper during preview would write AGENTS/skill files. + parent = self.store.root / "manager-workspaces" + path = parent / hashlib.sha256(session["channel_id"].encode()).hexdigest()[:24] + if (parent.is_symlink() or path.is_symlink() or not path.is_dir() + or not path.resolve().is_relative_to(self.store.root.resolve())): + raise ValueError("the original private steward workspace is unavailable") + return path.resolve() + + def _goal(self, goal_id: str) -> dict[str, Any] | None: + goal = load_goal_from_registry(self.controller.registry_path, goal_id) + if goal is None: + return None + return {"goal_id": goal["id"], "goal_instance_id": goal.get("goal_instance_id"), + "workspace_path": str(Path(goal["repo"]).resolve()), + "creation_operation_id": goal.get("creation_operation_id"), + "registered_agents": registered_agent_ids_for_goal(goal)} + + def prepare(self, binding_id: str, *, session_id: str | None = None, + expected_plan_id: str | None = None, execute: bool = False) -> dict[str, Any]: + session = self._session(binding_id, session_id) + saved = session["steward_context"] + # The local owner is a real web origin, not a fabricated provider event. + observed = self.bindings.resolve(binding_id=binding_id, session=session, + include_observation=True, origin="web", source_ref=saved["source_ref"], + sender_ref=saved["operator_ref"], private_human_message=False, session_context=saved) + with self.controller.lock, exclusive_file_lock(self.bindings.path, operation="prepare_native_steward"): + current = self.bindings.read() + binding = next((row for row in current["bindings"] if row["binding_id"] == binding_id), None) + if binding != observed["binding"]: + raise ValueError("the original conversation grant changed before preparation") + session = self._session(binding_id, session["session_id"]) + workspace = self._workspace(session) + goal_id = f"steward-agent-{binding_id}" + params = {"binding": binding, "binding_revision": current["revision"], "session": session, + "observation": observed["observation"], "workspace_path": str(workspace), + "queued_turn_count": len(self.store.queued_turns(session["session_id"])), + "runtime_profile": {key: session.get(key) for key in + ("manager_runtime_profile", "manager_runtime_sandbox", "manager_runtime_standing_grant")}, + "goal": self._goal(goal_id)} + plan = self.bindings._core("collaboration.steward.managed_mode.plan", params) + if not execute: + return {"ok": True, "applied": False, "plan": plan} + # Lost-response recovery reads the exact native lifetime/reference; + # it never creates another Goal or reuses an unrelated existing one. + if plan["already_prepared"]: + return {"ok": True, "applied": True, "idempotent": True, "plan": plan, + "actor": binding["managed_actor"], "scheduling_enabled": False} + if expected_plan_id != plan["plan_id"]: + raise ValueError("review the current steward preparation preview before applying") + from ...bootstrap import bootstrap_project + from ...configure_goal import configure_goal + from ...control_plane.coordination.new_goal_storage_transport import initialize_goal_storage_target + + if params["goal"] is None: + created = bootstrap_project(project=workspace, registry_path=self.controller.registry_path, + runtime_root=self.controller.coordination_runtime_root, goal_id=goal_id, + objective="Maintain the original private steward's authorized monitors and periodic reports; preserve original requests, scope and delivery evidence.", + domain="personal-steward", role="primary", parent_goal_id=None, state_file=None, + goal_doc=None, adapter_kind="generic_project_goal_v0", adapter_status="connected", + next_probe=None, spawn_allowed=False, max_children=0, allowed_domains=[], write_scope=[], + preserve_todos=True, display_name="Steward monitoring", force=False, dry_run=False, + sync_global=False, creation_operation_id=plan["creation_operation_id"]) + if not created.get("ok"): + raise ValueError("native steward Goal creation did not complete") + else: + # Native creation can publish registration before initializing + # storage. Its existing owner recovers that same creation; + # bootstrap itself correctly rejects every registered id. + goal = load_goal_from_registry(self.controller.registry_path, goal_id) + initialize_goal_storage_target(self.controller.coordination_runtime_root, goal, + registry_path=self.controller.registry_path) + configure_goal(registry_path=self.controller.registry_path, goal_id=goal_id, + registered_agents=[plan["agent_id"]], execute=True) + # A queued user request wins this idle boundary. Preserve the + # native creation receipt for recovery; do not publish a new host. + session = self._session(binding_id, session["session_id"]) + if self._workspace(session) != workspace: + raise ValueError("the original private steward workspace changed") + result = self.bindings._core("collaboration.conversation.binding", { + **params, "current": current, "expected_revision": current["revision"], + "operation": "bind_managed_actor", "binding_id": binding_id, + "observation": self.bindings.observe(binding["transport_ref"]), + "session": session, "queued_turn_count": len(self.store.queued_turns(session["session_id"])), + "goal": self._goal(goal_id)}) + if result["changed"]: + _atomic_write_json(self.bindings.path, result["state"]) + readback = self.bindings.read() + if readback != result["state"]: + raise OSError("managed steward identity publication did not verify") + actor = next(row["managed_actor"] for row in readback["bindings"] if row["binding_id"] == binding_id) + return {"ok": True, "applied": True, "idempotent": False, "plan": plan, "actor": actor, + "scheduling_enabled": False, "model_invoked": False} + + def health(self, binding_id: str) -> dict[str, Any]: + """Metadata projection only; it neither verifies grants nor enables work.""" + try: + binding = next(row for row in self.bindings.read()["bindings"] if row["binding_id"] == binding_id) + actor = binding.get("managed_actor") + if actor is None: + return {"available": True, "prepared": False, "scheduling_enabled": False} + goal = self._goal(actor["goal_id"]) + session = self._session(binding_id, actor["session_id"]) + if (not goal or goal["goal_instance_id"] != actor["goal_instance_id"] + or goal["creation_operation_id"] != actor["creation_operation_id"] + or actor["agent_id"] not in goal["registered_agents"] + or session.get("status") == "closed" + or goal["workspace_path"] != str(self._workspace(session))): + raise ValueError("native steward identity unavailable") + return {"available": True, "prepared": True, "scheduling_enabled": False, + "goal_id": actor["goal_id"], "agent_id": actor["agent_id"], "session_id": actor["session_id"]} + except (KeyError, StopIteration, OSError, ValueError): + return {"available": False, "error_code": "managed_steward_identity_unavailable"} diff --git a/loopx/chat_server.py b/loopx/chat_server.py index 9a1dfb99ab..7951928e56 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -1674,6 +1674,7 @@ def do_POST(self) -> None: PRIVATE_CONVERSATIONS_PATH: self._private_conversation_connect, PRIVATE_CONVERSATIONS_PATH + "/agent-targets": self._private_conversation_agent_target, PRIVATE_CONVERSATIONS_PATH + "/owner-notifications": self._private_conversation_owner_notifications, + PRIVATE_CONVERSATIONS_PATH + "/managed-steward": self._private_conversation_managed_steward, **self._configuration_post_routes(), **self._ssh_source_post_routes(), } diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index ecc4d22390..060bf73a13 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/loopx/control_plane/collaboration/conversation_binding.ts @@ -3,6 +3,7 @@ import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts"; import {requireInteger, requireJsonObject, requireNonEmptyString} from "../runtime_decode.ts"; import {requireAutomationCadenceEligibility} from "../quota/automation_cadence.ts"; import {normalizeProjectContext, normalizeStewardContext, normalizeStewardGoalScope} from "./conversation_scope.ts"; +import {bindStewardManagedActor, normalizeStewardManagedActor} from "./steward_managed_mode.ts"; /** Core owns the context and audience grant. A provider supplies verified, * opaque identity observations; neither a message nor a model selects them. @@ -154,6 +155,9 @@ function binding(value: unknown): JsonObject { throw new EffectRuntimeRequestError("ambiguous Agent target grant"); } } + if (row.managed_actor !== undefined && (row.context_kind !== "steward" || group || row.executor_endpoint_id !== "codex")) { + throw new EffectRuntimeRequestError("managed identity requires a private native steward"); + } return { schema_version: BINDING_SCHEMA, binding_id: ref(row.binding_id, "binding identity"), transport_ref: token(row.transport_ref, "transport reference"), @@ -166,6 +170,7 @@ function binding(value: unknown): JsonObject { ...(row.context_kind === "steward" ? {goal_ids: normalizeStewardGoalScope(row.goal_ids), ...(row.goal_scope !== undefined ? {goal_scope: row.goal_scope} : {})} : {}), ...(agents.length ? {agent_targets: agents} : {}), + ...(row.managed_actor !== undefined ? {managed_actor: normalizeStewardManagedActor(row.managed_actor)} : {}), }; } @@ -270,6 +275,14 @@ export function planConversationBinding(params: JsonObject): JsonObject { const goals = normalizeStewardGoalScope([...(previous.goal_ids as string[]), goal.goal_id]); if (JSON.stringify(goals) === JSON.stringify(previous.goal_ids)) return {changed: false, state: current}; rows = current.bindings.map(row => row.binding_id === id ? {...row, goal_ids: goals} : row); + } else if (params.operation === "bind_managed_actor") { + const id = ref(params.binding_id, "binding identity"); + const previous = current.bindings.find(row => row.binding_id === id); + if (!previous) throw new EffectRuntimeRequestError("the original steward binding is unavailable"); + const actor = bindStewardManagedActor({...params, binding: previous}); + if (JSON.stringify(actor) === JSON.stringify(previous.managed_actor)) return {changed: false, state: current}; + if (previous.managed_actor !== undefined) throw new EffectRuntimeRequestError("retain the original managed steward identity"); + rows = current.bindings.map(row => row.binding_id === id ? {...row, managed_actor: actor} : row); } else if (params.operation === "disconnect") { const id = ref(params.binding_id, "binding identity"); rows = current.bindings.filter(row => row.binding_id !== id); diff --git a/loopx/control_plane/collaboration/steward_managed_mode.ts b/loopx/control_plane/collaboration/steward_managed_mode.ts new file mode 100644 index 0000000000..ea8d5fb53d --- /dev/null +++ b/loopx/control_plane/collaboration/steward_managed_mode.ts @@ -0,0 +1,106 @@ +/** Source-bound preparation of one native execution identity for a steward. + * + * This owner prepares/binds identity only. Goal bootstrap, Agent registration, + * scheduling and model execution retain their existing owners. In particular, + * a prepared identity never advertises a running monitor or an enabled host. + */ +import { createHash } from "node:crypto"; +import type { JsonObject } from "../runtime_decode.ts"; +import { requireJsonObject, requireNonEmptyString } from "../runtime_decode.ts"; +import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; + +export const STEWARD_MANAGED_ACTOR_SCHEMA = "loopx_steward_managed_actor_v0"; +const AGENT_ID = "steward"; + +function text(value: unknown, name: string, expression?: RegExp): string { + const result = requireNonEmptyString(value, name); + if (result.length > 200 || (expression && !expression.test(result))) { + throw new EffectRuntimeRequestError(`invalid ${name}`); + } + return result; +} + +export function normalizeStewardManagedActor(value: unknown): JsonObject { + const row = requireJsonObject(value, "managed steward identity"); + const fields = ["schema_version", "goal_id", "goal_instance_id", "agent_id", "session_id", "source_ref", "creation_operation_id"]; + if (Object.keys(row).some(key => !fields.includes(key)) || row.schema_version !== STEWARD_MANAGED_ACTOR_SCHEMA + || row.agent_id !== AGENT_ID) throw new EffectRuntimeRequestError("invalid managed steward identity"); + return {schema_version: STEWARD_MANAGED_ACTOR_SCHEMA, + goal_id: text(row.goal_id, "managed Goal", /^[a-z0-9][a-z0-9._-]{0,119}$/), + // Ordinary registries have no source-session lifetime token. Preserve + // null, as the existing native Goal/Agent binding contract does; never + // fabricate a token. Source-session registries retain their exact token. + goal_instance_id: row.goal_instance_id === null ? null : text(row.goal_instance_id, "managed Goal lifetime"), agent_id: AGENT_ID, + session_id: text(row.session_id, "original steward Session", /^[a-f0-9]{32}$/), + source_ref: text(row.source_ref, "original steward source", /^[a-f0-9]{24}$/), + creation_operation_id: text(row.creation_operation_id, "managed Goal creation")}; +} + +function basis(params: JsonObject) { + const binding = requireJsonObject(params.binding, "steward binding"); + const session = requireJsonObject(params.session, "original steward Session"); + const context = requireJsonObject(session.steward_context, "original steward authority"); + const observation = requireJsonObject(params.observation, "current provider observation"); + const id = text(binding.binding_id, "binding identity", /^[a-f0-9]{24}$/); + const source = text(context.source_ref, "steward source", /^[a-f0-9]{24}$/); + if (binding.context_kind !== "steward" || binding.grant !== "portfolio_read" || binding.enabled !== true + || binding.audience === "group" || binding.executor_endpoint_id !== "codex" + || session.goal_id !== "loopx-manager" || session.agent_id !== "codex" + || session.session_mode === "attached_host" || session.status === "closed" + || session.channel_id !== `manager.external.native.${id}.${source}` + || session.active_turn_id || params.queued_turn_count !== 0) { + throw new EffectRuntimeRequestError("prepare the original idle private steward, retaining its current work"); + } + for (const key of ["binding_id", "provider_ref", "operator_ref", "project_ref"]) { + if (context[key] !== binding[key]) throw new EffectRuntimeRequestError("steward authority changed"); + } + if (observation.verified !== true || observation.transport_ref !== binding.transport_ref + || observation.provider_ref !== binding.provider_ref || observation.operator_ref !== binding.operator_ref) { + throw new EffectRuntimeRequestError("the original provider and owner must be verified"); + } + const workspace = requireNonEmptyString(params.workspace_path, "private steward workspace"); + const sessionId = text(session.session_id, "original steward Session", /^[a-f0-9]{32}$/); + const goalId = `steward-agent-${id}`; + const creation = `steward-managed:${id}:${source}`; + const goal = params.goal == null ? null : requireJsonObject(params.goal, "native steward Goal"); + if (goal && (goal.goal_id !== goalId || goal.workspace_path !== workspace || goal.creation_operation_id !== creation)) { + throw new EffectRuntimeRequestError("the target Goal belongs to another creation or workspace"); + } + const actor = binding.managed_actor == null ? null : normalizeStewardManagedActor(binding.managed_actor); + if (actor && (!goal || actor.goal_id !== goalId || actor.session_id !== sessionId || actor.source_ref !== source + || actor.creation_operation_id !== creation || actor.goal_instance_id !== goal.goal_instance_id + || !Array.isArray(goal.registered_agents) || !goal.registered_agents.includes(AGENT_ID))) { + throw new EffectRuntimeRequestError("the prepared steward identity or its original Goal lifetime is unavailable"); + } + return {binding, session, id, source, workspace, sessionId, goalId, creation, goal, actor}; +} + +export function planStewardManagedMode(params: JsonObject): JsonObject { + const current = basis(params); + if (!Number.isSafeInteger(params.binding_revision) || Number(params.binding_revision) < 0) { + throw new EffectRuntimeRequestError("current conversation revision is required"); + } + // A creation by this exact operation can be recovered after bootstrap or + // registration loses its response. Mutable Goal state is checked above, + // not included in the immutable pre-creation consent basis. + const fingerprint = [current.id, current.binding.provider_ref, current.binding.operator_ref, + current.binding.project_ref, current.source, current.sessionId, current.session.upstream_thread_id ?? null, + current.workspace, params.binding_revision, params.runtime_profile ?? null]; + const planId = createHash("sha256").update(JSON.stringify(fingerprint)).digest("hex"); + return {schema_version: "loopx_steward_managed_mode_plan_v0", plan_id: planId, + binding_id: current.id, session_id: current.sessionId, goal_id: current.goalId, + agent_id: AGENT_ID, creation_operation_id: current.creation, workspace_path: current.workspace, + already_prepared: current.actor !== null, model_invoked: false, scheduling_enabled: false, + execution_authority_granted: false}; +} + +export function bindStewardManagedActor(params: JsonObject): JsonObject { + const current = basis(params); + const goal = current.goal; + if (!goal || !Array.isArray(goal.registered_agents) || !goal.registered_agents.includes(AGENT_ID)) { + throw new EffectRuntimeRequestError("native Goal and Agent registration must be read back before binding"); + } + return normalizeStewardManagedActor({schema_version: STEWARD_MANAGED_ACTOR_SCHEMA, + goal_id: current.goalId, goal_instance_id: goal.goal_instance_id, agent_id: AGENT_ID, + session_id: current.sessionId, source_ref: current.source, creation_operation_id: current.creation}); +} diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 250795a133..b896ef84fb 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -626,6 +626,7 @@ export function createEffectRuntimeHandlers( ["collaboration.steward.session_identity", lazyHandler(() => import("./collaboration/conversation_scope.ts"), ({stewardConversationIdentity}) => stewardConversationIdentity)], ["collaboration.steward.command", lazyHandler(() => import("./collaboration/conversation_binding.ts"), ({stewardCommand}) => stewardCommand)], ["collaboration.steward.authorize_creation", lazyHandler(() => import("./collaboration/conversation_binding.ts"), ({authorizeStewardCreation}) => authorizeStewardCreation)], + ["collaboration.steward.managed_mode.plan", lazyHandler(() => import("./collaboration/steward_managed_mode.ts"), ({planStewardManagedMode}) => planStewardManagedMode)], ["collaboration.conversation.binding", lazyHandler(() => import("./collaboration/conversation_binding.ts"), ({planConversationBinding}) => planConversationBinding)], ["collaboration.conversation.bound_context", lazyHandler(() => import("./collaboration/conversation_binding.ts"), ({resolveBoundConversation}) => resolveBoundConversation)], ["collaboration.conversation.local_sessions", lazyHandler(() => import("./collaboration/conversation_binding.ts"), ({listLocalBoundSessions}) => listLocalBoundSessions)], diff --git a/loopx/extensions/lark/private_conversation_api.py b/loopx/extensions/lark/private_conversation_api.py index f5b7b9829c..74954e8d28 100644 --- a/loopx/extensions/lark/private_conversation_api.py +++ b/loopx/extensions/lark/private_conversation_api.py @@ -57,6 +57,10 @@ def _private_conversations(self) -> None: and hasattr(self.server.lark_private_conversations, "ingress") else {}), **({"owner_attention": self.server.lark_private_conversations.owner_attention.health(row["binding_id"])} if row["context_kind"] == "steward" and row.get("audience", "owner") == "owner" else {}), + **({"managed_steward": self.server.lark_private_conversations.managed_steward.health(row["binding_id"])} + if row["context_kind"] == "steward" and row.get("audience", "owner") == "owner" + and row["executor_endpoint_id"] == "codex" + and hasattr(self.server.lark_private_conversations, "managed_steward") else {}), **deliveries.get(row["binding_id"], {"pending_count": 0, "recovery_count": 0})} for row in current["bindings"]], "stop_recoveries": [{**row, @@ -137,6 +141,23 @@ def _private_conversation_disconnect(self) -> None: return self._private_conversations() + def _private_conversation_managed_steward(self) -> None: + try: + body = self._read_json() + if (set(body) - {"binding_id", "session_id", "plan_id", "execute"} + or not isinstance(body.get("binding_id"), str) + or not isinstance(body.get("execute", False), bool) + or ("session_id" in body and not isinstance(body["session_id"], str)) + or ("plan_id" in body and not isinstance(body["plan_id"], str))): + raise ValueError("use the original steward's current preparation preview") + result = self.server.lark_private_conversations.managed_steward.prepare( + body["binding_id"], session_id=body.get("session_id"), + expected_plan_id=body.get("plan_id"), execute=body.get("execute", False)) + except (ValueError, KeyError, OSError) as exc: + self._send_error(str(exc), status=409, error_code="managed_steward_preparation_unavailable") + return + self._send_json(result) + def _private_conversation_agent_target(self) -> None: try: body = self._read_json() diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index 4fb163359e..90b1eea297 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -57,6 +57,8 @@ def __init__(self, *, controller: Any, runtime_root: Path, runner: Any, cli_bin: self.owner_attention = LarkOwnerAttention(self) from .private_ingress import PrivateIngressObserver self.ingress = PrivateIngressObserver(self) + from ...capabilities.native_chat.steward_managed_mode import StewardManagedMode + self.managed_steward = StewardManagedMode(controller, self.bindings) def profiles(self) -> dict[str, dict[str, str]]: """Discover current App configuration, never authorize a conversation. diff --git a/tests/control_plane_ts/steward_managed_mode.test.ts b/tests/control_plane_ts/steward_managed_mode.test.ts new file mode 100644 index 0000000000..233d2d418c --- /dev/null +++ b/tests/control_plane_ts/steward_managed_mode.test.ts @@ -0,0 +1,100 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import {bindStewardManagedActor, planStewardManagedMode} from "../../loopx/control_plane/collaboration/steward_managed_mode.ts"; +import {planConversationBinding} from "../../loopx/control_plane/collaboration/conversation_binding.ts"; + +function fixture() { + const binding = {schema_version: "loopx_chat_conversation_binding_v0", binding_id: "b".repeat(24), + transport_ref: "steward-app", provider_ref: "c".repeat(24), operator_ref: "d".repeat(24), + project_ref: "a".repeat(24), context_kind: "steward", grant: "portfolio_read", + executor_endpoint_id: "codex", enabled: true, goal_scope: "all_registered", goal_ids: []}; + const source = "e".repeat(24); + const session = {session_id: "f".repeat(32), agent_id: "codex", goal_id: "loopx-manager", status: "ready", + upstream_thread_id: "original-thread", channel_id: `manager.external.native.${binding.binding_id}.${source}`, + steward_context: {binding_id: binding.binding_id, provider_ref: binding.provider_ref, + operator_ref: binding.operator_ref, project_ref: binding.project_ref, source_ref: source}}; + const params = {binding, session, binding_revision: 4, queued_turn_count: 0, + observation: {transport_ref: binding.transport_ref, provider_ref: binding.provider_ref, + operator_ref: binding.operator_ref, verified: true}, workspace_path: "/runtime/chat/manager-workspaces/original"}; + const goal = {goal_id: `steward-agent-${binding.binding_id}`, goal_instance_id: "original-goal-lifetime", + workspace_path: params.workspace_path, registered_agents: ["steward"], + creation_operation_id: `steward-managed:${binding.binding_id}:${source}`}; + return {params, goal}; +} + +test("preparation names one persistent steward, without model or scheduling authority", () => { + const {params, goal} = fixture(); + const before = structuredClone(params); + const first = planStewardManagedMode(params); + assert.deepEqual(params, before); + assert.equal(first.goal_id, goal.goal_id); + assert.equal(first.agent_id, "steward"); + assert.equal(first.session_id, params.session.session_id); + assert.equal(first.model_invoked, false); + assert.equal(first.scheduling_enabled, false); + assert.equal(first.execution_authority_granted, false); + // An exact creation's response can be lost before binding; recover the + // same consent basis instead of requiring a second Goal or model. + assert.equal(planStewardManagedMode({...params, goal}).plan_id, first.plan_id); + assert.notEqual(planStewardManagedMode({...params, binding_revision: 5}).plan_id, first.plan_id); +}); + +test("ordinary, group, attached and foreign-source conversations cannot acquire a steward actor", () => { + const {params} = fixture(); + for (const input of [ + {...params, binding: {...params.binding, context_kind: "project", grant: "workspace_write"}}, + {...params, binding: {...params.binding, audience: "group"}}, + {...params, session: {...params.session, session_mode: "attached_host"}}, + {...params, session: {...params.session, goal_id: "another-goal"}}, + {...params, session: {...params.session, channel_id: "manager.external.other"}}, + {...params, session: {...params.session, steward_context: {...params.session.steward_context, operator_ref: "0".repeat(24)}}}, + ]) assert.throws(() => planStewardManagedMode(input)); +}); + +test("busy work, an ordered user queue and unverified provider identity stop preparation", () => { + const {params} = fixture(); + for (const input of [ + {...params, session: {...params.session, active_turn_id: "accepted-original"}}, + {...params, queued_turn_count: 1}, + {...params, observation: {...params.observation, verified: false}}, + {...params, observation: {...params.observation, provider_ref: "0".repeat(24)}}, + ]) assert.throws(() => planStewardManagedMode(input)); +}); + +test("only the original native creation, workspace and registered Agent establish a binding", () => { + const {params, goal} = fixture(); + for (const invalid of [undefined, {...goal, creation_operation_id: "other-operation"}, + {...goal, workspace_path: "/other-project"}, {...goal, goal_id: "other-goal"}, + {...goal, registered_agents: ["project-worker"]}]) { + assert.throws(() => bindStewardManagedActor({...params, goal: invalid})); + } + const actor = bindStewardManagedActor({...params, goal}); + assert.equal(actor.goal_instance_id, goal.goal_instance_id); + assert.equal(actor.session_id, params.session.session_id); + assert.throws(() => planStewardManagedMode({...params, binding: {...params.binding, managed_actor: actor}, + goal: {...goal, goal_instance_id: "replacement-lifetime"}})); + assert.throws(() => planStewardManagedMode({...params, binding: {...params.binding, managed_actor: actor}})); + assert.throws(() => planStewardManagedMode({...params, binding: {...params.binding, managed_actor: actor}, + goal: {...goal, registered_agents: []}})); +}); + +test("publication changes only the native actor reference; scope configuration retains it", () => { + const {params, goal} = fixture(); + const current = {schema_version: "loopx_chat_conversation_bindings_v0", revision: 4, bindings: [params.binding]}; + const result = planConversationBinding({...params, goal, current, expected_revision: 4, + binding_id: params.binding.binding_id, operation: "bind_managed_actor"}); + const next = result.state as typeof current; + const prepared = next.bindings[0] as typeof params.binding & {managed_actor: Record}; + assert.deepEqual({...prepared, managed_actor: undefined}, {...params.binding, managed_actor: undefined}); + assert.equal(next.revision, 5); + const replay = planConversationBinding({...params, goal, binding: prepared, current: next, + binding_id: prepared.binding_id, expected_revision: 5, operation: "bind_managed_actor"}); + assert.equal(replay.changed, false); + const configured = planConversationBinding({current: next, expected_revision: 5, operation: "configure", + binding: {...prepared, goal_scope: "selected"}, observation: params.observation, + available_projects: [{kind: "project_workspace", project_ref: prepared.project_ref, + workspace_path: "/authorized/workspace", audience: "local_owner", grant: "workspace_read"}]}); + assert.deepEqual((configured.state as typeof next).bindings[0].managed_actor, prepared.managed_actor); + assert.throws(() => planConversationBinding({current: next, expected_revision: 5, operation: "configure", + binding: {...prepared, context_kind: "project", grant: "workspace_read"}, observation: params.observation})); +}); diff --git a/tests/test_native_steward_managed_mode.py b/tests/test_native_steward_managed_mode.py new file mode 100644 index 0000000000..086218a708 --- /dev/null +++ b/tests/test_native_steward_managed_mode.py @@ -0,0 +1,187 @@ +"""Real native File/SQLite creation behind the original private conversation. + +Only the existing synthetic model/provider are fake. These tests do not launch +or qualify a scheduled report, and never migrate an active user's Goal. +""" +import json + +import pytest + +from test_native_steward_private import steward as steward_fixture, finish +from loopx.capabilities.machine_configuration.builtins import build_builtin_machine_configuration_registry +from loopx.capabilities.machine_configuration.store import configure_machine_configuration +from loopx.control_plane.todos.list_readback import list_goal_todos +from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime + +steward = steward_fixture + + +def original(steward): + store, runtime, provider, transport, binding, capture, _ = steward + text = "Retain this original private conversation and its unfinished work." + transport.admit("steward-app", provider.event("steward-app", "original", text)) + row = next(item for item in transport.core.pending() if item["message"] == text) + assert finish(runtime, row)["status"] == "completed" + return row["session_id"], transport.managed_steward + + +def storage(runtime, provider): + args = {"runtime_root": runtime.coordination_runtime_root, + "registry": build_builtin_machine_configuration_registry(), + "configuration": {"schema_version": "loopx_machine_configuration_v0", "namespaces": { + "goal_storage": {"schema_version": "loopx_goal_storage_defaults_v1", + "new_goal_provider": provider, "canonical_creation": True, "new_goal_handoff_mode": "hard_lease"}}}} + plan = configure_machine_configuration(**args, execute=False) + configured = configure_machine_configuration(**args, execute=True, expected_plan_revision=plan["plan_revision"]) + assert configured["readback_verified"] + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_native_preparation_preserves_original_and_creates_one_real_actor(steward, tmp_path, monkeypatch, provider): + isolate_sqlite_runtime(tmp_path, monkeypatch) + store, runtime, _, transport, binding, capture, _ = steward + sid, service = original(steward) + storage(runtime, provider) + before_session = store.load_session(sid) + before_history, before_calls = store.messages(sid), capture.read_bytes() + before_registry = runtime.registry_path.read_bytes() + preview = service.prepare(binding["binding_id"], session_id=sid) + assert not preview["applied"] + assert runtime.registry_path.read_bytes() == before_registry + assert capture.read_bytes() == before_calls + applied = service.prepare(binding["binding_id"], session_id=sid, + expected_plan_id=preview["plan"]["plan_id"], execute=True) + actor = applied["actor"] + assert applied["applied"] and not applied["scheduling_enabled"] and not applied["model_invoked"] + assert actor["session_id"] == sid and actor["agent_id"] == "steward" + assert store.load_session(sid) == before_session + assert store.messages(sid) == before_history and capture.read_bytes() == before_calls + goals = json.loads(runtime.registry_path.read_text())["goals"] + assert len(goals) == 1 and goals[0]["id"] == actor["goal_id"] + assert goals[0]["repo"] != before_session["steward_context"]["workspace_path"] + todos = list_goal_todos(registry_path=runtime.registry_path, goal_id=actor["goal_id"], + agent_id="steward", role="agent", runtime_root_arg=str(runtime.coordination_runtime_root)) + assert todos["ok"] and todos["authority_read"]["source_authority"] == f"{provider}_v0" + assert todos["todo_count"] == 0 + replay = service.prepare(binding["binding_id"], session_id=sid, + expected_plan_id=preview["plan"]["plan_id"], execute=True) + assert replay["idempotent"] and replay["actor"] == actor + assert capture.read_bytes() == before_calls + assert service.health(binding["binding_id"])["prepared"] is True + # Changing only the existing portfolio scope must not detach the actor. + transport.bindings.configure(transport_ref="steward-app", project_ref=binding["project_ref"], + executor_endpoint_id="codex", context_kind="steward", goal_scope="selected") + assert transport.bindings.read()["bindings"][-1]["managed_actor"] == actor + + +@pytest.mark.parametrize("fault", ["queued", "provider", "stale_preview", "foreign_session"]) +def test_rejected_preparation_has_no_native_goal_or_model_effect(steward, fault): + store, runtime, provider, _, binding, capture, _ = steward + sid, service = original(steward) + preview = service.prepare(binding["binding_id"], session_id=sid) + if fault == "queued": + store.create_queued_turn(sid, client_turn_id="owner-followup", message="Keep my request first", origin="web") + elif fault == "provider": + provider.owner_ready = False + elif fault == "foreign_session": + sid = "0" * 32 + before_registry, before_calls = runtime.registry_path.read_bytes(), capture.read_bytes() + with pytest.raises((ValueError, KeyError)): + service.prepare(binding["binding_id"], session_id=sid, execute=True, + expected_plan_id="wrong" if fault == "stale_preview" else preview["plan"]["plan_id"]) + assert runtime.registry_path.read_bytes() == before_registry + assert capture.read_bytes() == before_calls + assert service.health(binding["binding_id"])["prepared"] is False + + +def test_preparation_recovers_native_creation_after_registration_loses_response(steward, tmp_path, monkeypatch): + isolate_sqlite_runtime(tmp_path, monkeypatch) + _, runtime, _, _, binding, capture, _ = steward + sid, service = original(steward) + storage(runtime, "sqlite") + preview = service.prepare(binding["binding_id"], session_id=sid) + before = capture.read_bytes() + import loopx.configure_goal as module + configure = module.configure_goal + + def fail_registration(**kwargs): + raise OSError("registration response lost") + + monkeypatch.setattr(module, "configure_goal", fail_registration) + with pytest.raises(OSError, match="response lost"): + service.prepare(binding["binding_id"], session_id=sid, + expected_plan_id=preview["plan"]["plan_id"], execute=True) + first = json.loads(runtime.registry_path.read_text())["goals"] + assert len(first) == 1 + monkeypatch.setattr(module, "configure_goal", configure) + recovered = service.prepare(binding["binding_id"], session_id=sid, + expected_plan_id=preview["plan"]["plan_id"], execute=True) + assert recovered["actor"]["goal_instance_id"] == first[0].get("goal_instance_id") + assert len(json.loads(runtime.registry_path.read_text())["goals"]) == 1 + assert capture.read_bytes() == before + + +def test_http_preparation_requires_current_preview_and_reads_original_binding(steward, tmp_path, monkeypatch): + import http.client + import threading + from types import SimpleNamespace + from loopx.chat_server import ChatHTTPServer, ChatRequestHandler + + isolate_sqlite_runtime(tmp_path, monkeypatch) + store, runtime, _, transport, binding, capture, _ = steward + sid, _ = original(steward) + storage(runtime, "sqlite") + server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler) + server.chat_store, server.runtime_controller = store, runtime + server.registry_path, server.runtime_root_override = runtime.registry_path, store.root.parent + server.lark_private_conversations = transport + server.lark_goal_topic_runtime = SimpleNamespace(health_snapshot=lambda: {}, close=lambda: None) + server.verbose = False + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + + def request(path, body=None, origin=None): + connection = http.client.HTTPConnection(*server.server_address, timeout=20) + try: + headers = {"Content-Type": "application/json"} + if origin: + headers["Origin"] = origin + connection.request("GET" if body is None else "POST", path, + None if body is None else json.dumps(body), headers=headers) + response = connection.getresponse() + return response.status, json.loads(response.read()) + finally: + connection.close() + + base = "/api/chat/lark/private-conversations" + route = base + "/managed-steward" + before_history, before_calls = store.messages(sid), capture.read_bytes() + try: + assert request(route, {"binding_id": binding["binding_id"]}, "https://foreign.example")[0] == 403 + assert request(route, {"binding_id": binding["binding_id"], "goal_id": "borrowed-project"})[0] == 409 + code, preview = request(route, {"binding_id": binding["binding_id"], "execute": False}) + assert code == 200 and not preview["applied"] + assert not json.loads(runtime.registry_path.read_text())["goals"] + # Changing the existing scope is a real source revision, not a forged + # expected digest. A stale preview must not create a Goal or model. + transport.bindings.configure(transport_ref="steward-app", project_ref=binding["project_ref"], + executor_endpoint_id="codex", context_kind="steward", goal_scope="selected") + body = {"binding_id": binding["binding_id"], "execute": True, + "session_id": sid, "plan_id": preview["plan"]["plan_id"]} + assert request(route, body)[0] == 409 + assert not json.loads(runtime.registry_path.read_text())["goals"] + code, preview = request(route, {"binding_id": binding["binding_id"], "execute": False}) + assert code == 200 + code, applied = request(route, {**body, "plan_id": preview["plan"]["plan_id"]}) + assert code == 200 and applied["applied"] + code, readback = request(base) + assert code == 200 + row = next(row for row in readback["connections"] if row["binding_id"] == binding["binding_id"]) + assert row["managed_steward"] == {"available": True, "prepared": True, "scheduling_enabled": False, + "goal_id": applied["actor"]["goal_id"], "agent_id": "steward", "session_id": sid} + assert store.messages(sid) == before_history and capture.read_bytes() == before_calls + assert request(route, {**body, "plan_id": preview["plan"]["plan_id"]})[1]["idempotent"] + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2)