diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index 45f45fdb21..64338a29fc 100644 --- a/docs/product/periodic-reports.md +++ b/docs/product/periodic-reports.md @@ -23,6 +23,23 @@ The full frozen source, authored response, bundle, candidate and body must still match; an unsupported or altered presentation remains unavailable. New reports always use the shared source-link rule above. +A report refused because its model allowance was exhausted can receive one +new, correlated Turn when the original host independently reads a stable native +failure containing only its original user input and observes usable allowance +for the current authentication. A reset time or successful authentication RPC +alone does not qualify. The shared typed conversation owner also checks the +original failed kernel receipt, unchanged actor, subscription, window and Task, +and the absence of any authored response, generation or delivery attempt. +The original Chat Turn and failed kernel journal remain unchanged. The new Turn +uses ordinary admission, lease, quota, validation and original-route publication; +user work retains priority. A second failure is retained, never chained into +another automatic attempt. Unknown effects, unavailable native metadata, +revocation, cancellation and an unavailable allowance retain the existing hold. +The admitted continuation refreshes only its own native adapter to reload +current authentication while retaining its native thread. This applies to +original steward reports and ordinary managed Session reports; authentication +selection and unrelated clients retain their existing owners. + The following synthetic reports use the production formatters. Text views use the compiled App Markdown component and its stylesheet; HTML views use the standalone report artifact. The same normalized records retain their outcomes, diff --git a/loopx/capabilities/native_chat/steward_managed_mode.py b/loopx/capabilities/native_chat/steward_managed_mode.py index 3bc3cd427f..8eb1a92592 100644 --- a/loopx/capabilities/native_chat/steward_managed_mode.py +++ b/loopx/capabilities/native_chat/steward_managed_mode.py @@ -255,6 +255,7 @@ def admit(self, request: dict[str, Any], *, turn: dict[str, Any] | None = None, settlement_recovery: dict[str, Any] | None = None, selection_recovery: dict[str, Any] | None = None) -> dict[str, Any]: from .conversation_bindings import ChatConversationBindings + from .steward_turn import StewardTurnHost if self.controller.closed.is_set(): raise ValueError("the original Chat host is stopped") actor = request["actor"] @@ -284,6 +285,7 @@ def admit(self, request: dict[str, Any], *, turn: dict[str, Any] | None = None, "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 {}), + **StewardTurnHost(self.controller).report_quota_basis(request), "queued_turn_count": len(self.store.queued_turns(actor["session_id"]))}) def inspect(self, session_id: str, *, agent_id: str | None = None, @@ -351,7 +353,19 @@ def submit(self, session_id: str, *, agent_id: str, operation_id: str, if frozen != request: raise ValueError("managed Turn request identity conflict; inspect the original operation") from .steward_turn import StewardTurnHost - selection_recovery = StewardTurnHost(controller).report_selection_recovery(existing) if report_window is not None else None + host = StewardTurnHost(controller) + if report_window is not None: + from .steward_turn import report_turn + retained = report_turn(self.store, request["actor"], report_window) + if retained is not None and retained["turn_id"] != existing["turn_id"]: + return retained, False + quota_request = host.report_quota_request(existing) + if quota_request is not None: + request = quota_request + client_id += ".quota-recovery" + existing = None + selection_recovery = host.report_selection_recovery(existing) if existing is not None and report_window is not None else None + if existing is not None: if selection_recovery is None: return existing, False # Failed/unknown work is never replayed. client_id += ".admission-retry" diff --git a/loopx/capabilities/native_chat/steward_reports.py b/loopx/capabilities/native_chat/steward_reports.py index c889289154..4355017740 100644 --- a/loopx/capabilities/native_chat/steward_reports.py +++ b/loopx/capabilities/native_chat/steward_reports.py @@ -73,13 +73,12 @@ def _published_window(self, window: dict[str, Any]) -> bool: return bool(cursor and trigger["selected_trigger_id"] in cursor["covered_trigger_ids"]) def _known_generation_failure(self, actor: dict[str, Any], window: dict[str, Any]) -> dict[str, str] | None: - from .steward_turn import native_steward_client_id + from .steward_turn import report_turn folder = attempt_directory(self.root, window) if (not (folder / "editorial-response.json").is_file() or (folder / "private-generation.json").exists() or (folder / "private-delivery.json").exists()): return None - turn = self.controller.store.turn_for_client(actor["session_id"], - native_steward_client_id(actor, "report." + window["window_id"])) + turn = report_turn(self.controller.store, actor, window) request = (turn or {}).get("loopx_request") or {} native = ((turn or {}).get("response") or {}).get("native_loopx") or {} if (turn and turn.get("status") == "completed" and native.get("status") == "failed" @@ -519,9 +518,9 @@ def publish(self, session_id: str, window: dict[str, Any]) -> bool: import hashlib from ..periodic_report.private_generation import validate_private_generation from ..periodic_report.incremental import commit_periodic_report_publication_cursor - from .steward_turn import native_steward_client_id + from .steward_turn import report_turn actor = self.selected(session_id)["actor"] - turn = self.controller.store.turn_for_client(session_id, native_steward_client_id(actor, "report." + window["window_id"])) + turn = report_turn(self.controller.store, actor, window) if not turn: return False request = turn.get("loopx_request") or {} @@ -574,9 +573,8 @@ def health(self, session_id: str) -> dict[str, Any]: journal = read_cadence_journal(runtime_root=self.root, goal_id=actor["goal_id"]) window = journal["window"] if journal else None folder = attempt_directory(self.root, window) if window else None - from .steward_turn import native_steward_client_id - turn = self.controller.store.turn_for_client(session_id, - native_steward_client_id(actor, "report." + window["window_id"])) if window else None + from .steward_turn import report_turn + turn = report_turn(self.controller.store, actor, window) if window else None native = ((turn or {}).get("response") or {}).get("native_loopx") or {} failed = bool(turn and (turn["status"] in {"failed", "interrupted", "timed_out"} or native.get("status") == "failed")) diff --git a/loopx/capabilities/native_chat/steward_turn.py b/loopx/capabilities/native_chat/steward_turn.py index 91b1ef3634..82478a1a74 100644 --- a/loopx/capabilities/native_chat/steward_turn.py +++ b/loopx/capabilities/native_chat/steward_turn.py @@ -1,8 +1,9 @@ """Original-Session app-server transport for the existing governed Turn driver. -No scheduler, task store, consumer or CLI process is created here. The caller +No scheduler, task store or message consumer is created here. The caller selects native work; the existing Turn command owns fresh admission, validation, -writeback, quota and scheduler handoff. Chat retains its own execution fence. +writeback, quota and scheduler handoff. Provider metadata uses the existing +read-only native control client. Chat retains its own execution fence. """ from __future__ import annotations @@ -25,6 +26,20 @@ def native_steward_client_id(actor: Mapping[str, Any], turn_instance_id: str) -> return "native." + hashlib.sha256(json.dumps([actor, turn_instance_id], sort_keys=True).encode()).hexdigest()[:40] +def report_turn(store: Any, actor: Mapping[str, Any], window: Mapping[str, Any]) -> dict[str, Any] | None: + """Read the one bounded continuation, retaining the original failed Turn.""" + operation = "report." + window["window_id"] + original = store.turn_for_client(actor["session_id"], native_steward_client_id(actor, operation)) + continued = store.turn_for_client(actor["session_id"], native_steward_client_id(actor, operation + ".quota-recovery")) + if continued is not None: + request = continued.get("loopx_request") or {} + if (not original or (request.get("report_quota_recovery") or {}).get("original_turn_id") != original["turn_id"] + or request.get("turn_instance_id") != original["loopx_request"]["turn_instance_id"] + ".quota-recovery"): + raise ValueError("the report quota continuation changed its original identity") + return continued + return original + + class StewardTurnHost: def __init__(self, controller: Any) -> None: self.controller = controller @@ -96,11 +111,57 @@ def scope_id(ids): **({"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 {}), + **self.report_quota_basis(request), **({"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_quota_basis(self, request: Mapping[str, Any]) -> dict[str, Any]: + """Transport original journal observations; the typed owner admits work.""" + recovery = request.get("report_quota_recovery") + if recovery is None: + return {} + from ...control_plane.turn_driver import inspect_loopx_turn_journal + from ..periodic_report.private_generation import attempt_directory + from .steward_reports import report_service + import time + actor = request["actor"] + original = self.store.load_turn(actor["session_id"], recovery["original_turn_id"]) + result = ((original or {}).get("response") or {}).get("native_loopx") or {} + reports, owner = report_service(self.controller, actor) + _, window = reports.current_window(owner) + folder = attempt_directory(reports.root, window) if window else None + return {"report_quota_basis": {"original": original, + "inspection": inspect_loopx_turn_journal(reports.root, goal_id=actor["goal_id"], + agent_id=actor["agent_id"], turn_key=result["resume_turn_key"]), + "generation_present": folder is None or any((folder / name).exists() for name in + ("editorial-response.json", "private-generation.json", "private-delivery.json")), + "observed_at": time.time()}} + + def report_quota_request(self, original: Mapping[str, Any]) -> dict[str, Any] | None: + """Observe quota recovery, never resume or rewrite the failed native Turn.""" + result = (original.get("response") or {}).get("native_loopx") or {} + request = original.get("loopx_request") or {} + if (original.get("status") != "completed" or request.get("work_kind") != "report" + or request.get("report_quota_recovery") is not None + or result.get("result_kind") != "host_failure" or not original.get("upstream_turn_id")): + return None + from .codex_adapter import CodexAppServerAdapter + from .codex_context import checked_session_home + session = self.store.load_session(original["session_id"]) + if not session or not session.get("codex_home") or not session.get("upstream_thread_id"): + return None + proof = CodexAppServerAdapter.failed_request_observation(codex_bin=self.controller.codex_bin, + codex_home=Path(session["codex_home"]), + model_auth_home=checked_session_home(self.controller.codex_home, session), + thread_ids=[session["upstream_thread_id"]], turn_id=original["upstream_turn_id"]) + if (not proof or proof.get("only_original_user_item") is not True + or proof.get("codex_error_info") != "usageLimitExceeded" or proof.get("quota_available") is not True): + return None + return {**request, "turn_instance_id": request["turn_instance_id"] + ".quota-recovery", + "report_quota_recovery": {"original_turn_id": original["turn_id"], "native_observation": proof}} + 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.""" @@ -260,8 +321,7 @@ def recover_report_settlement(self, actor: dict[str, Any], window: dict[str, Any from ..periodic_report.private_generation import attempt_directory, validate_private_generation from .steward_reports import report_service with self.controller.lock, self._report_fence({"actor": actor}): - original = self.store.turn_for_client(actor["session_id"], - native_steward_client_id(actor, "report." + window["window_id"])) + original = report_turn(self.store, actor, window) if not original: return request = original.get("loopx_request") or {} @@ -340,6 +400,16 @@ def submit(self, binding_id: str, *, turn_instance_id: str, todo_id: str | None frozen.setdefault("work_kind", "native_turn") if frozen != request or work_kind == "monitor" and existing["loopx_request"].get("monitor_deadline") != monitor_deadline: raise ValueError("native steward request identity conflict") + if work_kind == "report": + retained = report_turn(self.store, actor, report_window) + if retained is not None and retained["turn_id"] != existing["turn_id"]: + return retained, False + quota_request = self.report_quota_request(existing) + if quota_request is not None: + request = quota_request + client_id = native_steward_client_id(actor, request["turn_instance_id"]) + existing = None + if existing is not None: monitor_retry = work_kind == "monitor" and self._monitor_selection_failed_before_poll(existing) if not monitor_retry and work_kind != "report": return existing, False diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py index 91dd98224d..c608186299 100644 --- a/loopx/chat_runtime.py +++ b/loopx/chat_runtime.py @@ -812,6 +812,11 @@ def _ensure_adapter_locked( if self.project_contexts.conversation_bindings is not None: from .capabilities.native_chat.external_conversations import ChatExternalConversations recovery = ChatExternalConversations(self).queued_recovery_context(session, active, accepted_turn_id) + native_quota_refresh = bool(active and active_id == accepted_turn_id + and active.get("status") == "queued" and active.get("origin") == "native_loopx" + and active.get("loopx_execution") is True + and (active.get("loopx_request") or {}).get("work_kind") == "report" + and isinstance((active.get("loopx_request") or {}).get("report_quota_recovery"), dict)) with self.lock: current = self.adapters.get(session_id) manager_profile_changed = bool( @@ -830,6 +835,7 @@ def _ensure_adapter_locked( and current.healthcheck() and not manager_profile_changed and not recovery.get("quota_attempt") + and not native_quota_refresh and (isinstance(current, CodexAppServerAdapter) and current.session.read_only_review) == bool(recovery.get("read_only_review")) and model_allocation is None and not settings_changed diff --git a/loopx/control_plane/collaboration/steward_managed_mode.ts b/loopx/control_plane/collaboration/steward_managed_mode.ts index 441efdf3a8..9aba7f43e4 100644 --- a/loopx/control_plane/collaboration/steward_managed_mode.ts +++ b/loopx/control_plane/collaboration/steward_managed_mode.ts @@ -13,6 +13,64 @@ import {goalAgentConversationIdentity, resolveConversationScope} from "./convers import {readTurnSelectionRejection} from "../turn_driver/selection_rejection.ts"; import {EffectiveAction} from "../quota/effective_action.generated.ts"; +/** One new report Turn after a provider-proven untouched quota refusal. + * This does not reopen the original Chat/kernel Turn or relax its recovery. + * Native metadata and usable allowance are host observations, never model input. + */ +function admitReportQuotaRecovery(params: JsonObject, request: JsonObject, session: JsonObject): void { + if (request.report_quota_recovery === undefined) { + if (params.report_quota_basis != null) throw new EffectRuntimeRequestError("quota recovery lacks its original report"); + return; + } + const recovery = requireJsonObject(request.report_quota_recovery, "report quota continuation"); + const observation = requireJsonObject(recovery.native_observation, "native quota observation"); + const basis = requireJsonObject(params.report_quota_basis, "original quota failure basis"); + const original = requireJsonObject(basis.original, "original failed Chat Turn"); + const prior = requireJsonObject(original.loopx_request, "original report request"); + const result = requireJsonObject(requireJsonObject(original.response, "original response").native_loopx, "original failure"); + const failure = requireJsonObject(result.host_failure, "original host failure"); + const effects = requireJsonObject(result.effects, "original effects"); + const receipt = requireJsonObject(result.receipt, "original native receipt"); + const inspection = requireJsonObject(basis.inspection, "original native inspection"); + const recorded = requireJsonObject(inspection.recorded_effects, "original recorded effects"); + const actor = requireJsonObject(request.actor, "original report actor"); + const lineage = requireJsonObject(receipt.lineage, "original report lineage"); + const keys = Object.keys(request).filter(key => key !== "report_quota_recovery"); + const accepting = params.turn == null && params.report_settlement_recovery == null; + if (request.work_kind !== "report" || prior.work_kind !== "report" || prior.report_quota_recovery !== undefined + || request.resume_turn_key !== undefined || prior.resume_turn_key !== undefined + || Object.keys(recovery).sort().join() !== "native_observation,original_turn_id" + || typeof original.turn_id !== "string" || !/^[a-f0-9]{32}$/.test(original.turn_id) + || recovery.original_turn_id !== original.turn_id || original.session_id !== session.session_id + || original.origin !== "native_loopx" || original.loopx_execution !== true || original.status !== "completed" + || original.delta_count !== 0 || typeof original.upstream_turn_id !== "string" || !original.upstream_turn_id + || Object.keys(prior).length !== keys.length + || keys.some(key => key === "turn_instance_id" + ? request[key] !== String(prior[key]) + ".quota-recovery" + : JSON.stringify(request[key]) !== JSON.stringify(prior[key])) + || result.schema_version !== "loopx_turn_execution_v0" || result.ok !== false || result.mode !== "run_once" + || result.status !== "failed" || result.result_kind !== "host_failure" + || failure.schema_version !== "loopx_turn_host_failure_v0" || failure.attempt !== 1 + || effects.host_invoked !== true + || ["state_written", "quota_spent", "scheduler_acknowledged"].some(key => effects[key] !== false || recorded[key] !== false) + || receipt.ok !== true || receipt.failed_phase !== "host_execute" || receipt.status !== "failed" + || JSON.stringify(receipt.completed_phases) !== "[]" || JSON.stringify(inspection.completed_phases) !== "[]" + || lineage.goal_id !== actor.goal_id || lineage.agent_id !== actor.agent_id || lineage.todo_id !== request.todo_id + || inspection.schema_version !== "loopx_turn_journal_inspection_v1" || inspection.ok !== true + || ["goal_matches", "owner_matches", "turn_key_matches", "journal_consistent"].some(key => inspection[key] !== true) + || inspection.journal_status !== "failed" + || ![null, true].includes(recorded.host_invoked as null | boolean) + || observation.thread_id !== session.upstream_thread_id || typeof observation.thread_id !== "string" || !observation.thread_id + || observation.turn_id !== original.upstream_turn_id || observation.only_original_user_item !== true + || observation.codex_error_info !== "usageLimitExceeded" || observation.quota_available !== true + || typeof observation.quota_observed_at !== "number" || !Number.isFinite(observation.quota_observed_at) + || typeof basis.observed_at !== "number" || !Number.isFinite(basis.observed_at) + || accepting && (basis.generation_present !== false + || basis.observed_at < observation.quota_observed_at || basis.observed_at - observation.quota_observed_at > 60)) { + throw new EffectRuntimeRequestError("report continuation requires its original native user-only quota refusal and current usable allowance"); + } +} + /** 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; @@ -151,7 +209,7 @@ export function admitManagedSessionTurn(params: JsonObject): JsonObject { const scheduledWork = request.work_kind === "scheduled_task"; const requestFields = ["operation", "host", "actor", "turn_instance_id", "todo_id", "validation_command", "work_kind", ...(observationOnly ? ["monitor_deadline", "monitor_lease_version"] : []), - ...(reportWork ? ["report_window_id", "report_subscription_revision"] : []), + ...(reportWork ? ["report_window_id", "report_subscription_revision", "report_quota_recovery"] : []), ...(scheduledWork ? ["schedule_window_id", "schedule_subscription_revision"] : [])]; if (Object.keys(request).some(key => !requestFields.includes(key)) || request.operation !== "native_turn" || request.host !== "codex-app-server" || !["native_turn", "monitor", "report", "scheduled_task"].includes(String(request.work_kind)) @@ -234,6 +292,7 @@ export function admitManagedSessionTurn(params: JsonObject): JsonObject { throw new EffectRuntimeRequestError("the original conversation is busy or no longer owns this Turn; retain its current work"); } admitReportSelectionRecovery(params, request, session.session_id); + admitReportQuotaRecovery(params, request, session); return {actor, workspace_path: goal.workspace_path, host: "codex-app-server", observation_only: observationOnly}; } @@ -337,7 +396,7 @@ export function admitStewardManagedTurn(params: JsonObject): JsonObject { } const request = requireJsonObject(params.request, "native steward Turn request"); const fields = ["operation", "host", "actor", "turn_instance_id", "todo_id", "validation_command", "work_kind", - "monitor_deadline", "monitor_lease_version", "report_window_id", "report_subscription_revision", "resume_turn_key"]; + "monitor_deadline", "monitor_lease_version", "report_window_id", "report_subscription_revision", "resume_turn_key", "report_quota_recovery"]; if (Object.keys(request).some(key => !fields.includes(key)) || request.operation !== "native_turn" || request.host !== "codex-app-server" || ![undefined, "native_turn", "monitor", "report"].includes(request.work_kind as string | undefined) @@ -352,6 +411,7 @@ export function admitStewardManagedTurn(params: JsonObject): JsonObject { throw new EffectRuntimeRequestError("the host's native validator must be a bounded argv"); } admitReportSelectionRecovery(params, request, current.sessionId); + admitReportQuotaRecovery(params, request, current.session); 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/extensions/lark/steward_reports.py b/loopx/extensions/lark/steward_reports.py index fed02e40c8..13576f7977 100644 --- a/loopx/extensions/lark/steward_reports.py +++ b/loopx/extensions/lark/steward_reports.py @@ -16,7 +16,7 @@ from ...file_lock import exclusive_file_lock from ...registry import atomic_write_json from ...capabilities.native_chat.steward_reports import StewardReports -from ...capabilities.native_chat.steward_turn import native_steward_client_id +from ...capabilities.native_chat.steward_turn import report_turn from ...capabilities.periodic_report.cadence_journal import read_cadence_journal from ...capabilities.periodic_report.private_generation import attempt_directory, validate_private_generation from ...capabilities.periodic_report.incremental import commit_periodic_report_publication_cursor @@ -50,8 +50,7 @@ def pending(self) -> bool: def _committed_turn(self, actor: dict[str, Any], window: dict[str, Any]) -> dict[str, Any] | None: # Same stable client identity as the original native host. Reading a # retained completion/event never replays an unknown or failed Turn. - client = native_steward_client_id(actor, "report." + window["window_id"]) - turn = self.controller.store.turn_for_client(actor["session_id"], client) + turn = report_turn(self.controller.store, actor, window) if not turn: return None request = turn.get("loopx_request") or {} diff --git a/tests/control_plane_ts/steward_reports.test.ts b/tests/control_plane_ts/steward_reports.test.ts index d14e0d9545..6114d5dd79 100644 --- a/tests/control_plane_ts/steward_reports.test.ts +++ b/tests/control_plane_ts/steward_reports.test.ts @@ -232,6 +232,72 @@ test("a model report requires the exact live calendar window and original task, ]) assert.throws(() => admitStewardManagedTurn(invalid)); }); +test("a quota refusal admits one fresh report identity only with native no-effects and usable-allowance evidence", () => { + const p = original(); + const prior = {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 effects = {host_invoked: true, state_written: false, quota_spent: false, scheduler_acknowledged: false}; + const result = {schema_version: "loopx_turn_execution_v0", mode: "run_once", ok: false, + status: "failed", result_kind: "host_failure", effects, + host_failure: {schema_version: "loopx_turn_host_failure_v0", kind: "unknown", attempt: 1}, + receipt: {ok: true, failed_phase: "host_execute", status: "failed", completed_phases: [], + lineage: {goal_id: p.actor.goal_id, agent_id: "steward", todo_id: prior.todo_id}}}; + const source = {turn_id: "1".repeat(32), session_id: p.session.session_id, status: "completed", + origin: "native_loopx", loopx_execution: true, delta_count: 0, upstream_turn_id: "quota-native-turn", + loopx_request: prior, response: {native_loopx: result}}; + const observation = {thread_id: "original-native-thread", turn_id: source.upstream_turn_id, + only_original_user_item: true, codex_error_info: "usageLimitExceeded", quota_available: true, quota_observed_at: 100}; + const recovery = {original_turn_id: source.turn_id, native_observation: observation}; + const request = {...prior, turn_instance_id: prior.turn_instance_id + ".quota-recovery", report_quota_recovery: recovery}; + 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: "failed", completed_phases: [], recorded_effects: {...effects, host_invoked: null}}; + const basis = {original: source, inspection, generation_present: false, observed_at: 101}; + const input = {...p, session: {...p.session, upstream_thread_id: observation.thread_id}, request, + report_quota_basis: basis, + 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}, + report_subscription: {enabled: true, profile_preset: "portfolio-progress", + route_ref: `steward-private:${p.binding.binding_id}:${p.source}`, effective_revision: prior.report_subscription_revision}, + report_todo: {todo_id: prior.todo_id, target_key: prior.report_window_id, + claimed_by: "steward", action_kind: "generate_periodic_report", status: "open"}}; + const before = structuredClone(input); + assert.equal(admitStewardManagedTurn(input).observation_only, false); + assert.deepEqual(input, before); + for (const invalid of [ + {...input, report_quota_basis: null}, {...input, request: {...request, report_quota_recovery: null}}, + {...input, request: {...request, turn_instance_id: prior.turn_instance_id}}, + {...input, request: {...request, todo_id: "todo_other"}}, + {...input, request: {...request, report_window_id: "another-window"}}, + {...input, request: {...request, report_subscription_revision: "changed"}}, + {...input, report_subscription: {...input.report_subscription, enabled: false}}, + {...input, goal_active: false}, {...input, binding: {...input.binding, grant: "revoked"}}, + {...input, session: {...input.session, active_turn_id: "user-work"}}, {...input, queued_turn_count: 1}, + {...input, session: {...input.session, upstream_thread_id: "another-thread"}}, + ...["quota_available", "only_original_user_item"].map(key => ({...input, request: {...request, + report_quota_recovery: {...recovery, native_observation: {...observation, [key]: false}}}})), + ...["other", "serverOverloaded", null].map(codex_error_info => ({...input, request: {...request, + report_quota_recovery: {...recovery, native_observation: {...observation, codex_error_info}}}})), + ...[0, 102, "100", null].map(quota_observed_at => ({...input, request: {...request, + report_quota_recovery: {...recovery, native_observation: {...observation, quota_observed_at}}}})), + {...input, request: {...request, report_quota_recovery: {...recovery, original_turn_id: "2".repeat(32)}}}, + {...input, report_quota_basis: {...basis, original: {...source, delta_count: 1}}}, + {...input, report_quota_basis: {...basis, original: {...source, status: "running"}}}, + {...input, report_quota_basis: {...basis, original: {...source, + loopx_request: {...prior, report_quota_recovery: recovery}}}}, + {...input, report_quota_basis: {...basis, generation_present: true}}, + ...["state_written", "quota_spent", "scheduler_acknowledged"].map(key => ({...input, + report_quota_basis: {...basis, original: {...source, response: {native_loopx: {...result, + effects: {...effects, [key]: true}}}}}})), + ...["goal_matches", "owner_matches", "turn_key_matches", "journal_consistent"].map(key => ({...input, + report_quota_basis: {...basis, inspection: {...inspection, [key]: false}}})), + {...input, report_quota_basis: {...basis, inspection: {...inspection, completed_phases: ["host_execute"]}}}, + {...input, report_quota_basis: {...basis, inspection: {...inspection, + recorded_effects: {...inspection.recorded_effects, state_written: null}}}}, + ]) assert.throws(() => admitStewardManagedTurn(invalid)); +}); + test("a retained pre-publication launch refusal can resume only its original native transaction", () => { const p = original(); const prior = {operation: "native_turn", host: "codex-app-server", actor: p.actor, diff --git a/tests/test_native_steward_reports.py b/tests/test_native_steward_reports.py index 44c2f2ddfd..4f7897c587 100644 --- a/tests/test_native_steward_reports.py +++ b/tests/test_native_steward_reports.py @@ -462,6 +462,117 @@ 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("recovery_succeeds", [False, True]) +def test_report_quota_failure_continues_once_after_native_no_effects_and_available_allowance( + steward, tmp_path, monkeypatch, provider, local_session, recovery_succeeds): + import time + from datetime import timezone + from loopx.chat_agent import CodexChatAgentError + from loopx.capabilities.native_chat.codex_adapter import CodexAppServerAdapter + from loopx.control_plane.turn_driver.journal_store import turn_journal_path + from loopx.capabilities.periodic_report import cadence_runtime + import loopx.capabilities.native_chat.steward_reports as report_host + 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) + store, runtime, _, transport, binding, capture, _ = steward + 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"] + install_model(steward, tmp_path, sid) + calls = [] + + def exhausted(adapter, _prompt, event_sink): + calls.append(adapter.upstream_thread_id) + event_sink("turn.started", {"upstream_turn_id": "quota-refused-native-turn"}) + raise CodexChatAgentError("The provider allowance was exhausted.") + + real_start = CodexAppServerAdapter.start_turn + monkeypatch.setattr(CodexAppServerAdapter, "start_turn", exhausted) + original = finish(runtime, pump_steward_monitors(runtime)[0]) + native = original["response"]["native_loopx"] + assert native["status"] == "failed" and native["result_kind"] == "host_failure" + assert native["host_failure"]["kind"] == "unknown" + original_path = store._turn_path(sid, original["turn_id"]) + journal_path = turn_journal_path(reports.root, goal_id=actor["goal_id"], turn_key=native["resume_turn_key"]) + frozen = original_path.read_bytes(), journal_path.read_bytes() + _, window = reports.current_window(owner) + observations = [] + proof = {"thread_id": store.load_session(sid)["upstream_thread_id"], + "turn_id": original["upstream_turn_id"], "only_original_user_item": True, + "codex_error_info": "usageLimitExceeded", "quota_available": False, + "quota_observed_at": time.time()} + + def observe(**kwargs): + observations.append(kwargs) + return dict(proof) + + monkeypatch.setattr(CodexAppServerAdapter, "failed_request_observation", staticmethod(observe)) + runtime._steward_monitor_next_check.clear() + assert pump_steward_monitors(runtime) == [] # Reset time alone is not usable quota. + assert len(calls) == 1 + proof.update(quota_available=True, quota_observed_at=time.time()) + if recovery_succeeds: + monkeypatch.setattr(CodexAppServerAdapter, "start_turn", real_start) + runtime._steward_monitor_next_check.clear() + old_adapter = runtime.adapters[sid] + old_thread = store.load_session(sid)["upstream_thread_id"] + continued = pump_steward_monitors(runtime) + assert len(continued) == 1, "the original report must not remain stranded after proven quota recovery" + result = finish(runtime, continued[0]) + assert runtime.adapters[sid] is not old_adapter + assert old_adapter.session.process.poll() is not None + assert store.load_session(sid)["upstream_thread_id"] == old_thread + assert result["turn_id"] != original["turn_id"] + request = result["loopx_request"] + assert request["report_window_id"] == window["window_id"] + assert request["todo_id"] == original["loopx_request"]["todo_id"] + assert request["actor"] == original["loopx_request"]["actor"] + assert request["turn_instance_id"] == original["loopx_request"]["turn_instance_id"] + ".quota-recovery" + assert result["response"]["native_loopx"]["status"] == ("committed" if recovery_succeeds else "failed"), result + assert frozen == (original_path.read_bytes(), journal_path.read_bytes()) + assert observations and all(x["turn_id"] == original["upstream_turn_id"] for x in observations) + assert all(x["thread_ids"] == [proof["thread_id"]] for x in observations) + if not recovery_succeeds: + assert not reports.health(owner)["publication_verified"] + assert len(calls) == 2 + elif local_session: + assert reports.health(owner)["publication_verified"] + else: + assert transport.steward_reports.reconcile() == 1 + assert reports.health(owner)["publication_verified"] + before = capture.read_bytes(), len(observations), store.messages(sid) + runtime._steward_monitor_next_check.clear() + assert pump_steward_monitors(runtime) == [] + assert before == (capture.read_bytes(), len(observations), store.messages(sid)) + if recovery_succeeds: + point += timedelta(days=1) + runtime._steward_monitor_next_check.clear() + following = pump_steward_monitors(runtime) + assert len(following) == 1 + next_result = finish(runtime, following[0]) + assert next_result["response"]["native_loopx"]["status"] == "committed", next_result + assert "report_quota_recovery" not in next_result["loopx_request"] + assert next_result["loopx_request"]["report_window_id"] != window["window_id"] + assert len(observations) == before[1] # A new window does not inherit old recovery. + if not local_session: + assert transport.steward_reports.reconcile() == 1 + assert reports.health(owner)["publication_verified"] + assert frozen == (original_path.read_bytes(), journal_path.read_bytes()) + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) @pytest.mark.parametrize("local_session", [False, True]) @pytest.mark.parametrize("replan_outcome", ["settled", "wait", "invalid_semantic", "already_admitted", "retained_probe"])