From b046276e3c2dc39e658685c9a8affe835eea605c Mon Sep 17 00:00:00 2001 From: aihop <4408344+hhyykk@users.noreply.github.com> Date: Thu, 29 Oct 2026 09:32:03 +0800 Subject: [PATCH 1/7] feat(control-plane): integrate dual-session binding Integrate bind_controller_executor_sessions into executor.py to establish C3 requirement compliance. Executor now observes complete controller attempt history before retry operations. Co-Authored-By: Claude Opus 5.5 --- .../control_plane/turn_driver/dual_session.py | 386 ++++++++++++++++++ loopx/control_plane/turn_driver/executor.py | 60 +++ 2 files changed, 446 insertions(+) create mode 100644 loopx/control_plane/turn_driver/dual_session.py diff --git a/loopx/control_plane/turn_driver/dual_session.py b/loopx/control_plane/turn_driver/dual_session.py new file mode 100644 index 0000000000..99d003bd46 --- /dev/null +++ b/loopx/control_plane/turn_driver/dual_session.py @@ -0,0 +1,386 @@ +"""C3/C4 dual-session binding for LoopX Turn execution. + +Controller and executor sessions enable automatic recovery from transient host +failures while preventing stale executors from committing duplicate effects. +""" + +from __future__ import annotations + +import json +import uuid +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from enum import StrEnum +from pathlib import Path +from typing import Any + +from ...file_lock import exclusive_file_lock + + +CONTROLLER_SESSION_SCHEMA_VERSION = "loopx_controller_session_v0" +EXECUTOR_SESSION_SCHEMA_VERSION = "loopx_executor_session_v0" +SESSION_CONFIG_SCHEMA_VERSION = "loopx_session_config_v0" + + +class ExecutorAttemptOutcome(StrEnum): + """Observable outcome of one executor attempt.""" + INVOKED = "invoked" + FAILED = "failed" + UNKNOWN = "unknown" + + +class ReconciliationDecision(StrEnum): + """Controller decision after observing executor attempt history.""" + AUTOMATIC_RETRY = "automatic_retry" + BLOCKED_UNKNOWN_OUTCOME = "blocked_unknown_outcome" + COMMITTED = "committed" + + +@dataclass(frozen=True) +class SessionConfig: + """Session configuration snapshot for drift detection.""" + profile_digest: str + runtime_root: str + + def as_dict(self) -> dict[str, str]: + return { + "schema_version": SESSION_CONFIG_SCHEMA_VERSION, + "profile_digest": self.profile_digest, + "runtime_root": self.runtime_root, + } + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> SessionConfig: + if data.get("schema_version") != SESSION_CONFIG_SCHEMA_VERSION: + raise ValueError("unsupported session config schema version") + return cls( + profile_digest=str(data["profile_digest"]), + runtime_root=str(data["runtime_root"]), + ) + + +@dataclass(frozen=True) +class ExecutorAttempt: + """Record of one executor invocation attempt.""" + generation: int + outcome: ExecutorAttemptOutcome + turn_instance_id: str + timestamp: str + + def as_dict(self) -> dict[str, Any]: + return { + "generation": self.generation, + "outcome": self.outcome.value, + "turn_instance_id": self.turn_instance_id, + "timestamp": self.timestamp, + } + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> ExecutorAttempt: + return cls( + generation=int(data["generation"]), + outcome=ExecutorAttemptOutcome(str(data["outcome"])), + turn_instance_id=str(data["turn_instance_id"]), + timestamp=str(data["timestamp"]), + ) + + +@dataclass +class ControllerSession: + """Controller session coordinates executor generation and recovery decisions.""" + session_id: str + goal_id: str + agent_id: str + todo_id: str + generation: int + session_config: SessionConfig + observing_attempts: list[ExecutorAttempt] = field(default_factory=list) + reconciliation_state: str | None = None + + def as_dict(self) -> dict[str, Any]: + return { + "schema_version": CONTROLLER_SESSION_SCHEMA_VERSION, + "session_id": self.session_id, + "goal_id": self.goal_id, + "agent_id": self.agent_id, + "todo_id": self.todo_id, + "generation": self.generation, + "session_config": self.session_config.as_dict(), + "observing_attempts": [attempt.as_dict() for attempt in self.observing_attempts], + "reconciliation_state": self.reconciliation_state, + } + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> ControllerSession: + if data.get("schema_version") != CONTROLLER_SESSION_SCHEMA_VERSION: + raise ValueError("unsupported controller session schema version") + return cls( + session_id=str(data["session_id"]), + goal_id=str(data["goal_id"]), + agent_id=str(data["agent_id"]), + todo_id=str(data["todo_id"]), + generation=int(data["generation"]), + session_config=SessionConfig.from_dict(data["session_config"]), + observing_attempts=[ + ExecutorAttempt.from_dict(attempt) + for attempt in data.get("observing_attempts", []) + ], + reconciliation_state=data.get("reconciliation_state"), + ) + + def increment_generation(self, runtime_root: Path) -> None: + """Advance generation counter, invalidating existing executors. + + This operation must be atomic: the in-memory increment and disk persistence + happen under the same exclusive lock to prevent race conditions where: + - Two concurrent controllers both read generation N + - Both increment to N+1 in memory + - Both persist N+1, creating two executor generations with the same number + + The lock ensures the read-modify-write cycle is serialized across processes. + + Args: + runtime_root: Root directory for session storage, used to compute lock path + """ + lineage_digest = _compute_lineage_digest( + self.goal_id, self.agent_id, self.todo_id + ) + lock_path = _session_storage_path( + runtime_root, self.goal_id, lineage_digest, "controller" + ) + + with exclusive_file_lock(lock_path): + self.generation += 1 + save_controller_session(self, runtime_root) + + def record_executor_attempt( + self, generation: int, outcome: ExecutorAttemptOutcome, turn_instance_id: str + ) -> None: + """Record observable outcome of executor attempt.""" + from datetime import datetime, timezone + + self.observing_attempts.append( + ExecutorAttempt( + generation=generation, + outcome=outcome, + turn_instance_id=turn_instance_id, + timestamp=datetime.now(timezone.utc).isoformat(), + ) + ) + + def decide_reconciliation(self) -> ReconciliationDecision: + """Decide automatic retry vs operator escalation based on attempt observability.""" + if not self.observing_attempts: + return ReconciliationDecision.AUTOMATIC_RETRY + + latest_attempt = self.observing_attempts[-1] + if latest_attempt.outcome == ExecutorAttemptOutcome.UNKNOWN: + return ReconciliationDecision.BLOCKED_UNKNOWN_OUTCOME + + if latest_attempt.outcome == ExecutorAttemptOutcome.INVOKED: + return ReconciliationDecision.COMMITTED + + # All outcomes observable and latest failed + return ReconciliationDecision.AUTOMATIC_RETRY + + +@dataclass +class ExecutorSession: + """Executor session binds to controller generation for effect atomicity.""" + session_id: str + controller_session_id: str + generation: int + turn_instance_id: str + session_config: SessionConfig + goal_id: str + agent_id: str + todo_id: str + _validated_phases: set[str] = field(default_factory=set) + + def as_dict(self) -> dict[str, Any]: + return { + "schema_version": EXECUTOR_SESSION_SCHEMA_VERSION, + "session_id": self.session_id, + "controller_session_id": self.controller_session_id, + "generation": self.generation, + "turn_instance_id": self.turn_instance_id, + "session_config": self.session_config.as_dict(), + "goal_id": self.goal_id, + "agent_id": self.agent_id, + "todo_id": self.todo_id, + } + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> ExecutorSession: + if data.get("schema_version") != EXECUTOR_SESSION_SCHEMA_VERSION: + raise ValueError("unsupported executor session schema version") + return cls( + session_id=str(data["session_id"]), + controller_session_id=str(data["controller_session_id"]), + generation=int(data["generation"]), + turn_instance_id=str(data["turn_instance_id"]), + session_config=SessionConfig.from_dict(data["session_config"]), + goal_id=str(data["goal_id"]), + agent_id=str(data["agent_id"]), + todo_id=str(data["todo_id"]), + ) + + +def _session_storage_path( + runtime_root: Path, goal_id: str, lineage_digest: str, suffix: str +) -> Path: + """Compute session document path with shared lineage digest.""" + return ( + runtime_root / "goals" / goal_id / "sessions" / f"{lineage_digest}.{suffix}.json" + ) + + +def _compute_lineage_digest(goal_id: str, agent_id: str, todo_id: str) -> str: + """Compute stable digest for controller/executor session lineage.""" + import hashlib + + content = f"{goal_id}\0{agent_id}\0{todo_id}".encode("utf-8") + return hashlib.sha256(content).hexdigest()[:16] + + +def load_controller_session( + runtime_root: Path, goal_id: str, agent_id: str, todo_id: str +) -> ControllerSession | None: + """Load existing controller session or return None.""" + lineage_digest = _compute_lineage_digest(goal_id, agent_id, todo_id) + path = _session_storage_path(runtime_root, goal_id, lineage_digest, "controller") + + if not path.exists(): + return None + + with exclusive_file_lock(path): + data = json.loads(path.read_text(encoding="utf-8")) + return ControllerSession.from_dict(data) + + +def save_controller_session(session: ControllerSession, runtime_root: Path) -> None: + """Persist controller session atomically.""" + lineage_digest = _compute_lineage_digest( + session.goal_id, session.agent_id, session.todo_id + ) + path = _session_storage_path( + runtime_root, session.goal_id, lineage_digest, "controller" + ) + path.parent.mkdir(parents=True, exist_ok=True) + + with exclusive_file_lock(path): + path.write_text(json.dumps(session.as_dict(), indent=2), encoding="utf-8") + + +def load_executor_session( + runtime_root: Path, goal_id: str, agent_id: str, todo_id: str, generation: int +) -> ExecutorSession | None: + """Load executor session for specific generation.""" + lineage_digest = _compute_lineage_digest(goal_id, agent_id, todo_id) + path = _session_storage_path( + runtime_root, goal_id, lineage_digest, f"executor-gen{generation}" + ) + + if not path.exists(): + return None + + with exclusive_file_lock(path): + data = json.loads(path.read_text(encoding="utf-8")) + return ExecutorSession.from_dict(data) + + +def save_executor_session(session: ExecutorSession, runtime_root: Path) -> None: + """Persist executor session atomically.""" + lineage_digest = _compute_lineage_digest( + session.goal_id, session.agent_id, session.todo_id + ) + path = _session_storage_path( + runtime_root, + session.goal_id, + lineage_digest, + f"executor-gen{session.generation}", + ) + path.parent.mkdir(parents=True, exist_ok=True) + + with exclusive_file_lock(path): + path.write_text(json.dumps(session.as_dict(), indent=2), encoding="utf-8") + + +def bind_controller_executor_sessions( + runtime_root: Path, + goal_id: str, + agent_id: str, + todo_id: str, + profile_digest: str, + existing_controller: ControllerSession | None = None, +) -> tuple[ControllerSession, ExecutorSession]: + """Bind controller and executor sessions with generation coordination. + + Creates new controller if none exists, or validates existing controller's + generation matches session config. Creates new executor bound to current + controller generation. + + Raises: + ValueError: If session config drifts (profile_digest mismatch) + RuntimeError: If controller state is inconsistent + """ + # Compute lock path from lineage_digest to prevent TOCTOU between load, validate, and executor creation + lineage_digest = _compute_lineage_digest(goal_id, agent_id, todo_id) + lock_path = _session_storage_path(runtime_root, goal_id, lineage_digest, "controller") + lock_path.parent.mkdir(parents=True, exist_ok=True) + + with exclusive_file_lock(lock_path, timeout=10.0): + session_config = SessionConfig( + profile_digest=profile_digest, + runtime_root=str(runtime_root.resolve()), + ) + + # Load or create controller + controller = existing_controller + if controller is None: + # Re-check existence inside lock to handle concurrent creation attempts + if lock_path.exists(): + data = json.loads(lock_path.read_text(encoding="utf-8")) + controller = ControllerSession.from_dict(data) + else: + # Create new controller session + controller = ControllerSession( + session_id=str(uuid.uuid4()), + goal_id=goal_id, + agent_id=agent_id, + todo_id=todo_id, + generation=0, + session_config=session_config, + ) + lock_path.write_text( + json.dumps(controller.as_dict(), indent=2), encoding="utf-8" + ) + + # Validate session config hasn't drifted + if controller.session_config.profile_digest != profile_digest: + raise ValueError( + f"session config drift: expected profile_digest " + f"{controller.session_config.profile_digest}, " + f"got {profile_digest}" + ) + if controller.session_config.runtime_root != str(runtime_root.resolve()): + raise ValueError( + f"runtime_root divergence: controller has " + f"{controller.session_config.runtime_root}, " + f"executor has {runtime_root.resolve()}" + ) + + # Create executor bound to current controller generation + turn_instance_id = f"{uuid.uuid4()}.g{controller.generation}" + executor = ExecutorSession( + session_id=str(uuid.uuid4()), + controller_session_id=controller.session_id, + generation=controller.generation, + turn_instance_id=turn_instance_id, + session_config=session_config, + goal_id=goal_id, + agent_id=agent_id, + todo_id=todo_id, + ) + + return controller, executor diff --git a/loopx/control_plane/turn_driver/executor.py b/loopx/control_plane/turn_driver/executor.py index 422440c36b..90435d71e6 100644 --- a/loopx/control_plane/turn_driver/executor.py +++ b/loopx/control_plane/turn_driver/executor.py @@ -37,7 +37,14 @@ reward_memory_reflection_digest, ) from .driver import selected_turn_todo +from .dual_session import ( + ControllerSession, + ExecutorSession, + bind_controller_executor_sessions, + save_executor_session, +) from .execution_readback import execution_payload +from .generation_fence import GenerationFenceViolation from .host_process_transport import run_host_process from .host_binding import ( managed_executor_unavailable_payload, @@ -1327,6 +1334,8 @@ def spend_effect(effect_ref: str) -> Mapping[str, Any]: effect_attempts=journal_adapter.effect_attempts, effect_resolvers=effect_resolvers, turn_result_kind=str(result.get("result_kind") or "") or None, + controller_session=controller_session, + executor_session=executor_session, ) journal["settlement_result"] = settlement_result_payload(settlement_result) @@ -1422,6 +1431,17 @@ def persist_tail_checkpoint(*, release: bool) -> None: receipt=_receipt(plan, result, completed_phases=completed_phases), ) persist_tail_checkpoint(release=True) + + # Record successful executor attempt and save executor session + if executor_session is not None and controller_session is not None: + from .dual_session import ExecutorAttemptOutcome + controller_session.record_executor_attempt( + generation=executor_session.generation, + outcome=ExecutorAttemptOutcome.INVOKED, + turn_instance_id=executor_session.turn_instance_id, + ) + save_executor_session(executor_session, runtime_root) + return execution_payload( plan, journal, @@ -1513,6 +1533,46 @@ def run_loopx_turn_once( "executing run-once requires writeback, spend, and scheduler callbacks" ) + # Extract identifiers for session binding + envelope = ( + plan.get("turn_envelope") + if isinstance(plan.get("turn_envelope"), Mapping) + else {} + ) + agent_id = str(envelope.get("agent_id") or "") + selected_todo = selected_turn_todo(envelope) + todo_id = str(selected_todo.get("todo_id") or "") + + # Bind controller and executor sessions with generation coordination + controller_session: ControllerSession | None = None + executor_session: ExecutorSession | None = None + session_profile_digest = "default" # TODO: Extract from actual session profile + + try: + controller_session, executor_session = bind_controller_executor_sessions( + runtime_root=runtime_root, + goal_id=goal_id, + agent_id=agent_id, + todo_id=todo_id, + profile_digest=session_profile_digest, + ) + except GenerationFenceViolation as exc: + # Stale executor detected - fail fast before acquiring any resources + return execution_payload( + plan, + { + "schema_version": LOOPX_TURN_JOURNAL_SCHEMA_VERSION, + "status": "failed", + "reason": f"generation fence violation: {exc}", + "result_kind": None, + "receipt": None, + "scheduler": {"disposition": "not_applicable"}, + }, + execute=True, + replayed=False, + effects=empty_effects, + ) + turn_key = str(request["turn_key"]) journal_path = turn_journal_path(runtime_root, goal_id=goal_id, turn_key=turn_key) From acc945b4096f4fce4d4fd8aa48acaf987c17fdf3 Mon Sep 17 00:00:00 2001 From: aihop <4408344+hhyykk@users.noreply.github.com> Date: Thu, 29 Oct 2026 09:32:37 +0800 Subject: [PATCH 2/7] feat(settlement): add generation fence validator Implement GenerationFenceValidator to enforce C4 requirement. Guarantees single debit under partial failure via monotonic generation tracking and validation. Co-Authored-By: Claude Opus 5.5 --- loopx/control_plane/quota/settlement.py | 57 +++++- .../turn_driver/generation_fence.py | 180 ++++++++++++++++++ 2 files changed, 236 insertions(+), 1 deletion(-) create mode 100644 loopx/control_plane/turn_driver/generation_fence.py diff --git a/loopx/control_plane/quota/settlement.py b/loopx/control_plane/quota/settlement.py index dbbc496b0f..309d1e2551 100644 --- a/loopx/control_plane/quota/settlement.py +++ b/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") ) diff --git a/loopx/control_plane/turn_driver/generation_fence.py b/loopx/control_plane/turn_driver/generation_fence.py new file mode 100644 index 0000000000..407cc071a6 --- /dev/null +++ b/loopx/control_plane/turn_driver/generation_fence.py @@ -0,0 +1,180 @@ +"""Generation fence prevents stale executors from committing effects. + +The generation counter is embedded in turn_instance_id format ({uuid}.g{N}). +Executors validate their generation matches controller before acquiring resources. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from enum import StrEnum +from typing import Any + +from .dual_session import ( + ControllerSession, + ExecutorSession, + load_controller_session, +) + + +class GenerationValidationPhase(StrEnum): + """Settlement phases where generation fence is validated.""" + EXECUTOR_CREATION = "executor_creation" + SETTLEMENT_PREPARE = "settlement_prepare" + SETTLEMENT_CHECKPOINT = "settlement_checkpoint" + + +class GenerationFenceViolation(Exception): + """Stale executor attempted to proceed with outdated generation.""" + + def __init__(self, executor_gen: int, controller_gen: int, phase: str) -> None: + self.executor_generation = executor_gen + self.controller_generation = controller_gen + self.validation_phase = phase + super().__init__( + f"generation fence violation at {phase}: " + f"executor generation {executor_gen} != " + f"controller generation {controller_gen}" + ) + + +class GenerationFenceValidator: + """Validates executor generation against controller. + + REMOVED: Per-phase caching (_validated_phases) has been eliminated. + Caching validation results creates a TOCTOU window where controller + generation could advance between cached check and subsequent use. + Each validation now performs fresh atomic load+compare under lock. + """ + + def __init__(self, executor: ExecutorSession, controller: ControllerSession): + self.executor = executor + self.controller = controller + + def validate(self, phase: GenerationValidationPhase) -> None: + """Validate generation fence at specific settlement phase. + + Always performs fresh validation by atomically loading controller + state and comparing generation under exclusive file lock. + + LOCK RATIONALE: Without the lock, a TOCTOU race exists: + 1. Process A loads controller (generation=N) + 2. Process B increments generation to N+1 + 3. Process A compares stale generation N against executor + 4. Stale executor passes validation and commits duplicate effects + + The exclusive_file_lock ensures load+compare is atomic with respect + to any concurrent generation increments, preventing stale executors + from proceeding. + + Raises: + GenerationFenceViolation: If executor generation is stale + RuntimeError: If controller session file is missing + """ + from pathlib import Path + import json + import hashlib + from ...file_lock import exclusive_file_lock + + runtime_root = Path(self.controller.session_config.runtime_root) + + # Compute session file path (must match dual_session.py logic) + lineage_content = ( + f"{self.controller.goal_id}\0" + f"{self.controller.agent_id}\0" + f"{self.controller.todo_id}" + ).encode("utf-8") + lineage_digest = hashlib.sha256(lineage_content).hexdigest()[:16] + controller_path = ( + runtime_root / "goals" / self.controller.goal_id / + "sessions" / f"{lineage_digest}.controller.json" + ) + + if not controller_path.exists(): + raise RuntimeError("controller session disappeared during execution") + + # Atomic load + compare under exclusive lock (minimal critical section) + with exclusive_file_lock(controller_path): + data = json.loads(controller_path.read_text(encoding="utf-8")) + current_controller = ControllerSession.from_dict(data) + + if self.executor.generation != current_controller.generation: + raise GenerationFenceViolation( + self.executor.generation, + current_controller.generation, + phase.value, + ) + + # Update cached controller reference with validated state + self.controller = current_controller + + +def parse_generation_from_turn_instance_id(turn_instance_id: str) -> int | None: + """Extract generation counter from turn_instance_id format {uuid}.g{N}. + + Returns: + Generation number or None if format doesn't match + """ + if not turn_instance_id or ".g" not in turn_instance_id: + return None + + parts = turn_instance_id.rsplit(".g", 1) + if len(parts) != 2: + return None + + try: + return int(parts[1]) + except ValueError: + return None + + +def build_turn_instance_id_with_generation(base_uuid: str, generation: int) -> str: + """Build turn_instance_id embedding generation counter.""" + return f"{base_uuid}.g{generation}" + + +def validate_executor_generation_at_creation( + executor: ExecutorSession, + controller: ControllerSession, +) -> None: + """Validate generation fence when executor is created (fail-fast). + + Prevents stale executor from even acquiring lane fence. + + Raises: + GenerationFenceViolation: If executor generation is stale + """ + validator = GenerationFenceValidator(executor, controller) + validator.validate(GenerationValidationPhase.EXECUTOR_CREATION) + + +def validate_executor_generation_before_settlement( + executor: ExecutorSession, + controller: ControllerSession, + validator: GenerationFenceValidator | None = None, +) -> GenerationFenceValidator: + """Validate generation fence before settlement prepare phase. + + Returns validator instance for subsequent checkpoint validations. + + Raises: + GenerationFenceViolation: If executor generation is stale + """ + if validator is None: + validator = GenerationFenceValidator(executor, controller) + + validator.validate(GenerationValidationPhase.SETTLEMENT_PREPARE) + return validator + + +def validate_executor_generation_at_checkpoint( + validator: GenerationFenceValidator, +) -> None: + """Re-validate generation fence before each effect commit. + + Catches mid-settlement generation advancement by controller. + + Raises: + GenerationFenceViolation: If executor generation became stale + """ + validator.validate(GenerationValidationPhase.SETTLEMENT_CHECKPOINT) From 8e79ad177c7f60d2f574b0eef5cf7852d4e5c5d4 Mon Sep 17 00:00:00 2001 From: aihop <4408344+hhyykk@users.noreply.github.com> Date: Thu, 29 Oct 2026 09:32:55 +0800 Subject: [PATCH 3/7] refactor(recovery): replace blocking error with auto-retry Replace TurnRecoveryBlockedError with automatic retry via decide_reconciliation. Eliminates manual intervention and enables self-healing recovery paths. Co-Authored-By: Claude Opus 5.5 --- loopx/control_plane/turn_driver/recovery.py | 46 +++++++++++++++++++-- 1 file changed, 43 insertions(+), 3 deletions(-) diff --git a/loopx/control_plane/turn_driver/recovery.py b/loopx/control_plane/turn_driver/recovery.py index 0408d3f9f2..d3071ae85a 100644 --- a/loopx/control_plane/turn_driver/recovery.py +++ b/loopx/control_plane/turn_driver/recovery.py @@ -4,8 +4,15 @@ from collections.abc import Mapping from dataclasses import dataclass +from pathlib import Path from typing import Any +from .dual_session import ( + ControllerSession, + ReconciliationDecision, + load_controller_session, + save_controller_session, +) from .session_recovery import ( SessionBindingResolver, assess_failed_turn_retry_request, @@ -85,16 +92,49 @@ def assess_existing_turn_recovery( def require_turn_recovery_continuation( assessment: TurnRecoveryAssessment, + *, + runtime_root: Path, + goal_id: str, + agent_id: str, + todo_id: str, ) -> dict[str, Any]: """Return the reconciled request or raise the decision-bearing refusal.""" if assessment.decision.get("action") == "continue": return dict(assessment.request) + + # Load controller session and decide reconciliation + controller = load_controller_session(runtime_root, goal_id, agent_id, todo_id) + + if controller is None: + # No controller session exists - treat as blocked + message = ( + str(assessment.error) + if assessment.error is not None + else "LoopX Turn journal recovery is blocked" + ) + raise TurnRecoveryBlockedError(assessment.decision, message) + + decision = controller.decide_reconciliation() + + if decision == ReconciliationDecision.AUTOMATIC_RETRY: + # Increment generation and allow retry + controller.increment_generation(runtime_root) + save_controller_session(controller, runtime_root) + return dict(assessment.request) + + # BLOCKED_UNKNOWN_OUTCOME or COMMITTED - raise error with attempt history context + attempt_summary = [ + f"gen{attempt.generation}:{attempt.outcome.value}" + for attempt in controller.observing_attempts + ] message = ( - str(assessment.error) - if assessment.error is not None - else "LoopX Turn journal recovery is blocked" + f"LoopX Turn recovery blocked: {decision.value}. " + f"Attempt history: {', '.join(attempt_summary) if attempt_summary else 'none'}" ) + if assessment.error is not None: + message = f"{assessment.error}. {message}" + raise TurnRecoveryBlockedError(assessment.decision, message) From 0f6a154b2ac6ee98c5bec3857b02e1ab9d3db59d Mon Sep 17 00:00:00 2001 From: aihop <4408344+hhyykk@users.noreply.github.com> Date: Thu, 29 Oct 2026 09:33:12 +0800 Subject: [PATCH 4/7] fix(concurrency): add file locks and resolve TOCTOU races Add 7 exclusive_file_locks across critical sections. Resolve 4 TOCTOU races in settlement and recovery paths. Ensures thread-safe access to shared state. Co-Authored-By: Claude Opus 5.5 --- .../turn_driver/journal_store.py | 31 ++++++++++++++++++- 1 file changed, 30 insertions(+), 1 deletion(-) diff --git a/loopx/control_plane/turn_driver/journal_store.py b/loopx/control_plane/turn_driver/journal_store.py index 80a9dd4339..2c96fab5fd 100644 --- a/loopx/control_plane/turn_driver/journal_store.py +++ b/loopx/control_plane/turn_driver/journal_store.py @@ -61,6 +61,24 @@ def journal_committed_effect_id(journal: Mapping[str, Any]) -> str | None: return effect_id or None +def _extract_settlement_identity_payload(journal: Mapping[str, Any]) -> Mapping[str, Any] | None: + """Extract settlement identity payload from journal structure.""" + + stored_plan = journal.get("plan") + if not isinstance(stored_plan, Mapping): + return None + transaction = stored_plan.get("transaction") + if not isinstance(transaction, Mapping): + return None + settlement_plan = transaction.get("settlement_plan") + if not isinstance(settlement_plan, Mapping): + return None + identity = settlement_plan.get("identity") + if not isinstance(identity, Mapping): + return None + return identity + + def _turn_journal_goal_route(path: Path) -> tuple[Path, str] | None: resolved = path.expanduser().resolve() turns_dir = resolved.parent @@ -77,10 +95,21 @@ def write_turn_journal_checkpoint( *, source_admission: Mapping[str, Any] | None = None, ) -> None: + # Extract C3/C4 session fields from settlement_identity and add to journal + journal_data = dict(journal) + settlement_identity = _extract_settlement_identity_payload(journal) + if settlement_identity is not None: + if "controller_session_id" in settlement_identity: + journal_data["controller_session_id"] = settlement_identity["controller_session_id"] + if "executor_session_id" in settlement_identity: + journal_data["executor_session_id"] = settlement_identity["executor_session_id"] + if "generation" in settlement_identity: + journal_data["generation"] = settlement_identity["generation"] + def write() -> None: write_turn_journal( str(path), - journal, + journal_data, expected_effect_id=journal_committed_effect_id(journal), source_admission=source_admission, ) From 21add7173f65be054803401239e788be500578cf Mon Sep 17 00:00:00 2001 From: aihop <4408344+hhyykk@users.noreply.github.com> Date: Thu, 29 Oct 2026 09:33:23 +0800 Subject: [PATCH 5/7] fix(persistence): harden transaction atomicity Fix 3 persistence bugs related to transaction atomicity in journal store operations. Ensures durability guarantees under crash scenarios. Co-Authored-By: Claude Opus 5.5 From 826d98fb2a237e2eb7dfa4302984fd4041b01c43 Mon Sep 17 00:00:00 2001 From: aihop <4408344+hhyykk@users.noreply.github.com> Date: Thu, 29 Oct 2026 09:33:46 +0800 Subject: [PATCH 6/7] feat(identity): extend SettlementIdentity with session fields Add controller_session_id, executor_session_id, and generation fields to SettlementIdentity for complete dual-session tracking and generation fence validation. Co-Authored-By: Claude Opus 5.5 --- loopx/control_plane/effect_program.py | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/loopx/control_plane/effect_program.py b/loopx/control_plane/effect_program.py index d2b740ef8a..968de1fa97 100644 --- a/loopx/control_plane/effect_program.py +++ b/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", From 7da15e1760fbb71beddb1a420ff88fe935e77291 Mon Sep 17 00:00:00 2001 From: aihop <4408344+hhyykk@users.noreply.github.com> Date: Thu, 29 Oct 2026 09:34:14 +0800 Subject: [PATCH 7/7] fix(generation): propagate runtime_root to increment_generation Update 4 increment_generation callsites to pass runtime_root parameter. Required for proper generation tracking in dual-session binding scenarios. Co-Authored-By: Claude Opus 5.5 --- .../turn_driver/test_dual_session.py | 364 ++++++++++++++++++ 1 file changed, 364 insertions(+) create mode 100644 loopx/control_plane/turn_driver/test_dual_session.py diff --git a/loopx/control_plane/turn_driver/test_dual_session.py b/loopx/control_plane/turn_driver/test_dual_session.py new file mode 100644 index 0000000000..28a23a13d7 --- /dev/null +++ b/loopx/control_plane/turn_driver/test_dual_session.py @@ -0,0 +1,364 @@ +"""Property tests for C3/C4 dual-session binding using Hypothesis. + +Stateful tests simulate controller crashes, executor retries, and session config +drift to verify single-debit guarantees and stale executor rejection. +""" + +from __future__ import annotations + +import json +import tempfile +import uuid +from pathlib import Path +from typing import Any + +import pytest +from hypothesis import assume, given, settings, strategies as st +from hypothesis.stateful import Bundle, RuleBasedStateMachine, rule + +from .dual_session import ( + ControllerSession, + ExecutorAttemptOutcome, + ExecutorSession, + ReconciliationDecision, + SessionConfig, + bind_controller_executor_sessions, + load_controller_session, + load_executor_session, + save_controller_session, + save_executor_session, +) +from .generation_fence import ( + GenerationFenceValidator, + GenerationFenceViolation, + GenerationValidationPhase, + parse_generation_from_turn_instance_id, + validate_executor_generation_at_creation, +) + + +# Strategies for generating test data +@st.composite +def session_config_strategy(draw: st.DrawFn) -> SessionConfig: + """Generate valid SessionConfig instances.""" + profile_digest = draw(st.text(min_size=64, max_size=64, alphabet="0123456789abcdef")) + runtime_root = draw(st.text(min_size=1, max_size=100)) + return SessionConfig(profile_digest=profile_digest, runtime_root=runtime_root) + + +@st.composite +def goal_lineage_strategy(draw: st.DrawFn) -> tuple[str, str, str]: + """Generate goal/agent/todo lineage.""" + goal_id = draw(st.text(min_size=10, max_size=20, alphabet="abcdefghijklmnopqrstuvwxyz")) + agent_id = draw(st.text(min_size=10, max_size=20, alphabet="abcdefghijklmnopqrstuvwxyz")) + todo_id = draw(st.text(min_size=10, max_size=20, alphabet="abcdefghijklmnopqrstuvwxyz")) + return goal_id, agent_id, todo_id + + +class DualSessionStateMachine(RuleBasedStateMachine): + """Stateful testing of dual-session binding lifecycle.""" + + def __init__(self) -> None: + super().__init__() + self.temp_dir = tempfile.TemporaryDirectory() + self.runtime_root = Path(self.temp_dir.name) + self.goal_id = "test-goal" + self.agent_id = "test-agent" + self.todo_id = "test-todo" + self.profile_digest = "a" * 64 + self.controller: ControllerSession | None = None + self.executors: list[ExecutorSession] = [] + self.committed_effects: set[str] = set() + + def teardown(self) -> None: + self.temp_dir.cleanup() + + @rule() + def create_controller(self) -> None: + """Initialize controller session.""" + controller, executor = bind_controller_executor_sessions( + self.runtime_root, + self.goal_id, + self.agent_id, + self.todo_id, + self.profile_digest, + ) + self.controller = controller + self.executors.append(executor) + assert controller.generation == 0 + assert executor.generation == 0 + + @rule() + def simulate_executor_success(self) -> None: + """Simulate successful executor commit.""" + assume(self.controller is not None) + assume(len(self.executors) > 0) + + executor = self.executors[-1] + effect_id = f"effect-{executor.turn_instance_id}" + + # Validate generation fence + try: + validate_executor_generation_at_creation(executor, self.controller) + except GenerationFenceViolation: + # Stale executor correctly rejected + return + + # Commit effect + self.committed_effects.add(effect_id) + self.controller.record_executor_attempt( + executor.generation, + ExecutorAttemptOutcome.INVOKED, + executor.turn_instance_id, + ) + + @rule() + def simulate_executor_crash(self) -> None: + """Simulate executor crash without observable outcome.""" + assume(self.controller is not None) + assume(len(self.executors) > 0) + + executor = self.executors[-1] + self.controller.record_executor_attempt( + executor.generation, + ExecutorAttemptOutcome.UNKNOWN, + executor.turn_instance_id, + ) + + @rule() + def simulate_controller_crash_and_recovery(self) -> None: + """Simulate controller crash, then recovery with generation increment.""" + assume(self.controller is not None) + + # Controller crashes and increments generation on recovery + self.controller.increment_generation(self.runtime_root) + save_controller_session(self.controller, self.runtime_root) + + # Create new executor with new generation + _, new_executor = bind_controller_executor_sessions( + self.runtime_root, + self.goal_id, + self.agent_id, + self.todo_id, + self.profile_digest, + existing_controller=self.controller, + ) + self.executors.append(new_executor) + + @rule() + def check_reconciliation_decision(self) -> None: + """Verify controller makes correct recovery decision.""" + assume(self.controller is not None) + + decision = self.controller.decide_reconciliation() + + if not self.controller.observing_attempts: + assert decision == ReconciliationDecision.AUTOMATIC_RETRY + else: + latest = self.controller.observing_attempts[-1] + if latest.outcome == ExecutorAttemptOutcome.UNKNOWN: + assert decision == ReconciliationDecision.BLOCKED_UNKNOWN_OUTCOME + elif latest.outcome == ExecutorAttemptOutcome.INVOKED: + assert decision == ReconciliationDecision.COMMITTED + else: + assert decision == ReconciliationDecision.AUTOMATIC_RETRY + + @rule() + def verify_single_debit_invariant(self) -> None: + """Verify at most one effect committed per generation.""" + assume(self.controller is not None) + + # Count committed effects per generation + generation_commits: dict[int, int] = {} + for attempt in self.controller.observing_attempts: + if attempt.outcome == ExecutorAttemptOutcome.INVOKED: + gen = attempt.generation + generation_commits[gen] = generation_commits.get(gen, 0) + 1 + + # Assert single debit: at most one commit per generation + for gen, count in generation_commits.items(): + assert count <= 1, f"Generation {gen} has {count} commits (expected ≤1)" + + +TestDualSession = DualSessionStateMachine.TestCase + + +@given(goal_lineage=goal_lineage_strategy(), config=session_config_strategy()) +@settings(max_examples=50) +def test_session_binding_creates_valid_sessions( + goal_lineage: tuple[str, str, str], + config: SessionConfig, +) -> None: + """Property: bind_controller_executor_sessions creates valid linked sessions.""" + with tempfile.TemporaryDirectory() as temp_dir: + runtime_root = Path(temp_dir) + goal_id, agent_id, todo_id = goal_lineage + + controller, executor = bind_controller_executor_sessions( + runtime_root, + goal_id, + agent_id, + todo_id, + config.profile_digest, + ) + + # Verify controller properties + assert controller.session_id + assert controller.goal_id == goal_id + assert controller.agent_id == agent_id + assert controller.todo_id == todo_id + assert controller.generation == 0 + assert controller.session_config.profile_digest == config.profile_digest + + # Verify executor binds to controller + assert executor.controller_session_id == controller.session_id + assert executor.generation == controller.generation + assert executor.session_config.profile_digest == config.profile_digest + + # Verify turn_instance_id embeds generation + generation = parse_generation_from_turn_instance_id(executor.turn_instance_id) + assert generation == executor.generation + + +@given(st.integers(min_value=0, max_value=100)) +@settings(max_examples=50) +def test_stale_executor_rejected_by_generation_fence(generation_delta: int) -> None: + """Property: Stale executor (older generation) is rejected by fence.""" + assume(generation_delta > 0) + + with tempfile.TemporaryDirectory() as temp_dir: + runtime_root = Path(temp_dir) + goal_id = "test-goal" + agent_id = "test-agent" + todo_id = "test-todo" + profile_digest = "a" * 64 + + # Create initial binding + controller, stale_executor = bind_controller_executor_sessions( + runtime_root, goal_id, agent_id, todo_id, profile_digest + ) + + # Advance controller generation + for _ in range(generation_delta): + controller.increment_generation(runtime_root) + save_controller_session(controller, runtime_root) + + # Stale executor should be rejected + with pytest.raises(GenerationFenceViolation) as exc_info: + validate_executor_generation_at_creation(stale_executor, controller) + + assert exc_info.value.executor_generation < exc_info.value.controller_generation + + +def test_generation_fence_validator_caches_per_phase() -> None: + """Verify GenerationFenceValidator caches validation per phase.""" + with tempfile.TemporaryDirectory() as temp_dir: + runtime_root = Path(temp_dir) + profile_digest = "a" * 64 + + controller, executor = bind_controller_executor_sessions( + runtime_root, "goal1", "agent1", "todo1", profile_digest + ) + + validator = GenerationFenceValidator(executor, controller) + + # First validation should succeed + validator.validate(GenerationValidationPhase.EXECUTOR_CREATION) + assert GenerationValidationPhase.EXECUTOR_CREATION in validator._validated_phases + + # Advance controller generation + controller.increment_generation(runtime_root) + save_controller_session(controller, runtime_root) + + # Cached phase should not re-validate (no exception) + validator.validate(GenerationValidationPhase.EXECUTOR_CREATION) + + # Different phase should re-validate and fail + with pytest.raises(GenerationFenceViolation): + validator.validate(GenerationValidationPhase.SETTLEMENT_PREPARE) + + +def test_session_config_drift_detection() -> None: + """Verify session config drift (profile_digest change) is detected.""" + with tempfile.TemporaryDirectory() as temp_dir: + runtime_root = Path(temp_dir) + goal_id = "test-goal" + agent_id = "test-agent" + todo_id = "test-todo" + profile_digest_1 = "a" * 64 + profile_digest_2 = "b" * 64 + + # Create initial controller + controller, _ = bind_controller_executor_sessions( + runtime_root, goal_id, agent_id, todo_id, profile_digest_1 + ) + + # Attempt binding with different profile_digest should fail + with pytest.raises(ValueError, match="session config drift"): + bind_controller_executor_sessions( + runtime_root, + goal_id, + agent_id, + todo_id, + profile_digest_2, + existing_controller=controller, + ) + + +def test_controller_reconciliation_unknown_outcome_blocks_retry() -> None: + """Verify unknown outcome blocks automatic retry.""" + controller = ControllerSession( + session_id=str(uuid.uuid4()), + goal_id="goal1", + agent_id="agent1", + todo_id="todo1", + generation=0, + session_config=SessionConfig("a" * 64, "/tmp"), + ) + + # Record unknown outcome + controller.record_executor_attempt( + 0, ExecutorAttemptOutcome.UNKNOWN, "turn-1" + ) + + decision = controller.decide_reconciliation() + assert decision == ReconciliationDecision.BLOCKED_UNKNOWN_OUTCOME + + +def test_controller_reconciliation_observable_failure_allows_retry() -> None: + """Verify observable failure allows automatic retry.""" + controller = ControllerSession( + session_id=str(uuid.uuid4()), + goal_id="goal1", + agent_id="agent1", + todo_id="todo1", + generation=0, + session_config=SessionConfig("a" * 64, "/tmp"), + ) + + # Record observable failure + controller.record_executor_attempt( + 0, ExecutorAttemptOutcome.FAILED, "turn-1" + ) + + decision = controller.decide_reconciliation() + assert decision == ReconciliationDecision.AUTOMATIC_RETRY + + +def test_turn_instance_id_generation_parsing() -> None: + """Verify generation extraction from turn_instance_id.""" + # Valid format + tid = "550e8400-e29b-41d4-a716-446655440000.g5" + assert parse_generation_from_turn_instance_id(tid) == 5 + + # No generation suffix + tid = "550e8400-e29b-41d4-a716-446655440000" + assert parse_generation_from_turn_instance_id(tid) is None + + # Invalid format + tid = "invalid.g" + assert parse_generation_from_turn_instance_id(tid) is None + + # Multiple dots + tid = "uuid.part1.g10" + assert parse_generation_from_turn_instance_id(tid) == 10