diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index c7d30cc4db..36d9efb89f 100644 --- a/docs/product/periodic-reports.md +++ b/docs/product/periodic-reports.md @@ -499,6 +499,10 @@ does not authorize replay of a historical attempt with unknown host effects. The original pump can resume one retained report after the first-party launch owner's exact `stale_goal_authority` refusal. That owner rejects before publishing `turn/start`; a missing answer or missing events alone is insufficient. The +host adapter records this guard-entry refusal through the native executor's +existing `contract_rejected` host-failure contract. The durable attempt remains +recorded; an exception after publication is not converted into that proof. +Older in-progress journals with unknown host effects remain blocked. The typed admission checks the original Chat request, canonical native transaction, one recorded attempt, no generation or settlement, and the current original subscription and grant. A separately correlated Chat attempt resumes the same diff --git a/loopx/capabilities/native_chat/steward_turn.py b/loopx/capabilities/native_chat/steward_turn.py index 1cfedcd960..0c93e830f7 100644 --- a/loopx/capabilities/native_chat/steward_turn.py +++ b/loopx/capabilities/native_chat/steward_turn.py @@ -15,6 +15,7 @@ from typing import Any from ...control_plane.goals.activation import goal_is_stopped +from ...control_plane.goals.first_party_host_admission import FirstPartyHostRuntimeRejected from ...file_lock import exclusive_file_lock from .steward_managed_mode import ManagedSessionTurn, ManagedTaskValidationUnavailable, StewardManagedMode @@ -262,7 +263,7 @@ def _report_launch_recovery(self, original: dict[str, Any]) -> dict[str, Any] | the typed conversation owner checks it with the current journal. """ result = (original.get("response") or {}).get("native_loopx") or {} - if result.get("error_code") != "stale_goal_authority": + if (result.get("host_failure") or {}).get("kind") != "contract_rejected": return None from ...control_plane.turn_driver import inspect_loopx_turn_journal from ...control_plane.turn_driver.journal_store import ( @@ -281,7 +282,7 @@ def _report_launch_recovery(self, original: dict[str, Any]) -> dict[str, Any] | agent_id=actor["agent_id"], turn_key=key, retry_failed=True) journal = load_turn_journal(turn_journal_path(root, goal_id=actor["goal_id"], turn_key=key)) _, window = StewardReports(self.controller).current_window(actor["goal_id"].removeprefix("steward-agent-")) - return {"original": original, "inspection": inspection, "turn_key": key, + return {"original": original, "inspection": inspection, "journal": journal, "turn_key": key, "host_attempt_count": (journal or {}).get("host_attempt_count"), "generation_present": window is None or (attempt_directory(root, window) / "private-generation.json").exists()} @@ -644,13 +645,22 @@ def launch(): # The existing app-server request owner holds this fence only # through turn/start publication, never through model waiting. with ExitStack() as stack: - if previous_guard is not None: - stack.enter_context(previous_guard()) - stack.enter_context(launch_scope()) - with (nullcontext() if actor.get("schema_version") == "loopx_managed_session_actor_v0" - else exclusive_file_lock(self.bindings.path, operation="native_steward_launch")): + try: + if previous_guard is not None: + stack.enter_context(previous_guard()) + stack.enter_context(launch_scope()) + stack.enter_context(nullcontext() if actor.get("schema_version") == "loopx_managed_session_actor_v0" + else exclusive_file_lock(self.bindings.path, operation="native_steward_launch")) current() - yield + except FirstPartyHostRuntimeRejected as exc: + if not report_work or exc.code != "stale_goal_authority": + raise + # Only guard entry is before publication. Let the + # existing native executor durably record its typed + # failure; exceptions after yield remain unknown. + from ...control_plane.turn_driver.host_failure import BuiltInHostError + raise BuiltInHostError(str(exc), failure_kind="contract_rejected") from exc + yield adapter.session._launch_guard = launch try: diff --git a/loopx/control_plane/collaboration/steward_managed_mode.ts b/loopx/control_plane/collaboration/steward_managed_mode.ts index 23454b7adf..096bdd011c 100644 --- a/loopx/control_plane/collaboration/steward_managed_mode.ts +++ b/loopx/control_plane/collaboration/steward_managed_mode.ts @@ -508,6 +508,9 @@ export function admitStewardManagedTurn(params: JsonObject): JsonObject { const prior = requireJsonObject(original.loopx_request, "original rejected request"); const result = requireJsonObject(requireJsonObject(original.response, "original response").native_loopx, "native rejection"); const journal = requireJsonObject(proof.inspection, "canonical original Turn inspection"); + const retained = requireJsonObject(proof.journal, "canonical rejected native Turn"); + const failure = requireJsonObject(retained.host_failure, "recorded pre-publication rejection"); + const receipt = requireJsonObject(retained.receipt, "recorded host failure receipt"); const decision = requireJsonObject(journal.recovery_decision, "original native recovery decision"); const effects = requireJsonObject(journal.recorded_effects, "original native effects"); const invocation = requireJsonObject(result.effects, "original invocation effects"); @@ -518,14 +521,22 @@ export function admitStewardManagedTurn(params: JsonObject): JsonObject { || Object.keys(prior).length !== keys.length || keys.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 !== "stale_goal_authority" - || result.error !== "first-party Host runtime rejected: stale_goal_authority" + || result.mode !== "run_once" || result.status !== "failed" || result.result_kind !== "host_failure" + || result.reason !== "first-party Host runtime rejected: stale_goal_authority" || result.resume_turn_key !== request.resume_turn_key - || proof.turn_key !== request.resume_turn_key || Object.values(invocation).some(value => value === true) + || proof.turn_key !== request.resume_turn_key || retained.turn_key !== request.resume_turn_key + || retained.status !== "failed" || retained.result_kind !== "host_failure" + || retained.reason !== result.reason || retained.host_result != null + || retained.host_attempt_count !== 1 || failure.kind !== "contract_rejected" + || failure.schema_version !== "loopx_turn_host_failure_v0" || failure.attempt !== 1 || failure.retryable !== false + || JSON.stringify(result.host_failure) !== JSON.stringify(retained.host_failure) + || receipt.failed_phase !== "host_execute" + || !Array.isArray(retained.completed_phases) || retained.completed_phases.length !== 0 + || invocation.host_invoked !== true || invocation.state_written !== false + || invocation.quota_spent !== false || invocation.scheduler_acknowledged !== false || journal.schema_version !== "loopx_turn_journal_inspection_v1" || journal.ok !== true || journal.goal_matches !== true || journal.owner_matches !== true || journal.turn_key_matches !== true - || journal.journal_consistent !== true || journal.journal_status !== "in_progress" + || journal.journal_consistent !== true || journal.journal_status !== "failed" || !Array.isArray(journal.completed_phases) || journal.completed_phases.length !== 0 || proof.host_attempt_count !== 1 || proof.generation_present !== false || effects.host_invoked !== null || effects.state_written !== false diff --git a/tests/control_plane_ts/steward_reports.test.ts b/tests/control_plane_ts/steward_reports.test.ts index 6114d5dd79..a758466e86 100644 --- a/tests/control_plane_ts/steward_reports.test.ts +++ b/tests/control_plane_ts/steward_reports.test.ts @@ -306,17 +306,21 @@ test("a retained pre-publication launch refusal can resume only its original nat const key = "sha256:" + "a".repeat(64); const request = {...prior, resume_turn_key: key}; const result = {schema_version: "loopx_turn_execution_v0", mode: "run_once", ok: false, - error_code: "stale_goal_authority", error: "first-party Host runtime rejected: stale_goal_authority", - effects_scope: "current_invocation", effects: {host_invoked: null, state_written: null, quota_spent: null}, + status: "failed", result_kind: "host_failure", reason: "first-party Host runtime rejected: stale_goal_authority", + host_failure: {schema_version: "loopx_turn_host_failure_v0", kind: "contract_rejected", attempt: 1, retryable: false}, + effects: {host_invoked: true, state_written: false, quota_spent: false, scheduler_acknowledged: false}, resume_turn_key: key}; 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: prior, response: {native_loopx: result}}; 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: "in_progress", completed_phases: [], + journal_status: "failed", completed_phases: [], recorded_effects: {host_invoked: null, state_written: false, quota_spent: false, scheduler_acknowledged: false}, recovery_decision: {action: "continue", reinvoke_host: true, resume_from: "host_execute"}}; - const proof = {original: source, inspection, turn_key: key, host_attempt_count: 1, generation_present: false}; + const journal = {status: "failed", result_kind: "host_failure", reason: result.reason, + turn_key: key, host_attempt_count: 1, completed_phases: [], host_failure: result.host_failure, + receipt: {failed_phase: "host_execute"}}; + const proof = {original: source, inspection, journal, turn_key: key, host_attempt_count: 1, generation_present: false}; const input = {...p, request, report_launch_recovery: proof, report_window: {window_id: prior.report_window_id, subscription_revision: prior.report_subscription_revision, goal_id: p.actor.goal_id, agent_id: p.actor.agent_id}, @@ -338,8 +342,12 @@ test("a retained pre-publication launch refusal can resume only its original nat {...input, report_launch_recovery: {...proof, original: {...source, upstream_turn_id: "actual-start"}}}, {...input, report_launch_recovery: {...proof, original: {...source, loopx_request: {...prior, todo_id: "todo_other"}}}}, - ...["transport_lost", "goal_launch_not_allowed", "unknown"].map(error_code => ({...input, - report_launch_recovery: {...proof, original: {...source, response: {native_loopx: {...result, error_code}}}}})), + ...["transport_lost", "goal_launch_not_allowed", "unknown"].map(reason => ({...input, + report_launch_recovery: {...proof, original: {...source, response: {native_loopx: {...result, reason}}}}})), + {...input, report_launch_recovery: {...proof, journal: {...journal, status: "in_progress"}}}, + {...input, report_launch_recovery: {...proof, journal: {...journal, host_result: {result_kind: "wait"}}}}, + {...input, report_launch_recovery: {...proof, journal: {...journal, host_failure: {...journal.host_failure, kind: "unknown"}}}}, + {...input, report_launch_recovery: {...proof, journal: {...journal, receipt: {failed_phase: "validation"}}}}, {...input, report_launch_recovery: {...proof, original: {...source, response: {native_loopx: {...result, effects: {state_written: true}}}}}}, ...["goal_matches", "owner_matches", "turn_key_matches", "journal_consistent"].map(key => ({...input, diff --git a/tests/test_native_steward_reports.py b/tests/test_native_steward_reports.py index 168c9e4e52..d392189cee 100644 --- a/tests/test_native_steward_reports.py +++ b/tests/test_native_steward_reports.py @@ -1079,7 +1079,10 @@ def concurrent_registry_update(self, payload, **kwargs): first = pump_steward_monitors(runtime)[0] rejected = finish(runtime, first) failure = rejected["response"]["native_loopx"] - assert failure["error_code"] == "stale_goal_authority", rejected + assert failure["status"] == "failed" and failure["result_kind"] == "host_failure", rejected + assert failure["host_failure"] == {"schema_version": "loopx_turn_host_failure_v0", + "kind": "contract_rejected", "attempt": 1, "retryable": False} + assert failure["reason"] == "first-party Host runtime rejected: stale_goal_authority" assert rejected["delta_count"] == 0 and rejected["upstream_turn_id"] is None native = turn_journal_path(reports.root, goal_id=actor["goal_id"], turn_key=failure["resume_turn_key"]) assert json.loads(native.read_text())["host_attempt_count"] == 1 @@ -1100,7 +1103,8 @@ def concurrent_registry_update(self, payload, **kwargs): completed = finish(runtime, retry) assert completed["status"] == "completed", {key: completed.get(key) for key in ("status", "error", "error_code", "response")} if reject_recovery: - assert completed["response"]["native_loopx"]["error_code"] == "stale_goal_authority" + assert completed["response"]["native_loopx"]["host_failure"]["kind"] == "contract_rejected" + assert completed["response"]["native_loopx"]["host_failure"]["attempt"] == 2 before = capture.read_bytes(), list(bot.writes), native.read_bytes() for _ in range(2): runtime._steward_monitor_next_check.clear() @@ -1130,6 +1134,51 @@ def concurrent_registry_update(self, payload, **kwargs): assert bot.writes == sent and capture.read_bytes() == calls +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("stage", ["unknown_before_publication", "after_publication"]) +def test_report_launch_recovery_never_reclassifies_unknown_or_published_work(steward, tmp_path, monkeypatch, provider, stage): + from loopx.chat_agent import CodexChatAgentSession + from loopx.control_plane.goals.first_party_host_admission import FirstPartyHostRuntimeRejected + from loopx.control_plane.turn_driver.journal_store import turn_journal_path + + store, runtime, bot, transport, _, capture, _ = steward + sid, actor, reports, _ = configured(steward, tmp_path, monkeypatch, provider) + install_model(steward, tmp_path, sid) + calls = capture.read_bytes() + write = CodexChatAgentSession._write + + def uncertain_write(self, payload, **kwargs): + if payload.get("method") != "turn/start": + return write(self, payload, **kwargs) + if stage == "unknown_before_publication": + raise OSError("publication outcome unavailable") + write(self, payload, **kwargs) + # The same exception after the guarded publication is deliberately not + # evidence that this host invocation did nothing. + raise FirstPartyHostRuntimeRejected("stale_goal_authority") + + with monkeypatch.context() as fault: + fault.setattr(CodexChatAgentSession, "_write", uncertain_write) + first = pump_steward_monitors(runtime)[0] + done = finish(runtime, first) + result = done["response"]["native_loopx"] + if stage == "unknown_before_publication": + assert result["host_failure"]["kind"] == "unknown" + else: + assert result["error_code"] == "stale_goal_authority" + assert "host_failure" not in result + native = turn_journal_path(reports.root, goal_id=actor["goal_id"], turn_key=result["resume_turn_key"]) + starts = [json.loads(line) for line in capture.read_bytes()[len(calls):].splitlines() + if json.loads(line).get("method") == "turn/start"] + assert len(starts) == (1 if stage == "after_publication" else 0) + before = native.read_bytes(), store.load_turn(sid, first["turn_id"]), capture.read_bytes(), list(bot.writes) + for _ in range(2): + runtime._steward_monitor_next_check.clear() + assert not pump_steward_monitors(runtime) + assert transport.steward_reports.reconcile() == 0 + assert (native.read_bytes(), store.load_turn(sid, first["turn_id"]), capture.read_bytes(), bot.writes) == before + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_report_source_revoked_during_collection_never_enters_native_execution(steward, tmp_path, monkeypatch, provider): _, runtime, bot, _, binding, capture, _ = steward