diff --git a/loopx/control_plane/quota/refresh_external_delivery.py b/loopx/control_plane/quota/refresh_external_delivery.py index 385813584..fa1e1d197 100644 --- a/loopx/control_plane/quota/refresh_external_delivery.py +++ b/loopx/control_plane/quota/refresh_external_delivery.py @@ -3,7 +3,70 @@ from typing import Any from ...rollout_event_log import append_rollout_event, build_rollout_event, rollout_event_log_path -from .settlement import QuotaSettlementReadback, settlement_result_payload +from .effect_program import ( + SettlementStepKind, + build_turn_scoped_cli_settlement_plan, +) +from .settlement import ( + QuotaSettlementReadback, + settlement_result_payload, + settlement_step_command, +) + +TURN_SETTLEMENT_OWED_SCHEMA_VERSION = "turn_settlement_owed_v0" + + +def turn_settlement_owed( + readback: QuotaSettlementReadback, + *, + command_prefix: str = "loopx", +) -> dict[str, Any] | None: + """Name the settlement step a committed writeback still owes. + + A turn-scoped writeback appends the accountable run, but the Turn is only + settled once the quota spend for the same effect identity exists. Reporting + the writeback as finished hid that debt until the next wake raised + ``unsettled_host_turn_recovery``, so the owed step is named here together + with the exact command that pays it under the same identity. + """ + + identity = readback.identity.value + if identity is None or readback.spend_run is not None: + return None + owed: dict[str, Any] = { + "schema_version": TURN_SETTLEMENT_OWED_SCHEMA_VERSION, + "kind": "quota_spend", + "effect_id": identity.effect_id, + "goal_id": identity.goal_id, + "agent_id": identity.agent_id, + "todo_id": identity.todo_id, + "replan_obligation_id": identity.replan_obligation_id, + "turn_instance_id": identity.turn_instance_id, + "reason": ( + "the writeback is committed but this Turn still owes its quota spend; " + "spend once for the same settlement identity and do not treat the " + "writeback alone as closeout" + ), + "recovery_does_not_spend": True, + } + if bool(identity.todo_id) == bool(identity.replan_obligation_id): + # A turn-scoped identity carries exactly one work item. Without it the + # debt is still owed, but no exact command can be offered. + return owed + plan = build_turn_scoped_cli_settlement_plan( + goal_id=identity.goal_id, + agent_id=identity.agent_id, + command_prefix=command_prefix, + todo_id=identity.todo_id, + replan_obligation_id=identity.replan_obligation_id, + scoped_cli_args="", + lifecycle_actor_args="", + turn_instance_id=identity.turn_instance_id, + ) + command = settlement_step_command(plan.as_dict(), SettlementStepKind.QUOTA_SPEND) + if command: + owed["command"] = command + return owed def finish_external_delivery_refresh( @@ -17,6 +80,13 @@ def finish_external_delivery_refresh( raise RuntimeError("TypeScript refresh external delivery result missing or invalid") payload["external_delivery"] = {k: v for k, v in plan.items() if k != "transition"} payload["external_sink_delivery_authorized"] = plan["authorized"] is True + if payload.get("ok") and not dry_run: + owed = turn_settlement_owed(readback) + if owed is not None: + # The committed writeback is not a closeout: name the step and the + # exact command this Turn still owes so the caller cannot mistake + # one for the other. + payload["settlement_owed"] = owed transition = plan.get("transition") if payload.get("ok") and transition and not dry_run: identity = readback.identity.value diff --git a/tests/control_plane/test_turn_settlement_owed_signal.py b/tests/control_plane/test_turn_settlement_owed_signal.py new file mode 100644 index 000000000..4a3fa321f --- /dev/null +++ b/tests/control_plane/test_turn_settlement_owed_signal.py @@ -0,0 +1,195 @@ +"""A committed writeback names the quota spend the Turn still owes.""" + +from __future__ import annotations + +import json +from pathlib import Path + +from loopx.control_plane.quota.effect_program import SettlementIdentity +from loopx.control_plane.quota.refresh_external_delivery import turn_settlement_owed +from loopx.control_plane.quota.settlement import read_heartbeat_settlement +from loopx.rollout_event_log import rollout_event_log_path +from loopx.state_refresh import refresh_state_run + +GOAL_ID = "goal-owed-spend" +AGENT_ID = "agent-owed-spend" +TODO_ID = "todo_owed_spend" +TURN_ID = "turn-owed-spend" + +STATE_TEXT = f"""# Active Goal State + +## Agent Todo + +- [ ] [P1] finish the owed spend signal + +""" + + +def _append_guard_receipt(runtime_root: Path, identity: SettlementIdentity) -> None: + path = rollout_event_log_path(runtime_root, GOAL_ID) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps( + { + "schema_version": "loopx_rollout_event_v0", + "event_id": "event-owed-guard", + "event_kind": "quota_should_run", + "goal_id": GOAL_ID, + "agent_id": AGENT_ID, + "run_id": TURN_ID, + "details": { + "todo_id": TODO_ID, + "settlement_effect_id": identity.effect_id, + }, + } + ) + + "\n", + encoding="utf-8", + ) + + +def _append_run_index_record(runtime_root: Path, record: dict) -> None: + path = runtime_root / "goals" / GOAL_ID / "runs" / "index.jsonl" + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(record) + "\n") + + +def _readback(runtime_root: Path, *, spent: bool): + identity = SettlementIdentity(GOAL_ID, AGENT_ID, TODO_ID, TURN_ID) + _append_guard_receipt(runtime_root, identity) + _append_run_index_record( + runtime_root, + { + "classification": "state_refreshed", + "delivery_outcome": "outcome_progress", + "goal_id": GOAL_ID, + "agent_id": AGENT_ID, + "todo_id": TODO_ID, + "turn_instance_id": TURN_ID, + }, + ) + if spent: + _append_run_index_record( + runtime_root, + { + "classification": "quota_slot_spent", + "goal_id": GOAL_ID, + "agent_id": AGENT_ID, + "todo_id": TODO_ID, + "turn_instance_id": TURN_ID, + }, + ) + readback = read_heartbeat_settlement( + runtime_root, + goal_id=GOAL_ID, + agent_id=AGENT_ID, + todo_id=TODO_ID, + turn_instance_id=TURN_ID, + ) + assert readback is not None + return identity, readback + + +def test_unspent_turn_names_the_owed_quota_spend(tmp_path: Path) -> None: + identity, readback = _readback(tmp_path / "runtime", spent=False) + + owed = turn_settlement_owed(readback) + + assert owed is not None + assert owed["schema_version"] == "turn_settlement_owed_v0" + assert owed["kind"] == "quota_spend" + assert owed["effect_id"] == identity.effect_id + assert owed["todo_id"] == TODO_ID + assert owed["turn_instance_id"] == TURN_ID + assert owed["recovery_does_not_spend"] is True + assert owed["command"].startswith("loopx quota spend-slot") + assert f"--todo-id {TODO_ID}" in owed["command"] + assert f"--turn-instance-id {TURN_ID}" in owed["command"] + assert "--execute" in owed["command"] + + +def test_spent_turn_owes_nothing(tmp_path: Path) -> None: + _identity, readback = _readback(tmp_path / "runtime", spent=True) + + assert readback.spend_run is not None + assert turn_settlement_owed(readback) is None + + +def _fixture(tmp_path: Path) -> tuple[Path, Path, Path]: + project = tmp_path / "project" + state_path = project / ".codex" / "goals" / GOAL_ID / "ACTIVE_GOAL_STATE.md" + state_path.parent.mkdir(parents=True) + state_path.write_text(STATE_TEXT, encoding="utf-8") + registry_path = tmp_path / "registry.json" + registry_path.write_text( + json.dumps( + { + "goals": [ + { + "id": GOAL_ID, + "status": "active", + "repo": str(project), + "state_file": str(state_path.relative_to(project)), + "coordination": { + "agent_model": "peer_v1", + "registered_agents": [AGENT_ID], + }, + "workspace_guard_policy": { + "peer_independent_worktree_required": False, + }, + } + ] + } + ), + encoding="utf-8", + ) + return registry_path, project, tmp_path / "runtime" + + +def test_refresh_state_result_carries_the_owed_step(tmp_path: Path) -> None: + registry_path, project, runtime_root = _fixture(tmp_path) + identity = SettlementIdentity(GOAL_ID, AGENT_ID, TODO_ID, TURN_ID) + _append_guard_receipt(runtime_root, identity) + + result = refresh_state_run( + registry_path=registry_path, + runtime_root_override=str(runtime_root), + goal_id=GOAL_ID, + project=project, + state_file=None, + classification="validated_progress", + recommended_action=None, + delivery_batch_scale="single_surface", + delivery_outcome="outcome_progress", + delivery_workspace_path=project, + todo_id=TODO_ID, + turn_instance_id=TURN_ID, + agent_id=AGENT_ID, + dry_run=False, + sync_global=False, + ) + + assert result["ok"] is True + assert result["appended"] is True + owed = result["settlement_owed"] + assert owed["effect_id"] == identity.effect_id + assert owed["kind"] == "quota_spend" + assert f"--turn-instance-id {TURN_ID}" in owed["command"] + assert "settlement_owed" not in refresh_state_run( + registry_path=registry_path, + runtime_root_override=str(runtime_root), + goal_id=GOAL_ID, + project=project, + state_file=None, + classification="validated_progress", + recommended_action=None, + delivery_batch_scale="single_surface", + delivery_outcome="outcome_progress", + delivery_workspace_path=project, + todo_id=TODO_ID, + turn_instance_id=TURN_ID, + agent_id=AGENT_ID, + dry_run=True, + sync_global=False, + )