CHANGELOG.md

@@ -191,6 +191,8 @@
 - Added `ReadOnlyViolationError` to `ChangeSetCapture` to prevent write artifacts on
   read-only plans.
 - Added CLI fail-fast guards on `plan execute` and `plan apply` for read-only plans.
+- Added `DecisionService` with record/list/tree helpers and `SnapshotStore` for hash-based
+  deduplication of context snapshots during plan execution.
 - Expanded CONTRIBUTING.md with detailed guidance on the issue creation process, label system,
   ticket lifecycle, pull request requirements, and review/merge process.
 - Added commit scope, quality, and message format guidelines to CONTRIBUTING.md.

benchmarks/decision_di_bench.py

@@ -11,10 +11,11 @@ from __future__ import annotations
 import sys
 from datetime import UTC, datetime
 from pathlib import Path
-from unittest.mock import MagicMock
+from unittest.mock import create_autospec
 
 try:
     from cleveragents.application.services.decision_service import DecisionService
+    from cleveragents.config.settings import Settings
     from cleveragents.domain.models.core.action import Action, ActionState
     from cleveragents.domain.models.core.decision import Decision, DecisionType
     from cleveragents.domain.models.core.plan import (
@@ -25,11 +26,11 @@ try:
         PlanTimestamps,
         ProcessingState,
     )
-    from cleveragents.infrastructure.database.models import Base
     from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
 except ModuleNotFoundError:
     sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
     from cleveragents.application.services.decision_service import DecisionService
+    from cleveragents.config.settings import Settings
     from cleveragents.domain.models.core.action import Action, ActionState
     from cleveragents.domain.models.core.decision import Decision, DecisionType
     from cleveragents.domain.models.core.plan import (
@@ -40,31 +41,16 @@ except ModuleNotFoundError:
         PlanTimestamps,
         ProcessingState,
     )
-    from cleveragents.infrastructure.database.models import Base
     from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
 
-from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
 
 _PLAN_ID = "01HV00000000000000DIBENCH1"
 
 
 def _make_uow() -> UnitOfWork:
     """Create a UoW backed by an in-memory SQLite database."""
-    engine = create_engine("sqlite:///:memory:", echo=False)
-    Base.metadata.create_all(engine)
-
-    uow = UnitOfWork.__new__(UnitOfWork)
-    uow.database_url = "sqlite:///:memory:"
-    uow._engine = engine
-    uow._session_factory = sessionmaker(
-        bind=engine,
-        expire_on_commit=False,
-        autoflush=False,
-        autocommit=False,
-    )
-    uow._database_initialized = True
-    uow._prompt_for_migration = None
+    uow = UnitOfWork("sqlite:///:memory:")
+    uow.init_database()
     return uow
 
 
@@ -106,21 +92,6 @@ def _seed_prerequisites(uow: UnitOfWork) -> None:
         plan_repo.create(plan)
 
 
-def _make_decision(
-    seq: int = 0,
-    parent_id: str | None = None,
-    dtype: DecisionType = DecisionType.PROMPT_DEFINITION,
-) -> Decision:
-    return Decision(
-        plan_id=_PLAN_ID,
-        parent_decision_id=parent_id,
-        sequence_number=seq,
-        decision_type=dtype,
-        question="Benchmark question?",
-        chosen_option="Benchmark option",
-    )
-
-
 class TimeDecisionServiceResolution:
     """Benchmark DecisionService resolution from the DI container."""
 
@@ -155,16 +126,19 @@ class TimeDecisionRecord:
     def setup(self) -> None:
         self.uow = _make_uow()
         _seed_prerequisites(self.uow)
-        self.svc = DecisionService(settings=MagicMock(), unit_of_work=self.uow)
+        mock_settings = create_autospec(Settings, instance=True)
+        mock_settings.database_url = "sqlite:///:memory:"
+        self.svc = DecisionService(settings=mock_settings, unit_of_work=self.uow)
         self._seq = 100
 
     def time_decision_record(self) -> None:
         self._seq += 1
-        d = _make_decision(
-            seq=self._seq,
-            dtype=DecisionType.STRATEGY_CHOICE,
+        self.svc.record_decision(
+            plan_id=_PLAN_ID,
+            decision_type=DecisionType.STRATEGY_CHOICE,
+            question="Benchmark question?",
+            chosen_option="Benchmark option",
         )
-        self.svc.record_decision(d)
 
 
 class TimeDecisionTreeRetrieval:
@@ -175,30 +149,34 @@ class TimeDecisionTreeRetrieval:
     def setup(self) -> None:
         self.uow = _make_uow()
         _seed_prerequisites(self.uow)
-        self.svc = DecisionService(settings=MagicMock(), unit_of_work=self.uow)
+        mock_settings = create_autospec(Settings, instance=True)
+        mock_settings.database_url = "sqlite:///:memory:"
+        self.svc = DecisionService(settings=mock_settings, unit_of_work=self.uow)
 
         # Build a small tree
-        root = _make_decision(seq=0)
-        self.svc.record_decision(root)
-        self.root_id = root.decision_id
+        root = self.svc.record_decision(
+            plan_id=_PLAN_ID,
+            decision_type=DecisionType.PROMPT_DEFINITION,
+            question="Benchmark question?",
+            chosen_option="Benchmark option",
+        )
 
-        seq = 1
         for _ in range(3):
-            child = _make_decision(
-                seq=seq,
-                parent_id=root.decision_id,
-                dtype=DecisionType.STRATEGY_CHOICE,
+            child = self.svc.record_decision(
+                plan_id=_PLAN_ID,
+                decision_type=DecisionType.STRATEGY_CHOICE,
+                question="Benchmark question?",
+                chosen_option="Benchmark option",
+                parent_decision_id=root.decision_id,
             )
-            self.svc.record_decision(child)
             for _ in range(2):
-                seq += 1
-                gc = _make_decision(
-                    seq=seq,
-                    parent_id=child.decision_id,
-                    dtype=DecisionType.IMPLEMENTATION_CHOICE,
+                self.svc.record_decision(
+                    plan_id=_PLAN_ID,
+                    decision_type=DecisionType.IMPLEMENTATION_CHOICE,
+                    question="Benchmark question?",
+                    chosen_option="Benchmark option",
+                    parent_decision_id=child.decision_id,
                 )
-                self.svc.record_decision(gc)
-            seq += 1
 
     def time_decision_tree_retrieval(self) -> None:
-        self.svc.get_decision_tree(self.root_id)
+        self.svc.get_tree(_PLAN_ID)

benchmarks/decision_recording_bench.py

@@ -0,0 +1,169 @@
+"""ASV benchmarks for DecisionService recording throughput.
+
+Measures record_decision, list_decisions, get_tree, and snapshot operations.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+try:
+    from cleveragents.application.services.decision_service import (
+        DecisionService,
+        SnapshotStore,
+    )
+    from cleveragents.domain.models.core.decision import (
+        ContextSnapshot,
+        DecisionType,
+    )
+except ModuleNotFoundError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+    from cleveragents.application.services.decision_service import (
+        DecisionService,
+        SnapshotStore,
+    )
+    from cleveragents.domain.models.core.decision import (
+        ContextSnapshot,
+        DecisionType,
+    )
+
+
+_PLAN_ID = "01HV00000000000000BENCH002"
+
+
+class TimeRecordDecision:
+    """Benchmark decision recording throughput."""
+
+    timeout = 60
+
+    def setup(self):
+        self.svc = DecisionService()
+        self._plan_counter = 0
+
+    def time_record_single(self):
+        self._plan_counter += 1
+        plan_id = f"01HV00000000000000BN{self._plan_counter:06d}"
+        self.svc.record_decision(
+            plan_id=plan_id,
+            decision_type=DecisionType.STRATEGY_CHOICE,
+            question="Benchmark question",
+            chosen_option="Benchmark choice",
+        )
+
+
+class TimeRecordBatch:
+    """Benchmark recording 50 decisions for one plan."""
+
+    timeout = 60
+
+    def time_record_50(self):
+        svc = DecisionService()
+        root = svc.record_decision(
+            plan_id=_PLAN_ID,
+            decision_type=DecisionType.PROMPT_DEFINITION,
+            question="Root",
+            chosen_option="Root choice",
+        )
+        for _ in range(49):
+            svc.record_decision(
+                plan_id=_PLAN_ID,
+                decision_type=DecisionType.STRATEGY_CHOICE,
+                question="Child",
+                chosen_option="Child choice",
+                parent_decision_id=root.decision_id,
+            )
+
+
+class TimeListDecisions:
+    """Benchmark list_decisions with 50 pre-seeded decisions."""
+
+    timeout = 60
+
+    def setup(self):
+        self.svc = DecisionService()
+        root = self.svc.record_decision(
+            plan_id=_PLAN_ID,
+            decision_type=DecisionType.PROMPT_DEFINITION,
+            question="Root",
+            chosen_option="Root choice",
+        )
+        for _ in range(49):
+            self.svc.record_decision(
+                plan_id=_PLAN_ID,
+                decision_type=DecisionType.STRATEGY_CHOICE,
+                question="Child",
+                chosen_option="Child choice",
+                parent_decision_id=root.decision_id,
+            )
+
+    def time_list_all(self):
+        self.svc.list_decisions(_PLAN_ID)
+
+    def time_list_by_type(self):
+        self.svc.list_by_type(_PLAN_ID, DecisionType.STRATEGY_CHOICE)
+
+
+class TimeGetTree:
+    """Benchmark get_tree with a 3-level tree (1 + 3 + 6 = 10 nodes)."""
+
+    timeout = 60
+
+    def setup(self):
+        self.svc = DecisionService()
+        root = self.svc.record_decision(
+            plan_id=_PLAN_ID,
+            decision_type=DecisionType.PROMPT_DEFINITION,
+            question="Root",
+            chosen_option="Root choice",
+        )
+        for _ in range(3):
+            child = self.svc.record_decision(
+                plan_id=_PLAN_ID,
+                decision_type=DecisionType.STRATEGY_CHOICE,
+                question="Child",
+                chosen_option="Child choice",
+                parent_decision_id=root.decision_id,
+            )
+            for _ in range(2):
+                self.svc.record_decision(
+                    plan_id=_PLAN_ID,
+                    decision_type=DecisionType.IMPLEMENTATION_CHOICE,
+                    question="Grandchild",
+                    chosen_option="GC choice",
+                    parent_decision_id=child.decision_id,
+                )
+
+    def time_get_tree(self):
+        self.svc.get_tree(_PLAN_ID)
+
+
+class TimeSnapshotStore:
+    """Benchmark SnapshotStore operations."""
+
+    timeout = 60
+
+    def setup(self):
+        self.store = SnapshotStore()
+        self._counter = 0
+        # Pre-seed 100 snapshots
+        for i in range(100):
+            snap = ContextSnapshot(
+                hot_context_hash=f"sha256:bench{i:04d}",
+                hot_context_ref=f"ref-{i}",
+            )
+            self.store.store(f"DEC_{i:06d}", snap)
+
+    def time_store_snapshot(self):
+        self._counter += 1
+        snap = ContextSnapshot(
+            hot_context_hash=f"sha256:new{self._counter:06d}",
+            hot_context_ref=f"ref-new-{self._counter}",
+        )
+        self.store.store(f"NEW_{self._counter:06d}", snap)
+
+    def time_get_by_hash(self):
+        self.store.get_by_hash("sha256:bench0050")
+
+    def time_get_snapshot(self):
+        self.store.get("DEC_000050")

docs/reference/decision_service.md

@@ -0,0 +1,148 @@
+# Decision Service Reference
+
+## Overview
+
+`DecisionService` provides the application-layer interface for recording
+decisions during plan execution, retrieving decision histories, and
+managing context snapshots.  It lives in
+`src/cleveragents/application/services/decision_service.py`.
+
+## Dual-Mode Persistence
+
+| Mode       | UnitOfWork | Storage                                        |
+|------------|------------|------------------------------------------------|
+| In-memory  | `None`     | Internal dicts (`_decisions`, `_plan_decisions`) |
+| Persisted  | provided   | DB via `DecisionRepository` + in-memory cache  |
+
+When a `UnitOfWork` is wired, mutations are written to the database
+first, then the in-memory cache is updated (write-through).
+
+## Constructor
+
+```python
+DecisionService(
+    settings: Settings | None = None,
+    unit_of_work: UnitOfWork | None = None,
+)
+```
+
+- **settings** – Application settings (optional for in-memory mode).
+- **unit_of_work** – When provided, decisions are persisted via
+  `DecisionRepository`.
+
+## Recording
+
+### `record_decision()`
+
+```python
+svc.record_decision(
+    plan_id="01HV...",
+    decision_type=DecisionType.STRATEGY_CHOICE,   # or a string
+    question="Which approach?",
+    chosen_option="Build a REST API",
+    parent_decision_id=None,            # optional
+    alternatives_considered=None,       # list[str]
+    confidence_score=None,              # 0.0-1.0
+    rationale="",                       # human-readable
+    actor_reasoning=None,               # raw LLM trace
+    context_snapshot=None,              # auto-captured if omitted
+    artifacts_produced=None,            # list[ArtifactRef]
+    is_correction=False,
+    corrects_decision_id=None,
+    correction_reason=None,
+)
+```
+
+- Auto-assigns a monotonically increasing **sequence number** per plan.
+- Auto-generates a **ULID** for `decision_id`.
+- Auto-captures a **context snapshot** (SHA-256 hash) when no explicit
+  snapshot is provided.
+- Stores the snapshot in the built-in `SnapshotStore`.
+
+**Raises:**
+
+- `ValidationError` – required fields missing or empty.
+- `DuplicateDecisionError` – decision ID already exists.
+
+## Retrieval
+
+| Method                                | Description                                |
+|---------------------------------------|--------------------------------------------|
+| `get_decision(decision_id)`           | Single decision by ULID                    |
+| `list_decisions(plan_id)`             | All decisions for a plan, ordered by seq   |
+| `list_by_type(plan_id, decision_type)`| Filtered by type, ordered by seq           |
+
+All raise `DecisionNotFoundError` when the target decision does not
+exist (where applicable).
+
+## Tree Operations
+
+| Method                        | Description                              |
+|-------------------------------|------------------------------------------|
+| `get_tree(plan_id)`           | BFS from root(s), level by level         |
+| `get_path_to_root(decision_id)` | Walk from a decision up to the root    |
+
+## Superseded Decisions
+
+| Method                                        | Description                           |
+|-----------------------------------------------|---------------------------------------|
+| `mark_superseded(decision_id, new_decision_id)` | Mark a decision as superseded       |
+| `get_superseded(plan_id)`                      | List all superseded decisions        |
+
+## Delete
+
+```python
+svc.delete_decision(decision_id)  # -> True
+```
+
+Removes from both persistence and the snapshot store.  Raises
+`DecisionNotFoundError` if not found.
+
+## Snapshot Store
+
+`SnapshotStore` manages `ContextSnapshot` objects keyed by decision ID
+with hash-based deduplication.
+
+| Method                          | Description                                  |
+|---------------------------------|----------------------------------------------|
+| `get_snapshot(decision_id)`     | Retrieve snapshot for a decision             |
+| `get_snapshots_for_plan(plan_id)` | All snapshots for a plan's decisions       |
+| `snapshots.get_by_hash(hash)`   | Decision IDs sharing the same context hash  |
+| `snapshots.store(id, snap)`     | Store a snapshot (internal)                  |
+| `snapshots.remove(id)`          | Remove a snapshot                            |
+
+## Statistics
+
+| Method                        | Description                      |
+|-------------------------------|----------------------------------|
+| `count_decisions(plan_id)`    | Number of decisions for a plan   |
+| `get_next_sequence(plan_id)`  | Next available sequence number   |
+
+## Custom Exceptions
+
+| Exception               | Base                     | Description                         |
+|--------------------------|--------------------------|-------------------------------------|
+| `DuplicateDecisionError` | `BusinessRuleViolation`  | Decision ID already exists          |
+| `DecisionNotFoundError`  | `ResourceNotFoundError`  | Decision not found                  |
+| `SequenceConflictError`  | `BusinessRuleViolation`  | Sequence number already used        |
+
+## Running Tests
+
+```bash
+# Behave BDD tests
+nox -s unit_tests -- features/decision_recording.feature
+
+# Robot Framework integration smoke tests
+nox -s integration_tests
+
+# ASV benchmarks
+nox -s benchmark
+```
+
+## Related
+
+- Domain model: `src/cleveragents/domain/models/core/decision.py`
+- Persistence: `src/cleveragents/infrastructure/database/repositories.py`
+  (`DecisionRepository`)
+- Database schema: `docs/reference/database_schema.md`
+- ADR: `docs/adr/ADR-033-decision-recording-protocol.md`

docs/reference/di.md

@@ -51,8 +51,8 @@ decision_service = providers.Factory(
 |-------------------------|--------------------------------------------|
 | `record_decision()`    | Persist a new decision                     |
 | `get_decision()`       | Retrieve a decision by ID                  |
-| `get_decisions_for_plan()` | Get all decisions for a plan           |
-| `get_decision_tree()`  | BFS traversal of the decision tree         |
+| `list_decisions()`     | Get all decisions for a plan              |
+| `get_tree()`           | BFS traversal of the decision tree         |
 | `get_path_to_root()`   | Walk from a decision up to the root        |
 | `mark_superseded()`    | Mark a decision as superseded              |
 | `list_by_type()`       | List decisions by type for a plan          |
@@ -92,17 +92,15 @@ decision_svc = container.decision_service()
 lifecycle_svc = container.plan_lifecycle_service()
 
 # Record a decision manually
-from cleveragents.domain.models.core.decision import Decision, DecisionType
+from cleveragents.domain.models.core.decision import DecisionType
 
-decision = Decision(
+decision = decision_svc.record_decision(
     plan_id="01HV...",
-    sequence_number=0,
     decision_type=DecisionType.STRATEGY_CHOICE,
     question="Which approach?",
     chosen_option="Approach A",
 )
-decision_svc.record_decision(decision)
 
 # Retrieve decisions
-decisions = decision_svc.get_decisions_for_plan("01HV...")
+decisions = decision_svc.list_decisions("01HV...")
 ```

features/decision_recording.feature

@@ -0,0 +1,403 @@
+Feature: Decision recording and snapshot store
+  As a developer
+  I want a DecisionService that records decisions with auto-sequencing and snapshots
+  So that plan decision trees are tracked with full context for replay
+
+  Background:
+    Given a decision service
+
+  # --- Recording ---
+
+  Scenario: Record a root decision with auto-sequencing
+    When I record a prompt_definition decision for plan "P1" with question "What is the plan prompt?"
+    Then the dsvc decision should be recorded successfully
+    And the dsvc decision sequence number should be 0
+    And the dsvc decision type should be "prompt_definition"
+    And the dsvc decision should be a root decision
+    And the dsvc decision should have a valid ULID as decision_id
+    And the dsvc decision context snapshot hash should not be empty
+
+  Scenario: Record multiple decisions with monotonic sequencing
+    When I record a prompt_definition decision for plan "P1" with question "Plan prompt"
+    And I record a strategy_choice decision for plan "P1" with question "Which approach?"
+    And I record an implementation_choice decision for plan "P1" with question "How to implement?"
+    Then the dsvc plan "P1" should have 3 decisions
+    And the dsvc decisions should have sequence numbers 0 1 2
+
+  Scenario: Record a decision with explicit context snapshot
+    Given a context snapshot with hash "sha256:abc123" and ref "s3://bucket/ctx"
+    When I record a strategy_choice decision for plan "P1" with the explicit snapshot
+    Then the dsvc decision context snapshot hash should be "sha256:abc123"
+    And the dsvc decision context snapshot ref should be "s3://bucket/ctx"
+
+  Scenario: Record a decision with confidence score
+    When I record a strategy_choice decision for plan "P1" with confidence 0.85
+    Then the dsvc decision confidence score should be 0.85
+
+  Scenario: Record a correction decision
+    When I record a prompt_definition decision for plan "P1" with question "Original"
+    And I record a correction decision for plan "P1" correcting the first decision
+    Then the dsvc correction decision is_correction should be true
+    And the dsvc correction decision corrects_decision_id should match the first decision
+
+  Scenario: Record a decision with alternatives
+    When I record a strategy_choice decision for plan "P1" with alternatives "Option A" "Option B" "Option C"
+    Then the dsvc decision should have 3 alternatives considered
+
+  Scenario: Record a decision with artifacts
+    When I record an implementation_choice decision for plan "P1" with artifacts
+    Then the dsvc decision should have 2 artifacts produced
+
+  Scenario: Recording with empty plan_id raises ValidationError
+    When I try to record a decision with empty plan_id
+    Then a dsvc validation error should be raised
+    And the dsvc error should mention "plan_id"
+
+  Scenario: Recording with empty question raises ValidationError
+    When I try to record a decision with empty question
+    Then a dsvc validation error should be raised
+    And the dsvc error should mention "question"
+
+  Scenario: Recording with empty chosen_option raises ValidationError
+    When I try to record a decision with empty chosen_option
+    Then a dsvc validation error should be raised
+    And the dsvc error should mention "chosen_option"
+
+  Scenario: Recording with invalid decision_type raises ValueError
+    When I try to record a decision with invalid type "nonexistent_type"
+    Then a dsvc value error should be raised
+    And the dsvc error should mention "nonexistent_type"
+
+  Scenario: Recording with confidence_score at boundaries
+    When I record a strategy_choice decision for plan "P1" with confidence 0.0
+    Then the dsvc decision confidence score should be 0.0
+    When I record a strategy_choice decision for plan "P1" with confidence 1.0
+    Then the dsvc decision confidence score should be 1.0
+
+  Scenario: Recording with negative confidence_score raises ValidationError
+    When I try to record a decision with confidence -0.1
+    Then a dsvc validation error should be raised
+
+  Scenario: Recording with confidence_score above 1.0 raises ValidationError
+    When I try to record a decision with confidence 1.1
+    Then a dsvc validation error should be raised
+
+  Scenario: Auto-capture generates a snapshot hash
+    When I record a strategy_choice decision for plan "P1" with question "Test auto-capture"
+    Then the dsvc decision context snapshot hash should start with "sha256:"
+
+  # --- Retrieval ---
+
+  Scenario: Get a decision by ID
+    When I record a prompt_definition decision for plan "P1" with question "Test get"
+    And I get the decision by its ID
+    Then the dsvc retrieved decision should match the recorded decision
+
+  Scenario: Get a non-existent decision raises DecisionNotFoundError
+    When I try to get decision "01NONEXISTENT00000000000000"
+    Then a dsvc decision not found error should be raised
+
+  Scenario: List decisions for a plan returns all decisions ordered
+    When I record a prompt_definition decision for plan "P1" with question "First"
+    And I record a strategy_choice decision for plan "P1" with question "Second"
+    And I record an implementation_choice decision for plan "P1" with question "Third"
+    And I list decisions for plan "P1"
+    Then the dsvc decision list should have 3 entries
+    And the dsvc decision list should be ordered by sequence number
+
+  Scenario: List decisions for empty plan returns empty list
+    When I list decisions for plan "P_EMPTY"
+    Then the dsvc decision list should have 0 entries
+
+  Scenario: List decisions by type filters correctly
+    When I record a prompt_definition decision for plan "P1" with question "Root"
+    And I record a strategy_choice decision for plan "P1" with question "Strategy 1"
+    And I record a strategy_choice decision for plan "P1" with question "Strategy 2"
+    And I record an implementation_choice decision for plan "P1" with question "Impl"
+    And I filter decisions for plan "P1" by type "strategy_choice"
+    Then the dsvc decision list should have 2 entries
+
+  # --- Tree operations ---
+
+  Scenario: Get tree returns BFS order from root
+    When I record a prompt_definition decision for plan "P1" with question "Root"
+    And I record a strategy_choice decision for plan "P1" with parent as child "Child 1"
+    And I record a strategy_choice decision for plan "P1" with parent as child "Child 2"
+    And I record an implementation_choice decision for plan "P1" with second parent as child "Grandchild"
+    And I get the tree for plan "P1"
+    Then the dsvc tree should have 4 decisions
+    And the dsvc first tree decision should be a root decision
+    And the dsvc tree should be in BFS level order
+
+  Scenario: Get tree for empty plan returns empty list
+    When I get the tree for plan "P_EMPTY"
+    Then the dsvc tree should have 0 decisions
+
+  Scenario: Get path to root navigates upward
+    When I record a prompt_definition decision for plan "P1" with question "Root"
+    And I record a strategy_choice decision for plan "P1" with parent as child "Level 1"
+    And I record an implementation_choice decision for plan "P1" with second parent as child "Level 2"
+    And I get the path to root from the last decision
+    Then the dsvc path should have 3 decisions
+    And the dsvc first path decision should be the last recorded decision
+    And the dsvc last path decision should be the root
+
+  Scenario: Get path to root for non-existent decision raises error
+    When I try to get path to root for decision "01NONEXISTENT00000000000000"
+    Then a dsvc decision not found error should be raised
+
+  # --- Superseded decisions ---
+
+  Scenario: Mark a decision as superseded
+    When I record a prompt_definition decision for plan "P1" with question "Original"
+    And I record a strategy_choice decision for plan "P1" with question "Replacement"
+    And I mark the first decision as superseded by the second
+    Then the dsvc first decision should be superseded
+    And the dsvc first decision superseded_by should match the second decision
+
+  Scenario: Get superseded decisions for a plan
+    When I record a prompt_definition decision for plan "P1" with question "Original"
+    And I record a strategy_choice decision for plan "P1" with question "Replacement"
+    And I mark the first decision as superseded by the second
+    And I get superseded decisions for plan "P1"
+    Then the dsvc superseded list should have 1 entry
+
+  Scenario: Mark non-existent decision raises DecisionNotFoundError
+    When I try to mark decision "01NONEXISTENT00000000000000" as superseded
+    Then a dsvc decision not found error should be raised
+
+  # --- Delete ---
+
+  Scenario: Delete a decision
+    When I record a prompt_definition decision for plan "P1" with question "To delete"
+    And I delete the recorded decision
+    Then the dsvc decision should be deleted
+    And the dsvc plan "P1" should have 0 decisions
+
+  Scenario: Delete non-existent decision raises DecisionNotFoundError
+    When I try to delete decision "01NONEXISTENT00000000000000"
+    Then a dsvc decision not found error should be raised
+
+  # --- Snapshot store ---
+
+  Scenario: Snapshot is stored when decision is recorded
+    When I record a strategy_choice decision for plan "P1" with question "Snapshot test"
+    And I get the snapshot for the recorded decision
+    Then the dsvc snapshot should not be None
+
+  Scenario: Get snapshots for plan returns all snapshots
+    When I record a prompt_definition decision for plan "P1" with question "First"
+    And I record a strategy_choice decision for plan "P1" with question "Second"
+    And I get snapshots for plan "P1"
+    Then the dsvc snapshots dict should have 2 entries
+
+  Scenario: Snapshot hash deduplication index works
+    Given a context snapshot with hash "sha256:samehash" and ref "ref1"
+    When I record a strategy_choice decision for plan "P1" with the explicit snapshot
+    And I record a strategy_choice decision for plan "P2" with the same explicit snapshot
+    And I query the snapshot store by hash "sha256:samehash"
+    Then the dsvc hash query should return 2 decision IDs
+
+  Scenario: Snapshot is removed when decision is deleted
+    When I record a prompt_definition decision for plan "P1" with question "Ephemeral"
+    And I delete the recorded decision
+    And I get the snapshot for the deleted decision
+    Then the dsvc snapshot should be None
+
+  # --- Statistics ---
+
+  Scenario: Count decisions for a plan
+    When I record a prompt_definition decision for plan "P1" with question "One"
+    And I record a strategy_choice decision for plan "P1" with question "Two"
+    Then the dsvc decision count for plan "P1" should be 2
+
+  Scenario: Get next sequence returns the expected value
+    Then the dsvc next sequence for plan "P_NEW" should be 0
+    When I record a prompt_definition decision for plan "P_NEW" with question "First"
+    Then the dsvc next sequence for plan "P_NEW" should be 1
+
+  # --- DuplicateDecisionError ---
+
+  Scenario: DuplicateDecisionError contains decision_id
+    Given a DuplicateDecisionError for decision "DUP123"
+    Then the dsvc duplicate error decision_id should be "DUP123"
+    And the dsvc duplicate error message should contain "DUP123"
+
+  # --- SequenceConflictError ---
+
+  Scenario: SequenceConflictError contains plan_id and sequence
+    Given a SequenceConflictError for plan "P1" sequence 5
+    Then the dsvc sequence error plan_id should be "P1"
+    And the dsvc sequence error sequence_number should be 5
+    And the dsvc sequence error message should contain "P1"
+
+  # --- SnapshotStore standalone ---
+
+  Scenario: SnapshotStore remove returns False for unknown decision
+    Given a standalone snapshot store
+    When I remove snapshot for decision "UNKNOWN"
+    Then the dsvc snapshot remove result should be False
+
+  Scenario: SnapshotStore get returns None for unknown decision
+    Given a standalone snapshot store
+    When I get snapshot for decision "UNKNOWN"
+    Then the dsvc standalone snapshot should be None
+
+  Scenario: SnapshotStore get_by_hash returns empty for unknown hash
+    Given a standalone snapshot store
+    When I query the standalone store by hash "sha256:nope"
+    Then the dsvc standalone hash query should return 0 decision IDs
+
+  Scenario: DecisionService string type coercion works
+    When I record a decision for plan "P1" with string type "strategy_choice"
+    Then the dsvc decision type should be "strategy_choice"
+
+  # --- Edge-case coverage ---
+
+  Scenario: Snapshot stored without a hash skips hash index
+    Given a standalone snapshot store
+    And a context snapshot with no hash
+    When I store the hashless snapshot for decision "D1"
+    And I get standalone snapshot for decision "D1"
+    Then the dsvc standalone retrieved snapshot should not be None
+    And the dsvc standalone retrieved snapshot hash should be empty
+
+  Scenario: Snapshot store remove cleans up hash index
+    Given a standalone snapshot store
+    And a context snapshot with hash "sha256:removeme" and ref "ref"
+    When I store the explicit snapshot for decision "D_REM"
+    And I remove snapshot for decision "D_REM"
+    Then the dsvc snapshot remove result should be True
+    And the dsvc standalone hash query for "sha256:removeme" should return 0 decision IDs
+
+  Scenario: Snapshot store remove last entry deletes hash bucket
+    Given a standalone snapshot store
+    And a context snapshot with hash "sha256:onlyone" and ref "ref"
+    When I store the explicit snapshot for decision "D_ONLY"
+    And I remove snapshot for decision "D_ONLY"
+    Then the dsvc snapshot remove result should be True
+    And the dsvc standalone hash query for "sha256:onlyone" should return 0 decision IDs
+
+  Scenario: Snapshot store remove snapshot with no hash
+    Given a standalone snapshot store
+    And a context snapshot with no hash
+    When I store the hashless snapshot for decision "D_NOHASH"
+    And I remove snapshot for decision "D_NOHASH"
+    Then the dsvc snapshot remove result should be True
+
+  Scenario: Snapshot store remove with multiple entries keeps remaining
+    Given a standalone snapshot store
+    And a context snapshot with hash "sha256:shared" and ref "ref"
+    When I store the explicit snapshot for decision "D_KEEP1"
+    And I store the explicit snapshot for decision "D_KEEP2"
+    And I remove snapshot for decision "D_KEEP1"
+    Then the dsvc snapshot remove result should be True
+    And the dsvc standalone hash query for "sha256:shared" should return 1 decision IDs
+
+  Scenario: Get tree when no decision is a root returns all decisions
+    When I record a decision for plan "P1" with a forced non-root parent
+    And I get the tree for plan "P1"
+    Then the dsvc tree should have 1 decisions
+
+  Scenario: List by type with string coercion filters correctly
+    When I record a prompt_definition decision for plan "P1" with question "Type1"
+    And I record a strategy_choice decision for plan "P1" with question "Type2"
+    And I filter decisions for plan "P1" by type "prompt_definition"
+    Then the dsvc decision list should have 1 entries
+
+  Scenario: Recording with whitespace-only plan_id raises ValidationError
+    When I try to record a decision with whitespace plan_id
+    Then a dsvc validation error should be raised
+    And the dsvc error should mention "plan_id"
+
+  Scenario: Recording with whitespace-only question raises ValidationError
+    When I try to record a decision with whitespace question
+    Then a dsvc validation error should be raised
+    And the dsvc error should mention "question"
+
+  Scenario: Recording with whitespace-only chosen_option raises ValidationError
+    When I try to record a decision with whitespace chosen_option
+    Then a dsvc validation error should be raised
+    And the dsvc error should mention "chosen_option"
+
+  Scenario: SnapshotStore list_for_plan skips missing snapshots
+    Given a standalone snapshot store
+    When I list snapshots for decisions with missing entries
+    Then the dsvc standalone plan snapshots should have 0 entries
+
+  # --- Persisted-mode (database-backed) ---
+
+  Scenario: Persisted record and get decision round-trips through database
+    Given a persisted decision service
+    When I record a prompt_definition decision for plan "P1" with question "Persisted root"
+    And I get the decision by its ID
+    Then the dsvc retrieved decision should match the recorded decision
+
+  Scenario: Persisted get non-existent decision raises DecisionNotFoundError
+    Given a persisted decision service
+    When I try to get decision "01NONEXISTENT00000000000000"
+    Then a dsvc decision not found error should be raised
+
+  Scenario: Persisted list decisions returns ordered results from database
+    Given a persisted decision service
+    When I record a prompt_definition decision for plan "P1" with question "First persisted"
+    And I record a strategy_choice decision for plan "P1" with question "Second persisted"
+    And I list decisions for plan "P1"
+    Then the dsvc decision list should have 2 entries
+    And the dsvc decision list should be ordered by sequence number
+
+  Scenario: Persisted list by type filters through database
+    Given a persisted decision service
+    When I record a prompt_definition decision for plan "P1" with question "Root"
+    And I record a strategy_choice decision for plan "P1" with question "Strategy"
+    And I record a strategy_choice decision for plan "P1" with question "Strategy 2"
+    And I filter decisions for plan "P1" by type "strategy_choice"
+    Then the dsvc decision list should have 2 entries
+
+  Scenario: Persisted get path to root navigates through database
+    Given a persisted decision service
+    When I record a prompt_definition decision for plan "P1" with question "Root"
+    And I record a strategy_choice decision for plan "P1" with parent as child "Level 1"
+    And I record an implementation_choice decision for plan "P1" with second parent as child "Level 2"
+    And I get the path to root from the last decision
+    Then the dsvc path should have 3 decisions
+    And the dsvc last path decision should be the root
+
+  Scenario: Persisted get superseded returns superseded decisions from database
+    Given a persisted decision service
+    When I record a prompt_definition decision for plan "P1" with question "Original"
+    And I record a strategy_choice decision for plan "P1" with question "Replacement"
+    And I mark the first decision as superseded by the second
+    And I get superseded decisions for plan "P1"
+    Then the dsvc superseded list should have 1 entry
+
+  Scenario: Persisted mark superseded updates database and cache
+    Given a persisted decision service
+    When I record a prompt_definition decision for plan "P1" with question "Old"
+    And I record a strategy_choice decision for plan "P1" with question "New"
+    And I mark the first decision as superseded by the second
+    Then the dsvc first decision should be superseded
+    And the dsvc first decision superseded_by should match the second decision
+
+  Scenario: Persisted delete removes decision from database
+    Given a persisted decision service
+    When I record a prompt_definition decision for plan "P1" with question "Ephemeral"
+    And I delete the recorded decision
+    Then the dsvc decision should be deleted
+    And the dsvc plan "P1" should have 0 decisions
+
+  Scenario: Persisted duplicate decision raises DuplicateDecisionError
+    Given a persisted decision service
+    When I record a prompt_definition decision for plan "P1" with question "Original"
+    And I try to store a duplicate of the recorded decision
+    Then a dsvc duplicate decision error should be raised
+
+  Scenario: Persisted sequence resumes after service restart
+    Given a persisted decision service
+    When I record a prompt_definition decision for plan "P1" with question "Before restart"
+    And I record a strategy_choice decision for plan "P1" with question "Also before restart"
+    Then the dsvc decisions should have sequence numbers 0 1
+    When I recreate the decision service with the same database
+    And I record a strategy_choice decision for plan "P1" with question "After restart"
+    Then the dsvc decision sequence number should be 2
+    And the dsvc next sequence for plan "P1" should be 3

features/decision_service_coverage.feature

@@ -22,18 +22,15 @@ Feature: DecisionService application-layer coverage
 
   Scenario: dsvc- record_decision persists and returns the decision
     Given dsvc- a DecisionService with a mocked UnitOfWork
-    And dsvc- a sample root Decision object
-    When dsvc- I call record_decision with the sample decision
+    When dsvc- I call record_decision with plan_id and required args
     Then dsvc- the UoW transaction should have been entered
-    And dsvc- ctx.decisions.create should have been called with the decision
-    And dsvc- the returned decision should be the same object
+    And dsvc- ctx.decisions.create should have been called once
+    And dsvc- the returned decision should have the correct plan_id
 
-  Scenario: dsvc- record_decision logs info and debug messages
+  Scenario: dsvc- record_decision logs info on success
     Given dsvc- a DecisionService with a mocked UnitOfWork and captured logger
-    And dsvc- a sample root Decision object
-    When dsvc- I call record_decision with the sample decision
-    Then dsvc- the logger should have recorded an info call with "recording_decision"
-    And dsvc- the logger should have recorded a debug call with "decision_recorded"
+    When dsvc- I call record_decision with plan_id and required args
+    Then dsvc- the logger should have recorded an info call with "decision.recorded"
 
   # ------------------------------------------------------------------
   # get_decision
@@ -46,57 +43,38 @@ Feature: DecisionService application-layer coverage
     Then dsvc- ctx.decisions.get should have been called with the ID
     And dsvc- the returned value should be the expected Decision
 
-  Scenario: dsvc- get_decision returns None when not found
+  Scenario: dsvc- get_decision raises DecisionNotFoundError when not found
     Given dsvc- a DecisionService with a mocked UnitOfWork
     And dsvc- the mock repo get method returns None
-    When dsvc- I call get_decision with an unknown ID
-    Then dsvc- the returned value should be None
-
-  Scenario: dsvc- get_decision logs a debug message
-    Given dsvc- a DecisionService with a mocked UnitOfWork and captured logger
-    And dsvc- the mock repo get method returns a Decision
-    When dsvc- I call get_decision with a known ID
-    Then dsvc- the logger should have recorded a debug call with "getting_decision"
+    When dsvc- I call get_decision with an unknown ID expecting not-found
+    Then dsvc- a DecisionNotFoundError should have been raised
 
   # ------------------------------------------------------------------
-  # get_decisions_for_plan
+  # list_decisions
   # ------------------------------------------------------------------
 
-  Scenario: dsvc- get_decisions_for_plan returns list of decisions
+  Scenario: dsvc- list_decisions returns list of decisions
     Given dsvc- a DecisionService with a mocked UnitOfWork
     And dsvc- the mock repo get_by_plan method returns 3 decisions
-    When dsvc- I call get_decisions_for_plan with a plan ID
+    When dsvc- I call list_decisions with a plan ID
     Then dsvc- ctx.decisions.get_by_plan should have been called with the plan ID
     And dsvc- the returned list should have 3 decisions
 
-  Scenario: dsvc- get_decisions_for_plan returns empty list when none exist
+  Scenario: dsvc- list_decisions returns empty list when none exist
     Given dsvc- a DecisionService with a mocked UnitOfWork
     And dsvc- the mock repo get_by_plan method returns 0 decisions
-    When dsvc- I call get_decisions_for_plan with a plan ID
+    When dsvc- I call list_decisions with a plan ID
     Then dsvc- the returned list should have 0 decisions
 
-  Scenario: dsvc- get_decisions_for_plan logs a debug message
-    Given dsvc- a DecisionService with a mocked UnitOfWork and captured logger
-    And dsvc- the mock repo get_by_plan method returns 3 decisions
-    When dsvc- I call get_decisions_for_plan with a plan ID
-    Then dsvc- the logger should have recorded a debug call with "getting_decisions_for_plan"
-
   # ------------------------------------------------------------------
-  # get_decision_tree
+  # get_tree
   # ------------------------------------------------------------------
 
-  Scenario: dsvc- get_decision_tree returns BFS-ordered list
+  Scenario: dsvc- get_tree returns BFS-ordered list
     Given dsvc- a DecisionService with a mocked UnitOfWork
-    And dsvc- the mock repo get_tree method returns 4 decisions
-    When dsvc- I call get_decision_tree with a root ID
-    Then dsvc- ctx.decisions.get_tree should have been called with the root ID
-    And dsvc- the returned tree list should have 4 decisions
-
-  Scenario: dsvc- get_decision_tree logs a debug message
-    Given dsvc- a DecisionService with a mocked UnitOfWork and captured logger
-    And dsvc- the mock repo get_tree method returns 4 decisions
-    When dsvc- I call get_decision_tree with a root ID
-    Then dsvc- the logger should have recorded a debug call with "getting_decision_tree"
+    And dsvc- the mock repo get_by_plan method returns 4 decisions as a tree
+    When dsvc- I call get_tree with a plan ID
+    Then dsvc- the returned tree list should have 4 decisions
 
   # ------------------------------------------------------------------
   # get_path_to_root
@@ -109,18 +87,13 @@ Feature: DecisionService application-layer coverage
     Then dsvc- ctx.decisions.get_path_to_root should have been called with the leaf ID
     And dsvc- the returned path list should have 3 decisions
 
-  Scenario: dsvc- get_path_to_root logs a debug message
-    Given dsvc- a DecisionService with a mocked UnitOfWork and captured logger
-    And dsvc- the mock repo get_path_to_root method returns 3 decisions
-    When dsvc- I call get_path_to_root with a leaf ID
-    Then dsvc- the logger should have recorded a debug call with "getting_path_to_root"
-
   # ------------------------------------------------------------------
   # mark_superseded
   # ------------------------------------------------------------------
 
   Scenario: dsvc- mark_superseded updates and returns the decision
     Given dsvc- a DecisionService with a mocked UnitOfWork
+    And dsvc- the mock repo has the replacement decision in cache
     And dsvc- the mock repo update_superseded_by method returns a superseded Decision
     When dsvc- I call mark_superseded with old and new IDs
     Then dsvc- ctx.decisions.update_superseded_by should have been called with both IDs
@@ -128,9 +101,10 @@ Feature: DecisionService application-layer coverage
 
   Scenario: dsvc- mark_superseded logs info with both IDs
     Given dsvc- a DecisionService with a mocked UnitOfWork and captured logger
+    And dsvc- the mock repo has the replacement decision in cache
     And dsvc- the mock repo update_superseded_by method returns a superseded Decision
     When dsvc- I call mark_superseded with old and new IDs
-    Then dsvc- the logger should have recorded an info call with "marking_superseded"
+    Then dsvc- the logger should have recorded an info call with "decision.superseded"
 
   # ------------------------------------------------------------------
   # list_by_type
@@ -148,9 +122,3 @@ Feature: DecisionService application-layer coverage
     And dsvc- the mock repo list_by_type method returns 0 decisions
     When dsvc- I call list_by_type with plan ID and type "error_recovery"
     Then dsvc- the returned type list should have 0 decisions
-
-  Scenario: dsvc- list_by_type logs a debug message
-    Given dsvc- a DecisionService with a mocked UnitOfWork and captured logger
-    And dsvc- the mock repo list_by_type method returns 2 decisions
-    When dsvc- I call list_by_type with plan ID and type "strategy_choice"
-    Then dsvc- the logger should have recorded a debug call with "listing_by_type"

features/steps/decision_di_wiring_steps.py

@@ -10,16 +10,33 @@ existing step definitions.
 from __future__ import annotations
 
 import os
-from unittest.mock import MagicMock
+from typing import TYPE_CHECKING
+from unittest.mock import create_autospec
 
 from behave import given, then, when
 from behave.runner import Context
 
+if TYPE_CHECKING:
+    from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
+
 # -------------------------------------------------------------------
 # Helpers
 # -------------------------------------------------------------------
 
 
+def _make_test_uow() -> UnitOfWork:
+    """Create a UnitOfWork backed by an in-memory SQLite for testing.
+
+    Uses ``UnitOfWork.__init__`` + ``init_database()`` rather than the
+    ``__new__()`` anti-pattern to stay in sync with UoW internals.
+    """
+    from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
+
+    uow = UnitOfWork("sqlite:///:memory:")
+    uow.init_database()
+    return uow
+
+
 def _ensure_test_db_env(context: Context) -> None:
     """Set up a test database URL if not already set."""
     if not hasattr(context, "_decdi_env_saved"):
@@ -81,34 +98,15 @@ def step_check_decision_service(context: Context) -> None:
 @given("decdi- a PlanLifecycleService with a DecisionService")
 def step_create_lifecycle_with_decision(context: Context) -> None:
     """Build an in-memory PlanLifecycleService wired with DecisionService."""
-    from sqlalchemy import create_engine
-    from sqlalchemy.orm import sessionmaker
-
     from cleveragents.application.services.decision_service import DecisionService
     from cleveragents.application.services.plan_lifecycle_service import (
         PlanLifecycleService,
     )
-    from cleveragents.infrastructure.database.models import Base
-    from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
+    from cleveragents.config.settings import Settings
 
-    # Build a fresh in-memory DB
-    engine = create_engine("sqlite:///:memory:", echo=False)
-    Base.metadata.create_all(engine)
+    uow = _make_test_uow()
 
-    # Create a UoW that skips migration
-    uow = UnitOfWork.__new__(UnitOfWork)
-    uow.database_url = "sqlite:///:memory:"
-    uow._engine = engine
-    uow._session_factory = sessionmaker(
-        bind=engine,
-        expire_on_commit=False,
-        autoflush=False,
-        autocommit=False,
-    )
-    uow._database_initialized = True
-    uow._prompt_for_migration = None
-
-    mock_settings = MagicMock()
+    mock_settings = create_autospec(Settings, instance=True)
     mock_settings.database_url = "sqlite:///:memory:"
 
     decision_svc = DecisionService(settings=mock_settings, unit_of_work=uow)
@@ -143,12 +141,12 @@ def step_start_strategize(context: Context) -> None:
 
 @then("decdi- a strategy_choice decision should be recorded for the plan")
 def step_check_strategize_decision(context: Context) -> None:
-    decisions = context.decision_svc.get_decisions_for_plan(context.plan_id)
+    decisions = context.decision_svc.list_decisions(context.plan_id)
     strategy_decisions = [
         d for d in decisions if str(d.decision_type) == "strategy_choice"
     ]
-    assert len(strategy_decisions) >= 1, (
-        f"Expected at least 1 strategy_choice decision, got {len(strategy_decisions)}"
+    assert len(strategy_decisions) == 1, (
+        f"Expected exactly 1 strategy_choice decision, got {len(strategy_decisions)}"
     )
 
 
@@ -192,12 +190,12 @@ def step_start_execute(context: Context) -> None:
 
 @then("decdi- an implementation_choice decision should be recorded for the plan")
 def step_check_execute_decision(context: Context) -> None:
-    decisions = context.decision_svc.get_decisions_for_plan(context.plan_id)
+    decisions = context.decision_svc.list_decisions(context.plan_id)
     impl_decisions = [
         d for d in decisions if str(d.decision_type) == "implementation_choice"
     ]
-    assert len(impl_decisions) >= 1, (
-        f"Expected at least 1 implementation_choice decision, got {len(impl_decisions)}"
+    assert len(impl_decisions) == 1, (
+        f"Expected exactly 1 implementation_choice decision, got {len(impl_decisions)}"
     )
 
 

features/steps/decision_service_coverage_steps.py

@@ -18,7 +18,10 @@ from behave import given, then, when
 from behave.runner import Context
 from ulid import ULID
 
-from cleveragents.application.services.decision_service import DecisionService
+from cleveragents.application.services.decision_service import (
+    DecisionNotFoundError,
+    DecisionService,
+)
 from cleveragents.domain.models.core.decision import (
     Decision,
     DecisionType,
@@ -126,7 +129,6 @@ def step_dsvc_check_uow(context: Context) -> None:
 
 @then("dsvc- the service should have a bound structlog logger")
 def step_dsvc_check_logger(context: Context) -> None:
-    # The _logger is a BoundLogger with service="decision"
     assert hasattr(context.dsvc_service, "_logger"), "Service missing _logger attribute"
 
 
@@ -174,37 +176,34 @@ def step_dsvc_service_with_logger(context: Context) -> None:
 # ---------------------------------------------------------------------------
 
 
-@given("dsvc- a sample root Decision object")
-def step_dsvc_sample_decision(context: Context) -> None:
-    context.dsvc_decision = _make_decision(plan_id=_PLAN_ID)
-
-
-@when("dsvc- I call record_decision with the sample decision")
+@when("dsvc- I call record_decision with plan_id and required args")
 def step_dsvc_call_record(context: Context) -> None:
     context.dsvc_result = context.dsvc_service.record_decision(
-        context.dsvc_decision,
+        plan_id=_PLAN_ID,
+        decision_type=DecisionType.PROMPT_DEFINITION,
+        question="Which approach?",
+        chosen_option="REST API",
     )
 
 
 @then("dsvc- the UoW transaction should have been entered")
 def step_dsvc_txn_entered(context: Context) -> None:
-    # The fact that ctx.decisions.create was called proves we entered
-    # the transaction context manager successfully.  We verify that
-    # the mock context's decisions repo was accessed.
     assert context.dsvc_decisions.create.called, (
         "Transaction context was not entered — create was never called"
     )
 
 
-@then("dsvc- ctx.decisions.create should have been called with the decision")
+@then("dsvc- ctx.decisions.create should have been called once")
 def step_dsvc_create_called(context: Context) -> None:
-    context.dsvc_decisions.create.assert_called_once_with(context.dsvc_decision)
+    assert context.dsvc_decisions.create.call_count == 1, (
+        f"Expected create to be called once, got {context.dsvc_decisions.create.call_count}"
+    )
 
 
-@then("dsvc- the returned decision should be the same object")
-def step_dsvc_record_returns_same(context: Context) -> None:
-    assert context.dsvc_result is context.dsvc_decision, (
-        "record_decision should return the same Decision instance"
+@then("dsvc- the returned decision should have the correct plan_id")
+def step_dsvc_record_returns_correct_plan(context: Context) -> None:
+    assert context.dsvc_result.plan_id == _PLAN_ID, (
+        f"Expected plan_id '{_PLAN_ID}', got '{context.dsvc_result.plan_id}'"
     )
 
 
@@ -220,13 +219,6 @@ def step_dsvc_logger_info(context: Context, event: str) -> None:
     assert event in events, f"Expected info event '{event}' in {events}"
 
 
-@then('dsvc- the logger should have recorded a debug call with "{event}"')
-def step_dsvc_logger_debug(context: Context, event: str) -> None:
-    debug_calls = context.dsvc_logger.debug.call_args_list
-    events = [c.args[0] if c.args else c.kwargs.get("event", "") for c in debug_calls]
-    assert event in events, f"Expected debug event '{event}' in {events}"
-
-
 # ---------------------------------------------------------------------------
 # get_decision
 # ---------------------------------------------------------------------------
@@ -251,9 +243,13 @@ def step_dsvc_call_get(context: Context) -> None:
     context.dsvc_result = context.dsvc_service.get_decision(_DECISION_ID)
 
 
-@when("dsvc- I call get_decision with an unknown ID")
+@when("dsvc- I call get_decision with an unknown ID expecting not-found")
 def step_dsvc_call_get_unknown(context: Context) -> None:
-    context.dsvc_result = context.dsvc_service.get_decision("NONEXISTENT_ID_12345678")
+    try:
+        context.dsvc_service.get_decision("NONEXISTENT_ID_12345678")
+        context.dsvc_error = None
+    except DecisionNotFoundError as exc:
+        context.dsvc_error = exc
 
 
 @then("dsvc- ctx.decisions.get should have been called with the ID")
@@ -268,13 +264,14 @@ def step_dsvc_get_returns_expected(context: Context) -> None:
     )
 
 
-@then("dsvc- the returned value should be None")
-def step_dsvc_result_is_none(context: Context) -> None:
-    assert context.dsvc_result is None, f"Expected None, got {context.dsvc_result!r}"
+@then("dsvc- a DecisionNotFoundError should have been raised")
+def step_dsvc_not_found_raised(context: Context) -> None:
+    assert context.dsvc_error is not None, "Expected DecisionNotFoundError but got None"
+    assert isinstance(context.dsvc_error, DecisionNotFoundError)
 
 
 # ---------------------------------------------------------------------------
-# get_decisions_for_plan
+# list_decisions
 # ---------------------------------------------------------------------------
 
 
@@ -297,9 +294,9 @@ def step_dsvc_mock_get_by_plan(context: Context, count: int) -> None:
     context.dsvc_expected_count = count
 
 
-@when("dsvc- I call get_decisions_for_plan with a plan ID")
-def step_dsvc_call_get_for_plan(context: Context) -> None:
-    context.dsvc_result = context.dsvc_service.get_decisions_for_plan(_PLAN_ID)
+@when("dsvc- I call list_decisions with a plan ID")
+def step_dsvc_call_list_decisions(context: Context) -> None:
+    context.dsvc_result = context.dsvc_service.list_decisions(_PLAN_ID)
 
 
 @then("dsvc- ctx.decisions.get_by_plan should have been called with the plan ID")
@@ -315,34 +312,48 @@ def step_dsvc_list_count(context: Context, count: int) -> None:
 
 
 # ---------------------------------------------------------------------------
-# get_decision_tree
+# get_tree
 # ---------------------------------------------------------------------------
 
-
-@given("dsvc- the mock repo get_tree method returns {count:d} decisions")
-def step_dsvc_mock_get_tree(context: Context, count: int) -> None:
-    root = _make_decision(decision_id=_ROOT_ID, plan_id=_PLAN_ID)
-    decisions = [root] + [
-        _make_decision(
-            plan_id=_PLAN_ID,
-            sequence_number=i,
-            decision_type=DecisionType.STRATEGY_CHOICE,
-            parent_decision_id=_ROOT_ID,
-        )
-        for i in range(1, count)
-    ]
-    context.dsvc_decisions.get_tree.return_value = decisions
-    context.dsvc_expected_tree_count = count
+_CHILD1_ID = str(ULID())
+_CHILD2_ID = str(ULID())
+_GRANDCHILD_ID = str(ULID())
 
 
-@when("dsvc- I call get_decision_tree with a root ID")
+@given("dsvc- the mock repo get_by_plan method returns 4 decisions as a tree")
+def step_dsvc_mock_get_tree(context: Context) -> None:
+    root = _make_decision(
+        decision_id=_ROOT_ID,
+        plan_id=_PLAN_ID,
+        sequence_number=0,
+    )
+    child1 = _make_decision(
+        decision_id=_CHILD1_ID,
+        plan_id=_PLAN_ID,
+        sequence_number=1,
+        decision_type=DecisionType.STRATEGY_CHOICE,
+        parent_decision_id=_ROOT_ID,
+    )
+    child2 = _make_decision(
+        decision_id=_CHILD2_ID,
+        plan_id=_PLAN_ID,
+        sequence_number=2,
+        decision_type=DecisionType.STRATEGY_CHOICE,
+        parent_decision_id=_ROOT_ID,
+    )
+    grandchild = _make_decision(
+        decision_id=_GRANDCHILD_ID,
+        plan_id=_PLAN_ID,
+        sequence_number=3,
+        decision_type=DecisionType.IMPLEMENTATION_CHOICE,
+        parent_decision_id=_CHILD1_ID,
+    )
+    context.dsvc_decisions.get_by_plan.return_value = [root, child1, child2, grandchild]
+
+
+@when("dsvc- I call get_tree with a plan ID")
 def step_dsvc_call_get_tree(context: Context) -> None:
-    context.dsvc_result = context.dsvc_service.get_decision_tree(_ROOT_ID)
-
-
-@then("dsvc- ctx.decisions.get_tree should have been called with the root ID")
-def step_dsvc_get_tree_called(context: Context) -> None:
-    context.dsvc_decisions.get_tree.assert_called_once_with(_ROOT_ID)
+    context.dsvc_result = context.dsvc_service.get_tree(_PLAN_ID)
 
 
 @then("dsvc- the returned tree list should have {count:d} decisions")
@@ -398,6 +409,23 @@ def step_dsvc_path_count(context: Context, count: int) -> None:
 # ---------------------------------------------------------------------------
 
 
+@given("dsvc- the mock repo has the replacement decision in cache")
+def step_dsvc_seed_replacement_in_cache(context: Context) -> None:
+    """Seed the replacement decision into the service's in-memory cache.
+
+    ``mark_superseded`` now validates that ``new_decision_id`` exists
+    before proceeding.  We add the replacement to ``_decisions`` so
+    the validation lookup succeeds.
+    """
+    replacement = _make_decision(
+        decision_id=_NEW_DECISION_ID,
+        plan_id=_PLAN_ID,
+        sequence_number=99,
+        decision_type=DecisionType.STRATEGY_CHOICE,
+    )
+    context.dsvc_service._decisions[_NEW_DECISION_ID] = replacement
+
+
 @given("dsvc- the mock repo update_superseded_by method returns a superseded Decision")
 def step_dsvc_mock_superseded(context: Context) -> None:
     context.dsvc_superseded_decision = _make_decision(

features/steps/plan_explain_cli_coverage_steps.py

@@ -217,7 +217,7 @@ def step_pec_mock_tree_decisions(context: Context) -> None:
     context.pec_plan_id = str(ULID())
     decisions = _make_tree_decisions()
     svc = MagicMock()
-    svc.get_decisions_for_plan.return_value = decisions
+    svc.list_decisions.return_value = decisions
     context.pec_container = _mock_container_with_decision_svc(svc)
 
 
@@ -226,7 +226,7 @@ def step_pec_mock_deep_decisions(context: Context) -> None:
     context.pec_plan_id = str(ULID())
     decisions = _make_deep_decisions()
     svc = MagicMock()
-    svc.get_decisions_for_plan.return_value = decisions
+    svc.list_decisions.return_value = decisions
     context.pec_container = _mock_container_with_decision_svc(svc)
 
 
@@ -260,7 +260,7 @@ def step_pec_mock_superseded_decisions(context: Context) -> None:
         ),
     ]
     svc = MagicMock()
-    svc.get_decisions_for_plan.return_value = decisions
+    svc.list_decisions.return_value = decisions
     context.pec_container = _mock_container_with_decision_svc(svc)
 
 
@@ -268,7 +268,7 @@ def step_pec_mock_superseded_decisions(context: Context) -> None:
 def step_pec_mock_empty_decisions(context: Context) -> None:
     context.pec_plan_id = str(ULID())
     svc = MagicMock()
-    svc.get_decisions_for_plan.return_value = []
+    svc.list_decisions.return_value = []
     context.pec_container = _mock_container_with_decision_svc(svc)
 
 

robot/decision_recording.robot

@@ -0,0 +1,58 @@
+*** Settings ***
+Documentation     Smoke tests for DecisionService recording and snapshot store
+Resource          ${CURDIR}/common.resource
+Suite Setup       Setup Test Environment
+Suite Teardown    Cleanup Test Environment
+
+*** Variables ***
+${HELPER_SCRIPT}    robot/helper_decision_recording.py
+
+*** Test Cases ***
+Record And Retrieve Decision
+    [Documentation]    Record a decision via DecisionService and retrieve by ID
+    [Tags]    service    decision    recording
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    record-retrieve    cwd=${WORKSPACE}    timeout=30s
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    record-retrieve-ok
+
+Record Multiple With Sequencing
+    [Documentation]    Record 3 decisions and verify monotonic sequence numbers
+    [Tags]    service    decision    sequencing
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    sequencing    cwd=${WORKSPACE}    timeout=30s
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    sequencing-ok
+
+Snapshot Auto-Capture
+    [Documentation]    Record a decision and verify context snapshot is auto-captured
+    [Tags]    service    decision    snapshot
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    snapshot-capture    cwd=${WORKSPACE}    timeout=30s
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    snapshot-capture-ok
+
+Decision Tree BFS Via Service
+    [Documentation]    Build a 3-node tree and verify BFS order via get_tree
+    [Tags]    service    decision    tree
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    tree-bfs    cwd=${WORKSPACE}    timeout=30s
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    tree-bfs-ok
+
+Mark Decision Superseded Via Service
+    [Documentation]    Mark a decision as superseded via the service layer
+    [Tags]    service    decision    superseded
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    superseded    cwd=${WORKSPACE}    timeout=30s
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    superseded-ok
+
+Delete Decision Via Service
+    [Documentation]    Delete a decision and verify snapshot is also removed
+    [Tags]    service    decision    delete
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    delete    cwd=${WORKSPACE}    timeout=30s
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    delete-ok
+
+Snapshot Hash Deduplication
+    [Documentation]    Verify hash-based deduplication in SnapshotStore
+    [Tags]    service    decision    snapshot    dedup
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    snapshot-dedup    cwd=${WORKSPACE}    timeout=30s
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    snapshot-dedup-ok

robot/helper_decision_di.py

@@ -13,19 +13,15 @@ from __future__ import annotations
 import os
 import sys
 from pathlib import Path
-from unittest.mock import MagicMock
+from unittest.mock import create_autospec
 
 # Ensure src is importable when run from workspace root
 sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
 
-from sqlalchemy import create_engine  # noqa: I001
-from sqlalchemy.orm import sessionmaker
-
-from cleveragents.application.services.decision_service import DecisionService
+from cleveragents.application.services.decision_service import DecisionService  # noqa: I001
 from cleveragents.application.services.plan_lifecycle_service import (
     PlanLifecycleService,
 )
-from cleveragents.infrastructure.database.models import Base
 from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
 
 
@@ -36,20 +32,8 @@ from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
 
 def _make_uow() -> UnitOfWork:
     """Create a UoW backed by an in-memory SQLite database."""
-    engine = create_engine("sqlite:///:memory:", echo=False)
-    Base.metadata.create_all(engine)
-
-    uow = UnitOfWork.__new__(UnitOfWork)
-    uow.database_url = "sqlite:///:memory:"
-    uow._engine = engine
-    uow._session_factory = sessionmaker(
-        bind=engine,
-        expire_on_commit=False,
-        autoflush=False,
-        autocommit=False,
-    )
-    uow._database_initialized = True
-    uow._prompt_for_migration = None
+    uow = UnitOfWork("sqlite:///:memory:")
+    uow.init_database()
     return uow
 
 
@@ -84,7 +68,9 @@ def _resolve_service() -> None:
 def _record_integration() -> None:
     """Verify decision recording during plan lifecycle transitions."""
     uow = _make_uow()
-    mock_settings = MagicMock()
+    from cleveragents.config.settings import Settings
+
+    mock_settings = create_autospec(Settings, instance=True)
     mock_settings.database_url = "sqlite:///:memory:"
 
     decision_svc = DecisionService(settings=mock_settings, unit_of_work=uow)
@@ -108,7 +94,7 @@ def _record_integration() -> None:
     # Start strategize -> should record a strategy_choice decision
     lifecycle_svc.start_strategize(plan_id)
 
-    decisions = decision_svc.get_decisions_for_plan(plan_id)
+    decisions = decision_svc.list_decisions(plan_id)
     strategy_decisions = [
         d for d in decisions if str(d.decision_type) == "strategy_choice"
     ]
@@ -128,7 +114,7 @@ def _record_integration() -> None:
     # Start execute -> should record an implementation_choice decision
     lifecycle_svc.start_execute(plan_id)
 
-    decisions = decision_svc.get_decisions_for_plan(plan_id)
+    decisions = decision_svc.list_decisions(plan_id)
     impl_decisions = [
         d for d in decisions if str(d.decision_type) == "implementation_choice"
     ]

robot/helper_decision_recording.py

@@ -0,0 +1,245 @@
+"""Helper script for Robot Framework decision recording smoke tests.
+
+Usage:
+    python robot/helper_decision_recording.py <subcommand>
+
+Subcommands:
+    record-retrieve    Record + retrieve a decision via DecisionService
+    sequencing         Verify monotonic sequencing
+    snapshot-capture   Verify auto-captured context snapshot
+    tree-bfs           Build a tree and verify BFS via get_tree
+    superseded         Mark a decision as superseded via service
+    delete             Delete a decision and verify snapshot removal
+    snapshot-dedup     Verify hash-based deduplication in SnapshotStore
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+# Ensure src is importable when run from workspace root
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from cleveragents.application.services.decision_service import (
+    DecisionService,
+    SnapshotStore,
+)
+from cleveragents.domain.models.core.decision import (
+    ContextSnapshot,
+    DecisionType,
+)
+
+_PLAN_ID = "01HV000000000000000000RS01"
+
+
+# ---------------------------------------------------------------------------
+# Subcommands
+# ---------------------------------------------------------------------------
+
+
+def _record_retrieve():
+    svc = DecisionService()
+    d = svc.record_decision(
+        plan_id=_PLAN_ID,
+        decision_type=DecisionType.PROMPT_DEFINITION,
+        question="What approach?",
+        chosen_option="Build API",
+    )
+    got = svc.get_decision(d.decision_id)
+    assert got.decision_id == d.decision_id
+    assert got.is_root
+    assert str(got.decision_type) == "prompt_definition"
+    print("record-retrieve-ok")
+
+
+def _sequencing():
+    svc = DecisionService()
+    d0 = svc.record_decision(
+        plan_id=_PLAN_ID,
+        decision_type=DecisionType.PROMPT_DEFINITION,
+        question="First",
+        chosen_option="Chosen first",
+    )
+    d1 = svc.record_decision(
+        plan_id=_PLAN_ID,
+        decision_type=DecisionType.STRATEGY_CHOICE,
+        question="Second",
+        chosen_option="Chosen second",
+        parent_decision_id=d0.decision_id,
+    )
+    d2 = svc.record_decision(
+        plan_id=_PLAN_ID,
+        decision_type=DecisionType.IMPLEMENTATION_CHOICE,
+        question="Third",
+        chosen_option="Chosen third",
+        parent_decision_id=d0.decision_id,
+    )
+    assert d0.sequence_number == 0
+    assert d1.sequence_number == 1
+    assert d2.sequence_number == 2
+    assert svc.count_decisions(_PLAN_ID) == 3
+    print("sequencing-ok")
+
+
+def _snapshot_capture():
+    svc = DecisionService()
+    d = svc.record_decision(
+        plan_id=_PLAN_ID,
+        decision_type=DecisionType.STRATEGY_CHOICE,
+        question="Snapshot test",
+        chosen_option="Chosen",
+    )
+    snap = svc.get_snapshot(d.decision_id)
+    assert snap is not None
+    assert snap.hot_context_hash.startswith("sha256:")
+    print("snapshot-capture-ok")
+
+
+def _tree_bfs():
+    svc = DecisionService()
+    root = svc.record_decision(
+        plan_id=_PLAN_ID,
+        decision_type=DecisionType.PROMPT_DEFINITION,
+        question="Root",
+        chosen_option="Root choice",
+    )
+    child1 = svc.record_decision(
+        plan_id=_PLAN_ID,
+        decision_type=DecisionType.STRATEGY_CHOICE,
+        question="Child 1",
+        chosen_option="C1",
+        parent_decision_id=root.decision_id,
+    )
+    svc.record_decision(
+        plan_id=_PLAN_ID,
+        decision_type=DecisionType.STRATEGY_CHOICE,
+        question="Child 2",
+        chosen_option="C2",
+        parent_decision_id=root.decision_id,
+    )
+    svc.record_decision(
+        plan_id=_PLAN_ID,
+        decision_type=DecisionType.IMPLEMENTATION_CHOICE,
+        question="Grandchild",
+        chosen_option="GC",
+        parent_decision_id=child1.decision_id,
+    )
+    tree = svc.get_tree(_PLAN_ID)
+    assert len(tree) == 4, f"expected 4, got {len(tree)}"
+    assert tree[0].is_root
+
+    # Verify BFS level order: compute depths and check non-decreasing.
+    by_id = {d.decision_id: d for d in tree}
+
+    def _depth(node):
+        depth = 0
+        current = node
+        while current.parent_decision_id is not None:
+            parent = by_id.get(current.parent_decision_id)
+            if parent is None:
+                break
+            depth += 1
+            current = parent
+        return depth
+
+    depths = [_depth(d) for d in tree]
+    for i in range(1, len(depths)):
+        assert depths[i] >= depths[i - 1], (
+            f"BFS order violated at index {i}: "
+            f"depth {depths[i]} follows depth {depths[i - 1]}. "
+            f"Depths: {depths}"
+        )
+    print("tree-bfs-ok")
+
+
+def _superseded():
+    svc = DecisionService()
+    d1 = svc.record_decision(
+        plan_id=_PLAN_ID,
+        decision_type=DecisionType.PROMPT_DEFINITION,
+        question="Original",
+        chosen_option="Orig",
+    )
+    d2 = svc.record_decision(
+        plan_id=_PLAN_ID,
+        decision_type=DecisionType.STRATEGY_CHOICE,
+        question="Replacement",
+        chosen_option="New",
+        parent_decision_id=d1.decision_id,
+    )
+    updated = svc.mark_superseded(d1.decision_id, d2.decision_id)
+    assert updated.is_superseded
+    assert updated.superseded_by == d2.decision_id
+    superseded = svc.get_superseded(_PLAN_ID)
+    assert len(superseded) == 1
+    print("superseded-ok")
+
+
+def _delete():
+    svc = DecisionService()
+    d = svc.record_decision(
+        plan_id=_PLAN_ID,
+        decision_type=DecisionType.PROMPT_DEFINITION,
+        question="Ephemeral",
+        chosen_option="Gone",
+    )
+    did = d.decision_id
+    assert svc.get_snapshot(did) is not None
+    result = svc.delete_decision(did)
+    assert result is True
+    assert svc.get_snapshot(did) is None
+    assert svc.count_decisions(_PLAN_ID) == 0
+    print("delete-ok")
+
+
+def _snapshot_dedup():
+    store = SnapshotStore()
+    snap = ContextSnapshot(
+        hot_context_hash="sha256:samehash",
+        hot_context_ref="ref1",
+    )
+    store.store("DEC_A", snap)
+    store.store("DEC_B", snap)
+
+    ids = store.get_by_hash("sha256:samehash")
+    assert len(ids) == 2, f"expected 2, got {len(ids)}"
+    assert "DEC_A" in ids
+    assert "DEC_B" in ids
+
+    # Remove one
+    store.remove("DEC_A")
+    ids2 = store.get_by_hash("sha256:samehash")
+    assert len(ids2) == 1
+    assert "DEC_B" in ids2
+
+    print("snapshot-dedup-ok")
+
+
+# ---------------------------------------------------------------------------
+# Dispatcher
+# ---------------------------------------------------------------------------
+
+_COMMANDS = {
+    "record-retrieve": _record_retrieve,
+    "sequencing": _sequencing,
+    "snapshot-capture": _snapshot_capture,
+    "tree-bfs": _tree_bfs,
+    "superseded": _superseded,
+    "delete": _delete,
+    "snapshot-dedup": _snapshot_dedup,
+}
+
+
+def main():
+    if len(sys.argv) < 2:
+        raise SystemExit(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>")
+    command = sys.argv[1]
+    handler = _COMMANDS.get(command)
+    if handler is None:
+        raise SystemExit(f"Unknown command: {command}")
+    handler()
+
+
+if __name__ == "__main__":
+    main()

src/cleveragents/application/services/__init__.py

@@ -16,7 +16,11 @@ from cleveragents.application.services.correction_service import (
     CorrectionService,
 )
 from cleveragents.application.services.decision_service import (
+    DecisionNotFoundError,
     DecisionService,
+    DuplicateDecisionError,
+    SequenceConflictError,
+    SnapshotStore,
 )
 from cleveragents.application.services.invariant_service import (
     InvariantService,
@@ -117,9 +121,11 @@ __all__ = [
     "ConfigService",
     "ContextFragment",
     "CorrectionService",
+    "DecisionNotFoundError",
     "DecisionService",
     "DefaultValidationRunner",
     "DependencyCycleRule",
+    "DuplicateDecisionError",
     "DuplicateImportRule",
     "FileMergeOutcome",
     "InvariantService",
@@ -140,8 +146,10 @@ __all__ = [
     "SemanticValidationRule",
     "SemanticValidationService",
     "SemanticValidationSeverity",
+    "SequenceConflictError",
     "SkeletonCompressorService",
     "SkillRegistryService",
+    "SnapshotStore",
     "SpawnEntry",
     "SpawnMetadata",
     "SpawnResult",

src/cleveragents/application/services/decision_service.py

@@ -1,210 +1,772 @@
-"""Application-layer service for decision tree operations.
+"""Decision recording and snapshot store service.
 
-``DecisionService`` wraps the :class:`DecisionRepository` behind a
-thin application-layer façade, adding structured logging, transaction
-management via :class:`UnitOfWork`, and a consistent API surface for
-callers (e.g. ``PlanLifecycleService``).
+``DecisionService`` provides a clean interface for recording decisions
+during plan execution, retrieving decision histories, and managing
+context snapshots.
+
+Dual-mode persistence
+---------------------
+
+| Mode       | UnitOfWork | Storage                                 |
+|------------|------------|-----------------------------------------|
+| In-memory  | ``None``   | ``self._decisions`` / ``self._snapshots`` dicts |
+| Persisted  | provided   | DB via ``DecisionRepository`` + in-memory cache |
+
+In persisted mode the in-memory dicts act as a write-through cache:
+mutations are written to DB first, then the cache is updated.
 
 Based on:
-    - ADR-007 (Repository Pattern / Unit of Work)
-    - ADR-033 (Decision Recording Protocol)
-    - Forgejo issue #173 (Wire decision services into DI)
+    - Forgejo issue #172
+    - docs/specification.md L18345-L18521
+    - docs/adr/ADR-033-decision-recording-protocol.md
 """
 
 from __future__ import annotations
 
+import hashlib
+import json
+from collections import deque
 from typing import TYPE_CHECKING
 
 import structlog
 
-from cleveragents.domain.models.core.decision import Decision
+from cleveragents.core.exceptions import (
+    BusinessRuleViolation,
+    ResourceNotFoundError,
+    ValidationError,
+)
+from cleveragents.domain.models.core.decision import (
+    ArtifactRef,
+    ContextSnapshot,
+    Decision,
+    DecisionType,
+)
 
 if TYPE_CHECKING:
+    from cleveragents.config.settings import Settings
     from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
 
 logger = structlog.get_logger(__name__)
 
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+#: Maximum length for ``actor_reasoning`` (raw LLM trace).  Prevents
+#: unbounded storage from a misbehaving model while remaining generous
+#: enough for typical chain-of-thought traces.
+_MAX_ACTOR_REASONING: int = 100_000
+
+
+# ---------------------------------------------------------------------------
+# Custom exceptions
+# ---------------------------------------------------------------------------
+
+
+class DuplicateDecisionError(BusinessRuleViolation):
+    """Raised when a decision with the same ID already exists."""
+
+    def __init__(self, decision_id: str) -> None:
+        super().__init__(f"Decision '{decision_id}' already exists")
+        self.decision_id = decision_id
+
+
+class DecisionNotFoundError(ResourceNotFoundError):
+    """Raised when a decision cannot be found."""
+
+    def __init__(self, decision_id: str) -> None:
+        super().__init__(
+            resource_type="decision",
+            resource_id=decision_id,
+        )
+
+
+class SequenceConflictError(BusinessRuleViolation):
+    """Raised when a sequence number is already in use for a plan."""
+
+    def __init__(self, plan_id: str, sequence_number: int) -> None:
+        super().__init__(
+            f"Sequence number {sequence_number} already exists for plan '{plan_id}'"
+        )
+        self.plan_id = plan_id
+        self.sequence_number = sequence_number
+
+
+# ---------------------------------------------------------------------------
+# Snapshot store
+# ---------------------------------------------------------------------------
+
+
+class SnapshotStore:
+    """In-memory store for context snapshots keyed by decision_id.
+
+    Snapshots are stored alongside decisions but this dedicated store
+    provides hash-based deduplication and retrieval helpers.
+    """
+
+    def __init__(self) -> None:
+        self._snapshots: dict[str, ContextSnapshot] = {}
+        self._hash_index: dict[str, list[str]] = {}
+
+    def store(self, decision_id: str, snapshot: ContextSnapshot) -> None:
+        """Store a snapshot for a given decision.
+
+        Args:
+            decision_id: ULID of the decision.
+            snapshot: The context snapshot to store.
+        """
+        self._snapshots[decision_id] = snapshot
+        h = snapshot.hot_context_hash
+        if h:
+            self._hash_index.setdefault(h, []).append(decision_id)
+
+    def get(self, decision_id: str) -> ContextSnapshot | None:
+        """Retrieve a snapshot by decision ID.
+
+        Returns:
+            The snapshot, or ``None`` if not found.
+        """
+        return self._snapshots.get(decision_id)
+
+    def get_by_hash(self, context_hash: str) -> list[str]:
+        """Return decision IDs sharing the same context hash.
+
+        Args:
+            context_hash: The hot_context_hash to look up.
+
+        Returns:
+            List of decision IDs with matching hash.
+        """
+        return list(self._hash_index.get(context_hash, []))
+
+    def remove(self, decision_id: str) -> bool:
+        """Remove a snapshot.
+
+        Returns:
+            ``True`` if a snapshot was removed, ``False`` otherwise.
+        """
+        snapshot = self._snapshots.pop(decision_id, None)
+        if snapshot is None:
+            return False
+        h = snapshot.hot_context_hash
+        if h and h in self._hash_index:
+            ids = self._hash_index[h]
+            if decision_id in ids:
+                ids.remove(decision_id)
+            if not ids:
+                del self._hash_index[h]
+        return True
+
+    def list_for_plan(self, decisions: list[Decision]) -> dict[str, ContextSnapshot]:
+        """Return snapshots for a list of decisions.
+
+        Args:
+            decisions: Decisions to look up snapshots for.
+
+        Returns:
+            Dict mapping decision_id → snapshot for found entries.
+        """
+        result: dict[str, ContextSnapshot] = {}
+        for d in decisions:
+            snap = self._snapshots.get(d.decision_id)
+            if snap is not None:
+                result[d.decision_id] = snap
+        return result
+
+
+# ---------------------------------------------------------------------------
+# Decision service
+# ---------------------------------------------------------------------------
+
 
 class DecisionService:
-    """Coordinate decision persistence through the Unit of Work.
+    """Service for recording decisions and managing context snapshots.
 
-    All public methods open a transaction, delegate to
-    :class:`DecisionRepository`, and commit.  Callers do not need to
-    manage sessions directly.
-
-    Args:
-        settings: Application settings (unused currently but kept for
-            consistency with the service constructor convention).
-        unit_of_work: A :class:`UnitOfWork` instance used to open
-            transactional scopes.
+    Provides record, list, tree, and path-to-root helpers for the
+    decision subsystem.  Supports dual-mode persistence: in-memory
+    when no UnitOfWork is provided, or DB-backed via
+    ``DecisionRepository`` when a UnitOfWork is wired.
     """
 
     def __init__(
         self,
-        settings: object,
-        unit_of_work: UnitOfWork,
+        settings: Settings | None = None,
+        unit_of_work: UnitOfWork | None = None,
     ) -> None:
+        """Initialize the decision service.
+
+        Args:
+            settings: Application settings (optional for in-memory mode).
+            unit_of_work: Unit of Work for database transactions.
+                When provided, decisions are persisted via
+                ``DecisionRepository``.  When ``None``, the service
+                falls back to in-memory storage.
+        """
         self.settings = settings
         self.unit_of_work = unit_of_work
         self._logger = logger.bind(service="decision")
 
+        # In-memory storage (used as cache in persisted mode,
+        # or as primary storage when no UoW)
+        self._decisions: dict[str, Decision] = {}
+        self._plan_decisions: dict[str, list[str]] = {}
+        self._plan_sequence: dict[str, int] = {}
+        self._sequence_initialised: set[str] = set()
+
+        # Snapshot store
+        self.snapshots = SnapshotStore()
+
+    @property
+    def _persisted(self) -> bool:
+        """Return True when a UnitOfWork is wired for persistence."""
+        return self.unit_of_work is not None
+
     # ------------------------------------------------------------------
-    # record_decision
+    # Recording
     # ------------------------------------------------------------------
 
-    def record_decision(self, decision: Decision) -> Decision:
-        """Persist a new decision.
+    def record_decision(
+        self,
+        plan_id: str,
+        decision_type: DecisionType | str,
+        question: str,
+        chosen_option: str,
+        *,
+        parent_decision_id: str | None = None,
+        alternatives_considered: list[str] | None = None,
+        confidence_score: float | None = None,
+        rationale: str = "",
+        actor_reasoning: str | None = None,
+        context_snapshot: ContextSnapshot | None = None,
+        artifacts_produced: list[ArtifactRef] | None = None,
+        is_correction: bool = False,
+        corrects_decision_id: str | None = None,
+        correction_reason: str | None = None,
+    ) -> Decision:
+        """Record a new decision in the plan's decision tree.
 
-        Opens a UnitOfWork transaction, creates the decision via the
-        repository, and commits.
+        Automatically assigns a monotonically increasing sequence number,
+        generates a decision ID (ULID), and captures a context snapshot.
 
         Args:
-            decision: The :class:`Decision` domain object to persist.
+            plan_id: ULID of the plan.
+            decision_type: Type of decision being recorded.
+            question: What question was being answered.
+            chosen_option: The option that was chosen.
+            parent_decision_id: Optional parent in the decision tree.
+            alternatives_considered: Other options evaluated.
+            confidence_score: Confidence in the decision (0.0-1.0).
+            rationale: Human-readable rationale.
+            actor_reasoning: Raw LLM reasoning trace.
+            context_snapshot: Snapshot of context at decision time.
+            artifacts_produced: Artifacts created as side-effects.
+            is_correction: Whether this corrects another decision.
+            corrects_decision_id: ULID of the decision being corrected.
+            correction_reason: Why the correction was made.
 
         Returns:
-            The same :class:`Decision` instance (pass-through).
+            The recorded :class:`Decision`.
 
         Raises:
+            ValidationError: If required fields are missing or invalid.
             DuplicateDecisionError: If a decision with the same ID exists.
-            DatabaseError: On transient or unexpected DB errors.
         """
-        self._logger.info(
-            "recording_decision",
-            decision_id=decision.decision_id,
-            plan_id=decision.plan_id,
-            decision_type=str(decision.decision_type),
-            sequence_number=decision.sequence_number,
+        if not plan_id or not plan_id.strip():
+            raise ValidationError("plan_id must not be empty")
+        if not question or not question.strip():
+            raise ValidationError("question must not be empty")
+        if not chosen_option or not chosen_option.strip():
+            raise ValidationError("chosen_option must not be empty")
+        if actor_reasoning is not None and len(actor_reasoning) > _MAX_ACTOR_REASONING:
+            raise ValidationError(
+                f"actor_reasoning exceeds maximum length of {_MAX_ACTOR_REASONING}"
+            )
+
+        # Coerce string to enum if needed
+        if isinstance(decision_type, str):
+            decision_type = DecisionType(decision_type)
+
+        # Auto-assign sequence number
+        seq = self._next_sequence(plan_id)
+
+        # Build context snapshot
+        snapshot = context_snapshot or ContextSnapshot()
+        if not snapshot.hot_context_hash:
+            # Generate a hash from the question + chosen_option as a
+            # minimal snapshot when no explicit hash is provided
+            snapshot = self._auto_capture_snapshot(question, chosen_option, snapshot)
+
+        decision = Decision(
+            plan_id=plan_id,
+            decision_type=decision_type,
+            sequence_number=seq,
+            question=question,
+            chosen_option=chosen_option,
+            parent_decision_id=parent_decision_id,
+            alternatives_considered=alternatives_considered or [],
+            confidence_score=confidence_score,
+            rationale=rationale,
+            actor_reasoning=actor_reasoning,
+            context_snapshot=snapshot,
+            artifacts_produced=artifacts_produced or [],
+            is_correction=is_correction,
+            corrects_decision_id=corrects_decision_id,
+            correction_reason=correction_reason,
         )
-        with self.unit_of_work.transaction() as ctx:
-            ctx.decisions.create(decision)
-        self._logger.debug(
-            "decision_recorded",
+
+        self._store_decision(decision)
+
+        self._logger.info(
+            "decision.recorded",
             decision_id=decision.decision_id,
+            plan_id=plan_id,
+            decision_type=str(decision_type),
+            sequence=seq,
         )
         return decision
 
     # ------------------------------------------------------------------
-    # get_decision
+    # Retrieval
     # ------------------------------------------------------------------
 
-    def get_decision(self, decision_id: str) -> Decision | None:
-        """Retrieve a single decision by its ULID.
+    def get_decision(self, decision_id: str) -> Decision:
+        """Retrieve a single decision by ID.
 
         Args:
             decision_id: ULID of the decision.
 
         Returns:
-            The :class:`Decision` or ``None`` if not found.
+            The :class:`Decision`.
+
+        Raises:
+            DecisionNotFoundError: If the decision does not exist.
         """
-        self._logger.debug("getting_decision", decision_id=decision_id)
-        with self.unit_of_work.transaction() as ctx:
-            return ctx.decisions.get(decision_id)
+        if self._persisted and self.unit_of_work is not None:
+            with self.unit_of_work.transaction() as ctx:
+                result = ctx.decisions.get(decision_id)
+                if result is None:
+                    raise DecisionNotFoundError(decision_id)
+                return result
+        decision = self._decisions.get(decision_id)
+        if decision is None:
+            raise DecisionNotFoundError(decision_id)
+        return decision
 
-    # ------------------------------------------------------------------
-    # get_decisions_for_plan
-    # ------------------------------------------------------------------
-
-    def get_decisions_for_plan(self, plan_id: str) -> list[Decision]:
-        """Return all decisions for a plan ordered by sequence number.
+    def list_decisions(self, plan_id: str) -> list[Decision]:
+        """List all decisions for a plan, ordered by sequence number.
 
         Args:
             plan_id: ULID of the plan.
 
         Returns:
-            List of :class:`Decision` instances.
+            List of decisions sorted by sequence_number.
         """
-        self._logger.debug("getting_decisions_for_plan", plan_id=plan_id)
-        with self.unit_of_work.transaction() as ctx:
-            return ctx.decisions.get_by_plan(plan_id)
+        if self._persisted and self.unit_of_work is not None:
+            with self.unit_of_work.transaction() as ctx:
+                return ctx.decisions.get_by_plan(plan_id)
 
-    # ------------------------------------------------------------------
-    # get_decision_tree
-    # ------------------------------------------------------------------
+        ids = self._plan_decisions.get(plan_id, [])
+        decisions = [self._decisions[did] for did in ids if did in self._decisions]
+        return sorted(decisions, key=lambda d: d.sequence_number)
 
-    def get_decision_tree(self, root_id: str) -> list[Decision]:
-        """Retrieve the full decision tree rooted at *root_id* (BFS order).
+    def list_by_type(
+        self, plan_id: str, decision_type: DecisionType | str
+    ) -> list[Decision]:
+        """List decisions for a plan filtered by type.
 
         Args:
-            root_id: ULID of the root decision.
+            plan_id: ULID of the plan.
+            decision_type: The type to filter by.
 
         Returns:
-            List of :class:`Decision` instances (BFS order).
-
-        Raises:
-            DecisionNotFoundError: If the root decision does not exist.
+            Filtered list sorted by sequence_number.
         """
-        self._logger.debug("getting_decision_tree", root_id=root_id)
-        with self.unit_of_work.transaction() as ctx:
-            return ctx.decisions.get_tree(root_id)
+        if isinstance(decision_type, str):
+            decision_type = DecisionType(decision_type)
+
+        if self._persisted and self.unit_of_work is not None:
+            with self.unit_of_work.transaction() as ctx:
+                return ctx.decisions.list_by_type(plan_id, str(decision_type))
+
+        all_decisions = self.list_decisions(plan_id)
+        return [d for d in all_decisions if d.decision_type == decision_type]
 
     # ------------------------------------------------------------------
-    # get_path_to_root
+    # Tree operations
     # ------------------------------------------------------------------
 
+    def get_tree(self, plan_id: str) -> list[Decision]:
+        """Get the full decision tree for a plan via BFS from root.
+
+        Returns decisions in BFS order (root first, then children
+        level by level).
+
+        Args:
+            plan_id: ULID of the plan.
+
+        Returns:
+            List of decisions in BFS order.
+        """
+        all_decisions = self.list_decisions(plan_id)
+        if not all_decisions:
+            return []
+
+        # Find root(s) — decisions with no parent
+        roots = [d for d in all_decisions if d.is_root]
+        if not roots:
+            return all_decisions
+
+        # Build adjacency: parent_id → children
+        children_map: dict[str | None, list[Decision]] = {}
+        for d in all_decisions:
+            children_map.setdefault(d.parent_decision_id, []).append(d)
+
+        # BFS from roots
+        result: list[Decision] = []
+        queue: deque[Decision] = deque(roots)
+        visited: set[str] = set()
+
+        while queue:
+            node = queue.popleft()
+            if node.decision_id in visited:
+                continue
+            visited.add(node.decision_id)
+            result.append(node)
+            children = children_map.get(node.decision_id, [])
+            for child in sorted(children, key=lambda c: c.sequence_number):
+                if child.decision_id not in visited:
+                    queue.append(child)
+
+        # Include orphaned subtrees (decisions unreachable from any root).
+        # These can arise when a parent decision is deleted.  They are
+        # appended in sequence-number order after the BFS-reachable nodes.
+        if len(visited) < len(all_decisions):
+            orphans = [d for d in all_decisions if d.decision_id not in visited]
+            result.extend(orphans)
+
+        return result
+
     def get_path_to_root(self, decision_id: str) -> list[Decision]:
-        """Walk from *decision_id* up to the root (leaf → root order).
+        """Navigate from a decision up to the tree root.
+
+        Returns the path from the given decision to the root,
+        inclusive of both endpoints.
 
         Args:
             decision_id: ULID of the starting decision.
 
         Returns:
-            List of :class:`Decision` instances (leaf first).
+            List from the given decision up to the root.
 
         Raises:
-            DecisionNotFoundError: If the starting decision is missing.
+            DecisionNotFoundError: If the starting decision does not exist.
         """
-        self._logger.debug("getting_path_to_root", decision_id=decision_id)
-        with self.unit_of_work.transaction() as ctx:
-            return ctx.decisions.get_path_to_root(decision_id)
+        if self._persisted and self.unit_of_work is not None:
+            with self.unit_of_work.transaction() as ctx:
+                return ctx.decisions.get_path_to_root(decision_id)
 
-    # ------------------------------------------------------------------
-    # mark_superseded
-    # ------------------------------------------------------------------
+        current = self._decisions.get(decision_id)
+        if current is None:
+            raise DecisionNotFoundError(decision_id)
 
-    def mark_superseded(
-        self,
-        decision_id: str,
-        new_decision_id: str,
-    ) -> Decision:
-        """Mark a decision as superseded by a new one.
+        path: list[Decision] = [current]
+        visited: set[str] = {current.decision_id}
 
-        Args:
-            decision_id: ULID of the decision to mark.
-            new_decision_id: ULID of the replacement decision.
+        while current.parent_decision_id is not None:
+            parent = self._decisions.get(current.parent_decision_id)
+            if parent is None or parent.decision_id in visited:
+                break
+            visited.add(parent.decision_id)
+            path.append(parent)
+            current = parent
 
-        Returns:
-            Updated :class:`Decision`.
+        return path
 
-        Raises:
-            DecisionNotFoundError: If *decision_id* does not exist.
-        """
-        self._logger.info(
-            "marking_superseded",
-            decision_id=decision_id,
-            new_decision_id=new_decision_id,
-        )
-        with self.unit_of_work.transaction() as ctx:
-            return ctx.decisions.update_superseded_by(decision_id, new_decision_id)
-
-    # ------------------------------------------------------------------
-    # list_by_type
-    # ------------------------------------------------------------------
-
-    def list_by_type(self, plan_id: str, decision_type: str) -> list[Decision]:
-        """List decisions of a given type for a plan.
+    def get_superseded(self, plan_id: str) -> list[Decision]:
+        """Get all superseded decisions for a plan.
 
         Args:
             plan_id: ULID of the plan.
-            decision_type: String value of a :class:`DecisionType` enum
-                member (e.g. ``"strategy_choice"``).
 
         Returns:
-            List of :class:`Decision` instances.
+            List of decisions that have been superseded.
         """
-        self._logger.debug(
-            "listing_by_type",
-            plan_id=plan_id,
-            decision_type=decision_type,
+        if self._persisted and self.unit_of_work is not None:
+            with self.unit_of_work.transaction() as ctx:
+                return ctx.decisions.get_superseded(plan_id)
+
+        all_decisions = self.list_decisions(plan_id)
+        return [d for d in all_decisions if d.is_superseded]
+
+    # ------------------------------------------------------------------
+    # Mutations
+    # ------------------------------------------------------------------
+
+    def mark_superseded(self, decision_id: str, new_decision_id: str) -> Decision:
+        """Mark a decision as superseded by another.
+
+        Args:
+            decision_id: ULID of the decision to supersede.
+            new_decision_id: ULID of the replacement decision.
+
+        Returns:
+            The updated :class:`Decision` with superseded_by set.
+
+        Raises:
+            DecisionNotFoundError: If the decision or the replacement
+                decision does not exist.
+        """
+        # Validate the replacement decision exists.
+        replacement = self._decisions.get(new_decision_id)
+        if replacement is None and self._persisted and self.unit_of_work is not None:
+            with self.unit_of_work.transaction() as ctx:
+                replacement = ctx.decisions.get(new_decision_id)
+        if replacement is None:
+            raise DecisionNotFoundError(new_decision_id)
+
+        if self._persisted and self.unit_of_work is not None:
+            with self.unit_of_work.transaction() as ctx:
+                result = ctx.decisions.update_superseded_by(
+                    decision_id, new_decision_id
+                )
+                # Update cache
+                if decision_id in self._decisions:
+                    self._decisions[decision_id] = result
+                self._logger.info(
+                    "decision.superseded",
+                    decision_id=decision_id,
+                    superseded_by=new_decision_id,
+                )
+                return result
+
+        original = self._decisions.get(decision_id)
+        if original is None:
+            raise DecisionNotFoundError(decision_id)
+
+        updated = original.with_superseded_by(new_decision_id)
+        self._decisions[decision_id] = updated
+
+        self._logger.info(
+            "decision.superseded",
+            decision_id=decision_id,
+            superseded_by=new_decision_id,
         )
-        with self.unit_of_work.transaction() as ctx:
-            return ctx.decisions.list_by_type(plan_id, decision_type)
+        return updated
+
+    def delete_decision(self, decision_id: str) -> bool:
+        """Delete a decision.
+
+        Args:
+            decision_id: ULID of the decision to delete.
+
+        Returns:
+            ``True`` if the decision was deleted.
+
+        Raises:
+            DecisionNotFoundError: If the decision does not exist.
+        """
+        # Check cache first; in persisted mode also consult DB.
+        decision = self._decisions.get(decision_id)
+        if decision is None and self._persisted and self.unit_of_work is not None:
+            with self.unit_of_work.transaction() as ctx:
+                decision = ctx.decisions.get(decision_id)
+
+        if decision is None:
+            raise DecisionNotFoundError(decision_id)
+
+        # Delete from DB if persisted
+        if self._persisted and self.unit_of_work is not None:
+            with self.unit_of_work.transaction() as ctx:
+                ctx.decisions.delete(decision_id)
+
+        # Remove from cache
+        self._decisions.pop(decision_id, None)
+        plan_ids = self._plan_decisions.get(decision.plan_id, [])
+        if decision_id in plan_ids:
+            plan_ids.remove(decision_id)
+
+        self.snapshots.remove(decision_id)
+        self._logger.info("decision.deleted", decision_id=decision_id)
+        return True
+
+    # ------------------------------------------------------------------
+    # Snapshot helpers
+    # ------------------------------------------------------------------
+
+    def get_snapshot(self, decision_id: str) -> ContextSnapshot | None:
+        """Retrieve the context snapshot for a decision.
+
+        Args:
+            decision_id: ULID of the decision.
+
+        Returns:
+            The snapshot, or ``None`` if not stored separately.
+        """
+        return self.snapshots.get(decision_id)
+
+    def get_snapshots_for_plan(self, plan_id: str) -> dict[str, ContextSnapshot]:
+        """Retrieve all snapshots for a plan's decisions.
+
+        Args:
+            plan_id: ULID of the plan.
+
+        Returns:
+            Dict mapping decision_id → ContextSnapshot.
+        """
+        decisions = self.list_decisions(plan_id)
+        return self.snapshots.list_for_plan(decisions)
+
+    # ------------------------------------------------------------------
+    # Statistics
+    # ------------------------------------------------------------------
+
+    def count_decisions(self, plan_id: str) -> int:
+        """Return the number of decisions recorded for a plan.
+
+        In persisted mode uses ``SELECT COUNT(*)`` for efficiency.
+
+        Args:
+            plan_id: ULID of the plan.
+
+        Returns:
+            Decision count.
+        """
+        if self._persisted and self.unit_of_work is not None:
+            with self.unit_of_work.transaction() as ctx:
+                return ctx.decisions.count(plan_id)
+        return len(self.list_decisions(plan_id))
+
+    def get_next_sequence(self, plan_id: str) -> int:
+        """Return the next available sequence number for a plan.
+
+        This is useful for callers that need to know the next
+        sequence number before recording.  Triggers DB rehydration
+        on the first call for a given *plan_id*.
+
+        Args:
+            plan_id: ULID of the plan.
+
+        Returns:
+            The next sequence number.
+        """
+        if plan_id not in self._sequence_initialised:
+            self._rehydrate_sequence(plan_id)
+            self._sequence_initialised.add(plan_id)
+        return self._plan_sequence.get(plan_id, 0)
+
+    # ------------------------------------------------------------------
+    # Internal helpers
+    # ------------------------------------------------------------------
+
+    def _next_sequence(self, plan_id: str) -> int:
+        """Generate the next monotonically increasing sequence number.
+
+        On the first call for a given *plan_id* the counter is seeded from
+        the database (``MAX(sequence_number) + 1``) when a
+        :class:`UnitOfWork` is available.  Subsequent calls for the same
+        plan use the in-memory counter.
+
+        Raises:
+            SequenceConflictError: If the generated sequence number is
+                already present in the in-memory cache (indicates a
+                concurrency or logic bug).
+        """
+        if plan_id not in self._sequence_initialised:
+            self._rehydrate_sequence(plan_id)
+            self._sequence_initialised.add(plan_id)
+
+        seq = self._plan_sequence.get(plan_id, 0)
+
+        # Uniqueness guard: ensure no existing decision already holds
+        # this sequence number for the plan.
+        for did in self._plan_decisions.get(plan_id, []):
+            existing = self._decisions.get(did)
+            if existing is not None and existing.sequence_number == seq:
+                raise SequenceConflictError(plan_id, seq)
+
+        self._plan_sequence[plan_id] = seq + 1
+        return seq
+
+    def _rehydrate_sequence(self, plan_id: str) -> None:
+        """Seed the in-memory sequence counter from the database.
+
+        When a :class:`UnitOfWork` is wired, queries
+        ``MAX(sequence_number)`` for *plan_id* so that new decisions
+        continue the monotonic sequence after a service restart.  When
+        no persistence layer is available the counter starts at ``0``
+        (or stays at its current value if already set).
+        """
+        if plan_id in self._plan_sequence:
+            # Already has a counter — nothing to rehydrate.
+            return
+
+        if self._persisted and self.unit_of_work is not None:
+            with self.unit_of_work.transaction() as ctx:
+                max_seq = ctx.decisions.get_max_sequence_number(plan_id)
+            if max_seq is not None:
+                self._plan_sequence[plan_id] = max_seq + 1
+                return
+
+        # No DB or no existing rows — start from 0.
+        self._plan_sequence.setdefault(plan_id, 0)
+
+    def _store_decision(self, decision: Decision) -> None:
+        """Store a decision in both persistence and cache."""
+        decision_id = decision.decision_id
+        plan_id = decision.plan_id
+
+        if decision_id in self._decisions:
+            raise DuplicateDecisionError(decision_id)
+
+        # Persist to DB if available
+        if self._persisted and self.unit_of_work is not None:
+            with self.unit_of_work.transaction() as ctx:
+                ctx.decisions.create(decision)
+
+        # Update in-memory cache
+        self._decisions[decision_id] = decision
+        self._plan_decisions.setdefault(plan_id, []).append(decision_id)
+
+        # Store snapshot
+        self.snapshots.store(decision_id, decision.context_snapshot)
+
+    @staticmethod
+    def _auto_capture_snapshot(
+        question: str,
+        chosen_option: str,
+        base_snapshot: ContextSnapshot,
+    ) -> ContextSnapshot:
+        """Generate a minimal context snapshot hash.
+
+        When no explicit ``hot_context_hash`` is provided, we compute
+        a SHA-256 hash from the question + chosen option as a minimal
+        content-addressable identifier.
+
+        Args:
+            question: The decision question.
+            chosen_option: The chosen option.
+            base_snapshot: The base snapshot to augment.
+
+        Returns:
+            A new :class:`ContextSnapshot` with the hash populated.
+        """
+        content = json.dumps(
+            {"question": question, "chosen_option": chosen_option},
+            sort_keys=True,
+        )
+        context_hash = hashlib.sha256(content.encode()).hexdigest()[:16]
+        return ContextSnapshot(
+            hot_context_hash=f"sha256:{context_hash}",
+            hot_context_ref=base_snapshot.hot_context_ref,
+            relevant_resources=list(base_snapshot.relevant_resources),
+            actor_state_ref=base_snapshot.actor_state_ref,
+        )
+
+
+__all__ = [
+    "DecisionNotFoundError",
+    "DecisionService",
+    "DuplicateDecisionError",
+    "SequenceConflictError",
+    "SnapshotStore",
+]

src/cleveragents/application/services/plan_lifecycle_service.py

@@ -173,20 +173,11 @@ class PlanLifecycleService:
         self._actions: dict[str, Action] = {}
         self._plans: dict[str, Plan] = {}
 
-        # Decision sequence counter per plan (monotonic within a plan)
-        self._decision_seq: dict[str, int] = {}
-
     @property
     def _persisted(self) -> bool:
         """Return True when a UnitOfWork is wired for persistence."""
         return self.unit_of_work is not None
 
-    def _next_seq(self, plan_id: str) -> int:
-        """Return the next decision sequence number for *plan_id*."""
-        seq = self._decision_seq.get(plan_id, -1) + 1
-        self._decision_seq[plan_id] = seq
-        return seq
-
     def _try_record_decision(
         self,
         plan_id: str,
@@ -203,19 +194,14 @@ class PlanLifecycleService:
         if self.decision_service is None:
             return
 
-        from cleveragents.domain.models.core.decision import Decision as DecisionModel
-        from cleveragents.domain.models.core.decision import DecisionType as DT
-
         try:
-            decision = DecisionModel(
+            self.decision_service.record_decision(
                 plan_id=plan_id,
-                parent_decision_id=parent_decision_id,
-                sequence_number=self._next_seq(plan_id),
-                decision_type=DT(decision_type),
+                decision_type=decision_type,
                 question=question,
                 chosen_option=chosen_option,
+                parent_decision_id=parent_decision_id,
             )
-            self.decision_service.record_decision(decision)
         except Exception:
             self._logger.warning(
                 "decision_recording_failed",

src/cleveragents/cli/commands/plan.py

@@ -2870,7 +2870,7 @@ def tree_decisions_cmd(
 
     container = get_container()
     svc: DecisionService = container.resolve(_DS)
-    decisions = svc.get_decisions_for_plan(plan_id)
+    decisions = svc.list_decisions(plan_id)
     if not decisions:
         console.print(f"No decisions found for plan '{plan_id}'.")
         return

src/cleveragents/infrastructure/database/repositories.py

@@ -65,6 +65,7 @@ if TYPE_CHECKING:
     from cleveragents.domain.models.core.checkpoint import Checkpoint
 
 import structlog
+from sqlalchemy import func as sa_func
 from sqlalchemy.exc import DatabaseError as SQLAlchemyDatabaseError
 from sqlalchemy.exc import IntegrityError, OperationalError
 from sqlalchemy.orm import Session
@@ -5172,6 +5173,68 @@ class DecisionRepository:
                 f"Failed to delete decision {decision_id}: {exc}",
             ) from exc
 
+    # --- SEQUENCE ----------------------------------------------------------
+
+    @database_retry
+    def get_max_sequence_number(self, plan_id: str) -> int | None:
+        """Return the highest sequence number for *plan_id*, or ``None``.
+
+        Uses the composite index ``ix_decisions_plan_seq`` for an
+        efficient ``MAX()`` aggregate.
+
+        Args:
+            plan_id: ULID of the plan.
+
+        Returns:
+            The maximum ``sequence_number`` stored for the plan, or
+            ``None`` when no decisions exist for the plan.
+
+        Raises:
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            result: int | None = (
+                session.query(sa_func.max(DecisionModel.sequence_number))
+                .filter(DecisionModel.plan_id == plan_id)
+                .scalar()
+            )
+            return result
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            raise DatabaseError(
+                f"Failed to get max sequence number for plan {plan_id}: {exc}",
+            ) from exc
+
+    # --- COUNT -------------------------------------------------------------
+
+    @database_retry
+    def count(self, plan_id: str) -> int:
+        """Return the number of decisions for *plan_id*.
+
+        Uses ``COUNT(*)`` for efficiency instead of loading all rows.
+
+        Args:
+            plan_id: ULID of the plan.
+
+        Returns:
+            The decision count (``0`` when no decisions exist).
+
+        Raises:
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            result: int = (
+                session.query(sa_func.count(DecisionModel.decision_id))
+                .filter(DecisionModel.plan_id == plan_id)
+                .scalar()
+            ) or 0
+            return result
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            raise DatabaseError(
+                f"Failed to count decisions for plan {plan_id}: {exc}",
+            ) from exc
+
 
 # ---------------------------------------------------------------------------
 # Checkpoint Repository (Stage M6 - checkpointing and rollback)

vulture_whitelist.py

@@ -359,8 +359,8 @@ decision_service  # noqa: B018, F821
 plan_lifecycle_service  # noqa: B018, F821
 record_decision  # noqa: B018, F821
 get_decision  # noqa: B018, F821
-get_decisions_for_plan  # noqa: B018, F821
-get_decision_tree  # noqa: B018, F821
+list_decisions  # noqa: B018, F821
+get_tree  # noqa: B018, F821
 get_path_to_root  # noqa: B018, F821
 mark_superseded  # noqa: B018, F821
 list_by_type  # noqa: B018, F821