From 05e010d6d68eb5c602ea332a778aa25213b0c8c2 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 25 Oct 2026 17:49:12 +0800 Subject: [PATCH 1/2] fix(chat): continue reports after host-only admission stops Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/product/periodic-reports.md | 15 ++-- .../native_chat/steward_managed_mode.py | 8 ++- .../capabilities/native_chat/steward_turn.py | 70 ++++++++++++++++--- .../collaboration/steward_managed_mode.ts | 49 +++++++++++++ .../control_plane_ts/steward_reports.test.ts | 46 ++++++++++++ tests/test_native_steward_reports.py | 47 ++++++++++--- 6 files changed, 211 insertions(+), 24 deletions(-) diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index 0f6b66bc44..12950c510e 100644 --- a/docs/product/periodic-reports.md +++ b/docs/product/periodic-reports.md @@ -282,10 +282,17 @@ 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. +continuation of that window. It rechecks native admission without choosing +another business Task. When governance is still due, the canonical writeback +preview must independently accept the original Agent's semantic delta before +writeback and quota settlement. When the fresh guard no longer owes governance, +the host-only admission stop can request a new exact admission of the original +report Task without a governance model call. A different advisory Task in that +probe is never executed and grants no authority over the report. A retained stop +from the earlier host can receive one separately correlated continuation only +when its canonical journal proves that exact host-only path and no model work; +missing output alone is insufficient. Both paths retain the original native +report identity, task validator, report window and publication owner. 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, diff --git a/loopx/capabilities/native_chat/steward_managed_mode.py b/loopx/capabilities/native_chat/steward_managed_mode.py index 4d1d61fba7..3bc3cd427f 100644 --- a/loopx/capabilities/native_chat/steward_managed_mode.py +++ b/loopx/capabilities/native_chat/steward_managed_mode.py @@ -357,7 +357,13 @@ def submit(self, session_id: str, *, agent_id: str, operation_id: str, client_id += ".admission-retry" retry = self.store.turn_for_client(session_id, client_id) if retry is not None: - return retry, False + selection_recovery = StewardTurnHost(controller).report_selection_recovery(existing, probe_turn=retry) + if selection_recovery is None or "probe" not in selection_recovery: + return retry, False + client_id = "managed." + operation_id + ".admission-resume" + resumed = self.store.turn_for_client(session_id, client_id) + if resumed is not None: + return resumed, False commit = managed_session_goal_commit(store=self.store, registry_path=controller.registry_path, session_id=session_id) with commit(): diff --git a/loopx/capabilities/native_chat/steward_turn.py b/loopx/capabilities/native_chat/steward_turn.py index e5261b2c58..91b1ef3634 100644 --- a/loopx/capabilities/native_chat/steward_turn.py +++ b/loopx/capabilities/native_chat/steward_turn.py @@ -101,7 +101,8 @@ def scope_id(ids): 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: + def report_selection_recovery(self, original: dict[str, Any], *, + probe_turn: dict[str, Any] | None = None) -> 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") @@ -114,9 +115,28 @@ def report_selection_recovery(self, original: dict[str, Any]) -> dict[str, Any] 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 + proof = {"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()} + if probe_turn is not None: + from ...control_plane.turn_driver.executor import inspect_loopx_turn_journal + from ...control_plane.turn_driver.journal_store import ( + load_turn_journal, turn_journal_path, + ) + actor = request["actor"] + keys = {row["payload"].get("resume_turn_key") + for row in self.store.events_after(actor["session_id"], probe_turn["turn_id"], None) + if row.get("kind") == "native_loopx.replan_result"} + if len(keys) != 1 or not isinstance(next(iter(keys)), str): + return None + key = keys.pop() + # Transport canonical observations to the shared typed admission + # owner. Neither a missing model event nor Python prose decides. + proof["probe"] = {"turn": probe_turn, + "inspection": inspect_loopx_turn_journal(reports.root, goal_id=actor["goal_id"], + agent_id=actor["agent_id"], turn_key=key), + "journal": load_turn_journal(turn_journal_path(reports.root, goal_id=actor["goal_id"], turn_key=key))} + return proof def _report_launch_recovery(self, original: dict[str, Any]) -> dict[str, Any] | None: """Read the original typed launch refusal and its canonical transaction. @@ -202,7 +222,9 @@ def _task_execution_lease(self, request: Mapping[str, Any], turn_id: str, *, rec 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") or admission_retry else "" + suffix = (".validation-recovery." + str(version) if recovery else + ".admission-retry." + turn_id if admission_retry else + ".admission-retry" if request.get("resume_turn_key") 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", @@ -327,8 +349,16 @@ def submit(self, binding_id: str, *, turn_instance_id: str, todo_id: str | None client_id = native_steward_client_id(actor, turn_instance_id + ".admission-retry") retry = self.store.turn_for_client(actor["session_id"], client_id) if retry is not None: - return retry, False - if work_kind == "report": + if work_kind != "report": + return retry, False + selection_recovery = self.report_selection_recovery(existing, probe_turn=retry) + if selection_recovery is None or "probe" not in selection_recovery: + return retry, False + client_id = native_steward_client_id(actor, turn_instance_id + ".admission-resume") + resumed = self.store.turn_for_client(actor["session_id"], client_id) + if resumed is not None: + return resumed, False + if work_kind == "report" and selection_recovery is None: selection_recovery = self.report_selection_recovery(existing) if selection_recovery is None: launch_recovery = self._report_launch_recovery(existing) @@ -383,6 +413,19 @@ def execute(self, *, session_id: str, turn_id: str, adapter: Any, 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) + resumed_client = (original_client + ".admission-resume" + if request["actor"].get("schema_version") == "loopx_managed_session_actor_v0" + else native_steward_client_id(request["actor"], request["turn_instance_id"] + ".admission-resume")) + if selection_recovery and turn["client_turn_id"] == resumed_client: + retry_client = (original_client + ".admission-retry" + if request["actor"].get("schema_version") == "loopx_managed_session_actor_v0" + else native_steward_client_id(request["actor"], request["turn_instance_id"] + ".admission-retry")) + retry = self.store.turn_for_client(session_id, retry_client) + if retry is None: + raise ValueError("the original admission probe is unavailable") + selection_recovery = self.report_selection_recovery(original, probe_turn=retry) + if selection_recovery is None: + raise ValueError("the original admission probe has no canonical journal") admitted = self._admit(request, turn=turn, selection_recovery=selection_recovery) actor = admitted["actor"] local_session = actor.get("schema_version") == "loopx_managed_session_actor_v0" @@ -516,7 +559,7 @@ def launch(): result: dict[str, Any] = {} 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: + if selection_recovery and "probe" not in 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. @@ -531,10 +574,17 @@ def launch(): 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} + probe = self.report_selection_recovery(original, probe_turn=self.store.load_turn(session_id, turn_id)) + try: + if probe is None or "probe" not in probe: + raise ValueError("no canonical admission probe") + self._admit(request, turn=self.store.load_turn(session_id, turn_id), selection_recovery=probe) + except (OSError, ValueError, RuntimeError): + 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} + selection_recovery = probe governance_work = False if report_work: # Portfolio collection can take minutes and legitimately see diff --git a/loopx/control_plane/collaboration/steward_managed_mode.ts b/loopx/control_plane/collaboration/steward_managed_mode.ts index 5e9594dfbf..25799e3d5a 100644 --- a/loopx/control_plane/collaboration/steward_managed_mode.ts +++ b/loopx/control_plane/collaboration/steward_managed_mode.ts @@ -40,6 +40,55 @@ function admitReportSelectionRecovery(params: JsonObject, request: JsonObject, s || proof.generation_present !== false) { throw new EffectRuntimeRequestError("report selection recovery requires the original proven pre-execution replan refusal"); } + if (proof.probe == null) return; + // The original host returned before calling its model when a fresh guard + // no longer owed governance. Absence of output alone cannot prove that. + const probe = requireJsonObject(proof.probe, "original host admission probe"); + const observed = requireJsonObject(probe.turn, "original probe Chat Turn"); + const journal = requireJsonObject(probe.journal, "canonical probe journal"); + const inspection = requireJsonObject(probe.inspection, "canonical probe inspection"); + const recovery = requireJsonObject(inspection.recovery_decision, "probe recovery decision"); + const recorded = requireJsonObject(inspection.recorded_effects, "probe effects"); + const plan = requireJsonObject(journal.plan, "probe plan"); + const route = requireJsonObject(plan.route, "probe route"); + const envelope = requireJsonObject(plan.turn_envelope, "probe envelope"); + const action = requireJsonObject(envelope.action, "probe action"); + const selected = requireJsonObject(action.selected_todo, "probe selected task"); + const identity = requireJsonObject(requireJsonObject(requireJsonObject(plan.transaction, "probe transaction") + .settlement_plan, "probe settlement").identity, "probe identity"); + const host = requireJsonObject(journal.host, "probe host"); + const stopped = requireJsonObject(journal.host_result, "probe host result"); + const runningHere = params.turn != null && requireJsonObject(params.turn, "current Chat Turn").turn_id === observed.turn_id + && ["starting", "running"].includes(String(observed.status)); + if (observed.session_id !== sessionId || typeof observed.turn_id !== "string" || !/^[a-f0-9]{32}$/.test(observed.turn_id) + || observed.turn_id === original.turn_id + || observed.origin !== "native_loopx" || observed.loopx_execution !== true + || observed.delta_count !== 0 || observed.upstream_turn_id !== null || observed.error_code != null || observed.error != null + || JSON.stringify(observed.loopx_request) !== JSON.stringify(request) + || !(runningHere || observed.status === "completed" + && JSON.stringify(requireJsonObject(observed.response, "probe response").native_loopx) === JSON.stringify(result)) + || journal.schema_version !== "loopx_turn_journal_v0" || !ENVELOPED_SHA256_PATTERN.test(String(journal.turn_key)) + || journal.status !== "stopped" || journal.goal_id !== envelope.goal_id + || journal.host_attempt_count !== 1 || journal.result_kind !== "wait" + || JSON.stringify(journal.completed_phases) !== JSON.stringify(["host_execute", "typed_result", "validation"]) + || host.kind !== "codex-app-server" || host.executable !== "built-in" + || stopped.schema_version !== "loopx_turn_result_v0" || stopped.turn_key !== journal.turn_key + || stopped.result_kind !== "wait" || stopped.classification !== "report_replan_not_admitted" + || inspection.schema_version !== "loopx_turn_journal_inspection_v1" || inspection.ok !== true + || inspection.goal_matches !== true || inspection.owner_matches !== true || inspection.turn_key_matches !== true + || inspection.journal_consistent !== true || inspection.journal_status !== "stopped" + || recovery.action !== "return_existing" || recovery.reinvoke_host !== false + || recorded.host_invoked !== true || ["state_written", "quota_spent", "scheduler_acknowledged"].some(key => recorded[key] !== false) + || route.kind !== "ready_for_host" || route.effective_action !== "normal_run" + || envelope.effective_action !== "normal_run" || envelope.replan_action_packet != null + || action.must_attempt !== true || action.delivery_allowed !== true || selected.todo_id !== identity.todo_id + || identity.goal_id !== envelope.goal_id || identity.agent_id !== envelope.agent_id + || identity.goal_id !== requireJsonObject(request.actor, "report actor").goal_id + || identity.agent_id !== requireJsonObject(request.actor, "report actor").agent_id + || typeof identity.todo_id !== "string" || !/^todo_[A-Za-z0-9_-]{1,100}$/.test(identity.todo_id) + || identity.turn_instance_id !== request.turn_instance_id + ".replan") { + throw new EffectRuntimeRequestError("report continuation requires its canonical host-only admission stop; retain all model or unknown work"); + } } /** Registration and current Goal lifetime are observed by the host, never by diff --git a/tests/control_plane_ts/steward_reports.test.ts b/tests/control_plane_ts/steward_reports.test.ts index 8d30015e5c..d14e0d9545 100644 --- a/tests/control_plane_ts/steward_reports.test.ts +++ b/tests/control_plane_ts/steward_reports.test.ts @@ -327,4 +327,50 @@ test("only a proven original pre-execution replan refusal admits one correlated {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})); + + const turn_key = "sha256:" + "c".repeat(64); + const journal = {schema_version: "loopx_turn_journal_v0", goal_id: p.actor.goal_id, + turn_key, status: "stopped", host: {kind: "codex-app-server", executable: "built-in"}, + host_attempt_count: 1, result_kind: "wait", completed_phases: ["host_execute", "typed_result", "validation"], + host_result: {schema_version: "loopx_turn_result_v0", turn_key, result_kind: "wait", classification: "report_replan_not_admitted"}, + plan: {route: {kind: "ready_for_host", effective_action: "normal_run"}, + turn_envelope: {goal_id: p.actor.goal_id, agent_id: p.actor.agent_id, effective_action: "normal_run", + replan_action_packet: null, action: {must_attempt: true, delivery_allowed: true, selected_todo: {todo_id: request.todo_id}}}, + transaction: {settlement_plan: {identity: {goal_id: p.actor.goal_id, agent_id: p.actor.agent_id, + todo_id: request.todo_id, turn_instance_id: request.turn_instance_id + ".replan"}}}}}; + const inspection = {schema_version: "loopx_turn_journal_inspection_v1", ok: true, + goal_matches: true, owner_matches: true, turn_key_matches: true, journal_consistent: true, journal_status: "stopped", + recorded_effects: {host_invoked: true, state_written: false, quota_spent: false, scheduler_acknowledged: false}, + recovery_decision: {action: "return_existing", reinvoke_host: false}}; + const probe = {turn: {...source, turn_id: "c".repeat(32)}, journal, inspection}; + const resumed = {...input, report_selection_recovery: {...proof, probe}}; + assert.equal(admitStewardManagedTurn(resumed).observation_only, false); + // An unselected guard's advisory is not executed or inherited. The original + // report still has to pass a fresh native exact-Todo admission afterwards. + const other = structuredClone(journal); + other.plan.turn_envelope.action.selected_todo.todo_id = "todo_advisory"; + other.plan.transaction.settlement_plan.identity.todo_id = "todo_advisory"; + assert.equal(admitStewardManagedTurn({...input, + report_selection_recovery: {...proof, probe: {...probe, journal: other}}}).observation_only, false); + for (const changed of [ + {...probe, turn: {...probe.turn, turn_id: ""}}, {...probe, journal: {...journal, turn_key: "unknown"}}, + {...probe, turn: {...probe.turn, delta_count: 1}}, {...probe, turn: {...probe.turn, upstream_turn_id: "model"}}, + {...probe, turn: {...probe.turn, status: "running"}}, {...probe, turn: {...probe.turn, session_id: "another"}}, + {...probe, turn: {...probe.turn, response: {native_loopx: {...result, effects: {...effects, state_written: true}}}}}, + {...probe, journal: {...journal, host_attempt_count: 2}}, {...probe, journal: {...journal, status: "failed"}}, + {...probe, journal: {...journal, host: {...journal.host, executable: "external"}}}, + {...probe, journal: {...journal, host_result: {...journal.host_result, classification: "model_wait"}}}, + {...probe, journal: {...journal, completed_phases: [...journal.completed_phases, "durable_writeback"]}}, + {...probe, journal: {...journal, plan: {...journal.plan, turn_envelope: {...journal.plan.turn_envelope, + replan_action_packet: {schema_version: "autonomous_replan_obligation_v0"}}}}}, + {...probe, journal: {...journal, plan: {...journal.plan, turn_envelope: {...journal.plan.turn_envelope, + action: {...journal.plan.turn_envelope.action, selected_todo: {todo_id: "todo_other"}}}}}}, + {...probe, journal: {...journal, plan: {...journal.plan, transaction: {settlement_plan: {identity: { + ...journal.plan.transaction.settlement_plan.identity, turn_instance_id: "other.replan"}}}}}}, + ...["goal_matches", "owner_matches", "turn_key_matches", "journal_consistent"].map(key => + ({...probe, inspection: {...inspection, [key]: false}})), + ...["state_written", "quota_spent", "scheduler_acknowledged"].flatMap(key => [true, null].map(value => + ({...probe, inspection: {...inspection, recorded_effects: {...inspection.recorded_effects, [key]: value}}}))), + {...probe, inspection: {...inspection, recovery_decision: {...inspection.recovery_decision, reinvoke_host: true}}}, + ]) assert.throws(() => admitStewardManagedTurn({...input, report_selection_recovery: {...proof, probe: changed}})); }); diff --git a/tests/test_native_steward_reports.py b/tests/test_native_steward_reports.py index 775d507a97..44c2f2ddfd 100644 --- a/tests/test_native_steward_reports.py +++ b/tests/test_native_steward_reports.py @@ -464,7 +464,7 @@ def install_model(steward, tmp_path, sid, *, result_kind="validated_completion") @pytest.mark.parametrize("provider", ["file", "sqlite"]) @pytest.mark.parametrize("local_session", [False, True]) -@pytest.mark.parametrize("replan_outcome", ["settled", "wait", "invalid_semantic"]) +@pytest.mark.parametrize("replan_outcome", ["settled", "wait", "invalid_semantic", "already_admitted", "retained_probe"]) 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 @@ -507,6 +507,11 @@ def now(cls, tz=None): 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"] + if replan_outcome in {"already_admitted", "retained_probe"}: + # A past refusal is evidence, not current authority. The native owner + # can now admit this exact report without a semantic replan. + assert configure_goal(registry_path=runtime.registry_path, goal_id=actor["goal_id"], + execution_replan_after_turns=6, execute=True)["ok"] # 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", @@ -543,11 +548,32 @@ def now(cls, tz=None): 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]) + if replan_outcome == "retained_probe": + from loopx.capabilities.native_chat.steward_turn import StewardTurnHost + selection = StewardTurnHost.report_selection_recovery + + def legacy_selection(host, original, **kwargs): + # Exercise the real old host-only wait journal, not a fabricated + # success or an unknown failed model Turn. + return None if kwargs.get("probe_turn") is not None else selection(host, original) + + with monkeypatch.context() as legacy: + legacy.setattr(StewardTurnHost, "report_selection_recovery", legacy_selection) + stopped = finish(runtime, resumed[0]) + assert stopped["delta_count"] == 0 and stopped["upstream_turn_id"] is None + stopped_bytes = store._turn_path(sid, stopped["turn_id"]).read_bytes() + assert capture.read_bytes() == calls + runtime._steward_monitor_next_check.clear() + continued = pump_steward_monitors(runtime) + assert len(continued) == 1, "Only the canonical no-model admission probe may continue once" + done = finish(runtime, continued[0]) + assert store._turn_path(sid, stopped["turn_id"]).read_bytes() == stopped_bytes + else: + 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 len(replans) == (0 if replan_outcome == "retained_probe" else 1) + if replan_outcome not in {"settled", "already_admitted", "retained_probe"}: 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() @@ -570,10 +596,13 @@ def now(cls, tz=None): 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"] + if replan_outcome == "settled": + 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"] + elif replan_outcome == "already_admitted": + assert replans[0]["status"] == "stopped" and replans[0]["effects"]["quota_spent"] is False task = next(row for row in StewardMonitor(runtime).todos(actor) if row["todo_id"] == done["loopx_request"]["todo_id"]) assert task["status"] == "done" @@ -594,7 +623,7 @@ def now(cls, tz=None): 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 sum(row.get("method") == "turn/start" for row in tail) == (2 if replan_outcome == "settled" else 1) 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() From efa645d88956186a252315082ca55c5014b145d4 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 25 Oct 2026 18:01:20 +0800 Subject: [PATCH 2/2] fix(chat): reuse the canonical native action definition Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/control_plane/collaboration/steward_managed_mode.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/loopx/control_plane/collaboration/steward_managed_mode.ts b/loopx/control_plane/collaboration/steward_managed_mode.ts index 25799e3d5a..441efdf3a8 100644 --- a/loopx/control_plane/collaboration/steward_managed_mode.ts +++ b/loopx/control_plane/collaboration/steward_managed_mode.ts @@ -11,6 +11,7 @@ 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"; +import {EffectiveAction} from "../quota/effective_action.generated.ts"; /** A native refusal before execution can owe governance, never a business replay. */ function admitReportSelectionRecovery(params: JsonObject, request: JsonObject, sessionId: unknown): void { @@ -79,8 +80,8 @@ function admitReportSelectionRecovery(params: JsonObject, request: JsonObject, s || inspection.journal_consistent !== true || inspection.journal_status !== "stopped" || recovery.action !== "return_existing" || recovery.reinvoke_host !== false || recorded.host_invoked !== true || ["state_written", "quota_spent", "scheduler_acknowledged"].some(key => recorded[key] !== false) - || route.kind !== "ready_for_host" || route.effective_action !== "normal_run" - || envelope.effective_action !== "normal_run" || envelope.replan_action_packet != null + || route.kind !== "ready_for_host" || route.effective_action !== EffectiveAction.NORMAL_RUN + || envelope.effective_action !== EffectiveAction.NORMAL_RUN || envelope.replan_action_packet != null || action.must_attempt !== true || action.delivery_allowed !== true || selected.todo_id !== identity.todo_id || identity.goal_id !== envelope.goal_id || identity.agent_id !== envelope.agent_id || identity.goal_id !== requireJsonObject(request.actor, "report actor").goal_id