diff --git a/apps/presentation/dashboard/src/data/chat.ts b/apps/presentation/dashboard/src/data/chat.ts index bdae58f1a9..08f2ffe18f 100644 --- a/apps/presentation/dashboard/src/data/chat.ts +++ b/apps/presentation/dashboard/src/data/chat.ts @@ -1262,6 +1262,26 @@ export type LoopXModeSnapshot = { export function fetchLoopXMode(sessionId: string) { return requestJson(`/api/chat/sessions/${sessionId}/loopx`); } +export type ManagedNativeTurn = { + turn_id: string; status: string; error?: string; + response?: {message?: string; native_loopx?: {status?: string; result_kind?: string; reason?: string; error?: string}} | null; +}; +export type ManagedNativeTurnSnapshot = { + ok: true; session_id: string; available: boolean; registered_agents: string[]; + reason?: string; turn: ManagedNativeTurn | null; +}; +export function fetchManagedNativeTurn(sessionId: string, agentId?: string, operationId?: string) { + const query = new URLSearchParams(); + if (agentId) query.set("agent_id", agentId); + if (operationId) query.set("operation_id", operationId); + return requestJson(`/api/chat/sessions/${encodeURIComponent(sessionId)}/native-turn?${query}`); +} +export function startManagedNativeTurn(sessionId: string, agentId: string, operationId: string) { + return requestJson<{ok: true; turn: ManagedNativeTurn; created: boolean}>( + `/api/chat/sessions/${encodeURIComponent(sessionId)}/native-turn`, { + method: "POST", body: JSON.stringify({agent_id: agentId, operation_id: operationId}), + }); +} export type DelegationInventory = { items: Array<{record_id: string; operation_id: string | null; agent_id?: string; todo_id?: string; status: string; worker_active?: boolean; recovery_required: boolean | null; diff --git a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx index e71d1796c7..7dbd839ed7 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx @@ -46,6 +46,7 @@ import { useTypedActionReadback, useTypedActionProposalReadback } from "../../da import { ChannelHeader } from "./channel-header"; import { GoalLoopXMode } from "./goal-loopx-mode"; +import { ManagedSessionTurnControl } from "./steward-managed-mode"; import { GoalTeamResults } from "./goal-team-results"; import { GoalManagedResults } from "./goal-managed-results"; import { GoalResearchResults, type GoalResearchApi } from "./goal-research-results"; @@ -948,7 +949,7 @@ export function PersonalWorkspacePage({ /** The bound managed executor offers native exact-turn steering. */ conversationSupportsSteering?: boolean; conversationSessionId?: string; - conversationHistoryState?: ConversationHistoryStatus; + conversationHistoryState?: ConversationHistoryStatus & {refresh?: () => Promise}; agents?: WorkspaceAgentOption[]; callbacks?: PersonalWorkspaceCallbacks; goalArchiveLoadState?: WorkspaceGoalArchiveLoadState; @@ -2807,6 +2808,9 @@ export function PersonalWorkspacePage({ {!runtimeSourceUnavailable && selectedGoal && !selectedGoal.loadState && !agentWorkActive && selectedGoalTab === "overview" ? : !runtimeSourceUnavailable && selectedGoal?.loadState === "loading" && selectedGoalTab === "overview" ? : null} {!readOnly ? openSettings({ kind: "settings", tab: "machine" })} /> : null} + {selectedGoalId && selectedGoalTab === "chat" && !agentWorkActive && !readOnly && selectedAgentId === "codex" && conversationSessionId + ? {void conversationHistoryState?.refresh?.().catch(() => {}); void callbacks.onBackgroundRefresh?.();}}/> : null} {selectedGoalId && selectedGoalTab === "chat" && !agentWorkActive && !readOnly && selectedAgentId === "codex" && callbacks.onStartLoopX ? callbacks.onPrepareLoopX!(selectedAgentId, selectedGoalId)} 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 index e557398d2d..1c26d869b3 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/steward-managed-mode.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/steward-managed-mode.tsx @@ -1,5 +1,5 @@ -import type {PrivateConversation} from "../../data/chat"; -import {useEffect, useState} from "react"; +import {fetchManagedNativeTurn, startManagedNativeTurn, type ManagedNativeTurn, type ManagedNativeTurnSnapshot, type PrivateConversation} from "../../data/chat"; +import {useEffect, useRef, useState} from "react"; import {StewardReport} from "./steward-report"; /** Identity preparation is deliberately distinct from enabling scheduled work. */ @@ -44,3 +44,98 @@ export function StewardManagedMode({state, zh, busy, available, onPrepare, onMon } ; } + +/** One explicit segment; calendar scheduling remains with its existing owner. */ +export function ManagedSessionTurnControl({sessionId, zh, onReadback}: {sessionId: string; zh: boolean; onReadback?: () => void}) { + const cacheKey = `loopx.managed-turn.${sessionId}`; + const [intent, setIntent] = useState<{agent: string; id: string} | null>(() => { + try { + const saved = JSON.parse(sessionStorage.getItem(cacheKey) ?? "null"); + return saved && typeof saved.agent === "string" && typeof saved.id === "string" ? saved : null; + } catch {return null;} + }); + const [state, setState] = useState(null); + const [agent, setAgent] = useState(intent?.agent ?? ""); + const [turn, setTurn] = useState(null); + const [error, setError] = useState(""); + const [sending, setSending] = useState(false); + const [readRevision, setReadRevision] = useState(0); + const alive = useRef(true); + const requestInFlight = useRef(false); + const readbackRef = useRef(onReadback); + readbackRef.current = onReadback; + const observedTurn = useRef(""); + const terminal = Boolean(turn && ["completed", "failed", "interrupted"].includes(turn.status)); + const unresolved = Boolean(intent && !terminal); + useEffect(() => { + alive.current = true; + let reading = false; + let cancelled = false; + async function read() { + if (reading) return; + reading = true; + try { + const next = await fetchManagedNativeTurn(sessionId, agent || undefined, intent?.id); + if (cancelled) return; + setState(next); setTurn(next.turn); setError(""); + const identity = next.turn ? `${next.turn.turn_id}:${next.turn.status}` : ""; + if (identity && next.turn && ["completed", "failed", "interrupted"].includes(next.turn.status) + && observedTurn.current !== identity) {observedTurn.current = identity; readbackRef.current?.();} + if (!agent && next.registered_agents.length === 1) setAgent(next.registered_agents[0]); + } catch (failure) {if (!cancelled) setError(failure instanceof Error ? failure.message : String(failure));} + finally {reading = false;} + } + void read(); + const interval = unresolved ? window.setInterval(() => {if (!document.hidden) void read();}, 2500) : undefined; + return () => {cancelled = true; alive.current = false; if (interval) window.clearInterval(interval);}; + }, [sessionId, agent, intent?.id, unresolved, readRevision]); + async function run() { + if (requestInFlight.current || !agent) return; + const next = unresolved && intent ? intent : {agent, id: crypto.randomUUID()}; + requestInFlight.current = true; setSending(true); setError(""); + try { + // Persist correlation before POST. An unknown response, reload or retry + // keeps this identity; only a known terminal result permits new work. + sessionStorage.setItem(cacheKey, JSON.stringify(next)); setIntent(next); setTurn(null); + const accepted = await startManagedNativeTurn(sessionId, next.agent, next.id); + if (alive.current) setTurn(accepted.turn); + } catch (failure) {if (alive.current) setError(failure instanceof Error ? failure.message : String(failure));} + finally {requestInFlight.current = false; if (alive.current) setSending(false);} + } + const native = turn?.response?.native_loopx; + const replayHeld = terminal && native?.status !== "committed" && native?.status !== "stopped"; + const summaryState = error || replayHeld ? (zh ? " · 需要核验" : " · Review required") + : native?.result_kind === "user_action_required" ? (zh ? " · 需要处理" : " · Action required") + : turn && !terminal ? (zh ? " · 执行中" : " · Running") + : native?.status === "committed" ? (zh ? " · 已验收" : " · Validated") + : state && !state.available ? (zh ? " · 暂不可用" : " · Unavailable") : ""; + const status = !turn ? intent ? (zh ? "正在核对原请求；重试会保留同一请求。" : "Checking the original request; retry keeps its identity.") : "" + : !terminal ? (zh ? "本轮任务正在执行。" : "This work segment is running.") + : native?.status === "committed" ? native.result_kind === "validated_completion" + ? (zh ? "本任务已验收并完成，结果已回写。" : "This task was validated, completed and recorded.") + : (zh ? "本轮进展已验收并回写。" : "This segment's progress was validated and recorded.") + : native?.result_kind === "wait" ? (zh ? "本轮等待执行条件。" : "This segment is waiting for execution conditions.") + : native?.result_kind === "user_action_required" ? turn?.response?.message ?? (zh ? "本任务需要补充执行条件。" : "This task requires an execution condition.") + : (zh ? "本轮未完成，原执行记录已保留。" : "This segment did not complete; its original record is retained."); + return
+ {zh ? "按 LoopX 任务推进" : "Advance LoopX tasks"}{summaryState} +
+ {status ?

{status}

: null} + {replayHeld ?

{zh ? "请先核验原执行记录；此入口保留失败结果，不重放执行。" : "Review the original execution record first; this control retains failures without replaying them."}

: null} + {error ?

{error}

: null} + {native?.reason || native?.error || turn?.error || state && !state.available && !unresolved + ?
{zh ? "未执行原因" : "Execution details"}

{native?.reason || native?.error || turn?.error || state?.reason}

: null} + {error || state && !state.available ? : null} +
{zh ? "执行范围" : "Execution scope"}

{zh + ? "沿用当前会话的沙箱和原任务的独立验收声明，不升级写入权限。普通只读会话可核验已授权来源；缺少验收时不会启动模型或记为完成。本入口不启用周期调度。" + : "Keep this conversation's sandbox and the task's declared validation without adding write permission. Ordinary read-only conversations can verify authorized sources. Missing validation starts no model and cannot count as completion. This control does not enable scheduling."}

+
; +} diff --git a/docs/assets/personal-workspace/steward-reports/managed-session-after.png b/docs/assets/personal-workspace/steward-reports/managed-session-after.png new file mode 100644 index 0000000000..deb4d064ce Binary files /dev/null and b/docs/assets/personal-workspace/steward-reports/managed-session-after.png differ diff --git a/docs/assets/personal-workspace/steward-reports/managed-session-before.png b/docs/assets/personal-workspace/steward-reports/managed-session-before.png new file mode 100644 index 0000000000..4bc743994c Binary files /dev/null and b/docs/assets/personal-workspace/steward-reports/managed-session-before.png differ diff --git a/docs/assets/personal-workspace/steward-reports/managed-session-narrow.png b/docs/assets/personal-workspace/steward-reports/managed-session-narrow.png new file mode 100644 index 0000000000..cd89c51836 Binary files /dev/null and b/docs/assets/personal-workspace/steward-reports/managed-session-narrow.png differ diff --git a/docs/assets/personal-workspace/steward-reports/managed-session-ready.png b/docs/assets/personal-workspace/steward-reports/managed-session-ready.png new file mode 100644 index 0000000000..64fc592db7 Binary files /dev/null and b/docs/assets/personal-workspace/steward-reports/managed-session-ready.png differ diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index e9d6195a23..31762f0a84 100644 --- a/docs/product/periodic-reports.md +++ b/docs/product/periodic-reports.md @@ -434,6 +434,44 @@ not production-message or real-phone proof. ## Acceptance boundaries +### Ordinary managed Goal conversations + +In a local Goal's Codex conversation, **Advance LoopX tasks → Advance once** +can submit one bounded native Turn for an explicitly selected registered +Agent. It keeps the same Core Session, upstream thread, model settings and +sandbox. Ordinary Goal Chat is currently read-only: this path qualifies source +inspection and validation tasks, **not writable execution or a permission +upgrade**. For example, a finite task may close only when its independently +declared readiness check passes against the original source. + +The native Turn still selects the task and owns admission, writeback and quota. +The host resolves the selected task's existing immutable argv validation +declaration before model execution, retains its canonical digest/revision, and +rechecks that source and the execution lease before accepting completion. +The normal Todo completion owner runs its declared check as well. Missing +validation waits without starting a model; a failed check cannot complete the +task or spend a delivery slot. Model-reported progress alone is insufficient. +The API accepts only the registered Agent, operation identity and optional +existing Todo reference; it accepts no validation command or runtime profile. + +The control reads back the original Core Turn. Reload, an unknown POST response +and retry retain one operation identity; a failed or unknown execution is never +replayed. Busy/queued user work wins admission. Attached, external and private +steward conversations retain their own admission paths; stopped/replaced Goals, +changed validation and late results fail closed. Existing continuous Goal mode +is a separate path and is not converted by this action. + +This is a staged execution prerequisite. General writable managed execution, +natural-language task configuration and arbitrary calendar/monitor/report +composition remain unfinished. This control enables no recurring work and +does not change the original steward's subscription or publication route. +Packaged browser evidence uses a disposable real Core/File backend with a +synthetic model/provider; it does not certify real-model or natural-deadline +acceptance. See the [baseline](../assets/personal-workspace/steward-reports/managed-session-before.png) +and [validated readback](../assets/personal-workspace/steward-reports/managed-session-after.png). + +### Original private steward + The original-Session host bridge now runs a bounded native Turn through the existing Chat worker and Codex app-server connection. It uses the explicit `codex_app_server` outer-controller profile, not Codex CLI execution or a diff --git a/examples/personal-workspace-browser/steward_report_fixture.py b/examples/personal-workspace-browser/steward_report_fixture.py index ffd3ea1042..c07637d2fe 100644 --- a/examples/personal-workspace-browser/steward_report_fixture.py +++ b/examples/personal-workspace-browser/steward_report_fixture.py @@ -8,6 +8,7 @@ import argparse import json +import os from pathlib import Path import signal import sys @@ -21,6 +22,7 @@ def main() -> None: parser.add_argument("--output-dir", type=Path, required=True) 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.") args = parser.parse_args() repo = Path(__file__).resolve().parents[2] sys.path[:0] = [str(repo), str(repo / "tests")] @@ -41,13 +43,21 @@ def main() -> None: fixture = steward.__wrapped__(root) data = next(fixture) store, runtime, provider, transport, binding, capture, workspace = data - sid, managed = original(data) - storage(runtime, "file") - plan = managed.prepare(binding["binding_id"], session_id=sid)["plan"] - actor = managed.prepare(binding["binding_id"], session_id=sid, execute=True, - expected_plan_id=plan["plan_id"])["actor"] + managed_session = None + if args.managed_session: + import pytest + from test_native_steward_turn_host import managed_fixture + monkeypatch = pytest.MonkeyPatch() + _, managed_session, actor, managed_todo_id, artifact, _ = managed_fixture(data, root, monkeypatch) + sid, managed = actor["session_id"], transport.managed_steward + else: + sid, managed = original(data) + storage(runtime, "file") + plan = managed.prepare(binding["binding_id"], session_id=sid)["plan"] + actor = managed.prepare(binding["binding_id"], session_id=sid, execute=True, + expected_plan_id=plan["plan_id"])["actor"] + install_model(data, root, sid) transport.reconcile() - install_model(data, root, sid) if args.invalid_editorial: helper = root / "report_model.py" helper.write_text(helper.read_text().replace('"request_digest": request["request_digest"]', '"request_digest": "known-invalid"')) @@ -77,7 +87,10 @@ def main() -> None: signal.signal(signal.SIGINT, lambda *_: stop.set()) threading.Thread(target=server.serve_forever, daemon=True).start() (output / "ready.json").write_text(json.dumps({"url": f"http://127.0.0.1:{server.server_port}/chat/", - "binding_id": binding["binding_id"], "session_id": sid})) + "fixture_pid": os.getpid(), + "binding_id": binding["binding_id"], "session_id": sid, + **({"managed_session_id": managed_session["session_id"], "goal_id": actor["goal_id"], + "managed_todo_id": managed_todo_id} if managed_session else {})})) wake, outbound, delivery = None, None, None try: while not stop.wait(.2): @@ -97,6 +110,8 @@ def main() -> None: "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 {}), "synthetic_only": True}, ensure_ascii=False)) finally: if wake: @@ -106,6 +121,8 @@ def main() -> None: delivery.join(5) server.shutdown() server.server_close() + if managed_session: + monkeypatch.undo() try: next(fixture) except StopIteration: diff --git a/loopx/capabilities/native_chat/steward_managed_mode.py b/loopx/capabilities/native_chat/steward_managed_mode.py index b416041d0c..e2ced4e6c9 100644 --- a/loopx/capabilities/native_chat/steward_managed_mode.py +++ b/loopx/capabilities/native_chat/steward_managed_mode.py @@ -1,13 +1,15 @@ -"""Native Goal/Agent preparation for the original private steward. +"""Native identity preparation and managed Session admission in the Chat owner. 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. +second Chat Session is created here. The ordinary local entry reuses the +existing worker and sandbox; private steward preparation remains distinct. """ from __future__ import annotations import hashlib from pathlib import Path +import re from typing import Any from ...agent_registry import load_goal_from_registry, registered_agent_ids_for_goal @@ -169,3 +171,157 @@ def health(self, binding_id: str) -> dict[str, Any]: "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"} + + +class ManagedTaskValidationUnavailable(ValueError): + """A native selection has no usable caller-declared independent validator.""" + + +class ManagedSessionTurn: + """Thin local-owner entry to the existing Turn worker and task validator. + + No identity preparation, new Session, calendar or task policy lives here. + Attached and provider-bound conversations retain their original owners. + """ + def __init__(self, controller: Any) -> None: + self.controller, self.store = controller, controller.store + self.validation_basis: dict[str, Any] | None = None + self.task_validator = None + self.validation_unavailable = False + + def _request(self, session_id: str, agent_id: str, operation_id: str, + todo_id: str | None = None) -> dict[str, Any]: + if not re.fullmatch(r"[A-Za-z0-9._:-]{1,100}", operation_id): + raise ValueError("operation_id must be a bounded public token") + session = self.store.load_session(session_id) + if not session: + raise KeyError("chat session was not found") + goal = StewardManagedMode(self.controller, None)._goal(session.get("goal_id")) + if not goal: + raise ValueError("the original registered Goal is unavailable") + actor = {"schema_version": "loopx_managed_session_actor_v0", "goal_id": session["goal_id"], + "goal_instance_id": session.get("goal_instance_id"), "session_id": session_id, + "agent_id": agent_id, "workspace_path": goal["workspace_path"]} + return {"operation": "native_turn", "host": "codex-app-server", "actor": actor, + "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]: + 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"]) + 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)), + "queued_turn_count": len(self.store.queued_turns(actor["session_id"]))}) + + def inspect(self, session_id: str, *, agent_id: str | None = None, + operation_id: str | None = None) -> dict[str, Any]: + session = self.store.load_session(session_id) + if not session: + raise KeyError("chat session was not found") + goal = StewardManagedMode(self.controller, None)._goal(session.get("goal_id")) + agents = goal["registered_agents"] if goal else [] + result: dict[str, Any] = {"ok": True, "session_id": session_id, "registered_agents": agents, + "available": False, "turn": None} + if operation_id is not None: + if not re.fullmatch(r"[A-Za-z0-9._:-]{1,100}", operation_id): + raise ValueError("invalid operation identity") + # Readback of an accepted/unknown request is not another execution. + result["turn"] = self.store.turn_for_client(session_id, "managed." + operation_id) + try: + self.admit(self._request(session_id, agent_id or next(iter(agents), ""), "preview")) + result["available"] = True + except ValueError as exc: + result["reason"] = str(exc) + return result + + def submit(self, session_id: str, *, agent_id: str, operation_id: str, + todo_id: str | 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) + existing = self.store.turn_for_client(session_id, "managed." + operation_id) + if existing is not None: + if existing.get("loopx_request") != request: + raise ValueError("managed Turn request identity conflict; inspect the original operation") + return existing, False # Failed/unknown work is never replayed. + commit = managed_session_goal_commit(store=self.store, + registry_path=controller.registry_path, session_id=session_id) + with commit(): + 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="推进一次 LoopX 任务。", + origin="native_loopx", loopx_execution=True, loopx_request=request) + try: + session = self.store.load_session(session_id) + adapter = controller._ensure_adapter_locked(session, work_dir=Path(admitted["workspace_path"]), + objective="Advance this Goal's authorized work.", accepted_turn_id=accepted.turn["turn_id"]) + if Path(adapter.session.work_dir).resolve() != Path(admitted["workspace_path"]).resolve(): + raise ValueError("the original managed conversation workspace changed") + controller._start_accepted_turn_worker(session_id=session_id, turn_id=accepted.turn["turn_id"], + message=accepted.turn["message"], attachments=[], adapter=adapter, loopx_execution=True) + except Exception as exc: + controller._fail_queue_preparation(session_id, accepted.turn["turn_id"], exc) + raise + return accepted.turn, accepted.created + + def _validation_source(self, request: dict[str, Any], todo_id: str) -> dict[str, Any]: + from ...control_plane.todos.list_readback import list_goal_todos + readback = list_goal_todos(registry_path=self.controller.registry_path, goal_id=request["actor"]["goal_id"], + todo_id=todo_id, role="agent", runtime_root_arg=str(self.controller.coordination_runtime_root)) + rows = [readback["todo"]] if readback.get("todo") else [] + if len(rows) != 1 or rows[0].get("completion_validation_required") is not True: + raise ManagedTaskValidationUnavailable("the selected task needs its existing independent argv validation declaration") + return rows[0] + + def prepare_validation(self, request: dict[str, Any], native_request: dict[str, Any]) -> None: + from ...control_plane.turn_driver.driver import selected_turn_todo + from ...control_plane.turn_driver.command_validation import build_loopx_turn_command_validator + from ...control_plane.todos.completion_validation_store import read_completion_validation_declaration + selected = selected_turn_todo(native_request.get("turn_envelope") or {}) + if not selected.get("todo_id"): + raise ManagedTaskValidationUnavailable("select an existing native task with independent validation") + self.validation_basis = self._validation_source(request, selected["todo_id"]) + declaration = read_completion_validation_declaration(runtime_root=self.controller.coordination_runtime_root, + goal_id=request["actor"]["goal_id"], todo_id=selected["todo_id"], + expected_digest=self.validation_basis["completion_validation_sha256"]) + if not declaration or not declaration.get("validation_command_argv"): + raise ManagedTaskValidationUnavailable("the selected task requires its existing argv validation declaration") + self.task_validator = build_loopx_turn_command_validator(declaration["validation_command_argv"], + project=Path(request["actor"]["workspace_path"]), timeout_seconds=declaration.get("validation_timeout_seconds") or 20) + + def validator(self, request: dict[str, Any], *, turn_id: str): + """Resolve only the native selection's immutable, caller-declared argv. + + No browser or model argument may install a host validation command. + The captured canonical digest and fresh source must agree before and + after the existing command validator reads the task's real artifact. + """ + from ...control_plane.turn_driver.driver import selected_turn_todo + + def validate(plan, result): + actor = request["actor"] + selected = selected_turn_todo(plan.get("turn_envelope") or {}) + basis = self.validation_basis + if not basis or not self.task_validator or basis["todo_id"] != selected["todo_id"]: + raise ValueError("the original task validation source was not captured before model work") + + def current(): + self.admit(request, turn=self.store.load_turn(actor["session_id"], turn_id)) + fresh = self._validation_source(request, selected["todo_id"]) + if any(fresh.get(key) != basis.get(key) for key in + ("completion_validation_sha256", "completion_validation_revision")): + raise ValueError("the selected task's validation declaration changed") + + current() + receipt = self.task_validator(plan, result) + current() + return receipt + + return validate diff --git a/loopx/capabilities/native_chat/steward_turn.py b/loopx/capabilities/native_chat/steward_turn.py index e3764f4c9d..6bfcc8dbf9 100644 --- a/loopx/capabilities/native_chat/steward_turn.py +++ b/loopx/capabilities/native_chat/steward_turn.py @@ -15,7 +15,7 @@ from ...control_plane.goals.activation import goal_is_stopped from ...file_lock import exclusive_file_lock -from .steward_managed_mode import StewardManagedMode +from .steward_managed_mode import ManagedSessionTurn, ManagedTaskValidationUnavailable, StewardManagedMode def native_steward_client_id(actor: Mapping[str, Any], turn_instance_id: str) -> str: @@ -28,8 +28,6 @@ def __init__(self, controller: Any) -> None: self.controller = controller self.store = controller.store self.bindings = controller.project_contexts.conversation_bindings - if self.bindings is None: - raise ValueError("the original steward binding is unavailable") self.managed = StewardManagedMode(controller, self.bindings) def _monitor_selection_failed_before_poll(self, turn: dict[str, Any]) -> bool: @@ -50,6 +48,10 @@ def _admit(self, request: dict[str, Any], *, turn: dict[str, Any] | None = None, if self.controller.closed.is_set(): 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) + if self.bindings is None: + raise ValueError("the original steward binding is unavailable") session = self.store.load_session(actor["session_id"]) if not session or not session.get("steward_context"): raise ValueError("the original steward Session is unavailable") @@ -152,19 +154,19 @@ def validate(_plan: Mapping[str, Any], _result: Mapping[str, Any]) -> dict[str, "summary": "independent original report validation passed"} return validate - def _report_lease_proof(self, request: Mapping[str, Any], lease: Mapping[str, Any] | None) -> dict[str, Any]: + 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"], "--todo-id", request["todo_id"]]).get("lease") if (not lease or not observed or observed.get("status") != "active" or any(observed.get(key) != lease.get(key) for key in ("owner", "idempotency_key", "lease_epoch", "version"))): - raise ValueError("the original report execution lease changed") + raise ValueError("the original task execution lease changed") return {"task_lease_idempotency_key": observed["idempotency_key"], "task_lease_expected_version": observed["version"]} @contextmanager - def _report_execution_lease(self, request: Mapping[str, Any], turn_id: str, *, recovery: bool = False): + def _task_execution_lease(self, request: Mapping[str, Any], turn_id: str, *, recovery: bool = False): """Both generation and cached validation use the original Todo owner.""" from .steward_commands import native_command actor = request["actor"] @@ -184,13 +186,13 @@ def _report_execution_lease(self, request: Mapping[str, Any], turn_id: str, *, r retained = native_command(self.controller, ["task-lease", "inspect", "--goal-id", actor["goal_id"], "--todo-id", request["todo_id"]]).get("lease") if retained and retained.get("status") == "active": - proof = self._report_lease_proof(request, lease) + proof = self._task_lease_proof(request, lease) native_command(self.controller, ["task-lease", "release", "--goal-id", actor["goal_id"], "--todo-id", request["todo_id"], "--owner", actor["agent_id"], "--idempotency-key", proof["task_lease_idempotency_key"], "--expected-version", str(proof["task_lease_expected_version"])]) except (OSError, ValueError, RuntimeError): - self.store.append_event(actor["session_id"], turn_id, kind="native_loopx.report_lease_release_unconfirmed", + self.store.append_event(actor["session_id"], turn_id, kind="native_loopx.task_lease_release_unconfirmed", payload={"todo_id": request["todo_id"], "lease_epoch": lease["lease_epoch"]}) def recover_report_settlement(self, actor: dict[str, Any], window: dict[str, Any], todo: dict[str, Any]) -> None: @@ -237,7 +239,7 @@ def no_model(*_args): "--host", "generic-cli", "--resume-turn-key", key, "--retry-failed-turn", "--execute"]) args.host, args.session_scope = "codex-app-server", "agent" result: dict[str, Any] = {} - with (self._report_execution_lease(request, original["turn_id"], recovery=True) + with (self._task_execution_lease(request, original["turn_id"], recovery=True) if todo["status"] == "open" else nullcontext(None)) as lease: handle_turn_command(args, registry_path=self.controller.registry_path, runtime_root_arg=str(reports.root), output_format=lambda *_a, **_kw: "json", @@ -245,7 +247,7 @@ def no_model(*_args): native_session_binding_resolver=lambda _envelope: {"schema_version": "loopx_turn_session_binding_v0", "goal_id": actor["goal_id"], "agent_id": actor["agent_id"]}, native_task_validator=self._report_validator(request), - **({"native_completion_lease_resolver": lambda: self._report_lease_proof(request, lease)} if lease else {})) + **({"native_completion_lease_resolver": lambda: self._task_lease_proof(request, lease)} if lease else {})) # Retain the historical outer failure. The actual native owner # supplies the new event; no status or publication proof is forged. self.store.append_event(actor["session_id"], original["turn_id"], kind="native_loopx.result", payload=result) @@ -346,6 +348,7 @@ 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 if admitted["observation_only"]: from .steward_monitor import StewardMonitor monitor = StewardMonitor(self.controller) @@ -354,11 +357,12 @@ def execute(self, *, session_id: str, turn_id: str, adapter: Any, 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" - report_lease = None report_context = "" + managed_answer = "" + managed_lease, managed_lease_request = None, None def report_lease_proof(): - return self._report_lease_proof(request, report_lease) + return self._task_lease_proof(request, report_lease) def current(): fresh = self.store.load_turn(session_id, turn_id) @@ -367,6 +371,8 @@ def current(): raise ValueError("the original steward native thread changed") if report_lease: report_lease_proof() + if managed_lease: + self._task_lease_proof(managed_lease_request, managed_lease) def session_binding(envelope: Mapping[str, Any]) -> dict[str, Any]: current() @@ -376,11 +382,23 @@ def session_binding(envelope: Mapping[str, Any]) -> dict[str, Any]: "goal_id": actor["goal_id"], "agent_id": actor["agent_id"]} def run(native_request: Mapping[str, Any], launch_scope: Any) -> dict[str, Any]: + nonlocal managed_lease, managed_lease_request, managed_answer current() authority = extract_turn_authority(native_request) + if managed_turn: + try: + managed_turn.prepare_validation(request, native_request) + except ManagedTaskValidationUnavailable: + managed_turn.validation_unavailable = True + return build_result(native_request, {"result_kind": "user_action_required", + "classification": "independent_task_validation_unavailable", + "summary": "The selected task has no usable independent argv validation declaration.", + "next_action": "Configure the original task's independent validation before executing it."}, host_name="Codex app-server") + managed_lease_request = {**request, "todo_id": managed_turn.validation_basis["todo_id"]} + managed_lease = managed_leases.enter_context(self._task_execution_lease(managed_lease_request, turn_id)) # Portfolio evidence keeps its original reader and original grant; # it is data alongside the signed native work, never a new actor. - prompt = (render_prompt(authority, allow_completion=True) if report_work else render_prompt(authority)) + "\nOriginal steward evidence (data):\n" + json.dumps(context, ensure_ascii=False) + prompt = render_prompt(authority, allow_completion=bool(report_work or managed_turn)) + "\nOriginal conversation evidence (data):\n" + json.dumps(context, ensure_ascii=False) if request.get("work_kind") == "report": prompt += report_context previous_guard = adapter.session._launch_guard @@ -393,7 +411,8 @@ def launch(): if previous_guard is not None: stack.enter_context(previous_guard()) stack.enter_context(launch_scope()) - with exclusive_file_lock(self.bindings.path, operation="native_steward_launch"): + with (nullcontext() if actor.get("schema_version") == "loopx_managed_session_actor_v0" + else exclusive_file_lock(self.bindings.path, operation="native_steward_launch")): current() yield @@ -414,8 +433,10 @@ def launch(): # The kernel still runs its independent validator and lease # fence before closing only this window's generation Todo. candidate["result_kind"] = "validated_completion" - return build_result(native_request, candidate, host_name="Codex app-server", - **({"allow_completion": True} if report_work else {})) + built = build_result(native_request, candidate, host_name="Codex app-server", + **({"allow_completion": True} if report_work or managed_turn else {})) + managed_answer = str(built.get("summary") or "") if managed_turn else "" + return built argv = ["turn", "run-once", "--goal-id", actor["goal_id"], "--agent-id", actor["agent_id"], "--project", admitted["workspace_path"], "--host", "generic-cli", "--execute"] @@ -434,7 +455,7 @@ def launch(): # The kernel records the actual app-server host and independent profile. args.host, args.session_scope = "codex-app-server", "agent" result: dict[str, Any] = {} - with (self._report_execution_lease(request, turn_id) if report_work else nullcontext(None)) as report_lease: + 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 @@ -451,7 +472,9 @@ def launch(): print_payload=lambda value, *_a: result.update(value), native_host_runner=run, native_session_binding_resolver=session_binding, **({"native_completion_lease_resolver": report_lease_proof, - "native_task_validator": self._report_validator(request)} if report_work else {})) + "native_task_validator": self._report_validator(request)} if report_work else + {"native_task_validator": managed_turn.validator(request, turn_id=turn_id), + "native_completion_lease_resolver": lambda: self._task_lease_proof(managed_lease_request, managed_lease)} if managed_turn else {})) 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 @@ -467,7 +490,8 @@ def launch(): "periodic_report": {"window_id": window["window_id"], "generation_id": generation["generation_id"], "delivery_verified": False}} return {"schema_version": "loopx_chat_agent_response_v0", "message": - "本轮任务已完成结果回写。" if result.get("status") == "committed" else + "当前没有可执行且已配置独立验收的原生任务，尚未启动模型；请检查原任务的验收配置。" if managed_turn and managed_turn.validation_unavailable else + (managed_answer or "本轮任务已完成结果回写。") if result.get("status") == "committed" else "本轮等待后续执行条件。" if result.get("status") == "stopped" and result.get("result_kind") == "wait" else "本轮任务尚未完成；执行状态已保留。", "proposals": [], "gate": None, "native_loopx": result} diff --git a/loopx/chat_server.py b/loopx/chat_server.py index 78abf218e2..9f07fcac4b 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -948,6 +948,24 @@ def _session_snapshot(self, session_id: str) -> None: except KeyError: self._send_error("chat session was not found", status=404) + def _managed_native_turn(self, session_id: str, *, execute: bool = False) -> None: + from .capabilities.native_chat.steward_managed_mode import ManagedSessionTurn + service = ManagedSessionTurn(self.server.runtime_controller) + try: + if execute: + body = self._read_json() + if set(body) - {"agent_id", "operation_id", "todo_id"}: + raise ValueError("unsupported managed Turn arguments") + turn, created = service.submit(session_id, **body) + return self._send_json({"ok": True, "session_id": session_id, "turn": turn, "created": created}, status=202) + query = parse_qs(urlparse(self.path).query, keep_blank_values=True) + self._send_json(service.inspect(session_id, agent_id=query.get("agent_id", [None])[0], + operation_id=query.get("operation_id", [None])[0])) + except KeyError: + self._send_error("chat session was not found", status=404) + except (TypeError, ValueError, RuntimeError) as exc: + self._send_error(str(exc), status=409, error_code="managed_turn_unavailable") + def _list_sessions(self) -> None: query = parse_qs(urlparse(self.path).query, keep_blank_values=True) if "conversation_binding_id" in query: @@ -1636,6 +1654,8 @@ def do_GET(self) -> None: if len(action_parts) == 3 and action_parts[:2] == ["api", "actions"]: return self._action_snapshot(action_parts[2]) session_parts = path.strip("/").split("/") + if len(session_parts) == 5 and session_parts[:3] == ["api", "chat", "sessions"] and session_parts[4] == "native-turn": + return self._managed_native_turn(session_parts[3]) if len(session_parts) == 5 and session_parts[:3] == ["api", "chat", "sessions"] and session_parts[4] == "loopx": return handle_loopx_request(self, session_parts[3]) if len(session_parts) == 4 and session_parts[:3] == ["api", "chat", "sessions"]: @@ -1686,6 +1706,8 @@ def do_POST(self) -> None: if path in post_dispatch: return post_dispatch[path]() session_action_parts = path.strip("/").split("/") + if len(session_action_parts) == 5 and session_action_parts[:3] == ["api", "chat", "sessions"] and session_action_parts[4] == "native-turn": + return self._managed_native_turn(session_action_parts[3], execute=True) if len(session_action_parts) == 5 and session_action_parts[:3] == ["api", "chat", "sessions"] and session_action_parts[4] == "loopx": return handle_loopx_request(self, session_action_parts[3], apply=True) if len(session_action_parts) == 5 and session_action_parts[:3] == ["api", "chat", "sessions"] and session_action_parts[4] == "resume": diff --git a/loopx/control_plane/collaboration/steward_managed_mode.ts b/loopx/control_plane/collaboration/steward_managed_mode.ts index 6d8c3933fc..50d6ea2a10 100644 --- a/loopx/control_plane/collaboration/steward_managed_mode.ts +++ b/loopx/control_plane/collaboration/steward_managed_mode.ts @@ -9,6 +9,54 @@ import type { JsonObject } from "../runtime_decode.ts"; import { requireJsonObject, requireNonEmptyString } from "../runtime_decode.ts"; import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; import { ENVELOPED_SHA256_PATTERN } from "../content_digest.ts"; +import {resolveConversationScope} from "./conversation_scope.ts"; + +/** Local owner Sessions reuse the same governed Turn host. Registration is a + * selection, not a grant: the native Turn owner still admits the actual work. + * A bound steward keeps its stricter provider/source admission below. + */ +export function admitManagedSessionTurn(params: JsonObject): JsonObject { + const session = requireJsonObject(params.session, "managed Session"); + const goal = requireJsonObject(params.goal, "original Goal"); + 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.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) + || typeof goal.workspace_path !== "string" || !goal.workspace_path) { + throw new EffectRuntimeRequestError("select an active local Goal's managed Codex conversation and a registered Agent"); + } + 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" + || actor.goal_id !== goal.goal_id || actor.goal_instance_id !== goal.goal_instance_id + || actor.workspace_path !== goal.workspace_path + || actor.session_id !== session.session_id + || typeof actor.agent_id !== "string" || !/^[A-Za-z0-9][A-Za-z0-9._-]{0,119}$/.test(actor.agent_id) + || typeof actor.session_id !== "string" || !/^[a-f0-9]{32}$/.test(actor.session_id)) { + throw new EffectRuntimeRequestError("the native Turn must retain its original managed Session and Goal lifetime"); + } + const requestFields = ["operation", "host", "actor", "turn_instance_id", "todo_id", "validation_command", "work_kind"]; + if (Object.keys(request).some(key => !requestFields.includes(key)) + || request.operation !== "native_turn" || request.host !== "codex-app-server" || request.work_kind !== "native_turn" + || 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))) { + throw new EffectRuntimeRequestError("managed work uses the native task's existing validation, without caller commands"); + } + 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) + : session.active_turn_id || params.queued_turn_count !== 0) { + throw new EffectRuntimeRequestError("the original conversation is busy or no longer owns this Turn; retain its current work"); + } + return {actor, workspace_path: goal.workspace_path, host: "codex-app-server", observation_only: false}; +} export const STEWARD_MANAGED_ACTOR_SCHEMA = "loopx_steward_managed_actor_v0"; const AGENT_ID = "steward"; diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index f251b74ddb..6cb94c3ba0 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -628,6 +628,7 @@ export function createEffectRuntimeHandlers( ["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.steward.managed_turn.admit", lazyHandler(() => import("./collaboration/steward_managed_mode.ts"), ({admitStewardManagedTurn}) => admitStewardManagedTurn)], + ["collaboration.session.managed_turn.admit", lazyHandler(() => import("./collaboration/steward_managed_mode.ts"), ({admitManagedSessionTurn}) => admitManagedSessionTurn)], ["collaboration.steward.monitor.plan", lazyHandler(() => import("./collaboration/steward_managed_mode.ts"), ({planStewardMonitor}) => planStewardMonitor)], ["collaboration.steward.report.plan", lazyHandler(() => import("./collaboration/steward_reports.ts"), ({planStewardReport}) => planStewardReport)], ["collaboration.steward.monitor.observation", lazyHandler(() => import("./collaboration/steward_managed_mode.ts"), ({observeStewardPortfolio}) => observeStewardPortfolio)], diff --git a/tests/test_native_steward_turn_host.py b/tests/test_native_steward_turn_host.py index 689d0bca2d..7a2d6edd73 100644 --- a/tests/test_native_steward_turn_host.py +++ b/tests/test_native_steward_turn_host.py @@ -3,6 +3,7 @@ import io import json import sys +from pathlib import Path import pytest @@ -15,6 +16,186 @@ steward = steward_fixture +def managed_fixture(steward, tmp_path, monkeypatch, provider="file", *, validator=True): + isolate_sqlite_runtime(tmp_path, monkeypatch) + sid, managed = original(steward) + storage(steward[1], provider) + binding = steward[4] + 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"] + store, runtime, _, _, _, capture, _ = steward + from loopx.agent_registry import load_goal_from_registry + from loopx.capabilities.native_chat.steward_managed_mode import ManagedSessionTurn + goal = load_goal_from_registry(runtime.registry_path, actor["goal_id"]) + workspace = Path(goal["repo"]) + session, _ = runtime.open_session(goal_id=actor["goal_id"], agent_id="codex", work_dir=workspace, + objective="Observe the original task", channel_id=f"goal.{actor['goal_id']}", mode="new") + turn, _ = runtime.submit_turn(session_id=session["session_id"], client_turn_id="original-question", + message="Inspect this source and retain our original conversation.", work_dir=workspace, objective="Observe the original task") + assert finish(runtime, turn)["status"] == "completed" + observation = workspace / "managed-source.txt" + observation.write_text("ready") + declaration = [sys.executable, "-c", "from pathlib import Path; assert Path('managed-source.txt').read_text() == 'ready'"] + added = cli(runtime, "todo", "add", "--goal-id", actor["goal_id"], "--role", "agent", + "--text", "Validate the readiness source before completing this task.", "--claimed-by", actor["agent_id"], + "--action-kind", "validate", "--continuation-policy", "same_agent_non_delivery", + *(["--validation-command-json", json.dumps(declaration)] if validator else []), "--execute") + fake = Path(runtime.codex_bin) + candidate = {"result_kind": "validated_completion", "classification": "implementation", + "summary": "The synthetic readiness source is ready.", "next_action": "Continue the original authorized work.", + "delivery_batch_scale": "implementation", "delivery_outcome": "outcome_progress", + "vision_unchanged_reason": "The original task remains the current route.", + "work_evidence": ["managed-source.txt contains the original readiness signal"], + "validation_evidence": ["The task's independent validator checks the original readiness condition"]} + source = fake.read_text().replace('"message": "Runtime response."', '"message": ' + repr(json.dumps(candidate))) + fake.write_text(source) + runtime.adapters.pop(session["session_id"]).close_session() + return ManagedSessionTurn(runtime), session, actor, added["todo_id"], observation, capture + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_local_managed_session_reuses_native_turn_and_declared_validation(steward, tmp_path, monkeypatch, provider): + service, session, actor, todo_id, observation, capture = managed_fixture(steward, tmp_path, monkeypatch, provider) + store, runtime, *_ = steward + sid = session["session_id"] + before_sessions = store.list_sessions() + before_calls, thread = capture.read_bytes(), session["upstream_thread_id"] + before_history = store.messages(sid) + assert len(before_history) == 2 + assert service.inspect(sid)["available"] + accepted, created = service.submit(sid, agent_id=actor["agent_id"], operation_id="local-native-1", todo_id=todo_id) + assert created + completed = finish(runtime, accepted) + assert completed["status"] == "completed", completed + result = completed["response"]["native_loopx"] + assert result.get("status") == "committed", json.dumps(result) + assert result["effects"]["host_invoked"] and result["effects"]["quota_spent"] + assert result["validation"]["status"] == "passed" + assert observation.read_text() == "ready" + assert store.load_session(sid)["upstream_thread_id"] == thread + assert {row["session_id"] for row in store.list_sessions()} == {row["session_id"] for row in before_sessions} + assert store.messages(sid)[:len(before_history)] == before_history + calls = [json.loads(row) for row in capture.read_bytes()[len(before_calls):].splitlines()] + assert runtime.adapters[sid].session.sandbox == "read-only" + assert all(row.get("params", {}).get("sandbox", "read-only") == "read-only" for row in calls) + assert not [row for row in calls if row["method"] in {"thread/start", "thread/goal/set", "thread/inject_items"}] + after_calls = capture.read_bytes() + replay, created = service.submit(sid, agent_id=actor["agent_id"], operation_id="local-native-1", todo_id=todo_id) + assert not created and replay["turn_id"] == accepted["turn_id"] and capture.read_bytes() == after_calls + assert service.inspect(sid, operation_id="local-native-1")["turn"]["turn_id"] == accepted["turn_id"] + with pytest.raises(ValueError, match="identity conflict"): + service.submit(sid, agent_id=actor["agent_id"], operation_id="local-native-1", todo_id="todo_other") + + +@pytest.mark.parametrize("fault", ["busy", "queued", "attached", "foreign_goal", "stopped_goal", "private", "native_goal", "unregistered", "host_closed"]) +def test_local_managed_session_denials_never_launch_or_accept(steward, tmp_path, monkeypatch, fault): + service, session, actor, todo_id, _, capture = managed_fixture(steward, tmp_path, monkeypatch) + store, runtime, *_ = steward + sid = session["session_id"] + changes = {"busy": {"active_turn_id": "f" * 32}, + "native_goal": {"upstream_mode": "goal"}} + if fault in changes: + store.update_session(sid, **changes[fault]) + elif fault == "queued": + store.create_queued_turn(sid, client_turn_id="human-first", message="Keep my pending work first", origin="web") + elif fault in {"stopped_goal", "foreign_goal"}: + registry = json.loads(runtime.registry_path.read_text()) + registry["goals"][0]["status" if fault == "stopped_goal" else "goal_instance_id"] = "stopped" if fault == "stopped_goal" else "a-replaced-lifetime" + runtime.registry_path.write_text(json.dumps(registry)) + elif fault == "attached": + sid = store.create_session(goal_id=session["goal_id"], agent_id="codex", adapter_kind="codex_app_server", + upstream_thread_id="existing-attached-thread", session_mode="attached_host", host_surface="codex_app")["session_id"] + elif fault == "private": + sid = next(row["session_id"] for row in store.list_sessions() if row.get("goal_id") == "loopx-manager") + elif fault == "host_closed": + runtime.close() + before_calls, before_history = capture.read_bytes(), store.messages(sid) + with pytest.raises(ValueError): + service.submit(sid, agent_id="unregistered" if fault == "unregistered" else actor["agent_id"], + operation_id="denied", todo_id=todo_id) + assert capture.read_bytes() == before_calls and store.messages(sid) == before_history + + +@pytest.mark.parametrize("fault", ["missing_validator", "failed_validator", "stopped_after_model"]) +def test_managed_work_never_commits_without_current_independent_postcondition(steward, tmp_path, monkeypatch, fault): + service, session, actor, todo_id, observation, capture = managed_fixture(steward, tmp_path, monkeypatch, + validator=fault != "missing_validator") + store, runtime, *_ = steward + if fault == "failed_validator": + observation.write_text("not ready") + if fault == "stopped_after_model": + from loopx.capabilities.native_chat.codex_adapter import CodexAppServerAdapter + original_start = CodexAppServerAdapter.start_turn + + def stopped(adapter, message, sink): + response = original_start(adapter, message, sink) + registry = json.loads(runtime.registry_path.read_text()) + registry["goals"][0]["status"] = "stopped" + runtime.registry_path.write_text(json.dumps(registry)) + return response + + monkeypatch.setattr(CodexAppServerAdapter, "start_turn", stopped) + calls = capture.read_bytes() + accepted, _ = service.submit(session["session_id"], agent_id=actor["agent_id"], operation_id="unqualified", todo_id=todo_id) + completed = finish(runtime, accepted) + native = completed["response"]["native_loopx"] + assert native["status"] == ("stopped" if fault == "missing_validator" else "failed") and not native["effects"]["quota_spent"] + assert not native["effects"]["state_written"] + if fault == "missing_validator": + assert observation.read_text() == "ready" + assert not [json.loads(row) for row in capture.read_bytes()[len(calls):].splitlines() if json.loads(row)["method"] == "turn/start"] + after = capture.read_bytes() + assert not service.submit(session["session_id"], agent_id=actor["agent_id"], operation_id="unqualified", todo_id=todo_id)[1] + assert capture.read_bytes() == after + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_managed_turn_http_correlates_readback_without_accepting_commands(steward, tmp_path, monkeypatch, provider): + import http.client + import threading + from loopx.chat_server import ChatHTTPServer, ChatRequestHandler + + service, session, actor, todo_id, _, capture = managed_fixture(steward, tmp_path, monkeypatch, provider) + store, runtime, *_ = steward + server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler) + server.runtime_controller, server.chat_store, server.verbose = runtime, store, False + threading.Thread(target=server.serve_forever, daemon=True).start() + route = f"/api/chat/sessions/{session['session_id']}/native-turn" + + def request(body=None, suffix="", origin=None): + conn = http.client.HTTPConnection(*server.server_address, timeout=20) + try: + conn.request("GET" if body is None else "POST", route + suffix, + None if body is None else json.dumps(body), headers={"Content-Type": "application/json", + **({"Origin": origin} if origin else {})}) + result = conn.getresponse() + return result.status, json.loads(result.read()) + finally: + conn.close() + + try: + before = capture.read_bytes() + assert request()[1]["available"] + body = {"agent_id": actor["agent_id"], "operation_id": "http-original", "todo_id": todo_id} + assert request(body, origin="https://foreign.example")[0] == 403 + assert request({**body, "validation_command": ["arbitrary-command"]})[0] == 409 + assert capture.read_bytes() == before + status, accepted = request(body) + assert status == 202 and accepted["created"] + completed = finish(runtime, accepted["turn"]) + assert completed["response"]["native_loopx"]["status"] == "committed" + after = capture.read_bytes() + observed = request(suffix="?operation_id=http-original")[1]["turn"] + assert observed["turn_id"] == completed["turn_id"] and observed["status"] == "completed" + assert request(body)[1]["created"] is False + assert request({**body, "todo_id": "todo_different"})[0] == 409 + assert capture.read_bytes() == after + finally: + server.shutdown() + server.server_close() + + def test_native_task_validator_cannot_replace_a_cli_host(tmp_path): from loopx.cli import build_parser from loopx.cli_commands.turn import handle_turn_command