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
22 changes: 18 additions & 4 deletions loopx/control_plane/quota/blocked_wait.ts
Original file line number Diff line number Diff line change
Expand Up @@ -223,10 +223,22 @@ export function projectReceiptBoundWait(value: unknown): JsonObject {
const matches = todos.filter(row => row.todo_id === request.todo_id);
const todo = matches[0];
if (matches.length !== 1 || !todo || todo.role !== "agent" ||
todo.archive_state !== "active" || (todo.claimed_by && todo.claimed_by !== request.agent_id)) {
(todo.claimed_by && todo.claimed_by !== request.agent_id)) {
return {status: "none"};
}
const progress = jsonObject(request.settlement_progress);
// Ordinary completion is a validation receipt, not Turn settlement. The
// verified replay phase, rather than terminal Todo status alone, owns this
// recovery even after the row leaves the active presentation lane.
const completedSettlement = todo.status === "done" && todo.task_class === "advancement_task" &&
request.replay_phase === "settlement_pending" &&
progress?.schema_version === "quota_settlement_progress_v0";
if (completedSettlement && progress?.next_step === "durable_writeback" &&
(progress.state === "writeback_required" || progress.state === "writeback_receipt_required")) {
const reason = "The original Todo's completion is verified, but its Turn still owes durable writeback and settlement. Record only the validated original result, then spend once for that same identity. Inspect existing effects before retrying; do not repeat completed work, invent a successor, or rebind this Turn.";
return boundRecovery(request, "durable_writeback", reason);
}
if (todo.archive_state !== "active" && !completedSettlement) return {status: "none"};
if (progress?.schema_version === "quota_settlement_progress_v0" &&
progress.next_step === "quota_spend" &&
(progress.state === "spend_required" || progress.state === "spend_receipt_required")) {
Expand Down Expand Up @@ -262,10 +274,10 @@ export function projectReceiptBoundWait(value: unknown): JsonObject {

/** Recovery routes preserve the existing binding and grant no independent work. */
function boundRecovery(
request: JsonObject, repair: "lifecycle" | "blocked_writeback" | "quota_spend" | "dependency", reason: string, wait?: JsonObject,
request: JsonObject, repair: "lifecycle" | "blocked_writeback" | "durable_writeback" | "quota_spend" | "dependency", reason: string, wait?: JsonObject,
): JsonObject {
const obligation = repair === "blocked_writeback"
? "close_bound_wait_without_spend" : repair === "quota_spend"
? "close_bound_wait_without_spend" : repair === "quota_spend" || repair === "durable_writeback"
? "settle_bound_turn" : repair === "dependency"
? "repair_bound_dependency" : "repair_bound_monitor_lifecycle";
return {
Expand All @@ -279,7 +291,9 @@ function boundRecovery(
contract_obligation: obligation, must_attempt_work: true,
delivery_allowed: false, notify: "DONT_NOTIFY",
...(repair === "quota_spend" ? {spend_allowed_now: true} : {}),
spend_policy: repair === "quota_spend"
spend_policy: repair === "durable_writeback"
? "settle the original completed Todo after validated writeback; no independent work"
: repair === "quota_spend"
? "settle the original verified writeback once; no independent work or new writeback"
: "no spend for blocked closeout",
reason_code: repair === "blocked_writeback" ? "receipt_bound_pending_wait" : "unsettled_host_turn",
Expand Down
1 change: 1 addition & 0 deletions loopx/control_plane/quota/live_decision.py
Original file line number Diff line number Diff line change
Expand Up @@ -754,6 +754,7 @@ def build_live_quota_should_run_decision(
scheduler_execution_context=resolved_context,
monitor_phase=(receipt_bound_monitor_phase.value if receipt_bound_monitor_phase else None),
settlement_progress=settlement_readback.progress if settlement_readback else None,
replay_phase=receipt_bound_replay_phase,
)
if hook_dispatch["failures"]:
payload["capability_hook_dispatch"] = {
Expand Down
53 changes: 50 additions & 3 deletions loopx/control_plane/quota/unsettled_host_turn.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,14 +22,20 @@
from ..scheduler.execution_context import SchedulerExecutionContextResolution
from ..todos.todo_semantics import todo_item_task_class
from ..work_items.interaction_contract import (
_scoped_cli_args,
build_interaction_contract,
)
from .error_codes import (
CloseoutQueryUnavailableError,
HeartbeatReceiptIdentityConflictError,
)
from .accounting_admission import quota_accounting_admission
from .effect_program import ReceiptBoundReplayPhase
from .effect_program import (
ReceiptBoundReplayPhase,
SettlementStepKind,
build_turn_scoped_cli_settlement_plan,
settlement_step_command,
)

UNSETTLED_HOST_TURN_RECOVERY_SCHEMA_VERSION = "unsettled_host_turn_recovery_v0"

Expand Down Expand Up @@ -359,6 +365,7 @@ def apply_receipt_bound_wait_recovery(
scheduler_execution_context: Mapping[str, Any] | SchedulerExecutionContextResolution | None,
monitor_phase: str | None = None,
settlement_progress: dict[str, Any] | None = None,
replay_phase: ReceiptBoundReplayPhase | None = None,
) -> bool:
"""Read full provider facts only when a replay lost its executable binding."""
from ..todos.list_readback import list_goal_todos
Expand All @@ -370,13 +377,23 @@ def apply_receipt_bound_wait_recovery(
)
# Keep transitive dependencies from this coherent read. A direct-target
# slice loses longer cycles; compact facts keep prose out of the RPC.
items = compact_settlement_wait_todos(source["todos"], todo_id=todo_id)
source_items = source["todos"]
if not any(item.get("todo_id") == todo_id for item in source_items):
bound = _bound_todo_item(
registry_path=registry_path, runtime_root=runtime_root,
goal_id=goal_id, todo_id=todo_id,
)
if bound is None:
raise ValueError("The original receipt-bound Todo cannot be read; recover its lifecycle before settling this Turn")
source_items = [*source_items, bound]
items = compact_settlement_wait_todos(source_items, todo_id=todo_id)
verdict = effect_runtime_result("quota.settlement.read", {
"schema_version": "loopx_quota_receipt_bound_wait_request_v0",
"todos": items, "todo_id": todo_id, "agent_id": agent_id,
"turn_instance_id": turn_instance_id, "observed_at": now_utc_iso(),
"monitor_phase": monitor_phase,
"settlement_progress": settlement_progress,
"replay_phase": replay_phase.value if replay_phase else None,
}, large_local_snapshot=True)
if verdict.get("status") == "none":
return False
Expand All @@ -388,9 +405,39 @@ def apply_receipt_bound_wait_recovery(
from .settlement_precedence import clear_quota_action_projections

clear_quota_action_projections(payload)
return _apply_recovery_projection(payload, verdict=verdict, registry_path=registry_path, runtime_root=runtime_root,
_apply_recovery_projection(payload, verdict=verdict, registry_path=registry_path, runtime_root=runtime_root,
current_turn_instance_id=turn_instance_id, available_capabilities=available_capabilities,
scheduler_execution_context=scheduler_execution_context)
if verdict["recovery"].get("repair") == "durable_writeback":
# Transport the typed recovery through the existing command owner. A
# completed Todo is not an executable selected_todo or a new grant.
from ..work_items.action_selection_contract import render_cli_command_prefix

assert settlement_progress is not None
plan = build_turn_scoped_cli_settlement_plan(
goal_id=goal_id, agent_id=agent_id, todo_id=todo_id,
turn_instance_id=turn_instance_id,
command_prefix=render_cli_command_prefix(
registry_path=str(registry_path), runtime_root=str(runtime_root),
),
scoped_cli_args=_scoped_cli_args(
{"agent_id": agent_id}, available_capabilities=available_capabilities,
),
lifecycle_actor_args="",
quota_spend_source=settlement_progress["quota_spend_source"],
goal_ref=payload.get("goal_ref"),
).as_dict()
cli = payload["interaction_contract"]["cli_channel"]
cli.update({
"settlement_plan": plan,
"next_cli_actions": [
settlement_step_command(plan, kind)
for kind in (SettlementStepKind.DURABLE_WRITEBACK, SettlementStepKind.QUOTA_SPEND)
],
"spend_after_validation": True,
"quota_spend_source": settlement_progress["quota_spend_source"],
})
return True


def _apply_recovery_projection(
Expand Down
117 changes: 117 additions & 0 deletions tests/control_plane/test_quota_settlement_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -7238,6 +7238,123 @@ def test_same_turn_receipt_replay_defers_newly_due_higher_priority_monitor(
assert resumed_turn["selected_todo"]["todo_id"] == DUE_MONITOR_TODO_ID


@pytest.mark.parametrize("source_unavailable", [False, True])
def test_bound_recovery_does_not_infer_missing_original_todo(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, source_unavailable: bool,
) -> None:
from loopx.control_plane.quota import unsettled_host_turn as recovery
from loopx.control_plane.todos import list_readback

def read(**kwargs):
if kwargs.get("todo_id"):
if source_unavailable:
raise ValueError("exact source unavailable")
return {"todo": None}
return {"todos": []}

monkeypatch.setattr(list_readback, "list_goal_todos", read)
monkeypatch.setattr(recovery, "effect_runtime_result", lambda *args, **kwargs: pytest.fail("unknown source reached the decision owner"))
payload = {"ok": True}
with pytest.raises(ValueError, match="exact source unavailable|original receipt-bound Todo cannot be read"):
recovery.apply_receipt_bound_wait_recovery(
payload, registry_path=tmp_path / "registry.json", runtime_root=tmp_path,
goal_id=GOAL_ID, agent_id=AGENT_ID, todo_id=TODO_ID, turn_instance_id=TURN_ID,
available_capabilities=[], scheduler_execution_context=None,
)
assert payload == {"ok": True}


@pytest.mark.parametrize("provider", ["file", "sqlite"])
@pytest.mark.parametrize("archived_with_alternatives", [False, True])
def test_completed_todo_reentry_preserves_original_settlement_commands(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, archived_with_alternatives: bool,
) -> None:
from canonical_authority_fixture import isolate_sqlite_runtime

project, runtime, registry_path = _write_fixture(tmp_path)
state_path = _configure_read_only_todo(project)
if archived_with_alternatives:
_configure_selectable_alternative(project)
if provider == "sqlite":
isolate_sqlite_runtime(tmp_path, monkeypatch)
_initialize_fixture_authority(project, runtime, provider=provider)
guard_args = (
"quota", "should-run", "--codex-app", "--goal-id", GOAL_ID,
"--agent-id", AGENT_ID, "--turn-instance-id", TURN_ID,
)
rc, admitted = _run_cli(registry_path, runtime, *guard_args, "--todo-id", TODO_ID)
assert rc == 0, admitted
original_plan = admitted["interaction_contract"]["cli_channel"]["settlement_plan"]
rc, completed = _run_cli(
registry_path, runtime, "todo", "complete", "--goal-id", GOAL_ID,
"--todo-id", TODO_ID, "--agent-id", AGENT_ID,
"--turn-instance-id", TURN_ID, "--evidence", "evidence:verified-read-only-result",
)
assert rc == 0, completed
assert completed["changed"] is True
assert completed["completion_continuation"] == "active_goal"
assert _spend_run_count(runtime) == 0
if archived_with_alternatives:
rc, archived = _run_cli(
registry_path, runtime, "todo", "archive-completed", "--goal-id", GOAL_ID,
"--role", "agent", "--max-active-done", "0", "--execute",
)
assert rc == 0, archived
rc, refused = _run_cli(
registry_path, runtime, *guard_args, "--todo-id", ALTERNATIVE_TODO_ID,
)
assert rc == 1, refused
assert refused["error_code"] == "heartbeat_receipt_identity_conflict"
completed_state = state_path.read_bytes()

rc, resumed = _run_cli(registry_path, runtime, *guard_args)
assert rc == 0, resumed
cli = resumed["interaction_contract"]["cli_channel"]
plan = cli.get("settlement_plan")
assert plan is not None, resumed
assert plan["identity"] == original_plan["identity"]
assert resumed["heartbeat_receipt"]["settlement_identity"] == plan["identity"]
assert not resumed["normal_delivery_allowed"]
assert not resumed["interaction_contract"]["agent_channel"]["delivery_allowed"]
assert resumed.get("selected_todo") is None
assert cli["spend_allowed_now"] is False
assert _heartbeat_receipt_count(runtime, TURN_ID) == 1
assert state_path.read_bytes() == completed_state
commands = {step["kind"]: step["command_template"] for step in plan["ordered_steps"]}
assert cli["next_cli_actions"] == [commands["durable_writeback"], commands["quota_spend"]]
from loopx.control_plane.quota.turn_envelope import build_turn_envelope

envelope = build_turn_envelope(resumed)
assert envelope["writeback"]["settlement_plan"] == plan
assert envelope["writeback"]["next_cli_actions"] == cli["next_cli_actions"]
assert envelope["writeback"]["spend_allowed_now"] is False
refresh = commands["durable_writeback"].replace(
"<validated_progress>", "completed_read_only_result",
).replace("<scale>", "single_surface").replace("<outcome>", "outcome_progress")
refresh += (
" --vision-state vision_closed"
" --vision-acceptance 'The verified read-only result is durably recorded.'"
" --no-global-sync --suppress-external-sinks"
)
rc, written = _run_generated_cli(refresh, registry_path=registry_path)
assert rc == 0, written
rc, replayed = _run_generated_cli(refresh, registry_path=registry_path)
assert rc == 0, replayed
assert _classification_count(runtime, "completed_read_only_result") == 1
rc, owed = _run_cli(registry_path, runtime, *guard_args)
assert rc == 0, owed
assert owed["unsettled_host_turn_recovery"]["repair"] == "quota_spend"
assert owed["settlement_owed"]["todo_id"] == TODO_ID
for command in (owed["settlement_owed"]["command"], commands["quota_spend"]):
rc, spent = _run_generated_cli(command, registry_path=registry_path)
assert rc == 0, spent
assert _spend_run_count(runtime) == 1
rc, settled = _run_cli(registry_path, runtime, *guard_args)
assert rc == 0, settled
assert settled["effective_action"] == "heartbeat_settled_skip"
assert settled["interaction_contract"]["agent_channel"]["must_attempt"] is False


@pytest.mark.parametrize("provider", ["file", "sqlite"])
def test_read_only_settlement_omits_non_causal_delivery_workspace(
tmp_path: Path,
Expand Down
65 changes: 65 additions & 0 deletions tests/control_plane_ts/receipt_bound_wait.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,71 @@ test("a deferred Todo closes its original Turn rather than selecting a replaceme
assert.equal(recovery.turn_instance_id, "host-turn-1");
});

function completedRequest() {
const input = {...request(), replay_phase: "settlement_pending",
settlement_progress: {schema_version: "quota_settlement_progress_v0",
state: "writeback_required", next_step: "durable_writeback"}};
input.todos[0]!.status = "done";
return input;
}

for (const state of ["writeback_required", "writeback_receipt_required"]) {
for (const archive_state of ["active", "archive"]) {
test(`verified completion retains ${state} for its ${archive_state} original row`, () => {
const input = completedRequest();
input.settlement_progress.state = state;
input.todos[0]!.archive_state = archive_state;
const before = structuredClone(input);
const result = projectReceiptBoundWait(input);
const recovery = result.recovery as Record<string, unknown>;
assert.equal(recovery.repair, "durable_writeback");
assert.equal(recovery.binding_id, input.todo_id);
assert.equal(recovery.turn_instance_id, input.turn_instance_id);
const obligation = result.obligation as Record<string, unknown>;
assert.equal(obligation.delivery_allowed, false);
assert.equal(obligation.spend_allowed_now, undefined);
assert.deepEqual(input, before);
});
}
}

for (const replay_phase of [null, "open", "settled", "unknown"]) {
test(`completion status without verified pending phase ${replay_phase} grants no recovery`, () => {
assert.deepEqual(projectReceiptBoundWait({...completedRequest(), replay_phase}), {status: "none"});
});
}

test("archival cannot hide the original completed Turn's verified spend duty", () => {
const input = completedRequest();
input.todos[0]!.archive_state = "archive";
input.settlement_progress = {schema_version: "quota_settlement_progress_v0",
state: "spend_required", next_step: "quota_spend"};
const result = projectReceiptBoundWait(input);
assert.equal((result.recovery as Record<string, unknown>).repair, "quota_spend");
assert.equal((result.obligation as Record<string, unknown>).delivery_allowed, false);
});

for (const progress of [null, {},
{schema_version: "other", state: "writeback_required", next_step: "durable_writeback"},
{schema_version: "quota_settlement_progress_v0", state: "identity_required", next_step: "durable_writeback"},
{schema_version: "quota_settlement_progress_v0", state: "writeback_required", next_step: "quota_spend"},
{schema_version: "quota_settlement_progress_v0", state: "settled", next_step: null},
]) {
test(`completion with unknown or incompatible progress ${JSON.stringify(progress)} grants no recovery`, () => {
assert.deepEqual(projectReceiptBoundWait({...completedRequest(), settlement_progress: progress}), {status: "none"});
});
}

for (const patch of [{claimed_by: "agent-b"}, {role: "user"}, {task_class: "continuous_monitor"}]) {
test(`a mismatched completed row ${JSON.stringify(patch)} cannot borrow settlement recovery`, () => {
const input = completedRequest();
Object.assign(input.todos[0]!, patch);
assert.deepEqual(projectReceiptBoundWait(input), {status: "none"});
assert.deepEqual(projectReceiptBoundWait({...input, todos: []}), {status: "none"});
assert.deepEqual(projectReceiptBoundWait({...input, todos: [input.todos[0], input.todos[0]]}), {status: "none"});
});
}

for (const state of ["spend_required", "spend_receipt_required"]) {
test(`${state} preserves durable writeback before projecting a later dependency wait`, () => {
const input = {...request(), settlement_progress: {schema_version: "quota_settlement_progress_v0",
Expand Down
Loading