feat(orchestration): add durable v2 domain and storage

agent/agent/orchestration/config.py

@@ -1,6 +1,8 @@
 """Orchestration configuration."""
 
 from dataclasses import dataclass
+from math import isfinite
+from typing import Optional
 
 
 @dataclass(frozen=True)
@@ -9,9 +11,29 @@ class OrchestrationConfig:
     max_repair_continuations: int = 1
     default_worker_preset: str = "worker"
     default_validator_preset: str = "validator"
+    worker_timeout_seconds: Optional[float] = None
+    validator_timeout_seconds: Optional[float] = None
+    stop_grace_seconds: float = 5.0
 
     def __post_init__(self) -> None:
-        if self.max_parallel_tasks < 1:
+        if isinstance(self.max_parallel_tasks, bool) or self.max_parallel_tasks < 1:
             raise ValueError("max_parallel_tasks must be at least 1")
-        if self.max_repair_continuations < 0:
+        if (
+            isinstance(self.max_repair_continuations, bool)
+            or self.max_repair_continuations < 0
+        ):
             raise ValueError("max_repair_continuations cannot be negative")
+        for name, value in (
+            ("worker_timeout_seconds", self.worker_timeout_seconds),
+            ("validator_timeout_seconds", self.validator_timeout_seconds),
+        ):
+            if value is not None and (
+                isinstance(value, bool) or not isfinite(value) or value <= 0
+            ):
+                raise ValueError(f"{name} must be positive when configured")
+        if (
+            isinstance(self.stop_grace_seconds, bool)
+            or not isfinite(self.stop_grace_seconds)
+            or self.stop_grace_seconds < 0
+        ):
+            raise ValueError("stop_grace_seconds must be finite and non-negative")

agent/agent/orchestration/models.py

@@ -10,7 +10,8 @@ from __future__ import annotations
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from enum import Enum
-from typing import Any, Dict, List, Optional
+from math import isfinite
+from typing import Any, Dict, List, Optional, Tuple
 from uuid import uuid4
 
 
@@ -76,6 +77,84 @@ class ValidationVerdict(StrEnum):
     INCONCLUSIVE = "inconclusive"
 
 
+class ValidationMode(StrEnum):
+    AGENT = "agent"
+    DETERMINISTIC = "deterministic"
+
+
+@dataclass(frozen=True)
+class ValidationPlan:
+    """Frozen, serializable validation behavior for one ValidationReport."""
+
+    mode: ValidationMode = ValidationMode.AGENT
+    validator_preset: Optional[str] = "validator"
+    rule_ids: Tuple[str, ...] = ()
+    max_evidence_queries: int = 10
+    max_evidence_items_per_query: int = 50
+    evidence_timeout_seconds: float = 10.0
+
+    def __post_init__(self) -> None:
+        object.__setattr__(self, "rule_ids", tuple(self.rule_ids))
+        if self.mode == ValidationMode.AGENT and self.rule_ids:
+            raise ValueError("Agent validation plans cannot contain deterministic rules")
+        if (
+            self.mode == ValidationMode.AGENT
+            and self.validator_preset is not None
+            and not self.validator_preset.strip()
+        ):
+            raise ValueError("Agent validator_preset cannot be blank")
+        if self.mode == ValidationMode.DETERMINISTIC and self.validator_preset:
+            raise ValueError("Deterministic validation plans cannot select an agent preset")
+        if len(set(self.rule_ids)) != len(self.rule_ids):
+            raise ValueError("Validation rule_ids must be unique")
+        if any(not item.strip() for item in self.rule_ids):
+            raise ValueError("Validation rule_ids cannot be empty")
+        if (
+            isinstance(self.max_evidence_queries, bool)
+            or not isinstance(self.max_evidence_queries, int)
+            or self.max_evidence_queries < 0
+        ):
+            raise ValueError("max_evidence_queries must be a non-negative integer")
+        if (
+            isinstance(self.max_evidence_items_per_query, bool)
+            or not isinstance(self.max_evidence_items_per_query, int)
+            or self.max_evidence_items_per_query < 1
+        ):
+            raise ValueError("max_evidence_items_per_query must be a positive integer")
+        if (
+            isinstance(self.evidence_timeout_seconds, bool)
+            or not isinstance(self.evidence_timeout_seconds, (int, float))
+            or not isfinite(self.evidence_timeout_seconds)
+            or self.evidence_timeout_seconds <= 0
+        ):
+            raise ValueError("evidence_timeout_seconds must be finite and positive")
+
+    def to_dict(self) -> Dict[str, Any]:
+        return json_values(asdict(self))
+
+    @classmethod
+    def from_dict(
+        cls,
+        data: "ValidationPlan | Dict[str, Any]",
+    ) -> "ValidationPlan":
+        if isinstance(data, cls):
+            return data
+        mode = ValidationMode(data.get("mode", ValidationMode.AGENT.value))
+        return cls(
+            mode=mode,
+            validator_preset=data.get(
+                "validator_preset",
+                "validator" if mode == ValidationMode.AGENT else None,
+            ),
+            rule_ids=tuple(data.get("rule_ids", [])),
+            max_evidence_queries=int(data.get("max_evidence_queries", 10)),
+            max_evidence_items_per_query=int(
+                data.get("max_evidence_items_per_query", 50)
+            ),
+            evidence_timeout_seconds=float(data.get("evidence_timeout_seconds", 10.0)),
+        )
+
+
 class DecisionAction(StrEnum):
     ACCEPT = "accept"
     REPAIR = "repair"
@@ -89,6 +168,81 @@ class DecisionAction(StrEnum):
     SUPERSEDE = "supersede"
 
 
+class OperationKind(StrEnum):
+    DISPATCH = "dispatch"
+    REVALIDATE = "revalidate"
+
+
+class OperationStatus(StrEnum):
+    PENDING = "pending"
+    RUNNING = "running"
+    STOP_REQUESTED = "stop_requested"
+    STOPPED = "stopped"
+    COMPLETED = "completed"
+    FAILED = "failed"
+
+
+class FailureCode(StrEnum):
+    """Stable machine-readable reason for an execution-stage failure."""
+
+    UNSUPPORTED = "unsupported"
+    EXECUTOR_ERROR = "executor_error"
+    AGENT_FAILED = "agent_failed"
+    TIMEOUT = "timeout"
+    STOPPED = "stopped"
+    INTERRUPTED = "interrupted"
+    PROTOCOL_VIOLATION = "protocol_violation"
+
+
+@dataclass(frozen=True)
+class ExecutionStats:
+    """Per-stage usage; elapsed time is derived from the stage timestamps."""
+
+    primary_model: Optional[str] = None
+    total_tokens: Optional[int] = None
+    total_cost: Optional[float] = None
+    failure_code: Optional[FailureCode] = None
+
+    def __post_init__(self) -> None:
+        if self.primary_model is not None:
+            if not isinstance(self.primary_model, str) or not self.primary_model.strip():
+                raise ValueError(
+                    "ExecutionStats.primary_model must be a non-blank string"
+                )
+        if self.total_tokens is not None and (
+            isinstance(self.total_tokens, bool)
+            or not isinstance(self.total_tokens, int)
+            or self.total_tokens < 0
+        ):
+            raise ValueError("ExecutionStats.total_tokens must be a non-negative integer")
+        if self.total_cost is not None and (
+            isinstance(self.total_cost, bool)
+            or not isinstance(self.total_cost, (int, float))
+            or not isfinite(self.total_cost)
+            or self.total_cost < 0
+        ):
+            raise ValueError("ExecutionStats.total_cost must be finite and non-negative")
+        if self.failure_code is not None and not isinstance(
+            self.failure_code, FailureCode
+        ):
+            try:
+                object.__setattr__(self, "failure_code", FailureCode(self.failure_code))
+            except (TypeError, ValueError) as exc:
+                raise ValueError("ExecutionStats.failure_code is invalid") from exc
+
+    @classmethod
+    def from_dict(cls, data: "ExecutionStats | Dict[str, Any]") -> "ExecutionStats":
+        if isinstance(data, cls):
+            return data
+        failure_code = data.get("failure_code")
+        return cls(
+            primary_model=data.get("primary_model"),
+            total_tokens=data.get("total_tokens"),
+            total_cost=data.get("total_cost"),
+            failure_code=FailureCode(failure_code) if failure_code else None,
+        )
+
+
 @dataclass(frozen=True)
 class AcceptanceCriterion:
     criterion_id: str
@@ -245,13 +399,22 @@ class TaskAttempt:
     worker_preset: str
     execution_mode: str
     status: AttemptStatus = AttemptStatus.RUNNING
+    operation_id: Optional[str] = None
+    execution_epoch: int = 0
     continue_from_trace_id: Optional[str] = None
     snapshot_id: Optional[str] = None
     submission: Optional[AttemptSubmission] = None
+    execution_stats: Optional[ExecutionStats] = None
     error: Optional[str] = None
+    started_at: Optional[str] = None
+    completed_at: Optional[str] = None
     created_at: str = field(default_factory=utc_now)
     updated_at: str = field(default_factory=utc_now)
 
+    @property
+    def duration_ms(self) -> Optional[int]:
+        return _duration_ms(self.started_at, self.completed_at)
+
     @classmethod
     def from_dict(cls, data: Dict[str, Any]) -> "TaskAttempt":
         submission = data.get("submission")
@@ -263,10 +426,19 @@ class TaskAttempt:
             worker_preset=data.get("worker_preset", "worker"),
             execution_mode=data.get("execution_mode", "new"),
             status=AttemptStatus(data.get("status", AttemptStatus.RUNNING.value)),
+            operation_id=data.get("operation_id"),
+            execution_epoch=int(data.get("execution_epoch", 0)),
             continue_from_trace_id=data.get("continue_from_trace_id"),
             snapshot_id=data.get("snapshot_id"),
             submission=AttemptSubmission.from_dict(submission) if submission else None,
+            execution_stats=(
+                ExecutionStats.from_dict(data["execution_stats"])
+                if data.get("execution_stats") is not None
+                else None
+            ),
             error=data.get("error"),
+            started_at=data.get("started_at"),
+            completed_at=data.get("completed_at"),
             created_at=data.get("created_at", utc_now()),
             updated_at=data.get("updated_at", utc_now()),
         )
@@ -298,7 +470,12 @@ class ValidationReport:
     snapshot_id: str
     validator_trace_id: str
     validator_preset: str = "validator"
+    validation_plan: ValidationPlan = field(default_factory=ValidationPlan)
+    evidence_queries_used: int = 0
+    evidence_query_results: Dict[str, Dict[str, Any]] = field(default_factory=dict)
     status: ValidationRunStatus = ValidationRunStatus.PENDING
+    operation_id: Optional[str] = None
+    execution_epoch: int = 0
     verdict: Optional[ValidationVerdict] = None
     criterion_results: List[CriterionResult] = field(default_factory=list)
     summary: str = ""
@@ -306,10 +483,17 @@ class ValidationReport:
     unverified_claims: List[str] = field(default_factory=list)
     risks: List[str] = field(default_factory=list)
     recommendation: str = ""
+    execution_stats: Optional[ExecutionStats] = None
     error: Optional[str] = None
+    started_at: Optional[str] = None
+    completed_at: Optional[str] = None
     created_at: str = field(default_factory=utc_now)
     updated_at: str = field(default_factory=utc_now)
 
+    @property
+    def duration_ms(self) -> Optional[int]:
+        return _duration_ms(self.started_at, self.completed_at)
+
     @classmethod
     def from_dict(cls, data: Dict[str, Any]) -> "ValidationReport":
         verdict = data.get("verdict")
@@ -321,7 +505,15 @@ class ValidationReport:
             snapshot_id=data["snapshot_id"],
             validator_trace_id=data["validator_trace_id"],
             validator_preset=data.get("validator_preset", "validator"),
+            validation_plan=ValidationPlan.from_dict(data.get("validation_plan", {})),
+            evidence_queries_used=int(data.get("evidence_queries_used", 0)),
+            evidence_query_results={
+                str(key): dict(value)
+                for key, value in data.get("evidence_query_results", {}).items()
+            },
             status=ValidationRunStatus(data.get("status", ValidationRunStatus.PENDING.value)),
+            operation_id=data.get("operation_id"),
+            execution_epoch=int(data.get("execution_epoch", 0)),
             verdict=ValidationVerdict(verdict) if verdict else None,
             criterion_results=[CriterionResult.from_dict(x) for x in data.get("criterion_results", [])],
             summary=data.get("summary", ""),
@@ -329,7 +521,14 @@ class ValidationReport:
             unverified_claims=list(data.get("unverified_claims", [])),
             risks=list(data.get("risks", [])),
             recommendation=data.get("recommendation", ""),
+            execution_stats=(
+                ExecutionStats.from_dict(data["execution_stats"])
+                if data.get("execution_stats") is not None
+                else None
+            ),
             error=data.get("error"),
+            started_at=data.get("started_at"),
+            completed_at=data.get("completed_at"),
             created_at=data.get("created_at", utc_now()),
             updated_at=data.get("updated_at", utc_now()),
         )
@@ -364,6 +563,116 @@ class PlannerDecision:
         )
 
 
+@dataclass
+class BackgroundOperation:
+    operation_id: str
+    root_trace_id: str
+    kind: OperationKind
+    request: Dict[str, Any] = field(default_factory=dict)
+    request_fingerprint: str = ""
+    status: OperationStatus = OperationStatus.PENDING
+    task_ids: List[str] = field(default_factory=list)
+    attempt_ids: List[str] = field(default_factory=list)
+    validation_ids: List[str] = field(default_factory=list)
+    result_ref: Optional[Dict[str, Any]] = None
+    execution_epoch: int = 0
+    deadline_at: Optional[str] = None
+    error: Optional[str] = None
+    started_at: Optional[str] = None
+    completed_at: Optional[str] = None
+    created_at: str = field(default_factory=utc_now)
+    updated_at: str = field(default_factory=utc_now)
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "BackgroundOperation":
+        return cls(
+            operation_id=data["operation_id"],
+            root_trace_id=data["root_trace_id"],
+            kind=OperationKind(data["kind"]),
+            request=dict(data.get("request", {})),
+            request_fingerprint=str(data.get("request_fingerprint", "")),
+            status=OperationStatus(data.get("status", OperationStatus.PENDING.value)),
+            task_ids=list(data.get("task_ids", [])),
+            attempt_ids=list(data.get("attempt_ids", [])),
+            validation_ids=list(data.get("validation_ids", [])),
+            result_ref=dict(data["result_ref"]) if data.get("result_ref") is not None else None,
+            execution_epoch=int(data.get("execution_epoch", 0)),
+            deadline_at=data.get("deadline_at"),
+            error=data.get("error"),
+            started_at=data.get("started_at"),
+            completed_at=data.get("completed_at"),
+            created_at=data.get("created_at", utc_now()),
+            updated_at=data.get("updated_at", utc_now()),
+        )
+
+
+@dataclass(frozen=True)
+class CommandRecord:
+    command_id: str
+    operation: str
+    fingerprint: str
+    result_ref: Dict[str, Any] = field(default_factory=dict)
+    committed_revision: int = -1
+    operation_id: Optional[str] = None
+    created_at: str = field(default_factory=utc_now)
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "CommandRecord":
+        return cls(
+            command_id=data["command_id"],
+            operation=str(data.get("operation", "")),
+            fingerprint=str(data.get("fingerprint", "")),
+            result_ref=dict(data.get("result_ref", {})),
+            committed_revision=int(data.get("committed_revision", -1)),
+            operation_id=data.get("operation_id"),
+            created_at=data.get("created_at", utc_now()),
+        )
+
+
+@dataclass(frozen=True)
+class EventDraft:
+    event_type: str
+    payload: Dict[str, Any] = field(default_factory=dict)
+    command_id: Optional[str] = None
+    operation_id: Optional[str] = None
+
+
+@dataclass(frozen=True)
+class OrchestrationEvent:
+    schema_version: int
+    event_id: str
+    sequence: int
+    root_trace_id: str
+    ledger_revision: int
+    event_type: str
+    occurred_at: str
+    payload: Dict[str, Any] = field(default_factory=dict)
+    command_id: Optional[str] = None
+    operation_id: Optional[str] = None
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "OrchestrationEvent":
+        return cls(
+            schema_version=int(data.get("schema_version", 1)),
+            event_id=data["event_id"],
+            sequence=int(data["sequence"]),
+            root_trace_id=data["root_trace_id"],
+            ledger_revision=int(data["ledger_revision"]),
+            event_type=data["event_type"],
+            occurred_at=data["occurred_at"],
+            payload=dict(data.get("payload", {})),
+            command_id=data.get("command_id"),
+            operation_id=data.get("operation_id"),
+        )
+
+
+@dataclass(frozen=True)
+class EventPage:
+    events: List[OrchestrationEvent]
+    next_cursor: Optional[str]
+    has_more: bool = False
+
+
 @dataclass
 class TaskLedger:
     root_trace_id: str
@@ -373,13 +682,15 @@ class TaskLedger:
     attempts: Dict[str, TaskAttempt] = field(default_factory=dict)
     validations: Dict[str, ValidationReport] = field(default_factory=dict)
     decisions: Dict[str, PlannerDecision] = field(default_factory=dict)
+    operations: Dict[str, BackgroundOperation] = field(default_factory=dict)
+    command_records: Dict[str, CommandRecord] = field(default_factory=dict)
     focused_task_id: Optional[str] = None
     idempotency_results: Dict[str, Dict[str, Any]] = field(default_factory=dict)
     created_at: str = field(default_factory=utc_now)
     updated_at: str = field(default_factory=utc_now)
 
     def to_dict(self) -> Dict[str, Any]:
-        return _enum_values(asdict(self))
+        return json_values(asdict(self))
 
     @classmethod
     def from_dict(cls, data: Dict[str, Any]) -> "TaskLedger":
@@ -391,6 +702,8 @@ class TaskLedger:
             attempts={k: TaskAttempt.from_dict(v) for k, v in data.get("attempts", {}).items()},
             validations={k: ValidationReport.from_dict(v) for k, v in data.get("validations", {}).items()},
             decisions={k: PlannerDecision.from_dict(v) for k, v in data.get("decisions", {}).items()},
+            operations={k: BackgroundOperation.from_dict(v) for k, v in data.get("operations", {}).items()},
+            command_records={k: CommandRecord.from_dict(v) for k, v in data.get("command_records", {}).items()},
             focused_task_id=data.get("focused_task_id"),
             idempotency_results=dict(data.get("idempotency_results", {})),
             created_at=data.get("created_at", utc_now()),
@@ -408,7 +721,7 @@ class TaskCycleResult:
     error: Optional[str] = None
 
     def to_dict(self) -> Dict[str, Any]:
-        return _enum_values(asdict(self))
+        return json_values(asdict(self))
 
     @classmethod
     def from_dict(cls, data: Dict[str, Any]) -> "TaskCycleResult":
@@ -423,21 +736,38 @@ class TaskCycleResult:
         )
 
 
-def _enum_values(value: Any) -> Any:
+def _duration_ms(started_at: Optional[str], completed_at: Optional[str]) -> Optional[int]:
+    if not started_at or not completed_at:
+        return None
+    try:
+        started = datetime.fromisoformat(started_at.replace("Z", "+00:00"))
+        completed = datetime.fromisoformat(completed_at.replace("Z", "+00:00"))
+        if started.tzinfo is None:
+            started = started.replace(tzinfo=timezone.utc)
+        if completed.tzinfo is None:
+            completed = completed.replace(tzinfo=timezone.utc)
+    except (TypeError, ValueError):
+        return None
+    return max(0, int((completed - started).total_seconds() * 1000))
+
+
+def json_values(value: Any) -> Any:
     if isinstance(value, Enum):
         return value.value
     if isinstance(value, dict):
-        return {str(k): _enum_values(v) for k, v in value.items()}
+        return {str(k): json_values(v) for k, v in value.items()}
     if isinstance(value, (list, tuple)):
-        return [_enum_values(v) for v in value]
+        return [json_values(v) for v in value]
     return value
 
 
 __all__ = [
     "CompletionPolicy", "AgentRole", "TaskStatus", "AttemptStatus",
-    "ValidationRunStatus", "ValidationVerdict", "DecisionAction",
+    "ValidationRunStatus", "ValidationVerdict", "ValidationMode", "ValidationPlan", "DecisionAction",
+    "OperationKind", "OperationStatus", "FailureCode", "ExecutionStats",
     "AcceptanceCriterion", "TaskSpec", "TaskRecord", "TaskAttempt",
     "ArtifactRef", "ArtifactSnapshot", "AttemptSubmission", "CriterionResult",
-    "ValidationReport", "PlannerDecision", "TaskLedger", "TaskCycleResult",
-    "new_id", "utc_now",
+    "ValidationReport", "PlannerDecision", "BackgroundOperation", "CommandRecord",
+    "EventDraft", "OrchestrationEvent", "EventPage", "TaskLedger", "TaskCycleResult",
+    "json_values", "new_id", "utc_now",
 ]

agent/agent/orchestration/protocols.py

@@ -3,9 +3,17 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Any, Dict, Optional, Protocol
+from typing import Any, Dict, List, Optional, Protocol, Sequence
 
-from .models import ArtifactSnapshot, AttemptSubmission, TaskLedger
+from .models import (
+    ArtifactSnapshot,
+    AttemptSubmission,
+    BackgroundOperation,
+    EventDraft,
+    EventPage,
+    ExecutionStats,
+    TaskLedger,
+)
 
 
 @dataclass(frozen=True)
@@ -20,6 +28,7 @@ class WorkerRunResult:
     status: str
     summary: str = ""
     error: Optional[str] = None
+    execution_stats: Optional[ExecutionStats] = None
 
 
 @dataclass(frozen=True)
@@ -28,6 +37,7 @@ class ValidatorRunResult:
     status: str
     summary: str = ""
     error: Optional[str] = None
+    execution_stats: Optional[ExecutionStats] = None
 
 
 class TaskStore(Protocol):
@@ -37,17 +47,42 @@ class TaskStore(Protocol):
         ledger: TaskLedger,
         expected_revision: int,
         idempotency_key: Optional[str] = None,
+        event: Optional[EventDraft] = None,
     ) -> CommitResult: ...
+    async def list_events(
+        self,
+        root_trace_id: str,
+        cursor: Optional[str] = None,
+        limit: int = 100,
+    ) -> EventPage: ...
+    async def list_recoverable(self, root_trace_id: str) -> List[BackgroundOperation]: ...
 
 
 class ArtifactStore(Protocol):
-    async def freeze(self, attempt_id: str, submission: AttemptSubmission) -> ArtifactSnapshot: ...
-    async def get(self, snapshot_id: str) -> ArtifactSnapshot: ...
+    async def freeze(
+        self,
+        root_trace_id: str,
+        attempt_id: str,
+        submission: AttemptSubmission,
+    ) -> ArtifactSnapshot: ...
+    async def get(self, root_trace_id: str, snapshot_id: str) -> ArtifactSnapshot: ...
+    async def get_for_attempt(self, root_trace_id: str, attempt_id: str) -> Optional[ArtifactSnapshot]: ...
+    async def list_orphans(
+        self,
+        root_trace_id: str,
+        known_attempt_ids: Sequence[str],
+    ) -> List[ArtifactSnapshot]: ...
+    async def cleanup_orphans(
+        self,
+        root_trace_id: str,
+        known_attempt_ids: Sequence[str],
+    ) -> List[str]: ...
 
 
 class AgentExecutor(Protocol):
     async def run_worker(self, context: Dict[str, Any]) -> WorkerRunResult: ...
     async def run_validator(self, context: Dict[str, Any]) -> ValidatorRunResult: ...
+    async def stop(self, trace_id: str) -> bool: ...
 
 
 class ToolPolicy(Protocol):

agent/agent/orchestration/store.py

@@ -1,17 +1,31 @@
-"""Single-process filesystem adapters for orchestration ports."""
+"""Filesystem reference adapters for orchestration ports."""
 
 from __future__ import annotations
 
 import asyncio
+import base64
 import hashlib
 import json
 import os
 import tempfile
 from dataclasses import asdict
 from pathlib import Path
-from typing import Any, Dict, Optional
-
-from .models import ArtifactSnapshot, AttemptSubmission, TaskLedger, new_id, utc_now
+from typing import Any, Dict, List, Optional, Sequence, Tuple
+
+from filelock import FileLock
+
+from .models import (
+    ArtifactSnapshot,
+    AttemptSubmission,
+    BackgroundOperation,
+    EventDraft,
+    EventPage,
+    OperationStatus,
+    OrchestrationEvent,
+    TaskLedger,
+    json_values,
+    utc_now,
+)
 from .protocols import CommitResult
 
 
@@ -27,36 +41,50 @@ class RevisionConflict(TaskStoreError):
     pass
 
 
+class ArtifactConflict(TaskStoreError):
+    pass
+
+
 class FileSystemTaskStore:
-    """Atomic JSON ledger store with optimistic revisions and per-root locks."""
+    """Atomic JSON store with cross-process locking and optimistic revisions."""
 
-    _locks: Dict[str, asyncio.Lock] = {}
+    _EVENTS_KEY = "_events"
+    _NEXT_EVENT_SEQUENCE_KEY = "_next_event_sequence"
 
     def __init__(self, base_path: str = ".trace") -> None:
-        self.base_path = Path(base_path)
+        self.base_path = Path(base_path).resolve()
 
     def orchestration_dir(self, root_trace_id: str) -> Path:
-        return self.base_path / root_trace_id / "orchestration"
+        return _safe_child(self.base_path, root_trace_id, "orchestration")
 
     def ledger_path(self, root_trace_id: str) -> Path:
         return self.orchestration_dir(root_trace_id) / "ledger.json"
 
-    def _lock(self, root_trace_id: str) -> asyncio.Lock:
-        key = str(self.ledger_path(root_trace_id).resolve())
-        if key not in self._locks:
-            self._locks[key] = asyncio.Lock()
-        return self._locks[key]
+    def _lock_path(self, root_trace_id: str) -> Path:
+        return self.orchestration_dir(root_trace_id) / ".ledger.lock"
 
     async def load(self, root_trace_id: str) -> TaskLedger:
-        async with self._lock(root_trace_id):
-            return self._load_unlocked(root_trace_id)
+        return await asyncio.to_thread(self._load_locked, root_trace_id)
 
-    def _load_unlocked(self, root_trace_id: str) -> TaskLedger:
+    def _load_locked(self, root_trace_id: str) -> TaskLedger:
+        lock_path = self._lock_path(root_trace_id)
+        lock_path.parent.mkdir(parents=True, exist_ok=True)
+        with FileLock(str(lock_path)):
+            raw = self._read_raw(root_trace_id)
+            try:
+                return TaskLedger.from_dict(raw)
+            except (AttributeError, KeyError, TypeError, ValueError) as exc:
+                raise TaskStoreError(f"Cannot parse task ledger {self.ledger_path(root_trace_id)}: {exc}") from exc
+
+    def _read_raw(self, root_trace_id: str) -> Dict[str, Any]:
         path = self.ledger_path(root_trace_id)
         if not path.exists():
             raise TaskStoreNotFound(f"No task ledger for root trace {root_trace_id}")
         try:
-            return TaskLedger.from_dict(json.loads(path.read_text(encoding="utf-8")))
+            data = json.loads(path.read_text(encoding="utf-8"))
+            if not isinstance(data, dict):
+                raise TypeError("ledger root must be an object")
+            return data
         except (OSError, ValueError, TypeError, json.JSONDecodeError) as exc:
             raise TaskStoreError(f"Cannot load task ledger {path}: {exc}") from exc
 
@@ -65,25 +93,127 @@ class FileSystemTaskStore:
         ledger: TaskLedger,
         expected_revision: int,
         idempotency_key: Optional[str] = None,
+        event: Optional[EventDraft] = None,
+    ) -> CommitResult:
+        # idempotency_key remains accepted for V1 callers. Durable command
+        # replay is represented by TaskLedger.command_records in V2.
+        del idempotency_key
+        return await asyncio.to_thread(self._commit_locked, ledger, expected_revision, event)
+
+    def _commit_locked(
+        self,
+        ledger: TaskLedger,
+        expected_revision: int,
+        event: Optional[EventDraft],
     ) -> CommitResult:
         root_trace_id = ledger.root_trace_id
-        async with self._lock(root_trace_id):
-            path = self.ledger_path(root_trace_id)
-            current_revision = -1
-            if path.exists():
-                current_revision = self._load_unlocked(root_trace_id).revision
+        lock_path = self._lock_path(root_trace_id)
+        lock_path.parent.mkdir(parents=True, exist_ok=True)
+        with FileLock(str(lock_path)):
+            raw: Dict[str, Any] = {}
+            if self.ledger_path(root_trace_id).exists():
+                raw = self._read_raw(root_trace_id)
+            current_revision = int(raw.get("revision", -1))
             if current_revision != expected_revision:
                 raise RevisionConflict(
                     f"Task ledger {root_trace_id} revision conflict: "
                     f"expected {expected_revision}, actual {current_revision}"
                 )
 
-            ledger.revision = expected_revision + 1
-            ledger.updated_at = utc_now()
+            new_revision = expected_revision + 1
+            updated_at = utc_now()
+            data = ledger.to_dict()
+            data["revision"] = new_revision
+            data["updated_at"] = updated_at
+            events = self._read_events(raw)
+            next_sequence = self._next_sequence(raw, events)
+            if event and not self._event_already_recorded(events, event):
+                recorded = OrchestrationEvent(
+                    schema_version=1,
+                    event_id=_event_id(root_trace_id, next_sequence),
+                    sequence=next_sequence,
+                    root_trace_id=root_trace_id,
+                    ledger_revision=new_revision,
+                    event_type=event.event_type,
+                    occurred_at=utc_now(),
+                    payload=json_values(event.payload),
+                    command_id=_public_command_id(event.command_id),
+                    operation_id=event.operation_id,
+                )
+                events.append(json_values(asdict(recorded)))
+                next_sequence += 1
+            if events:
+                data[self._EVENTS_KEY] = events
+            data[self._NEXT_EVENT_SEQUENCE_KEY] = next_sequence
+
+            path = self.ledger_path(root_trace_id)
             path.parent.mkdir(parents=True, exist_ok=True)
-            self._atomic_json_write(path, ledger.to_dict())
+            self._atomic_json_write(path, data)
+            ledger.revision = new_revision
+            ledger.updated_at = updated_at
             return CommitResult(revision=ledger.revision, ledger=ledger)
 
+    async def list_events(
+        self,
+        root_trace_id: str,
+        cursor: Optional[str] = None,
+        limit: int = 100,
+    ) -> EventPage:
+        if not 1 <= limit <= 1000:
+            raise ValueError("Event page limit must be between 1 and 1000")
+        after = _decode_cursor(cursor, root_trace_id) if cursor else 0
+        return await asyncio.to_thread(self._list_events_locked, root_trace_id, after, limit, cursor)
+
+    def _list_events_locked(
+        self,
+        root_trace_id: str,
+        after: int,
+        limit: int,
+        prior_cursor: Optional[str],
+    ) -> EventPage:
+        lock_path = self._lock_path(root_trace_id)
+        lock_path.parent.mkdir(parents=True, exist_ok=True)
+        with FileLock(str(lock_path)):
+            raw = self._read_raw(root_trace_id)
+            try:
+                available = [
+                    OrchestrationEvent.from_dict(item)
+                    for item in self._read_events(raw)
+                    if int(item.get("sequence", 0)) > after
+                ]
+            except (KeyError, TypeError, ValueError) as exc:
+                raise TaskStoreError("Ledger durable event journal contains an invalid event") from exc
+        page = available[:limit]
+        next_cursor = _encode_cursor(root_trace_id, page[-1].sequence) if page else prior_cursor
+        return EventPage(events=page, next_cursor=next_cursor, has_more=len(available) > limit)
+
+    async def list_recoverable(self, root_trace_id: str) -> List[BackgroundOperation]:
+        ledger = await self.load(root_trace_id)
+        recoverable = {
+            OperationStatus.PENDING,
+            OperationStatus.RUNNING,
+            OperationStatus.STOP_REQUESTED,
+        }
+        return [operation for operation in ledger.operations.values() if operation.status in recoverable]
+
+    @classmethod
+    def _read_events(cls, raw: Dict[str, Any]) -> List[Dict[str, Any]]:
+        events = raw.get(cls._EVENTS_KEY, [])
+        if not isinstance(events, list) or any(not isinstance(item, dict) for item in events):
+            raise TaskStoreError("Ledger durable event journal is invalid")
+        return list(events)
+
+    @classmethod
+    def _next_sequence(cls, raw: Dict[str, Any], events: Sequence[Dict[str, Any]]) -> int:
+        fallback = max((int(item.get("sequence", 0)) for item in events), default=0) + 1
+        value = int(raw.get(cls._NEXT_EVENT_SEQUENCE_KEY, fallback))
+        return max(value, fallback, 1)
+
+    @staticmethod
+    def _event_already_recorded(events: Sequence[Dict[str, Any]], event: EventDraft) -> bool:
+        command_id = _public_command_id(event.command_id)
+        return bool(command_id) and any(item.get("command_id") == command_id for item in events)
+
     @staticmethod
     def _atomic_json_write(path: Path, data: Dict[str, Any]) -> None:
         fd, tmp_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent))
@@ -93,6 +223,7 @@ class FileSystemTaskStore:
                 handle.flush()
                 os.fsync(handle.fileno())
             os.replace(tmp_name, path)
+            _fsync_directory(path.parent)
         except Exception:
             try:
                 os.unlink(tmp_name)
@@ -102,98 +233,246 @@ class FileSystemTaskStore:
 
 
 class FileSystemArtifactStore:
-    """Immutable snapshot store bound to a root trace."""
-
-    _locks: Dict[str, asyncio.Lock] = {}
+    """Immutable snapshots with same-attempt first-write-wins semantics."""
 
     def __init__(self, base_path: str = ".trace", root_trace_id: Optional[str] = None) -> None:
-        self.base_path = Path(base_path)
+        self.base_path = Path(base_path).resolve()
         self.root_trace_id = root_trace_id
 
     def for_root(self, root_trace_id: str) -> "FileSystemArtifactStore":
+        """V1 compatibility adapter; new callers should pass root_trace_id."""
         return FileSystemArtifactStore(str(self.base_path), root_trace_id=root_trace_id)
 
-    def _artifacts_dir(self) -> Path:
+    def _artifacts_dir(self, root_trace_id: str) -> Path:
+        return _safe_child(self.base_path, root_trace_id, "orchestration", "artifacts")
+
+    def _bound_root(self) -> str:
         if not self.root_trace_id:
-            raise ValueError("FileSystemArtifactStore must be bound with for_root(root_trace_id)")
-        return self.base_path / self.root_trace_id / "orchestration" / "artifacts"
-
-    def _lock(self) -> asyncio.Lock:
-        key = str(self._artifacts_dir().resolve())
-        if key not in self._locks:
-            self._locks[key] = asyncio.Lock()
-        return self._locks[key]
-
-    async def freeze(self, attempt_id: str, submission: AttemptSubmission) -> ArtifactSnapshot:
-        normalized = {
-            "summary": submission.summary.strip(),
-            "artifact_refs": [asdict(x) for x in submission.artifact_refs],
-            "evidence_refs": [asdict(x) for x in submission.evidence_refs],
-        }
-        canonical = json.dumps(normalized, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
-        digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
-        snapshot = ArtifactSnapshot(
-            snapshot_id=new_id(),
-            attempt_id=attempt_id,
-            normalized_content=normalized,
-            sha256=digest,
-            artifact_refs=list(submission.artifact_refs),
-            evidence_refs=list(submission.evidence_refs),
-        )
-        path = self._artifacts_dir() / f"{snapshot.snapshot_id}.json"
-        async with self._lock():
-            path.parent.mkdir(parents=True, exist_ok=True)
-            if path.exists():
-                raise TaskStoreError(f"Snapshot already exists: {snapshot.snapshot_id}")
-            FileSystemTaskStore._atomic_json_write(path, _json_values(asdict(snapshot)))
-        return snapshot
-
-    async def get(self, snapshot_id: str) -> ArtifactSnapshot:
-        path = self._artifacts_dir() / f"{snapshot_id}.json"
-        async with self._lock():
+            raise ValueError("root_trace_id is required")
+        return self.root_trace_id
+
+    async def freeze(
+        self,
+        root_trace_id: str,
+        attempt_id: Any,
+        submission: Optional[AttemptSubmission] = None,
+    ) -> ArtifactSnapshot:
+        if submission is None:
+            # V1 bound form: bound.freeze(attempt_id, submission)
+            submission = attempt_id
+            attempt_id = root_trace_id
+            root_trace_id = self._bound_root()
+        if not isinstance(attempt_id, str) or not isinstance(submission, AttemptSubmission):
+            raise TypeError("freeze requires root_trace_id, attempt_id and AttemptSubmission")
+        return await asyncio.to_thread(self._freeze_locked, root_trace_id, attempt_id, submission)
+
+    def _freeze_locked(
+        self,
+        root_trace_id: str,
+        attempt_id: str,
+        submission: AttemptSubmission,
+    ) -> ArtifactSnapshot:
+        artifacts_dir = self._artifacts_dir(root_trace_id)
+        artifacts_dir.mkdir(parents=True, exist_ok=True)
+        with FileLock(str(artifacts_dir / ".artifacts.lock")):
+            existing = self._find_for_attempt(artifacts_dir, attempt_id)
+            normalized, digest = _normalize_submission(submission)
+            if existing:
+                if existing.sha256 == digest:
+                    return existing
+                raise ArtifactConflict(f"Attempt {attempt_id} already has a different artifact snapshot")
+
+            snapshot = ArtifactSnapshot(
+                snapshot_id=_snapshot_id(attempt_id),
+                attempt_id=attempt_id,
+                normalized_content=normalized,
+                sha256=digest,
+                artifact_refs=list(submission.artifact_refs),
+                evidence_refs=list(submission.evidence_refs),
+            )
+            FileSystemTaskStore._atomic_json_write(
+                artifacts_dir / f"{snapshot.snapshot_id}.json",
+                json_values(asdict(snapshot)),
+            )
+            return snapshot
+
+    async def get(self, root_trace_id: str, snapshot_id: Optional[str] = None) -> ArtifactSnapshot:
+        if snapshot_id is None:
+            snapshot_id = root_trace_id
+            root_trace_id = self._bound_root()
+        return await asyncio.to_thread(self._get_locked, root_trace_id, snapshot_id)
+
+    def _get_locked(self, root_trace_id: str, snapshot_id: str) -> ArtifactSnapshot:
+        artifacts_dir = self._artifacts_dir(root_trace_id)
+        path = _safe_child(artifacts_dir, f"{snapshot_id}.json")
+        artifacts_dir.mkdir(parents=True, exist_ok=True)
+        with FileLock(str(artifacts_dir / ".artifacts.lock")):
             if not path.exists():
                 raise FileNotFoundError(f"Artifact snapshot not found: {snapshot_id}")
+            return self._read_snapshot(path)
+
+    async def get_for_attempt(
+        self,
+        root_trace_id: str,
+        attempt_id: Optional[str] = None,
+    ) -> Optional[ArtifactSnapshot]:
+        if attempt_id is None:
+            attempt_id = root_trace_id
+            root_trace_id = self._bound_root()
+        return await asyncio.to_thread(self._get_for_attempt_locked, root_trace_id, attempt_id)
+
+    def _get_for_attempt_locked(self, root_trace_id: str, attempt_id: str) -> Optional[ArtifactSnapshot]:
+        artifacts_dir = self._artifacts_dir(root_trace_id)
+        artifacts_dir.mkdir(parents=True, exist_ok=True)
+        with FileLock(str(artifacts_dir / ".artifacts.lock")):
+            return self._find_for_attempt(artifacts_dir, attempt_id)
+
+    async def list_orphans(
+        self,
+        root_trace_id: str,
+        known_attempt_ids: Sequence[str],
+    ) -> List[ArtifactSnapshot]:
+        artifacts_dir = self._artifacts_dir(root_trace_id)
+        return await asyncio.to_thread(self._list_orphans_locked, artifacts_dir, set(known_attempt_ids))
+
+    def _list_orphans_locked(self, artifacts_dir: Path, known: set[str]) -> List[ArtifactSnapshot]:
+        artifacts_dir.mkdir(parents=True, exist_ok=True)
+        with FileLock(str(artifacts_dir / ".artifacts.lock")):
+            return [snapshot for snapshot in self._read_all(artifacts_dir) if snapshot.attempt_id not in known]
+
+    async def cleanup_orphans(
+        self,
+        root_trace_id: str,
+        known_attempt_ids: Sequence[str],
+    ) -> List[str]:
+        artifacts_dir = self._artifacts_dir(root_trace_id)
+        return await asyncio.to_thread(self._cleanup_orphans_locked, artifacts_dir, set(known_attempt_ids))
+
+    def _cleanup_orphans_locked(self, artifacts_dir: Path, known: set[str]) -> List[str]:
+        artifacts_dir.mkdir(parents=True, exist_ok=True)
+        with FileLock(str(artifacts_dir / ".artifacts.lock")):
+            orphans = [
+                (path, self._read_snapshot(path))
+                for path in sorted(artifacts_dir.glob("*.json"))
+            ]
+            orphans = [(path, snapshot) for path, snapshot in orphans if snapshot.attempt_id not in known]
+            for path, _snapshot in orphans:
+                # Delete only the directory entry that was actually scanned;
+                # snapshot JSON is untrusted and must not select a path.
+                path.unlink()
+            if orphans:
+                _fsync_directory(artifacts_dir)
+            return [snapshot.snapshot_id for _path, snapshot in orphans]
+
+    def _find_for_attempt(self, artifacts_dir: Path, attempt_id: str) -> Optional[ArtifactSnapshot]:
+        return next((item for item in self._read_all(artifacts_dir) if item.attempt_id == attempt_id), None)
+
+    def _read_all(self, artifacts_dir: Path) -> List[ArtifactSnapshot]:
+        return [self._read_snapshot(path) for path in sorted(artifacts_dir.glob("*.json"))]
+
+    @staticmethod
+    def _read_snapshot(path: Path) -> ArtifactSnapshot:
+        try:
             return ArtifactSnapshot.from_dict(json.loads(path.read_text(encoding="utf-8")))
+        except (OSError, ValueError, TypeError, json.JSONDecodeError) as exc:
+            raise TaskStoreError(f"Cannot load artifact snapshot {path}: {exc}") from exc
 
 
 class TraceEventSink:
-    """Append-only orchestration event stream."""
-
-    _locks: Dict[str, asyncio.Lock] = {}
+    """Best-effort legacy event mirror; durable events live in TaskStore."""
 
     def __init__(self, base_path: str = ".trace") -> None:
-        self.base_path = Path(base_path)
+        self.base_path = Path(base_path).resolve()
 
     async def emit(self, root_trace_id: str, event_type: str, payload: Dict[str, Any]) -> None:
-        path = self.base_path / root_trace_id / "orchestration" / "events.jsonl"
-        key = str(path.resolve())
-        lock = self._locks.setdefault(key, asyncio.Lock())
+        await asyncio.to_thread(self._emit_locked, root_trace_id, event_type, payload)
+
+    def _emit_locked(self, root_trace_id: str, event_type: str, payload: Dict[str, Any]) -> None:
+        directory = _safe_child(self.base_path, root_trace_id, "orchestration")
+        directory.mkdir(parents=True, exist_ok=True)
+        path = directory / "events.jsonl"
         event = {
-            "event_id": new_id(),
+            "event_id": _event_id(root_trace_id, os.urandom(8).hex()),
             "event_type": event_type,
             "root_trace_id": root_trace_id,
             "created_at": utc_now(),
-            "payload": payload,
+            "payload": json_values(payload),
         }
-        async with lock:
-            path.parent.mkdir(parents=True, exist_ok=True)
+        with FileLock(str(directory / ".events.lock")):
             with path.open("a", encoding="utf-8") as handle:
                 handle.write(json.dumps(event, ensure_ascii=False, sort_keys=True) + "\n")
                 handle.flush()
 
 
-def _json_values(value: Any) -> Any:
-    from enum import Enum
-    if isinstance(value, Enum):
-        return value.value
-    if isinstance(value, dict):
-        return {str(k): _json_values(v) for k, v in value.items()}
-    if isinstance(value, (list, tuple)):
-        return [_json_values(v) for v in value]
-    return value
+def _safe_child(base: Path, *parts: str) -> Path:
+    if any(not isinstance(part, str) or not part for part in parts):
+        raise ValueError("Path identifiers must be non-empty strings")
+    candidate = base.joinpath(*parts).resolve()
+    try:
+        candidate.relative_to(base.resolve())
+    except ValueError as exc:
+        raise ValueError("Path identifier escapes the configured base path") from exc
+    return candidate
+
+
+def _normalize_submission(submission: AttemptSubmission) -> Tuple[Dict[str, Any], str]:
+    normalized = {
+        "summary": submission.summary.strip(),
+        "artifact_refs": [json_values(asdict(item)) for item in submission.artifact_refs],
+        "evidence_refs": [json_values(asdict(item)) for item in submission.evidence_refs],
+    }
+    canonical = json.dumps(normalized, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    return normalized, hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+
+
+def _snapshot_id(attempt_id: str) -> str:
+    return "attempt-" + hashlib.sha256(attempt_id.encode("utf-8")).hexdigest()
+
+
+def _event_id(root_trace_id: str, sequence: Any) -> str:
+    source = f"{root_trace_id}:{sequence}".encode("utf-8")
+    return hashlib.sha256(source).hexdigest()
+
+
+def _public_command_id(command_id: Optional[str]) -> Optional[str]:
+    if command_id is None:
+        return None
+    return "cmd_" + hashlib.sha256(command_id.encode("utf-8")).hexdigest()
+
+
+def _encode_cursor(root_trace_id: str, sequence: int) -> str:
+    root_digest = hashlib.sha256(root_trace_id.encode("utf-8")).hexdigest()[:16]
+    raw = json.dumps({"v": 1, "root": root_digest, "after": sequence}, separators=(",", ":"))
+    return base64.urlsafe_b64encode(raw.encode("utf-8")).decode("ascii").rstrip("=")
+
+
+def _decode_cursor(cursor: str, root_trace_id: str) -> int:
+    try:
+        padding = "=" * (-len(cursor) % 4)
+        data = json.loads(base64.urlsafe_b64decode(cursor + padding).decode("utf-8"))
+        expected = hashlib.sha256(root_trace_id.encode("utf-8")).hexdigest()[:16]
+        if data.get("v") != 1 or data.get("root") != expected:
+            raise ValueError
+        sequence = int(data["after"])
+        if sequence < 0:
+            raise ValueError
+        return sequence
+    except (KeyError, TypeError, ValueError, UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise ValueError("Invalid event cursor") from exc
+
+
+def _fsync_directory(path: Path) -> None:
+    try:
+        descriptor = os.open(path, os.O_RDONLY)
+    except OSError:
+        return
+    try:
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)
 
 
 __all__ = [
-    "TaskStoreError", "TaskStoreNotFound", "RevisionConflict",
+    "TaskStoreError", "TaskStoreNotFound", "RevisionConflict", "ArtifactConflict",
     "FileSystemTaskStore", "FileSystemArtifactStore", "TraceEventSink",
 ]

agent/tests/test_orchestration_store.py

@@ -1,21 +1,57 @@
+import asyncio
 import json
+import multiprocessing
+from copy import deepcopy
 
 import pytest
 
 from agent.orchestration.models import (
     ArtifactRef,
     AttemptSubmission,
+    BackgroundOperation,
+    CommandRecord,
+    EventDraft,
+    OperationKind,
+    OperationStatus,
     TaskLedger,
     TaskStatus,
 )
 from agent.orchestration.state_machine import InvalidTaskTransition, transition
 from agent.orchestration.store import (
+    ArtifactConflict,
     FileSystemArtifactStore,
     FileSystemTaskStore,
     RevisionConflict,
+    TaskStoreError,
 )
 
 
+def _commit_from_process(base_path, mission, barrier, results):
+    async def run():
+        store = FileSystemTaskStore(base_path)
+        ledger = await store.load("root")
+        ledger.mission = mission
+        barrier.wait()
+        try:
+            await store.commit(ledger, expected_revision=0)
+        except RevisionConflict:
+            results.put("conflict")
+        else:
+            results.put("committed")
+
+    asyncio.run(run())
+
+
+def _freeze_from_process(base_path, barrier, results):
+    async def run():
+        store = FileSystemArtifactStore(base_path)
+        barrier.wait()
+        snapshot = await store.freeze("root", "attempt", AttemptSubmission(summary="same"))
+        results.put(snapshot.snapshot_id)
+
+    asyncio.run(run())
+
+
 @pytest.mark.asyncio
 async def test_ledger_atomic_revision_and_roundtrip(tmp_path):
     store = FileSystemTaskStore(str(tmp_path))
@@ -51,6 +87,319 @@ async def test_snapshot_is_immutable_and_hash_is_stable(tmp_path):
         ArtifactRef(uri="file:///mutable")
 
 
+@pytest.mark.asyncio
+async def test_same_attempt_snapshot_is_first_write_wins(tmp_path):
+    store = FileSystemArtifactStore(str(tmp_path))
+    submission = AttemptSubmission(
+        summary="delivered",
+        artifact_refs=[ArtifactRef(uri="file:///result", digest="abc")],
+    )
+
+    first = await store.freeze("root", "attempt-1", submission)
+    replay = await store.freeze("root", "attempt-1", submission)
+
+    assert replay == first
+    assert await store.get_for_attempt("root", "attempt-1") == first
+    with pytest.raises(ArtifactConflict, match="different artifact snapshot"):
+        await store.freeze(
+            "root",
+            "attempt-1",
+            AttemptSubmission(summary="changed", artifact_refs=submission.artifact_refs),
+        )
+
+
+@pytest.mark.asyncio
+async def test_concurrent_same_attempt_freeze_creates_one_snapshot(tmp_path):
+    first_store = FileSystemArtifactStore(str(tmp_path))
+    second_store = FileSystemArtifactStore(str(tmp_path))
+    submission = AttemptSubmission(summary="same")
+
+    first, second = await asyncio.gather(
+        first_store.freeze("root", "attempt-1", submission),
+        second_store.freeze("root", "attempt-1", submission),
+    )
+
+    assert first == second
+    artifact_files = list((tmp_path / "root" / "orchestration" / "artifacts").glob("*.json"))
+    assert len(artifact_files) == 1
+
+
+@pytest.mark.asyncio
+async def test_artifact_orphans_can_be_listed_and_explicitly_cleaned(tmp_path):
+    store = FileSystemArtifactStore(str(tmp_path))
+    kept = await store.freeze("root", "attempt-kept", AttemptSubmission(summary="kept"))
+    orphan = await store.freeze("root", "attempt-orphan", AttemptSubmission(summary="orphan"))
+
+    assert await store.list_orphans("root", ["attempt-kept"]) == [orphan]
+    assert await store.cleanup_orphans("root", ["attempt-kept"]) == [orphan.snapshot_id]
+    assert await store.get("root", kept.snapshot_id) == kept
+    with pytest.raises(FileNotFoundError):
+        await store.get("root", orphan.snapshot_id)
+
+
+@pytest.mark.asyncio
+async def test_operations_commands_and_old_ledgers_roundtrip(tmp_path):
+    store = FileSystemTaskStore(str(tmp_path))
+    ledger = TaskLedger(root_trace_id="root", mission="mission")
+    operation = BackgroundOperation(
+        operation_id="op-1",
+        root_trace_id="root",
+        kind=OperationKind.DISPATCH,
+        request={"task_ids": ["task-1"]},
+        request_fingerprint="fingerprint",
+        status=OperationStatus.RUNNING,
+    )
+    ledger.operations[operation.operation_id] = operation
+    ledger.command_records["start:key"] = CommandRecord(
+        command_id="start:key",
+        operation="start_dispatch",
+        fingerprint="fingerprint",
+        result_ref={"operation_id": "op-1"},
+        committed_revision=0,
+        operation_id="op-1",
+    )
+    await store.commit(ledger, -1)
+
+    loaded = await store.load("root")
+    assert loaded.operations["op-1"].kind is OperationKind.DISPATCH
+    assert loaded.command_records["start:key"].result_ref == {"operation_id": "op-1"}
+    assert [item.operation_id for item in await store.list_recoverable("root")] == ["op-1"]
+
+    old_path = tmp_path / "old-root" / "orchestration" / "ledger.json"
+    old_path.parent.mkdir(parents=True)
+    old_path.write_text(json.dumps({"root_trace_id": "old-root", "mission": "old"}), encoding="utf-8")
+    old = await store.load("old-root")
+    assert old.operations == {}
+    assert old.command_records == {}
+
+
+@pytest.mark.asyncio
+async def test_events_commit_atomically_and_page_with_opaque_cursor(tmp_path):
+    store = FileSystemTaskStore(str(tmp_path))
+    ledger = TaskLedger(root_trace_id="root", mission="mission")
+    await store.commit(
+        ledger,
+        -1,
+        event=EventDraft("ledger_created", {"mission": "mission"}, command_id="command-1"),
+    )
+    ledger.mission = "updated"
+    await store.commit(
+        ledger,
+        0,
+        event=EventDraft("mission_updated", {"mission": "updated"}, command_id="command-2"),
+    )
+
+    first_page = await store.list_events("root", limit=1)
+    assert [event.sequence for event in first_page.events] == [1]
+    assert first_page.events[0].schema_version == 1
+    assert first_page.has_more is True
+    assert first_page.next_cursor and "command" not in first_page.next_cursor
+    second_page = await store.list_events("root", cursor=first_page.next_cursor)
+    assert [event.event_type for event in second_page.events] == ["mission_updated"]
+    assert second_page.events[0].ledger_revision == 1
+
+    raw = json.loads((tmp_path / "root" / "orchestration" / "ledger.json").read_text())
+    assert len(raw["_events"]) == 2
+    assert "_events" not in (await store.load("root")).to_dict()
+    with pytest.raises(ValueError, match="Invalid event cursor"):
+        await store.list_events("other-root", cursor=first_page.next_cursor)
+
+
+@pytest.mark.asyncio
+async def test_event_command_replay_does_not_duplicate_event(tmp_path):
+    store = FileSystemTaskStore(str(tmp_path))
+    ledger = TaskLedger(root_trace_id="root", mission="mission")
+    await store.commit(ledger, -1, event=EventDraft("created", {}, command_id="command-1"))
+    ledger.mission = "same command replay"
+    await store.commit(ledger, 0, event=EventDraft("created", {}, command_id="command-1"))
+
+    page = await store.list_events("root")
+    assert len(page.events) == 1
+
+
+@pytest.mark.asyncio
+async def test_two_store_instances_enforce_compare_and_swap(tmp_path):
+    first_store = FileSystemTaskStore(str(tmp_path))
+    second_store = FileSystemTaskStore(str(tmp_path))
+    ledger = TaskLedger(root_trace_id="root", mission="mission")
+    await first_store.commit(ledger, -1)
+    first_copy = await first_store.load("root")
+    stale_copy = deepcopy(first_copy)
+    first_copy.mission = "winner"
+    stale_copy.mission = "stale"
+    await first_store.commit(first_copy, 0)
+    with pytest.raises(RevisionConflict):
+        await second_store.commit(stale_copy, 0)
+    assert (await second_store.load("root")).mission == "winner"
+
+
+def test_cross_process_lock_allows_only_one_cas_winner(tmp_path):
+    asyncio.run(
+        FileSystemTaskStore(str(tmp_path)).commit(
+            TaskLedger(root_trace_id="root", mission="initial"),
+            -1,
+        )
+    )
+    context = multiprocessing.get_context("spawn")
+    barrier = context.Barrier(2)
+    results = context.Queue()
+    processes = [
+        context.Process(
+            target=_commit_from_process,
+            args=(str(tmp_path), mission, barrier, results),
+        )
+        for mission in ("first", "second")
+    ]
+    for process in processes:
+        process.start()
+    for process in processes:
+        process.join(timeout=10)
+        assert process.exitcode == 0
+
+    assert sorted([results.get(timeout=1), results.get(timeout=1)]) == ["committed", "conflict"]
+    assert asyncio.run(FileSystemTaskStore(str(tmp_path)).load("root")).revision == 1
+
+
+def test_cross_process_same_attempt_freeze_writes_one_snapshot(tmp_path):
+    context = multiprocessing.get_context("spawn")
+    barrier = context.Barrier(2)
+    results = context.Queue()
+    processes = [
+        context.Process(
+            target=_freeze_from_process,
+            args=(str(tmp_path), barrier, results),
+        )
+        for _index in range(2)
+    ]
+    for process in processes:
+        process.start()
+    for process in processes:
+        process.join(timeout=10)
+        assert process.exitcode == 0
+
+    assert results.get(timeout=1) == results.get(timeout=1)
+    artifact_files = list((tmp_path / "root" / "orchestration" / "artifacts").glob("*.json"))
+    assert len(artifact_files) == 1
+
+
+@pytest.mark.asyncio
+async def test_store_rejects_paths_outside_base(tmp_path):
+    task_store = FileSystemTaskStore(str(tmp_path))
+    artifact_store = FileSystemArtifactStore(str(tmp_path))
+    with pytest.raises(ValueError, match="escapes"):
+        await task_store.load("../escape")
+    with pytest.raises(ValueError, match="escapes"):
+        await artifact_store.get("root", "../../escape")
+
+
+@pytest.mark.asyncio
+async def test_store_reports_corrupt_ledger_and_event_journal(tmp_path):
+    store = FileSystemTaskStore(str(tmp_path))
+    path = tmp_path / "root" / "orchestration" / "ledger.json"
+    path.parent.mkdir(parents=True)
+    path.write_text("{truncated", encoding="utf-8")
+    with pytest.raises(TaskStoreError, match="Cannot load task ledger"):
+        await store.load("root")
+
+    path.write_text(
+        json.dumps({"root_trace_id": "root", "mission": "mission", "_events": {}}),
+        encoding="utf-8",
+    )
+    with pytest.raises(TaskStoreError, match="event journal is invalid"):
+        await store.list_events("root")
+
+    with pytest.raises(ValueError, match="between 1 and 1000"):
+        await store.list_events("root", limit=0)
+
+
+@pytest.mark.asyncio
+async def test_store_wraps_nested_model_and_event_parse_errors(tmp_path):
+    store = FileSystemTaskStore(str(tmp_path))
+    path = tmp_path / "root" / "orchestration" / "ledger.json"
+    path.parent.mkdir(parents=True)
+    path.write_text(
+        json.dumps(
+            {
+                "root_trace_id": "root",
+                "mission": "mission",
+                "tasks": {"broken": {"task_id": "broken", "status": "not-a-status"}},
+            }
+        ),
+        encoding="utf-8",
+    )
+    with pytest.raises(TaskStoreError, match="Cannot parse task ledger"):
+        await store.load("root")
+
+    path.write_text(
+        json.dumps({"root_trace_id": "root", "mission": "mission", "tasks": []}),
+        encoding="utf-8",
+    )
+    with pytest.raises(TaskStoreError, match="Cannot parse task ledger"):
+        await store.load("root")
+
+    path.write_text(
+        json.dumps(
+            {
+                "root_trace_id": "root",
+                "mission": "mission",
+                "_events": [{"sequence": 1}],
+            }
+        ),
+        encoding="utf-8",
+    )
+    with pytest.raises(TaskStoreError, match="contains an invalid event"):
+        await store.list_events("root")
+
+
+@pytest.mark.asyncio
+async def test_failed_atomic_write_does_not_advance_caller_ledger(tmp_path, monkeypatch):
+    store = FileSystemTaskStore(str(tmp_path))
+    ledger = TaskLedger(root_trace_id="root", mission="mission")
+    await store.commit(ledger, expected_revision=-1, event=EventDraft("created"))
+    original_updated_at = ledger.updated_at
+    ledger.mission = "uncommitted"
+
+    def fail_write(_path, _data):
+        raise OSError("injected replace failure")
+
+    monkeypatch.setattr(FileSystemTaskStore, "_atomic_json_write", staticmethod(fail_write))
+    with pytest.raises(OSError, match="injected replace failure"):
+        await store.commit(ledger, expected_revision=0, event=EventDraft("updated"))
+
+    assert ledger.revision == 0
+    assert ledger.updated_at == original_updated_at
+    loaded = await store.load("root")
+    assert loaded.mission == "mission"
+    assert [event.event_type for event in (await store.list_events("root")).events] == ["created"]
+
+
+@pytest.mark.asyncio
+async def test_orphan_cleanup_never_trusts_snapshot_id_as_path(tmp_path):
+    store = FileSystemArtifactStore(str(tmp_path))
+    orchestration_dir = tmp_path / "root" / "orchestration"
+    artifacts_dir = orchestration_dir / "artifacts"
+    artifacts_dir.mkdir(parents=True)
+    ledger_path = orchestration_dir / "ledger.json"
+    ledger_path.write_text("sentinel", encoding="utf-8")
+    (artifacts_dir / "malicious.json").write_text(
+        json.dumps(
+            {
+                "snapshot_id": "../ledger",
+                "attempt_id": "orphan",
+                "normalized_content": {},
+                "sha256": "digest",
+                "artifact_refs": [],
+                "evidence_refs": [],
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    assert await store.cleanup_orphans("root", []) == ["../ledger"]
+    assert ledger_path.read_text(encoding="utf-8") == "sentinel"
+    assert not (artifacts_dir / "malicious.json").exists()
+
+
 def test_state_machine_rejects_shortcut_to_completed():
     assert transition(TaskStatus.PENDING, TaskStatus.RUNNING) == TaskStatus.RUNNING
     with pytest.raises(InvalidTaskTransition):