Merge pull request #3020 from huangruiteng/codex/m7-2-turn-driver-settlement-20260809

feat(turn): bind typed settlement runtime

Author: huangruiteng

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

@@ -213,6 +213,11 @@ their authority boundaries differ. A generic `Kleisli`, middleware stack,
 executor registry, or general `Effect` monad remains premature until shared
 execution ownership, not just similar packet fields, is proven.
 
+The shared settlement algebra is owned by the core `effect_program` module.
+Quota supplies the Codex App/CLI plan builder and compatibility re-exports;
+each runtime adapter composes the core algebra instead of inheriting a domain
+program or moving its execution authority into a generic base class.
+
 ### Handler Is Data, Not a Callable
 
 Runtime middleware receives a `handler` callable and decides whether to call

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

@@ -175,6 +175,8 @@ A => F[C]
 
 runtime 合同有两个一等调用方。默认 Codex App 路径通过跨 agent/host 边界的 data-encoded CLI effects 结算普通 LoopX turn。隔离 turn driver 通过 in-process callbacks 执行同一 settlement 形状。它们应共享 plan、receipt、effect identity 和 failure algebra，但不需要共享同一个 executor，因为它们的 authority boundary 不同。在共享执行所有权被证明之前，通用 `Kleisli`、middleware stack、executor registry 或通用 `Effect` monad 仍为时过早。
 
+共享 settlement algebra 由核心 `effect_program` 模块拥有。Quota 只提供 Codex App/CLI plan builder 与兼容 re-export；各 runtime adapter 组合核心 algebra，而不是继承领域 program，也不会把自己的执行权上移到通用基类。
+
 ### Handler 是数据，不是 Callable
 
 Runtime middleware 接收一个 `handler` callable，并决定是否调用、调用一次、重试、fallback 或短路。LoopX 无法跨 context 和 session 边界接收 model 或 host callable。相反，interpreter 在 packet 中返回 `next_effect`：CLI actions、scheduler ACK 和 failure hint。host 或下一个自动化 turn 调用这个 data-encoded handler。

loopx/cli_commands/turn.py

@@ -362,6 +362,10 @@ def handle_turn_command(
                 boundary = payload.get("boundary")
                 if isinstance(boundary, dict):
                     boundary.pop("opaque_session_handle_omitted", None)
+            else:
+                transaction = payload.get("transaction")
+                if isinstance(transaction, dict):
+                    transaction.pop("settlement_plan", None)
         elif args.turn_command == "run-once":
             if args.resume_turn_key:
                 if args.turn_instance_id:

loopx/control_plane/effect_program.py

@@ -1,8 +1,16 @@
 from __future__ import annotations
 
-from collections.abc import Mapping, Sequence
+from collections.abc import Callable, Mapping, Sequence
 from dataclasses import dataclass, field
-from typing import Any
+from enum import StrEnum
+from typing import Any, Generic, TypeVar, cast
+
+
+# Keep the established wire ids while moving their implementation to the
+# core-owned algebra shared by quota and Turn adapters.
+SETTLEMENT_IDENTITY_SCHEMA_VERSION = "quota_settlement_identity_v0"
+SETTLEMENT_PLAN_SCHEMA_VERSION = "quota_settlement_plan_v0"
+SETTLEMENT_RECEIPT_SCHEMA_VERSION = "quota_settlement_receipt_v0"
 
 
 @dataclass(frozen=True)
@@ -66,6 +74,184 @@ class EffectProgram:
     execution_mode: str | None = None
 
 
+class SettlementStepKind(StrEnum):
+    VALIDATION = "validation"
+    TODO_COMPLETION = "todo_completion"
+    DURABLE_WRITEBACK = "durable_writeback"
+    QUOTA_SPEND = "quota_spend"
+
+
+class SettlementFailureKind(StrEnum):
+    INVALID_IDENTITY = "invalid_identity"
+    RECEIPT_MISSING = "receipt_missing"
+    IDENTITY_MISMATCH = "identity_mismatch"
+    WRITEBACK_MISSING = "writeback_missing"
+    WRITEBACK_REJECTED = "writeback_rejected"
+    QUOTA_SPEND_REJECTED = "quota_spend_rejected"
+    CANCELLED = "cancelled"
+    PERMISSION_DENIED = "permission_denied"
+    BUDGET_REJECTED = "budget_rejected"
+
+
+@dataclass(frozen=True, slots=True)
+class SettlementIdentity:
+    goal_id: str
+    agent_id: str
+    todo_id: str
+    turn_instance_id: str
+
+    @property
+    def effect_id(self) -> str:
+        return f"{self.goal_id}:{self.agent_id}:{self.todo_id}:{self.turn_instance_id}"
+
+    def as_dict(self) -> dict[str, str]:
+        return {
+            "schema_version": SETTLEMENT_IDENTITY_SCHEMA_VERSION,
+            "effect_id": self.effect_id,
+            "goal_id": self.goal_id,
+            "agent_id": self.agent_id,
+            "todo_id": self.todo_id,
+            "turn_instance_id": self.turn_instance_id,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class SettlementReceipt:
+    step_kind: SettlementStepKind
+    status: str
+    effect_id: str
+    source_ref: str | None = None
+
+    def as_dict(self) -> dict[str, str]:
+        receipt = {
+            "schema_version": SETTLEMENT_RECEIPT_SCHEMA_VERSION,
+            "step_kind": self.step_kind.value,
+            "status": self.status,
+            "effect_id": self.effect_id,
+        }
+        if self.source_ref:
+            receipt["source_ref"] = self.source_ref
+        return receipt
+
+
+@dataclass(frozen=True, slots=True)
+class SettlementFailure:
+    kind: SettlementFailureKind
+    step_kind: SettlementStepKind
+    reason: str
+
+    def as_dict(self) -> dict[str, str]:
+        return {
+            "kind": self.kind.value,
+            "step_kind": self.step_kind.value,
+            "reason": self.reason,
+        }
+
+
+T = TypeVar("T")
+U = TypeVar("U")
+
+
+@dataclass(frozen=True, slots=True)
+class SettlementResult(Generic[T]):
+    value: T | None
+    receipts: tuple[SettlementReceipt, ...] = ()
+    failure: SettlementFailure | None = None
+
+    @classmethod
+    def pure(
+        cls,
+        value: T,
+        *,
+        receipts: tuple[SettlementReceipt, ...] = (),
+    ) -> SettlementResult[T]:
+        return cls(value=value, receipts=receipts)
+
+    @classmethod
+    def failed(
+        cls,
+        *,
+        kind: SettlementFailureKind,
+        step_kind: SettlementStepKind,
+        reason: str,
+        receipts: tuple[SettlementReceipt, ...] = (),
+    ) -> SettlementResult[T]:
+        return cls(
+            value=None,
+            receipts=receipts,
+            failure=SettlementFailure(
+                kind=kind,
+                step_kind=step_kind,
+                reason=reason,
+            ),
+        )
+
+    def bind(self, step: Callable[[T], SettlementResult[U]]) -> SettlementResult[U]:
+        if self.failure is not None:
+            return SettlementResult(
+                value=None,
+                receipts=self.receipts,
+                failure=self.failure,
+            )
+        next_result = step(cast(T, self.value))
+        return SettlementResult(
+            value=next_result.value,
+            receipts=(*self.receipts, *next_result.receipts),
+            failure=next_result.failure,
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class SettlementStep:
+    kind: SettlementStepKind
+    owner: str
+    precondition: str
+    idempotency_key_ref: str
+    expected_receipt: str
+    command_template: str | None = None
+    conditional: bool = False
+
+    def as_dict(self) -> dict[str, Any]:
+        step: dict[str, Any] = {
+            "kind": self.kind.value,
+            "owner": self.owner,
+            "precondition": self.precondition,
+            "idempotency_key_ref": self.idempotency_key_ref,
+            "expected_receipt": self.expected_receipt,
+        }
+        if self.command_template:
+            step["command_template"] = self.command_template
+        if self.conditional:
+            step["conditional"] = True
+        return step
+
+
+@dataclass(frozen=True, slots=True)
+class SettlementPlan:
+    identity: SettlementIdentity
+    steps: tuple[SettlementStep, ...]
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "schema_version": SETTLEMENT_PLAN_SCHEMA_VERSION,
+            "identity": self.identity.as_dict(),
+            "ordered_steps": [step.as_dict() for step in self.steps],
+            "host_handoff": {
+                "owner": "host",
+                "kind": "scheduler_handoff",
+                "inside_agent_settlement": False,
+            },
+        }
+
+
+def settlement_result_payload(result: SettlementResult[Any]) -> dict[str, Any]:
+    return {
+        "ok": result.failure is None,
+        "receipts": [receipt.as_dict() for receipt in result.receipts],
+        "failure": result.failure.as_dict() if result.failure else None,
+    }
+
+
 def _mapping(value: Any) -> Mapping[str, Any]:
     return value if isinstance(value, Mapping) else {}