Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions docs/product/periodic-reports.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
26 changes: 18 additions & 8 deletions loopx/capabilities/native_chat/steward_turn.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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 (
Expand All @@ -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()}

Expand Down Expand Up @@ -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:
Expand Down
21 changes: 16 additions & 5 deletions loopx/control_plane/collaboration/steward_managed_mode.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Expand All @@ -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
Expand Down
20 changes: 14 additions & 6 deletions tests/control_plane_ts/steward_reports.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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},
Expand All @@ -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,
Expand Down
53 changes: 51 additions & 2 deletions tests/test_native_steward_reports.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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()
Expand Down Expand Up @@ -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
Expand Down
Loading