fix(settlement): classify quota refusals from typed actions

docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md

@@ -386,6 +386,16 @@ Stop or narrow M7 when any kill criterion holds:
   adapter binds validation to the existing atomic lease write while pure
   eligibility, conflict, file-lock, and CAS rules remain task-lease-owned
   (#3091, #3095).
+- Quota-spend callback failures use the optional `quota_rejection_action` fact,
+  copied from the existing effective-action owner only when quota admission
+  itself refuses the spend. `throttled_skip` maps to `budget_rejected` in the
+  TypeScript settlement owner. A nested quota snapshot is not that fact;
+  writeback and terminal refusals retain their own failure kinds. Missing,
+  malformed or unknown actions retain `quota_spend_rejected`, even when the
+  message contains "budget". This changes the former text-based diagnostic
+  classification, not admission, charging or retry authority. Previously stored
+  failure receipts are read unchanged; new callbacks must not rely on message
+  wording to request a budget classification.
 - Scheduler apply, ACK, failure writeback, and cadence remain data-encoded host
   handoffs outside agent-owned settlement.
 - `interpret_quota_should_run_packet` and `interpret_turn_result_packet` remain

docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md

@@ -283,6 +283,7 @@ M7 只有在至少产生一个下列最终 effect 时才有理由存在：
 - 默认 Codex App / CLI quota 路径构建一份 typed settlement plan，把 validation、durable writeback、quota spend 和 conditional terminal closeout 绑定到原始 turn effect identity。final `no_followup` 是 spend 后 effect；普通 successor completion 仍属于 Todo lifecycle（#3016、#3033、#3034）。
 - 隔离 turn driver 通过自己的 callback executor 消费同一套 plan、identity、receipt、failure、replay 和 short-circuit algebra（#3020、#3023）；terminal closeout 单独写入 journal，因此 closeout 失败只重试 closeout，不重复 writeback/spend；loop controller 从已提交 receipt chain 派生 continuation，不再维护第二份 settlement truth（#3024）。
 - Task-lease acquire 是第一个有界采用该 algebra 的非 Turn 核心路径。adapter 把 validation 绑定到现有原子 lease write；纯 eligibility、conflict、file-lock 和 CAS 规则仍由 task-lease bounded context 持有（#3091、#3095）。
+- quota spend 的 callback 失败使用可选事实 `quota_rejection_action`，只有配额准入本身拒绝扣费时才从既有 effective-action owner 复制该值。TypeScript 结算 owner 将 `throttled_skip` 映射到 `budget_rejected`；嵌套 quota 快照不构成这项事实，writeback 和 terminal 拒绝保留各自的失败类型。值缺失、格式错误或未知时保留 `quota_spend_rejected`，即使说明文字包含 "budget"。这改变此前依赖文本的诊断分类，不改变准入、扣费或重试权限。已保存的失败回执保持原样读取；新 callback 不得依赖文案要求预算分类。
 - Scheduler apply、ACK、failure writeback 和 cadence 仍是 agent-owned settlement 之外的数据化 host handoff。
 - `interpret_quota_should_run_packet` 与 `interpret_turn_result_packet` 继续作为 packet lens；`EffectProgram` 和 `effect_program_from_ordered_steps` 继续为 bootstrap 与本地 scheduler construction 提供兼容的 ordered-step reader。
 - Outcome-continuity wait 已按因果关系判断。没有 material trigger 和 fresh evidence-linked path decision 的 `unchanged_with_reason` checkpoint，不能清除更早的 material checkpoint 或五条 Todo 完成长链 gap。这是有意的 qualification 行为，不是 watch-ACK 集成回归（#2998、#3009、#3022）。

loopx/control_plane/effect_program.ts

@@ -1,6 +1,7 @@
 import { EffectRuntimeRequestError } from "./effect_runtime_errors.ts";
 import { isStringLiteral } from "./runtime_decode.ts";
 import { decodeInteractionContract } from "./work_items/interaction_contract.ts";
+import { EffectiveAction } from "./quota/effective_action.generated.ts";
 
 export {
   RECEIPT_BOUND_MONITOR_PHASES,
@@ -907,13 +908,16 @@ export function settlementNextAction(options: {
   return { decision: "complete", step_kind: null, result: seeded };
 }
 
+/** Only the actual quota admission refusal carries a rejecting action.
+ * Missing/unknown facts retain a stage failure; diagnostic text is never authority. */
 export function callbackFailureKind(
   stepKind: SettlementStepKind,
-  reason: string,
+  payload: JsonObject,
 ): SettlementFailureKind {
   if (stepKind === "durable_writeback") return "writeback_rejected";
   if (stepKind === "terminal_closeout") return "terminal_closeout_rejected";
-  if (reason.toLowerCase().includes("budget")) return "budget_rejected";
+  if (stepKind === "quota_spend" &&
+      payload.quota_rejection_action === EffectiveAction.THROTTLED_SKIP) return "budget_rejected";
   return "quota_spend_rejected";
 }
 
@@ -932,7 +936,7 @@ export function commitStepPayload(options: {
       `${stepKind} callback rejected the settlement`;
     return {
       result: settlementFailed({
-        kind: callbackFailureKind(stepKind, reason),
+        kind: callbackFailureKind(stepKind, payload),
         step_kind: stepKind,
         reason,
       }),

loopx/control_plane/quota/slot_accounting.py

@@ -859,6 +859,10 @@ def build_quota_slot_preview_for_decision(
             "appended": False,
             "registry_mutated": False,
             "reason": before.get("reason") or "goal is not eligible for quota accounting preview",
+            # Transport the actual rejecting action, not an incidental quota
+            # snapshot on an earlier validation/workspace refusal. TS owns the
+            # mapping to settlement failure kinds; prose carries no such rule.
+            "quota_rejection_action": before.get("effective_action") if before.get("ok") is True else None,
             "before": before,
             "after": None,
         }

tests/architecture/test_settlement_receipt_source_boundaries.py

@@ -129,7 +129,7 @@ def _turn(identity, **overrides):
                 "quota_spend_payload": None,
                 "failed_provider_attempt": {
                     "step_kind": "quota_spend",
-                    "payload": {"ok": False, "reason": "budget exhausted"},
+                    "payload": {"ok": False, "quota_rejection_action": "throttled_skip", "reason": "quota exhausted"},
                 },
             },
             "budget_rejected",

tests/control_plane/test_quota_slot_accounting.py

@@ -120,6 +120,36 @@ def _poll(
     return payload
 
 
+@pytest.mark.parametrize("reason", ["额度已用尽", "quota limit reached", "budget exhausted"])
+def test_admission_refusal_transports_its_action_independently_of_prose(tmp_path: Path, reason: str) -> None:
+    preview = _preview(tmp_path, before_overrides={
+        "state": "throttled", "effective_action": "throttled_skip", "reason": reason,
+    })
+    assert preview["ok"] is False and preview["appended"] is False
+    assert preview["quota_rejection_action"] == "throttled_skip"
+    assert preview["reason"] == reason
+
+
+def test_invalid_decision_cannot_supply_a_quota_rejection_action(tmp_path: Path) -> None:
+    preview = _preview(tmp_path, before_overrides={"ok": False, "effective_action": "throttled_skip"})
+    assert preview["ok"] is False
+    assert preview["quota_rejection_action"] is None
+
+
+def test_binding_refusal_does_not_borrow_an_incidental_throttled_action(tmp_path: Path) -> None:
+    preview = build_quota_slot_preview_for_decision(
+        {}, goal_id=GOAL_ID, source="adapter", todo_id="todo_requested",
+        before={"ok": True, "should_run": False, "effective_action": "throttled_skip",
+                "selected_todo": {"todo_id": "todo_other"}},
+        after_decision=lambda _: pytest.fail("a rejected spend must not project success"),
+        quota_status_builder=lambda **_: pytest.fail("a rejected spend must not change quota"),
+        self_repair_spend_actions=frozenset(),
+    )
+    assert preview["ok"] is False
+    assert "quota_rejection_action" not in preview
+    assert "binding mismatch" in preview["reason"]
+
+
 def _spent(generated_at: str, *, agent_id: str | None = None) -> dict[str, Any]:
     payload: dict[str, Any] = {
         "generated_at": generated_at,

tests/control_plane_ts/effect_program.test.ts

@@ -380,12 +380,44 @@ test("commit reduction advances one phase only after a committed payload", () =>
     identity: identityInput,
     step_kind: "quota_spend",
     transaction_phases: ["validation", "durable_writeback", "quota_spend"],
-    payload: { ok: false, error: "budget exhausted" },
+    payload: { ok: false, quota_rejection_action: "throttled_skip", error: "quota exhausted" },
   });
   assert.equal(rejected.completed_phases, null);
   assert.equal(rejected.result.failure?.kind, "budget_rejected");
 });
 
+test("callback failure classification follows the rejecting action, never explanatory prose", () => {
+  const cases = [
+    [{quota_rejection_action: "throttled_skip", reason: "额度已用尽"}, "budget_rejected"],
+    [{quota_rejection_action: "throttled_skip", reason: "budget was exhausted"}, "budget_rejected"],
+    [{reason: "transport failed; budget was not exhausted"}, "quota_spend_rejected"],
+    [{reason: "budget exhausted"}, "quota_spend_rejected"],
+    [{quota_rejection_action: "blocked_wait", reason: "waiting for a budget report"}, "quota_spend_rejected"],
+    [{quota_rejection_action: "future_action", reason: "budget"}, "quota_spend_rejected"],
+    [{quota_rejection_action: 1, reason: "budget"}, "quota_spend_rejected"],
+    [{before: {effective_action: "throttled_skip"}, reason: "workspace refused"}, "quota_spend_rejected"],
+  ] as const;
+  for (const [fields, expected] of cases) {
+    const rejected = commitStepPayload({
+      identity: identityInput, step_kind: "quota_spend",
+      transaction_phases: ["validation", "durable_writeback", "quota_spend"],
+      payload: {ok: false, appended: false, ...fields},
+    });
+    assert.equal(rejected.result.failure?.kind, expected, JSON.stringify(fields));
+    assert.equal(rejected.completed_phases, null);
+    assert.deepEqual(rejected.result.receipts, []);
+  }
+  for (const [step_kind, expected] of [
+    ["durable_writeback", "writeback_rejected"],
+    ["terminal_closeout", "terminal_closeout_rejected"],
+  ] as const) {
+    assert.equal(commitStepPayload({
+      identity: identityInput, step_kind, transaction_phases: [step_kind],
+      payload: {ok: false, quota_rejection_action: "throttled_skip", reason: "budget"},
+    }).result.failure?.kind, expected);
+  }
+});
+
 test("runtime selects the next uncommitted effect instead of Python orchestration", () => {
   const next = settlementNextAction({
     identity: identityInput,

tests/test_loopx_turn_executor.py

@@ -3711,9 +3711,12 @@ def denied_spend() -> dict[str, object]:
     assert calls == {"host": 1, "writeback": 1, "spend": 2}
 
 
-def test_budget_rejection_is_typed_and_retry_does_not_repeat_writeback(
+def test_quota_admission_rejection_is_typed_and_retry_does_not_repeat_writeback(
     tmp_path: Path,
 ) -> None:
+    from loopx.control_plane.testing.quota_fixtures import quota_status_payload, quota_todo_item
+    from loopx.quota import spend_quota_slot
+
     plan = _plan()
     calls = {"host": 0, "writeback": 0, "spend": 0}
 
@@ -3727,11 +3730,12 @@ def writeback(_result: dict[str, object]) -> dict[str, object]:
 
     def reject_budget() -> dict[str, object]:
         calls["spend"] += 1
-        return {
-            "ok": False,
-            "appended": False,
-            "reason": "quota budget rejected",
-        }
+        status = quota_status_payload(
+            goal_id="fixture-goal", status="active", recommended_action="continue",
+            quota_state="throttled", quota_extra={"spent_slots": 10, "allowed_slots": 10},
+            agent_todo_items=[quota_todo_item(todo_id="todo_fixture", title="Deliver a result")],
+        )
+        return spend_quota_slot(status, goal_id="fixture-goal", source="adapter", execute=True)
 
     common = {
         "host_runner": host,
@@ -3756,6 +3760,14 @@ def reject_budget() -> dict[str, object]:
     assert failed["effects"]["state_written"] is True
     assert failed["effects"]["quota_spent"] is False
 
+    # Stored failure kinds remain historical facts, even if a later callback
+    # would return a differently worded refusal. Replay runs no provider.
+    replayed = run_loopx_turn_once(
+        plan, spend=lambda: pytest.fail("failed replay must not call the provider"), **common,
+    )
+    assert replayed["settlement_result"] == failed["settlement_result"]
+    assert calls == {"host": 1, "writeback": 1, "spend": 1}
+
     recovered = run_loopx_turn_once(
         plan,
         spend=lambda: (