feat(control-plane): implement C3/C4 dual-session binding

loopx/control_plane/effect_program.py

@@ -258,6 +258,9 @@ class SettlementIdentity:
     todo_id: str | None
     turn_instance_id: str
     replan_obligation_id: str | None = None
+    controller_session_id: str | None = None
+    executor_session_id: str | None = None
+    generation: int | None = None
     _runtime_identity: Mapping[str, Any] = field(
         init=False,
         repr=False,
@@ -281,6 +284,15 @@ def __post_init__(self) -> None:
         object.__setattr__(
             self, "replan_obligation_id", identity.get("replan_obligation_id")
         )
+
+        # Extract session IDs and generation from payload if present
+        if "controller_session_id" in payload:
+            object.__setattr__(self, "controller_session_id", str(payload["controller_session_id"]))
+        if "executor_session_id" in payload:
+            object.__setattr__(self, "executor_session_id", str(payload["executor_session_id"]))
+        if "generation" in payload:
+            object.__setattr__(self, "generation", int(payload["generation"]))
+
         object.__setattr__(self, "_runtime_identity", identity)
         object.__setattr__(self, "_runtime_payload", payload)
 
@@ -334,6 +346,13 @@ def from_runtime_payload(cls, payload: Any) -> SettlementIdentity:
             or payload.get("binding_id") != binding_id
         ):
             raise RuntimeError("TypeScript settlement identity shape mismatch")
+
+        # Extract session fields if present
+        controller_session_id = str(payload.get("controller_session_id") or "").strip() or None
+        executor_session_id = str(payload.get("executor_session_id") or "").strip() or None
+        generation_value = payload.get("generation")
+        generation = int(generation_value) if generation_value is not None else None
+
         instance = object.__new__(cls)
         object.__setattr__(instance, "goal_id", str(payload["goal_id"]))
         object.__setattr__(instance, "agent_id", str(payload["agent_id"]))
@@ -344,6 +363,9 @@ def from_runtime_payload(cls, payload: Any) -> SettlementIdentity:
             str(payload["turn_instance_id"]),
         )
         object.__setattr__(instance, "replan_obligation_id", replan_obligation_id)
+        object.__setattr__(instance, "controller_session_id", controller_session_id)
+        object.__setattr__(instance, "executor_session_id", executor_session_id)
+        object.__setattr__(instance, "generation", generation)
         object.__setattr__(
             instance,
             "_runtime_identity",

loopx/control_plane/quota/settlement.py

@@ -9,6 +9,17 @@
 from ...command_invocation import loopx_cli_argv
 from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result
 from ..settlement_driver import decode_settlement_result
+from ..turn_driver.dual_session import (
+    ControllerSession,
+    ExecutorSession,
+    load_controller_session,
+    load_executor_session,
+)
+from ..turn_driver.generation_fence import (
+    GenerationFenceValidator,
+    GenerationFenceViolation,
+    GenerationValidationPhase,
+)
 from .accounting_admission import quota_accounting_admission
 from .effect_program import (
     SETTLEMENT_IDENTITY_SCHEMA_VERSION,
@@ -332,8 +343,40 @@ def read_heartbeat_settlement(
     source_admission: Mapping[str, Any] | None = None,
     borrow_source_admission: bool = False,
     heartbeat_reentry_guard: dict[str, Any] | None = None,
+    validate_generation_fence: bool = True,
 ) -> QuotaSettlementReadback | None:
-    """Read one complete heartbeat settlement through the TS domain owner."""
+    """Read one complete heartbeat settlement through the TS domain owner.
+
+    Args:
+        validate_generation_fence: Enable generation fence validation at prepare phase.
+                                   Set to False to skip validation (testing only).
+    """
+
+    # Load dual-session for generation fence validation
+    validator: GenerationFenceValidator | None = None
+    if validate_generation_fence and agent_id and todo_id:
+        try:
+            controller = load_controller_session(
+                runtime_root, goal_id, agent_id, todo_id
+            )
+            executor = load_executor_session(
+                runtime_root, goal_id, agent_id, todo_id
+            )
+
+            if controller is not None and executor is not None:
+                validator = GenerationFenceValidator(executor, controller)
+                # Validate at settlement prepare phase before quota spend
+                validator.validate(GenerationValidationPhase.SETTLEMENT_PREPARE)
+        except GenerationFenceViolation as exc:
+            raise ValueError(
+                f"Generation fence violation at settlement prepare: "
+                f"executor gen {exc.executor_generation} != "
+                f"controller gen {exc.controller_generation}"
+            ) from exc
+        except Exception as exc:
+            # Log warning but don't block settlement read if session loading fails
+            # This maintains backward compatibility with pre-dual-session code
+            pass
 
     def read(admission: Mapping[str, Any] | None) -> Any:
         try:
@@ -410,6 +453,18 @@ def read(admission: Mapping[str, Any] | None) -> Any:
         return None
     if payload.get("found") is not True:
         raise RuntimeError("TypeScript quota settlement readback result shape mismatch")
+
+    # Validate at checkpoint phase before effect commit
+    if validator is not None:
+        try:
+            validator.validate(GenerationValidationPhase.SETTLEMENT_CHECKPOINT)
+        except GenerationFenceViolation as exc:
+            raise ValueError(
+                f"Generation fence violation at settlement checkpoint: "
+                f"executor gen {exc.executor_generation} != "
+                f"controller gen {exc.controller_generation}"
+            ) from exc
+
     workspace_causality = _optional_readback_record(
         payload.get("workspace_causality")
     )