fix(quota): align Turn settlement outcome guidance

loopx/control_plane/quota/settlement_plan.ts

@@ -7,7 +7,7 @@ import {
 import {VISION_MATERIAL_CLOSEOUT_HINT, visionAuthoringContract} from "../goals/vision_checkpoint.ts";
 import {parseExactGoalRef} from "../goals/goal_instance_identity.ts";
 import {requireJsonObject, requireNonEmptyString, requireStringArray} from "../runtime_decode.ts";
-import {DELIVERY_OUTCOMES} from "../work_items/delivery_outcome.ts";
+import {MATERIAL_DELIVERY_OUTCOMES} from "../work_items/delivery_outcome.ts";
 import {DELIVERY_BOUNDARIES} from "../turn_driver/delivery_continuity.ts";
 import {PROGRESS_RESULT_CLASSES} from "../work_items/replan_history.ts";
 
@@ -24,7 +24,9 @@ export function turnScopedCliSettlementPlan(params: JsonObject): SettlementPlan
   requireNonEmptyString(scales.map(scale => requireNonEmptyString(scale, "delivery batch scale")).join("|"),
     "delivery batch scale choices");
   const inputHint = " Fill placeholders from the actual result: classification is a descriptive label; " +
-    "scale=" + scales.join("|") + "; outcome=" + DELIVERY_OUTCOMES.join("|") + ". " +
+    "scale=" + scales.join("|") + "; outcome=" + MATERIAL_DELIVERY_OUTCOMES.join("|") + ". " +
+    "surface_only cannot settle this Turn. Do not relabel preparation-only work as progress; " +
+    "continue authorized work within existing limits or report a genuine typed blocker. " +
     "Set --delivery-boundary " + DELIVERY_BOUNDARIES.join("|") + " for the applicable case above. " +
     "Typed evidence: --progress-result-class " + PROGRESS_RESULT_CLASSES.join("|") +
     ". For blocked outcome_gap, --progress-work-item-id must match this plan's Todo/obligation id " +

tests/control_plane/test_quota_settlement_cli.py

@@ -2440,12 +2440,20 @@ def test_same_todo_writeback_reentry_only_settles_original_turn(
     assert Path(refreshed["json_path"]).read_bytes() == writeback_bytes
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
 def test_in_flight_progress_preserves_todo_across_heartbeat_settlements(
-    tmp_path: Path,
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str,
 ) -> None:
+    from canonical_authority_fixture import isolate_sqlite_runtime
+    from loopx.control_plane.work_items.delivery_outcome import (
+        qualifies_turn_scoped_settlement,
+    )
+
     project, runtime, registry_path = _write_fixture(tmp_path)
     first_turn_id = "turn-settlement-continuation-1"
-    _initialize_fixture_authority(project, runtime)
+    if provider == "sqlite":
+        isolate_sqlite_runtime(tmp_path, monkeypatch)
+    _initialize_fixture_authority(project, runtime, provider=provider)
     guard_rc, guard = _run_cli(
         registry_path,
         runtime,
@@ -2469,6 +2477,29 @@ def test_in_flight_progress_preserves_todo_across_heartbeat_settlements(
     writeback_step = guard["interaction_contract"]["cli_channel"]["settlement_plan"]["ordered_steps"][1]
     assert "vision_authoring" not in writeback_step
     assert "outcome_gap requires semantic_closeout" in writeback_step["precondition"]
+    offered = writeback_step["precondition"].split("; outcome=", 1)[1].split(".", 1)[0].split("|")
+    blocked = {
+        "schema_version": "typed_progress_observation_v0", "result_class": "blocked",
+        "work_item_id": TODO_ID, "blocker_id": "blocker_fixture", "evidence_ids": ["evidence:fixture"],
+    }
+    assert offered and all(qualifies_turn_scoped_settlement(
+        outcome, blocked, work_item_id=TODO_ID,
+    ) for outcome in offered)
+
+    # A caller must not escape an honest preparation result by renaming it.
+    # Refusing it leaves this original Turn available for actual progress.
+    refused_rc, refused = _run_cli(
+        registry_path, runtime, "refresh-state", "--goal-id", GOAL_ID,
+        "--classification", "preparation_only", "--delivery-batch-scale", "single_surface",
+        "--delivery-outcome", "surface_only", "--agent-id", AGENT_ID,
+        "--todo-id", TODO_ID, "--turn-instance-id", first_turn_id,
+        "--no-global-sync", "--suppress-external-sinks",
+    )
+    assert refused_rc == 1 and refused["appended"] is False, refused
+    assert "surface_only cannot settle a Turn" in refused["error"]
+    assert _classification_count(runtime, "preparation_only") == 0
+    assert _spend_run_count(runtime) == 0
+    assert "outcome_progress" in offered
 
     refresh_rc, refresh = _run_cli(
         registry_path,

tests/control_plane_ts/quota_settlement_plan.test.ts

@@ -4,6 +4,8 @@ import {settlementPlanPayload} from "../../loopx/control_plane/effect_program.ts
 import {visionAuthoringContract} from "../../loopx/control_plane/goals/vision_checkpoint.ts";
 import {projectMcpInteraction} from "../../loopx/control_plane/turn_driver/host_interaction.ts";
 import {turnScopedCliSettlementPlan} from "../../loopx/control_plane/quota/settlement_plan.ts";
+import {DELIVERY_OUTCOMES, decodeOptionalDeliveryOutcome, isTurnScopedSettlementOutcome}
+  from "../../loopx/control_plane/work_items/delivery_outcome.ts";
 
 const input = {
   identity: {goal_id: "goal", agent_id: "agent", todo_id: "todo", turn_instance_id: "turn"},
@@ -14,6 +16,30 @@ const input = {
   },
 };
 
+for (const lane of ["ordinary", "in_flight", "autonomous_replan"]) {
+  test(`${lane} settlement offers only outcomes that its writer can qualify`, () => {
+    const identity = lane === "autonomous_replan"
+      ? {goal_id: "goal", agent_id: "agent", replan_obligation_id: "replan", turn_instance_id: "turn"}
+      : input.identity;
+    const plan = turnScopedCliSettlementPlan({...input, identity,
+      ...(lane === "in_flight" ? {delivery_boundary: "in_flight_continuation"} : {}),
+    });
+    const guidance = plan.steps[1].precondition;
+    const offered = guidance.match(/; outcome=([a-z_|]+)\./)?.[1].split("|");
+    const binding = lane === "autonomous_replan" ? "replan" : "todo";
+    const blocked = {schema_version: "typed_progress_observation_v0", result_class: "blocked",
+      work_item_id: binding, blocker_id: "blocker", evidence_ids: ["evidence:failure"]};
+    // The writer's qualification rule, rather than the projection's source list,
+    // is the independent oracle for this Turn's advertised choices.
+    assert.deepEqual(offered, DELIVERY_OUTCOMES.filter(outcome =>
+      isTurnScopedSettlementOutcome(outcome, blocked, binding)));
+    assert.equal(isTurnScopedSettlementOutcome("outcome_gap", undefined, binding), false);
+    assert.equal(decodeOptionalDeliveryOutcome("surface_only"), "surface_only");
+    assert.equal(isTurnScopedSettlementOutcome("surface_only", blocked, binding), false);
+    assert.match(guidance, /do not relabel preparation-only work as progress/i);
+  });
+}
+
 test("ordinary completion is conditional on deliverable acceptance, not Turn accounting", () => {
   const plan = turnScopedCliSettlementPlan(input);
   assert.deepEqual(plan.steps.map(step => step.kind),