fix(quota): name the quota spend a committed writeback still owes

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

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
+  <!-- loopx:todo status=open task_class=advancement_task claimed_by={AGENT_ID} todo_id={TODO_ID} -->
+"""
+
+
+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,
+    )