fix(interaction): share closed-contract consistency checks

docs/reference/effect-interpreter-packet.md

@@ -48,6 +48,41 @@ journal replay 使用自己的判决，不再输出动作字段。host 提供的
 解释为 quota 决策。这只改变瞬时结果投射，不重写落盘的 host result、receipt
 或 journal plan。
 
+## Interaction consistency validation
+
+The TypeScript interaction-contract owner supplies the execution-consistency
+rules used by the Effect interpreter, scoped-override projection and Python
+scheduler arbitration. Delivery requires an attempt; a quiet no-op cannot
+coexist with required work or user action. Terminal, peer-blocked and
+monitor-only modes require `quiet_noop_allowed=true`, no required action or
+delivery, and no quota spend. An already-settled mode permits neither required
+work/user action, delivery nor another spend. Non-blocking notifications remain valid.
+Independent capability intent retains its own mode and existing authority.
+
+This strengthens transient contract validation: typed consumers now reject the
+closed-mode contradictions that scheduler arbitration already rejected, and
+all consumers reject a closed mode with `spend_allowed_now=true` or
+`spend_after_validation=true`. TurnEnvelope construction uses the same interpreter
+before signing its action. Missing spend metadata remains false for older packets;
+a present value must be boolean. Python retains structural diagnostics and cadence selection, while
+semantic rules come from the existing Effect runtime. Runtime unavailability
+cannot fall back to a second Python rule. No persisted receipt is rewritten and
+these checks do not grant execution, lease or charging authority.
+
+TypeScript interaction-contract owner 统一提供 Effect 解释器、scoped-override
+投影与 Python 调度仲裁使用的执行一致性规则。delivery 必须有 attempt；quiet
+no-op 不能与必须执行的工作或用户动作并存。terminal、peer-blocked、monitor-only
+模式必须允许 quiet no-op，不得要求动作、允许交付或扣费；已结算模式也不得重新
+要求工作或用户动作、允许交付或扣费。非阻塞通知仍然有效。
+独立 capability intent 保留自己的模式与现有权限。
+
+本次加强瞬时合同校验：类型化消费者开始拒绝此前调度仲裁已拒绝的关闭模式矛盾，
+消费者都拒绝关闭模式的 `spend_allowed_now=true` 或 `spend_after_validation=true`；
+TurnEnvelope 构造在签署动作前也使用同一解释器。旧 packet 缺失扣费字段时仍按
+false 处理，显式提供的值必须是布尔值。Python 保留结构诊断与 cadence
+选择，语义规则来自已有 Effect runtime；运行时不可用时不回退到另一套 Python
+判断。不重写历史回执，也不授予执行、租约或扣费权限。
+
 ## Turn Journal Lens
 
 `interpret_turn_journal` reads an existing fenced Turn journal and returns an

loopx/control_plane/effect_runtime_handlers.ts

@@ -745,6 +745,7 @@ export function createEffectRuntimeHandlers(
     ["turn.settlement.reduce", lazyHandler(() => import("./turn_driver/settlement.ts"), ({reduceTurnSettlementTransaction}) => reduceTurnSettlementTransaction)],
     ["turn.host_todo_completion.evaluate", lazyHandler(() => import("./turn_driver/host_todo_completion.ts"), ({evaluateHostTodoCompletion}) => evaluateHostTodoCompletion)],
     ["work_item.interaction_reads.project", lazyHandler(() => import("./work_items/interaction_contract.ts"), ({projectInteractionRequiredReads}) => projectInteractionRequiredReads)],
+    ["work_item.interaction_consistency.validate", lazyHandler(() => import("./work_items/interaction_contract.ts"), ({projectInteractionConsistencyValidation}) => projectInteractionConsistencyValidation)],
     ["work_item.context.plan", lazyHandler(() => import("./work_items/interaction_contract.ts"), ({planInteractionWorkContext}) => planInteractionWorkContext)],
     ["work_item.context.project", lazyHandler(() => import("./work_items/interaction_contract.ts"), ({projectInteractionWorkContext}) => projectInteractionWorkContext)],
     ["work_item.replan_settlement.project", lazyHandler(() => import("./work_items/replan_settlement.ts"), ({projectReplanSettlementContract}) => projectReplanSettlementContract)],

loopx/control_plane/scheduler/arbitration.py

@@ -160,20 +160,22 @@ def build_scheduler_arbitration(
         errors=errors,
     )
 
-    if delivery_allowed and not must_attempt:
-        errors.append("interaction_contract.delivery_without_attempt")
-    if quiet_noop_allowed and (must_attempt or delivery_allowed or user_required):
-        errors.append("interaction_contract.quiet_noop_conflicts_with_required_action")
-    if mode in {
-        "terminal_no_followup",
-        "peer_coordination_blocked",
-        "agent_monitor_only",
-    } and (
-        user_required or must_attempt or delivery_allowed or not quiet_noop_allowed
+    from ..effect_runtime import effect_runtime_result
+
+    cli_channel = _mapping(contract.get("cli_channel"))
+    spend_flags = {field: _required_bool(
+        cli_channel, field, error_prefix="interaction_contract.cli_channel", errors=errors,
+    ) if field in cli_channel else False for field in ("spend_allowed_now", "spend_after_validation")}
+    validation = effect_runtime_result("work_item.interaction_consistency.validate", {
+        "mode": mode, "user_required": user_required, "must_attempt": must_attempt,
+        "delivery_allowed": delivery_allowed, "quiet_noop_allowed": quiet_noop_allowed,
+        **spend_flags,
+    })
+    if not isinstance(validation, dict) or not isinstance(validation.get("errors"), list) or any(
+        not isinstance(error, str) for error in validation["errors"]
     ):
-        errors.append("interaction_contract.terminal_conflicts_with_open_action")
-    if mode == "heartbeat_settled_skip" and (user_required or must_attempt or delivery_allowed):
-        errors.append("interaction_contract.settled_conflicts_with_open_action")
+        raise TypeError("interaction consistency validation returned invalid diagnostics")
+    errors.extend(validation["errors"])
     disposition, reason_code = _classify_disposition(
         mode=mode,
         user_required=user_required,

loopx/control_plane/work_items/interaction_contract.ts

@@ -47,6 +47,63 @@ export interface InteractionContract extends JsonObject {
   cli_channel: JsonObject;
 }
 
+const INTERACTION_CONSISTENCY_DIAGNOSTICS = {
+  delivery: {code: "interaction_contract.delivery_without_attempt",
+    message: "interaction_contract.agent_channel cannot allow delivery without an attempt"},
+  quiet: {code: "interaction_contract.quiet_noop_conflicts_with_required_action",
+    message: "interaction_contract quiet no-op conflicts with a required action"},
+  terminal: {code: "interaction_contract.terminal_conflicts_with_open_action",
+    message: "interaction_contract closed mode conflicts with an open action"},
+  settled: {code: "interaction_contract.settled_conflicts_with_open_action",
+    message: "interaction_contract settled mode conflicts with an open action"},
+  spend: {code: "interaction_contract.settled_conflicts_with_spend",
+    message: "interaction_contract settled mode conflicts with quota spend"},
+} as const;
+
+interface InteractionConsistencyFacts {
+  mode: string;
+  user_required: boolean;
+  must_attempt: boolean;
+  delivery_allowed: boolean;
+  quiet_noop_allowed: boolean;
+  spend_allowed_now: boolean;
+  spend_after_validation: boolean;
+}
+
+/** The scheduler and typed consumers share these rules, not independent copies. */
+function interactionConsistencyIssues(facts: InteractionConsistencyFacts) {
+  const {mode, user_required, must_attempt, delivery_allowed, quiet_noop_allowed} = facts;
+  const spendAllowed = facts.spend_allowed_now || facts.spend_after_validation;
+  const issues: Array<(typeof INTERACTION_CONSISTENCY_DIAGNOSTICS)[keyof typeof INTERACTION_CONSISTENCY_DIAGNOSTICS]> = [];
+  if (delivery_allowed && !must_attempt) issues.push(INTERACTION_CONSISTENCY_DIAGNOSTICS.delivery);
+  if (quiet_noop_allowed && (must_attempt || delivery_allowed || user_required)) {
+    issues.push(INTERACTION_CONSISTENCY_DIAGNOSTICS.quiet);
+  }
+  if ((mode === "terminal_no_followup" || mode === "peer_coordination_blocked" || mode === "agent_monitor_only")
+      && (user_required || must_attempt || delivery_allowed || !quiet_noop_allowed || spendAllowed)) {
+    issues.push(INTERACTION_CONSISTENCY_DIAGNOSTICS.terminal);
+  }
+  if (mode === "heartbeat_settled_skip") {
+    if (user_required || must_attempt || delivery_allowed) issues.push(INTERACTION_CONSISTENCY_DIAGNOSTICS.settled);
+    if (spendAllowed) issues.push(INTERACTION_CONSISTENCY_DIAGNOSTICS.spend);
+  }
+  return issues;
+}
+
+/** Python normalizes its legacy structural diagnostics, then asks this owner. */
+export function projectInteractionConsistencyValidation(request: JsonObject): JsonObject {
+  if (typeof request.mode !== "string") throw new EffectRuntimeRequestError("interaction consistency mode must be a string");
+  return {errors: interactionConsistencyIssues({
+    mode: request.mode,
+    user_required: requireBoolean(request.user_required, "user_required"),
+    must_attempt: requireBoolean(request.must_attempt, "must_attempt"),
+    delivery_allowed: requireBoolean(request.delivery_allowed, "delivery_allowed"),
+    quiet_noop_allowed: requireBoolean(request.quiet_noop_allowed, "quiet_noop_allowed"),
+    spend_allowed_now: requireBoolean(request.spend_allowed_now, "spend_allowed_now"),
+    spend_after_validation: requireBoolean(request.spend_after_validation, "spend_after_validation"),
+  }).map(issue => issue.code)};
+}
+
 function decodeUserChannel(value: unknown): UserInteractionChannel {
   const channel = requireJsonObject(value, "interaction_contract.user_channel");
   const actionRequired = requireBoolean(
@@ -106,16 +163,10 @@ function decodeAgentChannel(
     channel.quiet_noop_allowed,
     "interaction_contract.agent_channel.quiet_noop_allowed",
   );
-  if (deliveryAllowed && !mustAttempt) {
-    throw new EffectRuntimeRequestError(
-      "interaction_contract.agent_channel cannot allow delivery without an attempt",
-    );
-  }
-  if (quietNoopAllowed && (mustAttempt || deliveryAllowed || userActionRequired)) {
-    throw new EffectRuntimeRequestError(
-      "interaction_contract quiet no-op conflicts with a required action",
-    );
-  }
+  const issues = interactionConsistencyIssues({mode: "", user_required: userActionRequired,
+    must_attempt: mustAttempt, delivery_allowed: deliveryAllowed,
+    quiet_noop_allowed: quietNoopAllowed, spend_allowed_now: false, spend_after_validation: false});
+  if (issues.length) throw new EffectRuntimeRequestError(issues[0]!.message);
   if (mustAttempt) {
     return {
       ...channel,
@@ -145,16 +196,23 @@ export function decodeInteractionContract(value: unknown): InteractionContract {
     contract.agent_channel,
     { userActionRequired: userChannel.action_required },
   );
+  const mode = requireNonEmptyString(contract.mode, "interaction_contract.mode");
+  const cliChannel = requireJsonObject(contract.cli_channel, "interaction_contract.cli_channel");
+  const issues = interactionConsistencyIssues({mode, user_required: userChannel.action_required,
+    must_attempt: agentChannel.must_attempt, delivery_allowed: agentChannel.delivery_allowed,
+    quiet_noop_allowed: agentChannel.quiet_noop_allowed,
+    spend_allowed_now: cliChannel.spend_allowed_now === undefined ? false
+      : requireBoolean(cliChannel.spend_allowed_now, "interaction_contract.cli_channel.spend_allowed_now"),
+    spend_after_validation: cliChannel.spend_after_validation === undefined ? false
+      : requireBoolean(cliChannel.spend_after_validation, "interaction_contract.cli_channel.spend_after_validation")});
+  if (issues.length) throw new EffectRuntimeRequestError(issues[0]!.message);
   return {
     ...contract,
     schema_version: INTERACTION_CONTRACT_SCHEMA_VERSION,
-    mode: requireNonEmptyString(contract.mode, "interaction_contract.mode"),
+    mode,
     user_channel: userChannel,
     agent_channel: agentChannel,
-    cli_channel: requireJsonObject(
-      contract.cli_channel,
-      "interaction_contract.cli_channel",
-    ),
+    cli_channel: cliChannel,
   };
 }
 

tests/control_plane/test_settled_replay_construction.py

@@ -10,6 +10,9 @@
 from loopx.presentation.renderers.quota_markdown import render_quota_should_run_markdown
 from loopx.control_plane.scheduler.execution_context import scheduler_execution_context_for_runtime_profile
 from loopx.control_plane.scheduler.arbitration import SchedulerDisposition, build_scheduler_arbitration
+from loopx.control_plane.effect_program import interpret_quota_should_run_packet
+from loopx.control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_result
+from loopx.control_plane.quota.turn_envelope import build_turn_envelope
 
 
 @pytest.mark.parametrize("profile", ["codex_app_heartbeat", "trae_app", "ark_managed_agent_goal"])
@@ -57,6 +60,53 @@ def test_settled_contract_cannot_carry_open_execution(channel, field):
     assert "interaction_contract.settled_conflicts_with_open_action" in result.errors
 
 
+@pytest.mark.parametrize("mode", ["heartbeat_settled_skip", "terminal_no_followup",
+                                  "peer_coordination_blocked", "agent_monitor_only"])
+def test_closed_contract_cannot_reopen_work_through_typed_consumers(mode):
+    payload = build_quota_should_run(
+        quota_status_payload(goal_id="settled-fixture", status="active", recommended_action="Continue"),
+        goal_id="settled-fixture", receipt_bound_replay_phase=ReceiptBoundReplayPhase.SETTLED,
+    )
+    contract = payload["interaction_contract"]
+    contract["mode"] = mode
+    contract["agent_channel"].update(must_attempt=True, delivery_allowed=True, quiet_noop_allowed=False)
+    assert build_scheduler_arbitration(payload).disposition is SchedulerDisposition.CONSISTENCY_REPAIR
+    with pytest.raises(EffectRuntimeRejected, match="conflicts with"):
+        interpret_quota_should_run_packet(payload)
+    with pytest.raises(EffectRuntimeRejected, match="conflicts with"):
+        build_turn_envelope(payload)
+    with pytest.raises(EffectRuntimeRejected, match="conflicts with"):
+        effect_runtime_result("quota.scoped_override.project", {
+            "override": {}, "interaction_contract": contract,
+            "selected_todo": {"todo_id": "todo_other", "text": "Advance other work"},
+        })
+
+
+@pytest.mark.parametrize("mode,error_code", [
+    ("heartbeat_settled_skip", "interaction_contract.settled_conflicts_with_spend"),
+    ("terminal_no_followup", "interaction_contract.terminal_conflicts_with_open_action"),
+    ("peer_coordination_blocked", "interaction_contract.terminal_conflicts_with_open_action"),
+    ("agent_monitor_only", "interaction_contract.terminal_conflicts_with_open_action"),
+])
+@pytest.mark.parametrize("spend_field", ["spend_allowed_now", "spend_after_validation"])
+def test_closed_contract_cannot_authorize_another_spend(mode, error_code, spend_field):
+    payload = build_quota_should_run(
+        quota_status_payload(goal_id="settled-fixture", status="active", recommended_action="Continue"),
+        goal_id="settled-fixture", receipt_bound_replay_phase=ReceiptBoundReplayPhase.SETTLED,
+    )
+    contract = payload["interaction_contract"]
+    contract["mode"] = mode
+    contract["agent_channel"]["quiet_noop_allowed"] = mode != "heartbeat_settled_skip"
+    contract["cli_channel"][spend_field] = True
+    result = build_scheduler_arbitration(payload)
+    assert result.disposition is SchedulerDisposition.CONSISTENCY_REPAIR
+    assert error_code in result.errors
+    with pytest.raises(EffectRuntimeRejected, match="conflicts with"):
+        interpret_quota_should_run_packet(payload)
+    with pytest.raises(EffectRuntimeRejected, match="conflicts with"):
+        build_turn_envelope(payload)
+
+
 
 @pytest.mark.parametrize("quota_state", ["eligible", "operator_gate", "waiting_external", "exhausted"])
 @pytest.mark.parametrize("monitor", [False, True])

tests/control_plane_ts/interaction_contract.test.ts

@@ -6,6 +6,7 @@ import type { JsonObject } from "../../loopx/control_plane/effect_program.ts";
 import { interpretQuotaShouldRunPacket } from "../../loopx/control_plane/effect_program.ts";
 import {
   decodeInteractionContract,
+  projectInteractionConsistencyValidation,
   projectInteractionRequiredReads,
   projectInteractionWorkContext,
   type AgentInteractionChannel,
@@ -20,6 +21,32 @@ const invalidDeliveryChannel: AgentInteractionChannel = {
 };
 void invalidDeliveryChannel;
 
+test("a closed contract can notify without requiring user action or spending", () => {
+  const contract = {schema_version: "loopx_interaction_contract_v0", mode: "terminal_no_followup",
+    user_channel: {action_required: false, notify: "NOTIFY"},
+    agent_channel: {must_attempt: false, delivery_allowed: false, quiet_noop_allowed: true},
+    cli_channel: {next_cli_actions: ["no quota spend until explicit goal resume"]}};
+  assert.deepEqual(decodeInteractionContract(contract), contract);
+});
+
+test("shared consistency validation retains diagnostic order and does not gate independent capability intent", () => {
+  const invalid = {mode: "heartbeat_settled_skip", user_required: true,
+    must_attempt: false, delivery_allowed: true, quiet_noop_allowed: true,
+    spend_allowed_now: false, spend_after_validation: true};
+  assert.deepEqual(projectInteractionConsistencyValidation(invalid), {errors: [
+    "interaction_contract.delivery_without_attempt",
+    "interaction_contract.quiet_noop_conflicts_with_required_action",
+    "interaction_contract.settled_conflicts_with_open_action",
+    "interaction_contract.settled_conflicts_with_spend",
+  ]});
+  assert.deepEqual(projectInteractionConsistencyValidation({
+    mode: "governed_capability_intent", user_required: true,
+    must_attempt: true, delivery_allowed: true, quiet_noop_allowed: false,
+    spend_allowed_now: false, spend_after_validation: false,
+  }), {errors: []});
+  assert.throws(() => projectInteractionConsistencyValidation({...invalid, must_attempt: "false"}), /must_attempt.*boolean/);
+});
+
 function readFacts(): JsonObject {
   return {required_reads: [{command: "read-existing-evidence", source: "existing"}],
     goal_id: "goal-requirements", goal_state_file: "/tmp/goal source/state's requirements.md",