diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index 7407792c25..0f6b66bc44 100644 --- a/docs/product/periodic-reports.md +++ b/docs/product/periodic-reports.md @@ -280,6 +280,18 @@ priority, replan, budget, stop and settlement policies remain effective. Pausing the steward prevents new scheduled model Turns without discarding accepted work or historical results. +If the native guard refuses the exact report Task before execution because a +semantic replan is due, the original idle worker can admit one correlated +continuation of that window. It first runs the original Agent's native replan, +without choosing another business Task. The canonical writeback preview must +independently accept its semantic delta before writeback and quota settlement; +only a committed replan permits a fresh admission of the retained report Task. +Ordinary delivery remains withheld during governance. The original return +reader then verifies the full report through its existing publication owner. +The refused Chat request remains in history. Unknown execution, partial effects, +a frozen generation, changed authorization or an unsuccessful continuation do +not authorize another model replay, window or message consumer. + A native status watch calls the existing monitor-poll owner under its original Chat Turn and task lease. It does not select a continuous monitor as a heartbeat advancement task, so open executable work does not prevent a due observation. diff --git a/loopx/capabilities/native_chat/steward_managed_mode.py b/loopx/capabilities/native_chat/steward_managed_mode.py index 99c46f95ff..4d1d61fba7 100644 --- a/loopx/capabilities/native_chat/steward_managed_mode.py +++ b/loopx/capabilities/native_chat/steward_managed_mode.py @@ -252,7 +252,8 @@ def _request(self, session_id: str, agent_id: str, operation_id: str, "work_kind": "native_turn", "validation_command": None} def admit(self, request: dict[str, Any], *, turn: dict[str, Any] | None = None, - settlement_recovery: dict[str, Any] | None = None) -> dict[str, Any]: + settlement_recovery: dict[str, Any] | None = None, + selection_recovery: dict[str, Any] | None = None) -> dict[str, Any]: from .conversation_bindings import ChatConversationBindings if self.controller.closed.is_set(): raise ValueError("the original Chat host is stopped") @@ -282,6 +283,7 @@ def admit(self, request: dict[str, Any], *, turn: dict[str, Any] | None = None, "goal_active": bool(goal and goal.get("status") == "active" and not goal_is_stopped(goal)), "monitor_todo": monitor, **report, **({"report_settlement_recovery": settlement_recovery} if settlement_recovery is not None else {}), + **({"report_selection_recovery": selection_recovery} if selection_recovery is not None else {}), "queued_turn_count": len(self.store.queued_turns(actor["session_id"]))}) def inspect(self, session_id: str, *, agent_id: str | None = None, @@ -339,6 +341,8 @@ def submit(self, session_id: str, *, agent_id: str, operation_id: str, request.update(work_kind="monitor", monitor_deadline=monitor_deadline, monitor_lease_version=retained["version"] if retained else 0) existing = self.store.turn_for_client(session_id, "managed." + operation_id) + selection_recovery = None + client_id = "managed." + operation_id if existing is not None: frozen = dict(existing.get("loopx_request") or {}) if monitor_deadline is not None: @@ -346,12 +350,19 @@ def submit(self, session_id: str, *, agent_id: str, operation_id: str, request["monitor_lease_version"] = frozen.get("monitor_lease_version") if frozen != request: raise ValueError("managed Turn request identity conflict; inspect the original operation") - return existing, False # Failed/unknown work is never replayed. + from .steward_turn import StewardTurnHost + selection_recovery = StewardTurnHost(controller).report_selection_recovery(existing) if report_window is not None else None + if selection_recovery is None: + return existing, False # Failed/unknown work is never replayed. + client_id += ".admission-retry" + retry = self.store.turn_for_client(session_id, client_id) + if retry is not None: + return retry, False 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, + admitted = self.admit(request, selection_recovery=selection_recovery) + accepted = self.store.accept_managed_turn(session_id, client_turn_id=client_id, message="Advance the currently admitted native LoopX work.", display_message="生成本期定时报告。" if report_window is not None else "执行本期定时任务。" if scheduled_window is not None else diff --git a/loopx/capabilities/native_chat/steward_turn.py b/loopx/capabilities/native_chat/steward_turn.py index 4174e127fd..e5261b2c58 100644 --- a/loopx/capabilities/native_chat/steward_turn.py +++ b/loopx/capabilities/native_chat/steward_turn.py @@ -46,12 +46,14 @@ def _monitor_selection_failed_before_poll(self, turn: dict[str, Any]) -> bool: def _admit(self, request: dict[str, Any], *, turn: dict[str, Any] | None = None, prepare_context: bool = False, settlement_recovery: dict[str, Any] | None = None, - launch_recovery: dict[str, Any] | None = None) -> dict[str, Any]: + launch_recovery: dict[str, Any] | None = None, + selection_recovery: dict[str, Any] | None = None) -> dict[str, Any]: 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, settlement_recovery=settlement_recovery) + return ManagedSessionTurn(self.controller).admit(request, turn=turn, + settlement_recovery=settlement_recovery, selection_recovery=selection_recovery) if self.bindings is None: raise ValueError("the original steward binding is unavailable") session = self.store.load_session(actor["session_id"]) @@ -93,11 +95,29 @@ def scope_id(ids): "request": request, "turn": turn, "scope": scope, "monitor_todo": monitor, **({"report_settlement_recovery": settlement_recovery} if settlement_recovery is not None else {}), **({"report_launch_recovery": launch_recovery} if launch_recovery is not None else {}), + **({"report_selection_recovery": selection_recovery} if selection_recovery is not None else {}), **({"preparing_model_context": True} if prepare_context else {}), **report}) if admitted["refresh_original_scope"]: self.store.update_session(session["session_id"], manager_authorization_scope_id=scope["current"]) return {**admitted, "observed_scope_id": scope["current"] if scope else None} + def report_selection_recovery(self, original: dict[str, Any]) -> dict[str, Any] | None: + """The typed owner qualifies this observation before any new worker.""" + result = (original.get("response") or {}).get("native_loopx") or {} + refusal = result.get("selection_rejection") + if (result.get("error_code") != "turn_todo_selection_deferred" or not isinstance(refusal, Mapping) + or refusal.get("reason_code") != "autonomous_replan" + or refusal.get("recovery_action") != "reenter_guard_without_selection"): + return None + from ..periodic_report.private_generation import attempt_directory + from .steward_reports import report_service + request = original["loopx_request"] + reports, owner = report_service(self.controller, request["actor"]) + _, window = reports.current_window(owner) + return {"original": original, "generation_present": window is None + or window["window_id"] != request["report_window_id"] + or (attempt_directory(reports.root, window) / "private-generation.json").exists()} + def _report_launch_recovery(self, original: dict[str, Any]) -> dict[str, Any] | None: """Read the original typed launch refusal and its canonical transaction. @@ -174,14 +194,15 @@ def _task_lease_proof(self, request: Mapping[str, Any], lease: Mapping[str, Any] "task_lease_expected_version": observed["version"]} @contextmanager - def _task_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, + admission_retry: bool = False): """Both generation and cached validation use the original Todo owner.""" from .steward_commands import native_command actor = request["actor"] retained = native_command(self.controller, ["task-lease", "inspect", "--goal-id", actor["goal_id"], "--todo-id", request["todo_id"]]).get("lease") version = retained["version"] if retained else 0 - suffix = ".validation-recovery." + str(version) if recovery else ".admission-retry" if request.get("resume_turn_key") else "" + suffix = ".validation-recovery." + str(version) if recovery else ".admission-retry" if request.get("resume_turn_key") or admission_retry else "" lease = native_command(self.controller, ["task-lease", "acquire", "--goal-id", actor["goal_id"], "--todo-id", request["todo_id"], "--owner", actor["agent_id"], "--idempotency-key", request["turn_instance_id"] + suffix, "--ttl-seconds", "3600", @@ -288,6 +309,7 @@ def submit(self, binding_id: str, *, turn_instance_id: str, todo_id: str | None client_id = native_steward_client_id(actor, turn_instance_id) existing = self.store.turn_for_client(actor["session_id"], client_id) launch_recovery = None + selection_recovery = None if existing is not None: # Never replay a failed/unknown model Turn. A committed Chat # request is observed; native journal recovery remains separate. @@ -307,10 +329,12 @@ def submit(self, binding_id: str, *, turn_instance_id: str, todo_id: str | None if retry is not None: return retry, False if work_kind == "report": - launch_recovery = self._report_launch_recovery(existing) - if launch_recovery is None: - return existing, False - request["resume_turn_key"] = existing["response"]["native_loopx"].get("resume_turn_key") + selection_recovery = self.report_selection_recovery(existing) + if selection_recovery is None: + launch_recovery = self._report_launch_recovery(existing) + if launch_recovery is None: + return existing, False + request["resume_turn_key"] = existing["response"]["native_loopx"].get("resume_turn_key") if work_kind == "monitor": from .steward_commands import native_command if not todo_id or not monitor_deadline: @@ -319,7 +343,7 @@ def submit(self, binding_id: str, *, turn_instance_id: str, todo_id: str | None "--todo-id", todo_id]).get("lease") request.update(monitor_deadline=monitor_deadline, monitor_lease_version=retained["version"] if retained else 0) - admitted = self._admit(request, launch_recovery=launch_recovery) + admitted = self._admit(request, launch_recovery=launch_recovery, selection_recovery=selection_recovery) session = self.store.load_session(actor["session_id"]) accepted = self.store.accept_managed_turn(actor["session_id"], client_turn_id=client_id, message="Advance the currently admitted native LoopX work.", @@ -353,7 +377,13 @@ def execute(self, *, session_id: str, turn_id: str, adapter: Any, turn = self.store.load_turn(session_id, turn_id) request = turn["loopx_request"] - admitted = self._admit(request, turn=turn) + original_client = ("managed." + request["turn_instance_id"].removeprefix("managed-session.") + if request["actor"].get("schema_version") == "loopx_managed_session_actor_v0" + else native_steward_client_id(request["actor"], request["turn_instance_id"])) + original = self.store.turn_for_client(session_id, original_client) if request.get("work_kind") == "report" else None + selection_recovery = (self.report_selection_recovery(original) + if original and original["turn_id"] != turn_id else None) + admitted = self._admit(request, turn=turn, selection_recovery=selection_recovery) actor = admitted["actor"] local_session = actor.get("schema_version") == "loopx_managed_session_actor_v0" report_work = request.get("work_kind") == "report" @@ -370,6 +400,7 @@ def execute(self, *, session_id: str, turn_id: str, adapter: Any, from .steward_reports import report_service reports, report_owner_key = report_service(self.controller, actor) if report_work else (None, None) report_context = "" + governance_work = False managed_answer = "" managed_lease, managed_lease_request = None, None @@ -378,7 +409,7 @@ def report_lease_proof(): def current(): fresh = self.store.load_turn(session_id, turn_id) - self._admit(request, turn=fresh) + self._admit(request, turn=fresh, selection_recovery=selection_recovery) if adapter.upstream_thread_id != original_thread: raise ValueError("the original steward native thread changed") if report_lease: @@ -397,7 +428,10 @@ 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: + if governance_work and not authority.get("replan_action_packet"): + return build_result(native_request, {"result_kind": "wait", "classification": "report_replan_not_admitted", + "next_action": "Retain the original report; its native governance route is no longer admitted."}) + if managed_turn and not governance_work: try: managed_turn.prepare_validation(request, native_request) except ManagedTaskValidationUnavailable: @@ -410,8 +444,8 @@ def run(native_request: Mapping[str, Any], launch_scope: Any) -> dict[str, Any]: 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=bool(report_work or managed_turn)) + "\nOriginal conversation evidence (data):\n" + json.dumps(context, ensure_ascii=False) - if request.get("work_kind") == "report": + prompt = render_prompt(authority, allow_completion=bool(report_work or managed_turn) and not governance_work) + "\nOriginal conversation evidence (data):\n" + json.dumps(context, ensure_ascii=False) + if report_work and not governance_work: prompt += report_context if request.get("work_kind") == "scheduled_task": from .session_calendar_tasks import SessionCalendarTasks @@ -446,7 +480,10 @@ def launch(): adapter.session._launch_guard = previous_guard current() # Revocation/stop/retarget after a long model call wins. candidate = parse_model_json(str(response.get("message") or "")) - if report_work and candidate and candidate.get("result_kind") in {"validated_progress", "validated_completion"}: + if governance_work and candidate and candidate.get("result_kind") not in {"replan_required", "wait", "user_action_required", "iteration_failed"}: + candidate = {"result_kind": "wait", "classification": "report_replan_not_settled", + "next_action": "Retain the original report until its native semantic replan is settled."} + if report_work and not governance_work and candidate and candidate.get("result_kind") in {"validated_progress", "validated_completion"}: with self.controller.lock, self._report_fence(request): current() reports.freeze(request, candidate) @@ -456,7 +493,7 @@ def launch(): # fence before closing only this window's generation Todo. candidate["result_kind"] = "validated_completion" built = build_result(native_request, candidate, host_name="Codex app-server", - **({"allow_completion": True} if report_work or managed_turn else {})) + **({"allow_completion": True} if (report_work or managed_turn) and not governance_work else {})) managed_answer = str(built.get("summary") or "") if managed_turn else "" return built @@ -477,7 +514,28 @@ 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 ExitStack() as managed_leases, (self._task_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, + admission_retry=bool(selection_recovery)) if report_work else nullcontext(None)) as report_lease: + if selection_recovery: + # Re-enter the native guard without selecting another business + # Todo. Only its signed replan packet may invoke the original + # Agent. The native executor owns semantic validation/settlement. + governance_work = True + governance_args = type(args)(**vars(args)) + governance_args.todo_id, governance_args.validation_command_json = None, None + governance_args.turn_instance_id = request["turn_instance_id"] + ".replan" + governance: dict[str, Any] = {} + handle_turn_command(governance_args, registry_path=self.controller.registry_path, + runtime_root_arg=str(self.controller.coordination_runtime_root), + output_format=lambda *_a, **_kw: "json", print_payload=lambda value, *_a: governance.update(value), + native_host_runner=run, native_session_binding_resolver=session_binding) + self.store.append_event(session_id, turn_id, kind="native_loopx.replan_result", payload=governance) + if governance.get("status") != "committed" or governance.get("result_kind") != "replan_required": + retained = selection_recovery["original"]["response"]["native_loopx"] + self.store.append_event(session_id, turn_id, kind="native_loopx.result", payload=retained) + return {"schema_version": "loopx_chat_agent_response_v0", "message": "原生重规划尚未完成；保留原报告及执行证据。", + "proposals": [], "gate": None, "native_loopx": retained} + governance_work = False if report_work: # Portfolio collection can take minutes and legitimately see # registry updates. Finish it before the native driver captures diff --git a/loopx/cli_commands/turn_decision.py b/loopx/cli_commands/turn_decision.py index 8610fd29ef..deba7b0095 100644 --- a/loopx/cli_commands/turn_decision.py +++ b/loopx/cli_commands/turn_decision.py @@ -31,6 +31,7 @@ periodic_report_pending_intent_interaction_hook, ) from ..control_plane.quota.live_decision import build_live_quota_should_run_decision +from ..control_plane.quota.effective_action import EffectiveAction from ..control_plane.quota.turn_envelope import build_turn_envelope from ..control_plane.quota.settlement import SettlementIdentity, build_turn_scoped_cli_settlement_plan from ..control_plane.quota.heartbeat_receipt import ( @@ -64,8 +65,16 @@ def _attach_original_replan_settlement( to the same shared settlement owner, not to a made-up Todo or Host session. """ obligation = replan_obligation_id_from_packet(decision.get("replan_action_packet")) - if (not turn_instance_id or not obligation or decision.get("selected_todo") - or decision.get("should_run") is not True): + if not turn_instance_id or not obligation or decision.get("should_run") is not True: + return decision + agent = (decision.get("interaction_contract") or {}).get("agent_channel") or {} + if (agent.get("must_attempt") is True and agent.get("delivery_allowed") is False + and decision.get("effective_action") == EffectiveAction.AUTONOMOUS_REPLAN_REQUIRED.value): + # A fallback business Todo is context, not the selected owner of a + # required semantic replan. Bind the existing obligation so launch, + # writeback and spend all retain that same governance identity. + decision.pop("selected_todo", None) + if decision.get("selected_todo"): return decision cli = (decision.get("interaction_contract") or {}).get("cli_channel") if not isinstance(cli, dict) or cli.get("settlement_plan"): diff --git a/loopx/cli_commands/turn_run_once.py b/loopx/cli_commands/turn_run_once.py index 8d48263823..995c837cdd 100644 --- a/loopx/cli_commands/turn_run_once.py +++ b/loopx/cli_commands/turn_run_once.py @@ -250,7 +250,7 @@ def append_settlement_event( project if isinstance( writeback_contract.get("delivery_workspace_causality"), dict - ) + ) and not (settlement_identity.replan_obligation_id and not selected_todo) else None ) completion_delivery_workspace = ( @@ -265,14 +265,13 @@ def append_settlement_event( else None ) - def writeback( + def refresh_candidate( result: dict[str, object], *, completion_todo_id: str | None = None, completion_turn_key: str | None = None, - effect_ref: str, + dry_run: bool = False, ) -> dict[str, object]: - require_effect_ref(effect_ref, SettlementStepKind.DURABLE_WRITEBACK) # The host workspace is execution context, not state authority. state_project = None result_kind = str(result.get("result_kind") or "") @@ -285,7 +284,7 @@ def writeback( # Diagnostics already live in the journal's typed host_result. # Do not overwrite task context before refresh can reject a result. progress_observation = result.get("progress_observation") - refresh = refresh_state_run( + return refresh_state_run( registry_path=registry_path, runtime_root_override=runtime_root_arg, goal_id=args.goal_id, @@ -320,10 +319,34 @@ def writeback( ), completion_todo_id=completion_todo_id, completion_turn_key=completion_turn_key, - dry_run=False, + dry_run=dry_run, sync_global=not bool(args.no_global_sync), goal_ref=goal_ref, ) + + # A Todo-less native governance Turn has no business artifact/argv. + # Its independent validator is the canonical semantic writeback preview, + # with the same captured obligation and fresh authority as the commit. + # Never treat a model's progress/completion assertion as that validation. + if task_validator is None and settlement_identity.replan_obligation_id and not selected_todo: + def validate_replan(_plan: Mapping[str, Any], result: Mapping[str, Any]) -> dict[str, Any]: + passed = (result.get("result_kind") == "replan_required" + and refresh_candidate(dict(result), dry_run=True).get("ok") is True) + return {"status": "passed" if passed else "failed", "validator_kind": "callback", + "summary": "independent native semantic replan preview " + ("passed" if passed else "failed"), + **({"recovery_kind": "repair_required"} if not passed else {})} + task_validator = validate_replan + + def writeback( + result: dict[str, object], + *, + completion_todo_id: str | None = None, + completion_turn_key: str | None = None, + effect_ref: str, + ) -> dict[str, object]: + require_effect_ref(effect_ref, SettlementStepKind.DURABLE_WRITEBACK) + refresh = refresh_candidate(result, completion_todo_id=completion_todo_id, + completion_turn_key=completion_turn_key) if refresh.get("ok") and ( refresh.get("appended") or refresh.get("idempotent_replay") diff --git a/loopx/control_plane/collaboration/steward_managed_mode.ts b/loopx/control_plane/collaboration/steward_managed_mode.ts index cdb7f136e2..5e9594dfbf 100644 --- a/loopx/control_plane/collaboration/steward_managed_mode.ts +++ b/loopx/control_plane/collaboration/steward_managed_mode.ts @@ -10,6 +10,37 @@ import { requireJsonObject, requireNonEmptyString } from "../runtime_decode.ts"; import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; import { ENVELOPED_SHA256_PATTERN } from "../content_digest.ts"; import {goalAgentConversationIdentity, resolveConversationScope} from "./conversation_scope.ts"; +import {readTurnSelectionRejection} from "../turn_driver/selection_rejection.ts"; + +/** A native refusal before execution can owe governance, never a business replay. */ +function admitReportSelectionRecovery(params: JsonObject, request: JsonObject, sessionId: unknown): void { + if (params.report_selection_recovery == null) return; + if (request.work_kind !== "report") + throw new EffectRuntimeRequestError("selection recovery belongs only to its original report"); + const proof = requireJsonObject(params.report_selection_recovery, "original selection refusal"); + const original = requireJsonObject(proof.original, "original refused Chat Turn"); + const prior = requireJsonObject(original.loopx_request, "original refused request"); + const result = requireJsonObject(requireJsonObject(original.response, "original response").native_loopx, "native refusal"); + const refusal = readTurnSelectionRejection(result.selection_rejection, request.todo_id); + const effects = requireJsonObject(result.effects, "original invocation effects"); + if (original.session_id !== sessionId || original.origin !== "native_loopx" + || original.loopx_execution !== true || original.status !== "completed" + || original.delta_count !== 0 || original.upstream_turn_id !== null + || Object.keys(prior).length !== Object.keys(request).length + || Object.keys(request).some(key => JSON.stringify(prior[key]) !== JSON.stringify(request[key])) + || result.schema_version !== "loopx_turn_execution_v0" || result.ok !== false + || result.mode !== "run_once" || result.effects_scope !== "current_invocation" + || result.error_code !== "turn_todo_selection_deferred" || result.resume_turn_key != null + || refusal.state !== "deferred" || refusal.reason_code !== "autonomous_replan" + || refusal.recovery_action !== "reenter_guard_without_selection" + || refusal.status_health_ok !== true || refusal.contract_error_count !== 0 + || !(refusal.delivery_preemptions as unknown[]).includes("autonomous_replan") + || Object.keys(effects).length !== 4 + || ["host_invoked", "state_written", "quota_spent", "scheduler_acknowledged"].some(key => effects[key] !== false) + || proof.generation_present !== false) { + throw new EffectRuntimeRequestError("report selection recovery requires the original proven pre-execution replan refusal"); + } +} /** Registration and current Goal lifetime are observed by the host, never by * a message or the selected executor/model. This creates no scheduling grant. */ @@ -152,6 +183,7 @@ export function admitManagedSessionTurn(params: JsonObject): JsonObject { : 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"); } + admitReportSelectionRecovery(params, request, session.session_id); return {actor, workspace_path: goal.workspace_path, host: "codex-app-server", observation_only: observationOnly}; } @@ -269,6 +301,7 @@ export function admitStewardManagedTurn(params: JsonObject): JsonObject { || request.validation_command.some(value => typeof value !== "string" || !value || value.includes("\0") || value.length > 10000))) { throw new EffectRuntimeRequestError("the host's native validator must be a bounded argv"); } + admitReportSelectionRecovery(params, request, current.sessionId); if (turn && (turn.session_id !== current.sessionId || turn.turn_id !== current.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))) { diff --git a/loopx/control_plane/turn_driver/driver.py b/loopx/control_plane/turn_driver/driver.py index 20321aa457..0e9889a56e 100644 --- a/loopx/control_plane/turn_driver/driver.py +++ b/loopx/control_plane/turn_driver/driver.py @@ -160,7 +160,18 @@ def _typed_route(envelope: Mapping[str, Any]) -> LoopXTurnRoute: must_attempt = action.get("must_attempt") is True if should_run: - if not delivery_allowed or not must_attempt: + try: + governance_allowed = ( + effective_action in REPLAN_ACTIONS + and not delivery_allowed + and turn_envelope_replan_settlement_identity(envelope, turn_instance_id=None) is not None + ) + except (ValueError, RuntimeError): + return LoopXTurnRoute.CONTRACT_ERROR + # Quota can require a semantic replan while withholding ordinary + # delivery. Honor that signed governance duty, without making any + # normal business route executable or replacing its original gates. + if not must_attempt or not (delivery_allowed or governance_allowed): return LoopXTurnRoute.BLOCKED try: registered_action = EffectiveAction(effective_action) diff --git a/tests/control_plane_ts/steward_reports.test.ts b/tests/control_plane_ts/steward_reports.test.ts index 31ffa96aab..8d30015e5c 100644 --- a/tests/control_plane_ts/steward_reports.test.ts +++ b/tests/control_plane_ts/steward_reports.test.ts @@ -288,3 +288,43 @@ test("a retained pre-publication launch refusal can resume only its original nat {...input, queued_turn_count: 1}, ]) assert.throws(() => admitStewardManagedTurn(invalid)); }); + +test("only a proven original pre-execution replan refusal admits one correlated report continuation", () => { + const p = original(); + const request = {operation: "native_turn", host: "codex-app-server", actor: p.actor, + turn_instance_id: "report.cadence_original", todo_id: "todo_report", validation_command: ["python", "validate.py"], + work_kind: "report", report_window_id: "cadence_original", report_subscription_revision: "sha256:original"}; + const refusal = {schema_version: "loopx_turn_selection_rejection_v0", source: "quota.should-run", + requested_todo_id: request.todo_id, state: "deferred", reason_code: "autonomous_replan", + recovery_action: "reenter_guard_without_selection", delivery_preemptions: ["autonomous_replan", "delivery_not_allowed"], + status_health_ok: true, contract_error_count: 0}; + const effects = {host_invoked: false, state_written: false, quota_spent: false, scheduler_acknowledged: false}; + const result = {schema_version: "loopx_turn_execution_v0", mode: "run_once", ok: false, + error_code: "turn_todo_selection_deferred", effects_scope: "current_invocation", effects, selection_rejection: refusal}; + const source = {session_id: p.session.session_id, status: "completed", origin: "native_loopx", loopx_execution: true, + delta_count: 0, upstream_turn_id: null, loopx_request: request, response: {native_loopx: result}}; + const proof = {original: source, generation_present: false}; + const input = {...p, request, report_selection_recovery: proof, + report_window: {window_id: request.report_window_id, subscription_revision: request.report_subscription_revision, + goal_id: p.actor.goal_id, agent_id: p.actor.agent_id}, + report_subscription: {enabled: true, profile_preset: "portfolio-progress", + route_ref: `steward-private:${p.binding.binding_id}:${p.source}`, effective_revision: request.report_subscription_revision}, + report_todo: {todo_id: request.todo_id, target_key: request.report_window_id, + claimed_by: "steward", action_kind: "generate_periodic_report", status: "open"}}; + assert.equal(admitStewardManagedTurn(input).observation_only, false); + for (const original of [ + {...source, status: "failed"}, {...source, delta_count: 1}, {...source, upstream_turn_id: "actual-model"}, + {...source, session_id: "other"}, {...source, loopx_request: {...request, todo_id: "todo_other"}}, + ...["host_invoked", "state_written", "quota_spent", "scheduler_acknowledged"].flatMap(key => [true, null].map(value => + ({...source, response: {native_loopx: {...result, effects: {...effects, [key]: value}}}}))), + ...[{state: "unavailable"}, {reason_code: "control_repair"}, {status_health_ok: false}, {contract_error_count: 1}, + {recovery_action: "inspect_original"}, {requested_todo_id: "todo_other"}].map(changed => + ({...source, response: {native_loopx: {...result, selection_rejection: {...refusal, ...changed}}}})), + {...source, response: {native_loopx: {...result, resume_turn_key: "sha256:" + "a".repeat(64)}}}, + ]) assert.throws(() => admitStewardManagedTurn({...input, report_selection_recovery: {...proof, original}})); + for (const changed of [ + {report_selection_recovery: {...proof, generation_present: true}}, + {report_subscription: {...input.report_subscription, enabled: false}}, {queued_turn_count: 1}, {goal_active: false}, + {request: {...request, work_kind: "native_turn"}}, + ]) assert.throws(() => admitStewardManagedTurn({...input, ...changed})); +}); diff --git a/tests/test_loopx_turn_driver.py b/tests/test_loopx_turn_driver.py index 6a8d28b103..1831446a96 100644 --- a/tests/test_loopx_turn_driver.py +++ b/tests/test_loopx_turn_driver.py @@ -1292,6 +1292,35 @@ def test_turn_plan_keeps_todoless_replan_for_controller(effective_action: str) - assert payload["turn_envelope"] == envelope +@pytest.mark.parametrize("hold", [None, "should_run", "must_attempt", "signature", "obligation", "business"]) +def test_signed_governance_can_run_while_normal_delivery_remains_withheld(hold): + from loopx.control_plane.quota.settlement import build_turn_scoped_cli_settlement_plan + obligation = "replan-" + "a" * 16 + settlement = build_turn_scoped_cli_settlement_plan(goal_id="fixture-goal", agent_id="codex-fixture", + replan_obligation_id=obligation, turn_instance_id="fixture.replan", command_prefix="loopx", + scoped_cli_args="", lifecycle_actor_args="").as_dict() + decision = {"ok": True, "goal_id": "fixture-goal", "agent_id": "codex-fixture", "decision": "run", + "should_run": hold != "should_run", "effective_action": "normal_run" if hold == "business" else "autonomous_replan_required", + "replan_action_packet": {"schema_version": "replan_action_packet_v0", "decision": "replan_required", "obligation_id": obligation}, + "interaction_contract": {"schema_version": "loopx_interaction_contract_v0", "mode": "autonomous_replan", + "user_channel": {"action_required": False, "notify": "DONT_NOTIFY"}, + "agent_channel": {"must_attempt": hold != "must_attempt", "delivery_allowed": False, "quiet_noop_allowed": False}, + "cli_channel": {"spend_after_validation": True, "settlement_plan": settlement}}} + envelope = build_turn_envelope(decision) + if hold == "signature": + envelope["action_signature"]["matches"] = False + if hold == "obligation": + envelope["replan_action_packet"]["obligation_id"] = "replan-" + "b" * 16 + plan = build_loopx_turn_plan(envelope, host="codex-app-server", execution_mode="isolated-headless", + session_scope="agent", turn_instance_id="fixture.replan") + assert plan["route"]["would_invoke_host"] is (hold is None) + assert plan["route"]["selected_todo"] is None + assert not any(plan["effects"].values()) + if hold is None: + assert plan["route"]["kind"] == "replan_required" + assert plan["transaction"]["settlement_plan"]["identity"]["replan_obligation_id"] == obligation + + @pytest.mark.parametrize("invalid_signature", [False, True]) def test_turn_plan_missing_todo_does_not_bypass_contract_errors( invalid_signature: bool, diff --git a/tests/test_native_steward_reports.py b/tests/test_native_steward_reports.py index e9a9265081..775d507a97 100644 --- a/tests/test_native_steward_reports.py +++ b/tests/test_native_steward_reports.py @@ -462,6 +462,147 @@ def install_model(steward, tmp_path, sid, *, result_kind="validated_completion") runtime.adapters.pop(sid).close_session() # Resume the same native thread with the synthetic response. +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("local_session", [False, True]) +@pytest.mark.parametrize("replan_outcome", ["settled", "wait", "invalid_semantic"]) +def test_report_pump_replans_only_proven_refusals_and_retains_unknown_work(steward, tmp_path, monkeypatch, provider, local_session, replan_outcome): + from datetime import timezone + from loopx.capabilities.periodic_report import cadence_runtime + from loopx.capabilities.native_chat.steward_monitor import StewardMonitor + import loopx.capabilities.native_chat.steward_reports as report_host + store, runtime, bot, transport, binding, capture, _ = steward + point = datetime.now(timezone.utc).replace(hour=9, minute=1, second=0, microsecond=0) - timedelta(days=2) + + class CalendarClock(datetime): + @classmethod + def now(cls, tz=None): + return point.astimezone(tz) if tz else point.replace(tzinfo=None) + monkeypatch.setattr(report_host, "datetime", CalendarClock) + monkeypatch.setattr(cadence_runtime, "datetime", CalendarClock) + if local_session: + session, reports, _, _ = configured_local_report(steward, tmp_path, monkeypatch, provider) + sid = owner = session["session_id"] + actor = reports.selected(sid)["actor"] + else: + sid, actor, reports, _ = configured(steward, tmp_path, monkeypatch, provider) + owner = binding["binding_id"] + from loopx.configure_goal import configure_goal + assert configure_goal(registry_path=runtime.registry_path, goal_id=actor["goal_id"], + execution_replan_after_turns=1, execute=True)["ok"] + install_model(steward, tmp_path, sid) + assert finish(runtime, pump_steward_monitors(runtime)[0])["response"]["native_loopx"]["status"] == "committed" + assert transport.steward_reports.reconcile() == (0 if local_session else 1) + assert reports.health(owner)["publication_verified"] + _, first_window = reports.current_window(owner) + first_folder = attempt_directory(reports.root, first_window) + first_body = (first_folder / "report.md").read_bytes() + point += timedelta(days=1) + runtime._steward_monitor_next_check.clear() + rejected = finish(runtime, pump_steward_monitors(runtime)[0]) + refusal = rejected["response"]["native_loopx"] + assert refusal["error_code"] == "turn_todo_selection_deferred", refusal + assert refusal["selection_rejection"]["reason_code"] == "autonomous_replan" + assert refusal["effects"] == {"host_invoked": False, "state_written": False, + "quota_spent": False, "scheduler_acknowledged": False} + original_bytes = store._turn_path(sid, rejected["turn_id"]).read_bytes() + _, second_window = reports.current_window(owner) + assert second_window["window_id"] != first_window["window_id"] + # The model is a transport oracle only. Native admission, the persisted + # missing-baseline trigger, semantic validation and settlement stay real. + vision = {"schema_version": "goal_vision_replan_contract_v0", "state": "active", + "vision_patch": {"vision_summary": "Maintain the original monitoring and reporting work.", + "acceptance_summary": "Independently validate each report and its full original private delivery; retain source gaps."}, + "path_delta": {"outcome": "replan", "prior_assumption": "The first report could keep an absent Agent vision unchanged.", + "observed_reality": "The first report was independently validated and delivered; native governance now requires a persisted baseline.", + "retained": ["The same subscriptions, report window, original actor and all native tasks."], + "changed": ["Record the original Agent's evidence-linked continuation before resuming the retained report."], + "evidence_refs": ["fixture:validated-original-report-and-provider-delivery"]}} + replan = {"result_kind": "replan_required", "classification": "original_report_native_replan", + "summary": "The original reporting route remains active with an evidence-linked baseline.", + "next_action": "Resume the retained native report task under a fresh exact selection.", + "delivery_batch_scale": "implementation", "delivery_outcome": "outcome_progress", + "work_evidence": ["The original report generation and independently read original private delivery."], + "validation_evidence": ["Native semantic validation checks the persisted baseline and current obligation."], + "path_delta_mode": "material_replan", "agent_vision_json": json.dumps(vision)} + if replan_outcome == "wait": + replan = {"result_kind": "wait", "classification": "fixture_missing_context", + "next_action": "Retain the original report until its required context is available."} + elif replan_outcome == "invalid_semantic": + vision["path_delta"]["evidence_refs"] = [] + replan["agent_vision_json"] = json.dumps(vision) + helper = tmp_path / "report_model.py" + helper.write_text(_REPORT_MODEL_SOURCE.replace("def fake_report_model(", "def report_only(") + + "\ndef fake_report_model(params):\n" + " import json\n" + " prompt = '\\n'.join(row.get('text', '') for row in params.get('input', []) if isinstance(row, dict))\n" + " authority = json.JSONDecoder().raw_decode(prompt.split('Turn authority JSON:', 1)[1].lstrip())[0]\n" + " if authority.get('replan_action_packet'):\n" + " return " + repr(json.dumps(replan)) + "\n" + " return report_only(params)\n") + calls, writes = capture.read_bytes(), list(bot.writes) + runtime._steward_monitor_next_check.clear() + resumed = pump_steward_monitors(runtime) + assert len(resumed) == 1, "The original idle producer must recover its proven pre-execution refusal" + done = finish(runtime, resumed[0]) + replans = [row["payload"] for row in store.events_after(sid, done["turn_id"], None) + if row["kind"] == "native_loopx.replan_result"] + assert len(replans) == 1 + if replan_outcome != "settled": + assert replans[0].get("status") != "committed" + assert replans[0]["effects"]["state_written"] is False and replans[0]["effects"]["quota_spent"] is False + assert not (attempt_directory(reports.root, second_window) / "private-generation.json").exists() + assert bot.writes == writes and transport.steward_reports.reconcile() == 0 + tail = [json.loads(line) for line in capture.read_bytes()[len(calls):].splitlines()] + assert sum(row.get("method") == "turn/start" for row in tail) == 1 + held = capture.read_bytes(), store.load_turn(sid, rejected["turn_id"]) + for _ in range(2): + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + assert (capture.read_bytes(), store.load_turn(sid, rejected["turn_id"])) == held + assert next(row for row in StewardMonitor(runtime).todos(actor) + if row["todo_id"] == rejected["loopx_request"]["todo_id"])["status"] == "open" + assert reports.current_window(owner)[1] == second_window + return + assert done.get("response") and done["response"]["native_loopx"].get("status") == "committed", { + "turn": done, "replan": [row["payload"] for row in store.events_after(sid, done["turn_id"], None) + if row["kind"] == "native_loopx.replan_result"]} + assert done["loopx_request"] == rejected["loopx_request"] + assert done["turn_id"] != rejected["turn_id"] + assert store._turn_path(sid, rejected["turn_id"]).read_bytes() == original_bytes + assert reports.current_window(owner)[1] == second_window + assert replans[0]["result_kind"] == "replan_required" and replans[0]["status"] == "committed" + assert replans[0]["validation"]["summary"] == "independent native semantic replan preview passed" + assert replans[0]["receipt"]["lineage"]["todo_id"] == "" + assert replans[0]["receipt"]["lineage"]["replan_obligation_id"] + task = next(row for row in StewardMonitor(runtime).todos(actor) + if row["todo_id"] == done["loopx_request"]["todo_id"]) + assert task["status"] == "done" + # The existing no-model recovery owner attaches the native commit to the + # original return reader; the refused outer Chat request stays immutable. + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + assert transport.steward_reports.reconcile() == (0 if local_session else 1) + assert reports.health(owner)["publication_verified"] + assert (first_folder / "report.md").read_bytes() == first_body + folder = attempt_directory(reports.root, second_window) + body = (folder / "report.md").read_text() + if local_session: + assert bot.writes == writes + assert len([row for row in store.messages(sid) if row["role"] == "agent" and row["text"] == body]) == 1 + else: + proof = json.loads((folder / "private-delivery.json").read_text())["deliveries"]["periodic_report." + second_window["window_id"]] + assert proof["verified"] and proof["text"] == body.strip() + assert len(bot.writes) > len(writes) # The original sender may split a full report. + tail = [json.loads(line) for line in capture.read_bytes()[len(calls):].splitlines()] + assert sum(row.get("method") == "turn/start" for row in tail) == 2 + assert not any(row.get("method") == "thread/start" for row in tail) + after = capture.read_bytes(), list(bot.writes) + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + assert transport.steward_reports.reconcile() == 0 + assert (capture.read_bytes(), bot.writes) == after + + def embedded_report_python(tmp_path, monkeypatch): """A real child interpreter without the embedding host's LoopX bootstrap.""" import subprocess