CHANGELOG.md

@@ -11,6 +11,10 @@
 - Fixed `examples/actors/graph_workflow.yaml` to use `actor_ref` instead of deprecated `actor_path`.
 - Added Robot smoke test for loading hierarchical actor YAML via `ActorLoader.discover()`.
 
+- Added decision persistence layer with DecisionRepository, DecisionModel, Alembic migration,
+  tree queries (BFS traversal, path-to-root), superseded lookup, and ordered decision path
+  retrieval. Includes Behave BDD scenarios, Robot Framework integration tests, and ASV
+  benchmarks. (#171)
 - Added token/cost tracking, budget enforcement (per-plan and per-day), provider fallback
   selection with capability filtering, and cost metadata for plan execution. New config keys
   `budget_per_plan`, `budget_per_day`, and `fallback_providers` control spending limits and

CONTRIBUTORS.md

@@ -2,6 +2,7 @@
 
 * Jeffrey Phillips Freeman <jeffrey.freeman@syncleus.com>
 * Brent E. Edwards <brent.edwards@cleverthis.com>
+* Hamza Khyari <hamza.khyari@cleverthis.com>
 * Luis Mendes <luis.p.mendes@gmail.com>
 
 # Details

alembic/versions/m4_001_decision_tables.py

@@ -0,0 +1,153 @@
+"""Add decision tree tables.
+
+Creates the ``decisions`` and ``decision_dependencies`` tables for
+the decision tree subsystem (Stage M3.2).
+
+Revision ID: m4_001_decision_tables
+Revises: d0_002_merge_changeset_and_locks
+Create Date: 2026-02-23
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "m4_001_decision_tables"
+down_revision: str | None = "d0_002_merge_changeset_and_locks"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Create decisions and decision_dependencies tables."""
+
+    # -- decisions ----------------------------------------------------------
+    op.create_table(
+        "decisions",
+        sa.Column("decision_id", sa.String(26), nullable=False),
+        sa.Column(
+            "plan_id",
+            sa.String(26),
+            sa.ForeignKey("v3_plans.plan_id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column(
+            "parent_decision_id",
+            sa.String(26),
+            sa.ForeignKey("decisions.decision_id", ondelete="SET NULL"),
+            nullable=True,
+        ),
+        sa.Column("sequence_number", sa.Integer, nullable=False),
+        sa.Column("decision_type", sa.String(30), nullable=False),
+        sa.Column("question", sa.Text, nullable=False),
+        sa.Column("chosen_option", sa.Text, nullable=False),
+        sa.Column("alternatives_considered_json", sa.Text, nullable=True),
+        sa.Column("confidence_score", sa.Float, nullable=True),
+        sa.Column("context_snapshot_json", sa.Text, nullable=False),
+        sa.Column("rationale", sa.Text, nullable=True),
+        sa.Column("actor_reasoning", sa.Text, nullable=True),
+        sa.Column("downstream_decision_ids_json", sa.Text, nullable=True),
+        sa.Column("downstream_plan_ids_json", sa.Text, nullable=True),
+        sa.Column("artifacts_produced_json", sa.Text, nullable=True),
+        sa.Column("created_at", sa.String(30), nullable=False),
+        sa.Column(
+            "is_correction",
+            sa.Boolean,
+            nullable=False,
+            server_default=sa.text("0"),
+        ),
+        sa.Column(
+            "corrects_decision_id",
+            sa.String(26),
+            sa.ForeignKey("decisions.decision_id", ondelete="SET NULL"),
+            nullable=True,
+        ),
+        sa.Column("correction_reason", sa.Text, nullable=True),
+        sa.Column(
+            "superseded_by",
+            sa.String(26),
+            sa.ForeignKey("decisions.decision_id", ondelete="SET NULL"),
+            nullable=True,
+        ),
+        sa.PrimaryKeyConstraint("decision_id"),
+        sa.CheckConstraint(
+            "decision_type IN ("
+            "'prompt_definition', 'invariant_enforced', "
+            "'strategy_choice', 'implementation_choice', "
+            "'resource_selection', 'subplan_spawn', "
+            "'subplan_parallel_spawn', 'tool_invocation', "
+            "'error_recovery', 'validation_response', "
+            "'user_intervention')",
+            name="ck_decisions_type",
+        ),
+        sa.CheckConstraint(
+            "confidence_score IS NULL OR "
+            "(confidence_score >= 0.0 AND confidence_score <= 1.0)",
+            name="ck_decisions_confidence",
+        ),
+    )
+    op.create_index("ix_decisions_plan", "decisions", ["plan_id"])
+    op.create_index("ix_decisions_parent", "decisions", ["parent_decision_id"])
+    op.create_index("ix_decisions_type", "decisions", ["decision_type"])
+    op.create_index("ix_decisions_created", "decisions", ["created_at"])
+    op.create_index(
+        "ix_decisions_plan_seq", "decisions", ["plan_id", "sequence_number"]
+    )
+
+    # -- decision_dependencies ----------------------------------------------
+    op.create_table(
+        "decision_dependencies",
+        sa.Column(
+            "source_decision_id",
+            sa.String(26),
+            sa.ForeignKey("decisions.decision_id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column(
+            "target_decision_id",
+            sa.String(26),
+            sa.ForeignKey("decisions.decision_id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column(
+            "relationship_type",
+            sa.String(30),
+            nullable=False,
+            server_default="influences",
+        ),
+        sa.Column("created_at", sa.String(30), nullable=False),
+        sa.PrimaryKeyConstraint("source_decision_id", "target_decision_id"),
+        sa.CheckConstraint(
+            "source_decision_id != target_decision_id",
+            name="ck_decision_deps_no_self_loop",
+        ),
+    )
+    op.create_index(
+        "ix_decision_deps_source",
+        "decision_dependencies",
+        ["source_decision_id"],
+    )
+    op.create_index(
+        "ix_decision_deps_target",
+        "decision_dependencies",
+        ["target_decision_id"],
+    )
+
+
+def downgrade() -> None:
+    """Drop decision_dependencies and decisions tables."""
+
+    # Drop decision_dependencies first (FK to decisions)
+    op.drop_index("ix_decision_deps_target", table_name="decision_dependencies")
+    op.drop_index("ix_decision_deps_source", table_name="decision_dependencies")
+    op.drop_table("decision_dependencies")
+
+    # Drop decisions
+    op.drop_index("ix_decisions_plan_seq", table_name="decisions")
+    op.drop_index("ix_decisions_created", table_name="decisions")
+    op.drop_index("ix_decisions_type", table_name="decisions")
+    op.drop_index("ix_decisions_parent", table_name="decisions")
+    op.drop_index("ix_decisions_plan", table_name="decisions")
+    op.drop_table("decisions")

benchmarks/decision_persistence_bench.py

@@ -0,0 +1,246 @@
+"""ASV benchmarks for DecisionRepository persistence operations.
+
+Measures create, get, get_by_plan, get_tree, and list_by_type performance.
+"""
+
+from __future__ import annotations
+
+import sys
+from datetime import UTC, datetime
+from pathlib import Path
+
+try:
+    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 (
+        NamespacedName,
+        Plan,
+        PlanIdentity,
+        PlanPhase,
+        PlanTimestamps,
+        ProcessingState,
+    )
+    from cleveragents.infrastructure.database.models import Base
+    from cleveragents.infrastructure.database.repositories import (
+        ActionRepository,
+        DecisionRepository,
+        LifecyclePlanRepository,
+    )
+except ModuleNotFoundError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+    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 (
+        NamespacedName,
+        Plan,
+        PlanIdentity,
+        PlanPhase,
+        PlanTimestamps,
+        ProcessingState,
+    )
+    from cleveragents.infrastructure.database.models import Base
+    from cleveragents.infrastructure.database.repositories import (
+        ActionRepository,
+        DecisionRepository,
+        LifecyclePlanRepository,
+    )
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+_PLAN_ID = "01HV00000000000000BENCH001"
+
+
+def _make_db():
+    """Create an in-memory SQLite DB with all tables and prerequisites."""
+    engine = create_engine("sqlite:///:memory:", echo=False)
+    Base.metadata.create_all(engine)
+    sm = sessionmaker(bind=engine)
+    session = sm()
+    factory = lambda: session  # noqa: E731
+
+    # Action prerequisite
+    action_repo = ActionRepository(session_factory=factory)
+    action = Action(
+        namespaced_name=NamespacedName.parse("local/bench-decision-action"),
+        description="Benchmark action",
+        definition_of_done="Done",
+        strategy_actor="local/s",
+        execution_actor="local/e",
+        state=ActionState.AVAILABLE,
+        created_at=datetime(2026, 1, 1, tzinfo=UTC),
+        updated_at=datetime(2026, 1, 1, tzinfo=UTC),
+    )
+    action_repo.create(action)
+    session.commit()
+
+    # Plan prerequisite
+    plan_repo = LifecyclePlanRepository(session_factory=factory)
+    now = datetime(2026, 3, 1, tzinfo=UTC)
+    plan = Plan(
+        identity=PlanIdentity(plan_id=_PLAN_ID, attempt=1),
+        namespaced_name=NamespacedName(namespace="local", name="bench-plan"),
+        action_name="local/bench-decision-action",
+        description="Benchmark plan",
+        definition_of_done="Done",
+        phase=PlanPhase.STRATEGIZE,
+        processing_state=ProcessingState.PROCESSING,
+        strategy_actor="local/s",
+        execution_actor="local/e",
+        timestamps=PlanTimestamps(created_at=now, updated_at=now),
+        created_by="bench",
+        tags=[],
+        reusable=True,
+        read_only=False,
+    )
+    plan_repo.create(plan)
+    session.commit()
+
+    return session, factory
+
+
+def _make_decision(seq=0, parent_id=None, dtype=DecisionType.PROMPT_DEFINITION):
+    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 TimeDecisionCreate:
+    """Benchmark decision creation."""
+
+    timeout = 60
+
+    def setup(self):
+        self.session, self.factory = _make_db()
+        self.repo = DecisionRepository(session_factory=self.factory)
+        self._seq = 100
+
+    def time_create_single(self):
+        self._seq += 1
+        d = _make_decision(seq=self._seq, dtype=DecisionType.STRATEGY_CHOICE)
+        self.repo.create(d)
+        self.session.commit()
+
+
+class TimeDecisionGet:
+    """Benchmark decision retrieval."""
+
+    timeout = 60
+
+    def setup(self):
+        self.session, self.factory = _make_db()
+        self.repo = DecisionRepository(session_factory=self.factory)
+        d = _make_decision()
+        self.repo.create(d)
+        self.session.commit()
+        self.decision_id = d.decision_id
+
+    def time_get_by_id(self):
+        self.repo.get(self.decision_id)
+
+
+class TimeDecisionGetByPlan:
+    """Benchmark get_by_plan with 50 pre-seeded decisions."""
+
+    timeout = 60
+
+    def setup(self):
+        self.session, self.factory = _make_db()
+        self.repo = DecisionRepository(session_factory=self.factory)
+        root = _make_decision(seq=0)
+        self.repo.create(root)
+        root_id = root.decision_id
+        for i in range(1, 50):
+            d = _make_decision(
+                seq=i,
+                parent_id=root_id,
+                dtype=DecisionType.STRATEGY_CHOICE,
+            )
+            self.repo.create(d)
+        self.session.commit()
+
+    def time_get_by_plan(self):
+        self.repo.get_by_plan(_PLAN_ID)
+
+
+class TimeDecisionTree:
+    """Benchmark get_tree with a 3-level tree."""
+
+    timeout = 60
+
+    def setup(self):
+        self.session, self.factory = _make_db()
+        self.repo = DecisionRepository(session_factory=self.factory)
+
+        root = _make_decision(seq=0)
+        self.repo.create(root)
+        self.root_id = root.decision_id
+
+        seq = 1
+        for _ in range(3):
+            child = _make_decision(
+                seq=seq,
+                parent_id=root.decision_id,
+                dtype=DecisionType.STRATEGY_CHOICE,
+            )
+            self.repo.create(child)
+            for _ in range(2):
+                seq += 1
+                gc = _make_decision(
+                    seq=seq,
+                    parent_id=child.decision_id,
+                    dtype=DecisionType.IMPLEMENTATION_CHOICE,
+                )
+                self.repo.create(gc)
+            seq += 1
+        self.session.commit()
+
+    def time_get_tree(self):
+        self.repo.get_tree(self.root_id)
+
+
+class TimeDecisionListByType:
+    """Benchmark list_by_type with pre-seeded typed decisions."""
+
+    timeout = 60
+
+    def setup(self):
+        self.session, self.factory = _make_db()
+        self.repo = DecisionRepository(session_factory=self.factory)
+
+        root = _make_decision(seq=0)
+        self.repo.create(root)
+        root_id = root.decision_id
+
+        for i in range(1, 20):
+            d = _make_decision(
+                seq=i,
+                parent_id=root_id,
+                dtype=DecisionType.STRATEGY_CHOICE,
+            )
+            self.repo.create(d)
+        for i in range(20, 30):
+            d = _make_decision(
+                seq=i,
+                parent_id=root_id,
+                dtype=DecisionType.IMPLEMENTATION_CHOICE,
+            )
+            self.repo.create(d)
+        self.session.commit()
+
+    def time_list_strategy_choices(self):
+        self.repo.list_by_type(_PLAN_ID, "strategy_choice")
+
+    def time_list_implementation_choices(self):
+        self.repo.list_by_type(_PLAN_ID, "implementation_choice")

docs/reference/database_schema.md

@@ -230,6 +230,82 @@ tracking, and optional compiled metadata.
 
 ---
 
+## Decision Tree Tables
+
+The decision tree subsystem consists of two tables introduced in Stage M3.2:
+
+| Table                    | Model                    | Description                      |
+|--------------------------|--------------------------|----------------------------------|
+| `decisions`              | `DecisionModel`          | Decision tree nodes              |
+| `decision_dependencies`  | `DecisionDependencyModel` | Decision influence DAG edges    |
+
+### decisions
+
+Primary key: `decision_id` (26-char ULID).
+FK to `v3_plans.plan_id` (CASCADE delete).  Self-referential FKs for
+`parent_decision_id`, `corrects_decision_id`, and `superseded_by` (SET NULL).
+
+| Column                         | Type        | Constraints                           | Description                             |
+|--------------------------------|-------------|---------------------------------------|-----------------------------------------|
+| `decision_id`                  | String(26)  | PRIMARY KEY                           | ULID identifier                         |
+| `plan_id`                      | String(26)  | NOT NULL, FK -> v3_plans CASCADE      | Parent plan reference                   |
+| `parent_decision_id`           | String(26)  | FK -> decisions SET NULL              | Parent decision in tree                 |
+| `sequence_number`              | Integer     | NOT NULL                              | Monotonic order within plan             |
+| `decision_type`                | String(30)  | NOT NULL, CHECK (11 enum values)      | Classification of the decision          |
+| `question`                     | Text        | NOT NULL                              | What question was being answered        |
+| `chosen_option`                | Text        | NOT NULL                              | The option that was chosen              |
+| `alternatives_considered_json` | Text        |                                       | JSON array of alternative options       |
+| `confidence_score`             | Float       | CHECK (0.0-1.0 or NULL)              | Confidence in the decision              |
+| `context_snapshot_json`        | Text        | NOT NULL                              | JSON blob of ContextSnapshot            |
+| `rationale`                    | Text        |                                       | Human-readable rationale                |
+| `actor_reasoning`              | Text        |                                       | Raw LLM reasoning trace                 |
+| `downstream_decision_ids_json` | Text        |                                       | JSON array of dependent decision ULIDs  |
+| `downstream_plan_ids_json`     | Text        |                                       | JSON array of spawned plan ULIDs        |
+| `artifacts_produced_json`      | Text        |                                       | JSON array of ArtifactRef objects       |
+| `created_at`                   | String(30)  | NOT NULL                              | ISO-8601 timestamp                      |
+| `is_correction`                | Boolean     | NOT NULL, DEFAULT FALSE               | Whether this is a correction            |
+| `corrects_decision_id`         | String(26)  | FK -> decisions SET NULL              | Decision being corrected                |
+| `correction_reason`            | Text        |                                       | Why the correction was made             |
+| `superseded_by`                | String(26)  | FK -> decisions SET NULL              | Decision that supersedes this one       |
+
+**Check constraints:**
+
+- `ck_decisions_type`: `decision_type IN ('prompt_definition', 'invariant_enforced', 'strategy_choice', 'implementation_choice', 'resource_selection', 'subplan_spawn', 'subplan_parallel_spawn', 'tool_invocation', 'error_recovery', 'validation_response', 'user_intervention')`
+- `ck_decisions_confidence`: `confidence_score IS NULL OR (confidence_score >= 0.0 AND confidence_score <= 1.0)`
+
+**Indexes:**
+
+- `ix_decisions_plan` on `(plan_id)`
+- `ix_decisions_parent` on `(parent_decision_id)`
+- `ix_decisions_type` on `(decision_type)`
+- `ix_decisions_created` on `(created_at)`
+- `ix_decisions_plan_seq` on `(plan_id, sequence_number)`
+
+### decision_dependencies
+
+Composite primary key: `(source_decision_id, target_decision_id)`.
+Both columns FK to `decisions.decision_id` with `CASCADE` delete.
+Stores `relationship_type` (default `influences`) and a self-loop
+check constraint.
+
+| Column                | Type        | Constraints                            | Description                     |
+|-----------------------|-------------|----------------------------------------|---------------------------------|
+| `source_decision_id`  | String(26)  | PK, FK -> decisions CASCADE            | Decision that influences        |
+| `target_decision_id`  | String(26)  | PK, FK -> decisions CASCADE            | Decision that is influenced     |
+| `relationship_type`   | String(30)  | NOT NULL, DEFAULT 'influences'         | Type of influence relationship  |
+| `created_at`          | String(30)  | NOT NULL                               | ISO-8601 timestamp              |
+
+**Check constraints:**
+
+- `ck_decision_deps_no_self_loop`: `source_decision_id != target_decision_id`
+
+**Indexes:**
+
+- `ix_decision_deps_source` on `(source_decision_id)`
+- `ix_decision_deps_target` on `(target_decision_id)`
+
+---
+
 ## Migration
 
 - **Revision**: `c1_001_tool_registry`

features/decision_persistence.feature

@@ -0,0 +1,149 @@
+Feature: Decision persistence via DecisionRepository
+  As a developer
+  I want decisions to persist correctly through the DecisionRepository
+  So that the decision tree is durable across operations
+
+  Background:
+    Given a fresh in-memory decision persistence database
+    And a prerequisite action "local/decision-action" exists for decisions
+    And a prerequisite plan exists for decisions
+
+  # ------------------------------------------------------------------
+  # Create and retrieve
+  # ------------------------------------------------------------------
+
+  Scenario: Create a root decision and retrieve it by ID
+    Given a new root decision with sequence 0
+    When I persist the decision via the decision repository
+    Then I can retrieve the decision by its ID
+    And the persisted decision type should be "prompt_definition"
+    And the persisted decision should be a root
+
+  Scenario: Create a child decision and retrieve it
+    Given a persisted root decision
+    And a new child decision of type "strategy_choice" with sequence 1
+    When I persist the decision via the decision repository
+    Then I can retrieve the decision by its ID
+    And the persisted decision should not be a root
+    And the persisted decision parent should match the root
+
+  Scenario: Persisted decision preserves all scalar fields
+    Given a new root decision with all fields populated
+    When I persist the decision via the decision repository
+    Then I can retrieve the decision by its ID
+    And the persisted decision question should match
+    And the persisted decision chosen_option should match
+    And the persisted decision confidence_score should be 0.85
+    And the persisted decision rationale should match
+
+  # ------------------------------------------------------------------
+  # JSON round-trip
+  # ------------------------------------------------------------------
+
+  Scenario: Alternatives considered round-trip through JSON
+    Given a new decision with 3 alternatives
+    When I persist the decision via the decision repository
+    Then I can retrieve the decision by its ID
+    And the persisted decision should have 3 alternatives
+
+  Scenario: Context snapshot round-trip through JSON
+    Given a new decision with a populated context snapshot
+    When I persist the decision via the decision repository
+    Then I can retrieve the decision by its ID
+    And the persisted decision context hash should be "sha256:persist-test"
+    And the persisted decision context should have 2 resources
+
+  Scenario: Artifacts produced round-trip through JSON
+    Given a new decision with 2 artifacts
+    When I persist the decision via the decision repository
+    Then I can retrieve the decision by its ID
+    And the persisted decision should have 2 artifacts
+
+  Scenario: Downstream IDs round-trip through JSON
+    Given a new decision with downstream IDs
+    When I persist the decision via the decision repository
+    Then I can retrieve the decision by its ID
+    And the persisted decision should have downstream decision IDs
+    And the persisted decision should have downstream plan IDs
+
+  # ------------------------------------------------------------------
+  # Get by plan
+  # ------------------------------------------------------------------
+
+  Scenario: Get decisions by plan returns ordered results
+    Given 3 persisted decisions for the same plan
+    When I get decisions by plan ID
+    Then I should get 3 decisions in sequence order
+
+  # ------------------------------------------------------------------
+  # Get tree (BFS)
+  # ------------------------------------------------------------------
+
+  Scenario: Get tree returns root and descendants in BFS order
+    Given a persisted decision tree with 3 levels
+    When I get the decision tree from the root
+    Then the tree should have 4 decisions
+    And the first decision in the tree should be the root
+
+  # ------------------------------------------------------------------
+  # Get path to root
+  # ------------------------------------------------------------------
+
+  Scenario: Get path to root walks from leaf to root
+    Given a persisted decision tree with 3 levels
+    When I get the path from a leaf to root
+    Then the path should start with the leaf
+    And the path should end with the root
+
+  # ------------------------------------------------------------------
+  # Correction and superseded
+  # ------------------------------------------------------------------
+
+  Scenario: Create a correction decision
+    Given a persisted root decision
+    And a correction decision that corrects the root
+    When I persist the decision via the decision repository
+    Then I can retrieve the decision by its ID
+    And the persisted decision is_correction should be true
+
+  Scenario: Update superseded_by on an existing decision
+    Given a persisted root decision
+    When I update the root decision superseded_by to a new ULID
+    Then the root decision should be marked as superseded
+
+  Scenario: Get superseded decisions for a plan
+    Given a persisted root decision that is superseded
+    When I get superseded decisions for the plan
+    Then I should get 1 superseded decision
+
+  # ------------------------------------------------------------------
+  # List by type
+  # ------------------------------------------------------------------
+
+  Scenario: List decisions by type filters correctly
+    Given 2 persisted decisions of type "strategy_choice"
+    And 1 persisted decision of type "implementation_choice"
+    When I list decisions by type "strategy_choice"
+    Then I should get 2 decisions of that type
+
+  # ------------------------------------------------------------------
+  # Delete
+  # ------------------------------------------------------------------
+
+  Scenario: Delete a decision by ID
+    Given a persisted root decision
+    When I delete the root decision
+    Then the root decision should no longer exist
+
+  Scenario: Delete a non-existent decision returns false
+    When I try to delete a non-existent decision
+    Then the decision delete result should be false
+
+  # ------------------------------------------------------------------
+  # Duplicate detection
+  # ------------------------------------------------------------------
+
+  Scenario: Creating a duplicate decision raises DuplicateDecisionError
+    Given a persisted root decision
+    When I try to persist the same decision again
+    Then a DuplicateDecisionError should be raised

features/steps/decision_persistence_steps.py

@@ -0,0 +1,608 @@
+"""Step definitions for decision_persistence.feature.
+
+Tests the DecisionRepository CRUD operations, JSON round-trips,
+tree traversal, correction workflows, and constraint enforcement.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+from behave import given, then, when  # type: ignore[import-untyped]
+from behave.runner import Context
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from ulid import ULID
+
+from cleveragents.domain.models.core.action import Action, ActionState
+from cleveragents.domain.models.core.decision import (
+    ArtifactRef,
+    ContextSnapshot,
+    Decision,
+    DecisionType,
+    ResourceRef,
+)
+from cleveragents.domain.models.core.plan import (
+    NamespacedName,
+    Plan,
+    PlanIdentity,
+    PlanPhase,
+    PlanTimestamps,
+    ProcessingState,
+)
+from cleveragents.infrastructure.database.models import Base
+from cleveragents.infrastructure.database.repositories import (
+    ActionRepository,
+    DecisionRepository,
+    DuplicateDecisionError,
+    LifecyclePlanRepository,
+)
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_PLAN_ID = "01HV000000000000000000DP01"
+
+
+def _setup_db(context: Context) -> None:
+    """Create an in-memory SQLite DB and attach repos."""
+    engine = create_engine("sqlite:///:memory:", echo=False)
+    Base.metadata.create_all(engine)
+    sm = sessionmaker(bind=engine)
+    session = sm()
+    context._dp_engine = engine
+    context._dp_session = session
+    context._dp_factory = lambda: session
+    context._dp_repo = DecisionRepository(session_factory=context._dp_factory)
+    context._dp_action_repo = ActionRepository(session_factory=context._dp_factory)
+    context._dp_plan_repo = LifecyclePlanRepository(
+        session_factory=context._dp_factory,
+    )
+
+
+def _make_decision(
+    plan_id: str = _PLAN_ID,
+    parent_decision_id: str | None = None,
+    sequence_number: int = 0,
+    decision_type: DecisionType = DecisionType.PROMPT_DEFINITION,
+    question: str = "What approach should we take?",
+    chosen_option: str = "Build a REST API",
+    confidence_score: float | None = None,
+    context_snapshot: ContextSnapshot | None = None,
+    alternatives_considered: list[str] | None = None,
+    artifacts_produced: list[ArtifactRef] | None = None,
+    downstream_decision_ids: list[str] | None = None,
+    downstream_plan_ids: list[str] | None = None,
+    rationale: str = "",
+    is_correction: bool = False,
+    corrects_decision_id: str | None = None,
+    correction_reason: str | None = None,
+    superseded_by: str | None = None,
+) -> Decision:
+    kwargs: dict[str, Any] = {
+        "plan_id": plan_id,
+        "sequence_number": sequence_number,
+        "decision_type": decision_type,
+        "question": question,
+        "chosen_option": chosen_option,
+        "rationale": rationale,
+        "is_correction": is_correction,
+    }
+    if parent_decision_id is not None:
+        kwargs["parent_decision_id"] = parent_decision_id
+    if confidence_score is not None:
+        kwargs["confidence_score"] = confidence_score
+    if context_snapshot is not None:
+        kwargs["context_snapshot"] = context_snapshot
+    if alternatives_considered is not None:
+        kwargs["alternatives_considered"] = alternatives_considered
+    if artifacts_produced is not None:
+        kwargs["artifacts_produced"] = artifacts_produced
+    if downstream_decision_ids is not None:
+        kwargs["downstream_decision_ids"] = downstream_decision_ids
+    if downstream_plan_ids is not None:
+        kwargs["downstream_plan_ids"] = downstream_plan_ids
+    if corrects_decision_id is not None:
+        kwargs["corrects_decision_id"] = corrects_decision_id
+    if correction_reason is not None:
+        kwargs["correction_reason"] = correction_reason
+    if superseded_by is not None:
+        kwargs["superseded_by"] = superseded_by
+    return Decision(**kwargs)
+
+
+def _create_prerequisite_action(context: Context) -> None:
+    ns = NamespacedName.parse("local/decision-action")
+    action = Action(
+        namespaced_name=ns,
+        description="Prerequisite action for decision tests",
+        definition_of_done="Done",
+        strategy_actor="local/s",
+        execution_actor="local/e",
+        state=ActionState.AVAILABLE,
+        created_at=datetime(2026, 1, 1, tzinfo=UTC),
+        updated_at=datetime(2026, 1, 1, tzinfo=UTC),
+    )
+    context._dp_action_repo.create(action)
+    context._dp_session.commit()
+
+
+def _create_prerequisite_plan(context: Context) -> None:
+    now = datetime(2026, 3, 1, tzinfo=UTC)
+    plan = Plan(
+        identity=PlanIdentity(plan_id=_PLAN_ID, attempt=1),
+        namespaced_name=NamespacedName(namespace="local", name="decision-plan"),
+        action_name="local/decision-action",
+        description="Decision test plan",
+        definition_of_done="All assertions pass",
+        phase=PlanPhase.STRATEGIZE,
+        processing_state=ProcessingState.PROCESSING,
+        strategy_actor="local/s",
+        execution_actor="local/e",
+        timestamps=PlanTimestamps(created_at=now, updated_at=now),
+        created_by="test-decision",
+        tags=[],
+        reusable=True,
+        read_only=False,
+    )
+    context._dp_plan_repo.create(plan)
+    context._dp_session.commit()
+
+
+# ---------------------------------------------------------------------------
+# Background
+# ---------------------------------------------------------------------------
+
+
+@given("a fresh in-memory decision persistence database")
+def step_fresh_db(context: Context) -> None:
+    _setup_db(context)
+
+
+@given('a prerequisite action "{action_name}" exists for decisions')
+def step_prereq_action(context: Context, action_name: str) -> None:
+    _create_prerequisite_action(context)
+
+
+@given("a prerequisite plan exists for decisions")
+def step_prereq_plan(context: Context) -> None:
+    _create_prerequisite_plan(context)
+
+
+# ---------------------------------------------------------------------------
+# Create and retrieve
+# ---------------------------------------------------------------------------
+
+
+@given("a new root decision with sequence {seq:d}")
+def step_new_root(context: Context, seq: int) -> None:
+    context._dp_decision = _make_decision(sequence_number=seq)
+
+
+@given("a persisted root decision")
+def step_persisted_root(context: Context) -> None:
+    d = _make_decision()
+    context._dp_repo.create(d)
+    context._dp_session.commit()
+    context._dp_root_decision = d
+    context._dp_root_id = d.decision_id
+
+
+@given('a new child decision of type "{dtype}" with sequence {seq:d}')
+def step_new_child(context: Context, dtype: str, seq: int) -> None:
+    context._dp_decision = _make_decision(
+        parent_decision_id=context._dp_root_id,
+        sequence_number=seq,
+        decision_type=DecisionType(dtype),
+    )
+
+
+@given("a new root decision with all fields populated")
+def step_new_full_root(context: Context) -> None:
+    context._dp_decision = _make_decision(
+        question="Full question?",
+        chosen_option="Full option",
+        confidence_score=0.85,
+        rationale="Thorough analysis",
+        alternatives_considered=["A", "B"],
+        context_snapshot=ContextSnapshot(
+            hot_context_hash="sha256:full",
+            hot_context_ref="store://full",
+            relevant_resources=[ResourceRef(resource_id=str(ULID()))],
+            actor_state_ref="checkpoint://full",
+        ),
+    )
+
+
+@when("I persist the decision via the decision repository")
+def step_persist_decision(context: Context) -> None:
+    context._dp_repo.create(context._dp_decision)
+    context._dp_session.commit()
+
+
+@then("I can retrieve the decision by its ID")
+def step_retrieve_by_id(context: Context) -> None:
+    d = context._dp_repo.get(context._dp_decision.decision_id)
+    assert d is not None, "Decision not found"
+    context._dp_retrieved = d
+
+
+@then('the persisted decision type should be "{dtype}"')
+def step_check_type(context: Context, dtype: str) -> None:
+    assert str(context._dp_retrieved.decision_type) == dtype
+
+
+@then("the persisted decision should be a root")
+def step_check_is_root(context: Context) -> None:
+    assert context._dp_retrieved.is_root
+
+
+@then("the persisted decision should not be a root")
+def step_check_not_root(context: Context) -> None:
+    assert not context._dp_retrieved.is_root
+
+
+@then("the persisted decision parent should match the root")
+def step_check_parent(context: Context) -> None:
+    assert context._dp_retrieved.parent_decision_id == context._dp_root_id
+
+
+@then("the persisted decision question should match")
+def step_check_question(context: Context) -> None:
+    assert context._dp_retrieved.question == context._dp_decision.question
+
+
+@then("the persisted decision chosen_option should match")
+def step_check_chosen(context: Context) -> None:
+    assert context._dp_retrieved.chosen_option == context._dp_decision.chosen_option
+
+
+@then("the persisted decision confidence_score should be {score}")
+def step_check_confidence(context: Context, score: str) -> None:
+    expected = float(score)
+    assert context._dp_retrieved.confidence_score == expected, (
+        f"Expected {expected}, got {context._dp_retrieved.confidence_score}"
+    )
+
+
+@then("the persisted decision rationale should match")
+def step_check_rationale(context: Context) -> None:
+    assert context._dp_retrieved.rationale == context._dp_decision.rationale
+
+
+# ---------------------------------------------------------------------------
+# JSON round-trip
+# ---------------------------------------------------------------------------
+
+
+@given("a new decision with {count:d} alternatives")
+def step_new_with_alts(context: Context, count: int) -> None:
+    alts = [f"Alternative {i}" for i in range(count)]
+    context._dp_decision = _make_decision(alternatives_considered=alts)
+
+
+@then("the persisted decision should have {count:d} alternatives")
+def step_check_alt_count(context: Context, count: int) -> None:
+    assert len(context._dp_retrieved.alternatives_considered) == count
+
+
+@given("a new decision with a populated context snapshot")
+def step_new_with_snapshot(context: Context) -> None:
+    snap = ContextSnapshot(
+        hot_context_hash="sha256:persist-test",
+        hot_context_ref="store://persist-test",
+        relevant_resources=[
+            ResourceRef(resource_id=str(ULID()), path="src/main.py"),
+            ResourceRef(resource_id=str(ULID())),
+        ],
+        actor_state_ref="checkpoint://persist",
+    )
+    context._dp_decision = _make_decision(context_snapshot=snap)
+
+
+@then('the persisted decision context hash should be "{expected}"')
+def step_check_ctx_hash(context: Context, expected: str) -> None:
+    assert context._dp_retrieved.context_snapshot.hot_context_hash == expected
+
+
+@then("the persisted decision context should have {count:d} resources")
+def step_check_ctx_resources(context: Context, count: int) -> None:
+    assert len(context._dp_retrieved.context_snapshot.relevant_resources) == count
+
+
+@given("a new decision with {count:d} artifacts")
+def step_new_with_artifacts(context: Context, count: int) -> None:
+    arts = [
+        ArtifactRef(artifact_path=f"src/file_{i}.py", artifact_type="file")
+        for i in range(count)
+    ]
+    context._dp_decision = _make_decision(artifacts_produced=arts)
+
+
+@then("the persisted decision should have {count:d} artifacts")
+def step_check_artifact_count(context: Context, count: int) -> None:
+    assert len(context._dp_retrieved.artifacts_produced) == count
+
+
+@given("a new decision with downstream IDs")
+def step_new_with_downstream(context: Context) -> None:
+    context._dp_decision = _make_decision(
+        downstream_decision_ids=[str(ULID()), str(ULID())],
+        downstream_plan_ids=[str(ULID())],
+    )
+
+
+@then("the persisted decision should have downstream decision IDs")
+def step_check_downstream_decisions(context: Context) -> None:
+    assert len(context._dp_retrieved.downstream_decision_ids) > 0
+
+
+@then("the persisted decision should have downstream plan IDs")
+def step_check_downstream_plans(context: Context) -> None:
+    assert len(context._dp_retrieved.downstream_plan_ids) > 0
+
+
+# ---------------------------------------------------------------------------
+# Get by plan
+# ---------------------------------------------------------------------------
+
+
+@given("{count:d} persisted decisions for the same plan")
+def step_n_decisions(context: Context, count: int) -> None:
+    for i in range(count):
+        dt = DecisionType.PROMPT_DEFINITION if i == 0 else DecisionType.STRATEGY_CHOICE
+        parent = None if i == 0 else context._dp_first_id
+        d = _make_decision(
+            sequence_number=i,
+            decision_type=dt,
+            parent_decision_id=parent,
+        )
+        context._dp_repo.create(d)
+        if i == 0:
+            context._dp_first_id = d.decision_id
+    context._dp_session.commit()
+
+
+@when("I get decisions by plan ID")
+def step_get_by_plan(context: Context) -> None:
+    context._dp_plan_decisions = context._dp_repo.get_by_plan(_PLAN_ID)
+
+
+@then("I should get {count:d} decisions in sequence order")
+def step_check_plan_decisions(context: Context, count: int) -> None:
+    decisions = context._dp_plan_decisions
+    assert len(decisions) == count, f"Expected {count}, got {len(decisions)}"
+    for i in range(len(decisions) - 1):
+        assert decisions[i].sequence_number <= decisions[i + 1].sequence_number
+
+
+# ---------------------------------------------------------------------------
+# Get tree
+# ---------------------------------------------------------------------------
+
+
+@given("a persisted decision tree with 3 levels")
+def step_decision_tree(context: Context) -> None:
+    # Root (level 0)
+    root = _make_decision(sequence_number=0)
+    context._dp_repo.create(root)
+    context._dp_tree_root_id = root.decision_id
+
+    # Level 1 — two children
+    child1 = _make_decision(
+        parent_decision_id=root.decision_id,
+        sequence_number=1,
+        decision_type=DecisionType.STRATEGY_CHOICE,
+    )
+    context._dp_repo.create(child1)
+
+    child2 = _make_decision(
+        parent_decision_id=root.decision_id,
+        sequence_number=2,
+        decision_type=DecisionType.STRATEGY_CHOICE,
+    )
+    context._dp_repo.create(child2)
+
+    # Level 2 — one grandchild under child1
+    grandchild = _make_decision(
+        parent_decision_id=child1.decision_id,
+        sequence_number=3,
+        decision_type=DecisionType.IMPLEMENTATION_CHOICE,
+    )
+    context._dp_repo.create(grandchild)
+    context._dp_tree_leaf_id = grandchild.decision_id
+
+    context._dp_session.commit()
+
+
+@when("I get the decision tree from the root")
+def step_get_tree(context: Context) -> None:
+    context._dp_tree = context._dp_repo.get_tree(context._dp_tree_root_id)
+
+
+@then("the tree should have {count:d} decisions")
+def step_check_tree_count(context: Context, count: int) -> None:
+    assert len(context._dp_tree) == count, (
+        f"Expected {count}, got {len(context._dp_tree)}"
+    )
+
+
+@then("the first decision in the tree should be the root")
+def step_check_tree_root(context: Context) -> None:
+    assert context._dp_tree[0].decision_id == context._dp_tree_root_id
+
+
+# ---------------------------------------------------------------------------
+# Get path to root
+# ---------------------------------------------------------------------------
+
+
+@when("I get the path from a leaf to root")
+def step_get_path(context: Context) -> None:
+    context._dp_path = context._dp_repo.get_path_to_root(
+        context._dp_tree_leaf_id,
+    )
+
+
+@then("the path should start with the leaf")
+def step_check_path_leaf(context: Context) -> None:
+    assert context._dp_path[0].decision_id == context._dp_tree_leaf_id
+
+
+@then("the path should end with the root")
+def step_check_path_root(context: Context) -> None:
+    assert context._dp_path[-1].decision_id == context._dp_tree_root_id
+
+
+# ---------------------------------------------------------------------------
+# Correction and superseded
+# ---------------------------------------------------------------------------
+
+
+@given("a correction decision that corrects the root")
+def step_correction_decision(context: Context) -> None:
+    context._dp_decision = _make_decision(
+        parent_decision_id=context._dp_root_id,
+        sequence_number=1,
+        decision_type=DecisionType.STRATEGY_CHOICE,
+        is_correction=True,
+        corrects_decision_id=context._dp_root_id,
+        correction_reason="Root was suboptimal",
+    )
+
+
+@then("the persisted decision is_correction should be true")
+def step_check_is_correction(context: Context) -> None:
+    assert context._dp_retrieved.is_correction
+
+
+@when("I update the root decision superseded_by to a new ULID")
+def step_update_superseded(context: Context) -> None:
+    new_id = str(ULID())
+    context._dp_superseding_id = new_id
+    context._dp_repo.update_superseded_by(context._dp_root_id, new_id)
+    context._dp_session.commit()
+
+
+@then("the root decision should be marked as superseded")
+def step_check_superseded(context: Context) -> None:
+    d = context._dp_repo.get(context._dp_root_id)
+    assert d is not None
+    assert d.is_superseded
+    assert d.superseded_by == context._dp_superseding_id
+
+
+@given("a persisted root decision that is superseded")
+def step_superseded_root(context: Context) -> None:
+    d = _make_decision(superseded_by=str(ULID()))
+    context._dp_repo.create(d)
+    context._dp_session.commit()
+    context._dp_root_id = d.decision_id
+
+
+@when("I get superseded decisions for the plan")
+def step_get_superseded(context: Context) -> None:
+    context._dp_superseded = context._dp_repo.get_superseded(_PLAN_ID)
+
+
+@then("I should get {count:d} superseded decision")
+def step_check_superseded_count(context: Context, count: int) -> None:
+    assert len(context._dp_superseded) == count
+
+
+# ---------------------------------------------------------------------------
+# List by type
+# ---------------------------------------------------------------------------
+
+
+@given('{count:d} persisted decisions of type "{dtype}"')
+def step_n_typed_decisions(context: Context, count: int, dtype: str) -> None:
+    dt = DecisionType(dtype)
+    for i in range(count):
+        # Use sequence offset to avoid collision with any prior root
+        seq = 10 + i
+        parent = None
+        if dt != DecisionType.PROMPT_DEFINITION:
+            # Need a root first
+            if not hasattr(context, "_dp_type_root_id"):
+                root = _make_decision(sequence_number=9)
+                context._dp_repo.create(root)
+                context._dp_session.commit()
+                context._dp_type_root_id = root.decision_id
+            parent = context._dp_type_root_id
+        d = _make_decision(
+            parent_decision_id=parent,
+            sequence_number=seq,
+            decision_type=dt,
+        )
+        context._dp_repo.create(d)
+    context._dp_session.commit()
+
+
+@given('{count:d} persisted decision of type "{dtype}"')
+def step_one_typed_decision(context: Context, count: int, dtype: str) -> None:
+    step_n_typed_decisions(context, count, dtype)
+
+
+@when('I list decisions by type "{dtype}"')
+def step_list_by_type(context: Context, dtype: str) -> None:
+    context._dp_typed = context._dp_repo.list_by_type(_PLAN_ID, dtype)
+
+
+@then("I should get {count:d} decisions of that type")
+def step_check_typed_count(context: Context, count: int) -> None:
+    assert len(context._dp_typed) == count, (
+        f"Expected {count}, got {len(context._dp_typed)}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Delete
+# ---------------------------------------------------------------------------
+
+
+@when("I delete the root decision")
+def step_delete_root(context: Context) -> None:
+    context._dp_delete_result = context._dp_repo.delete(context._dp_root_id)
+    context._dp_session.commit()
+
+
+@then("the root decision should no longer exist")
+def step_check_deleted(context: Context) -> None:
+    assert context._dp_delete_result is True
+    d = context._dp_repo.get(context._dp_root_id)
+    assert d is None
+
+
+@when("I try to delete a non-existent decision")
+def step_delete_missing(context: Context) -> None:
+    context._dp_delete_result = context._dp_repo.delete(str(ULID()))
+
+
+@then("the decision delete result should be false")
+def step_check_delete_false(context: Context) -> None:
+    assert context._dp_delete_result is False
+
+
+# ---------------------------------------------------------------------------
+# Duplicate detection
+# ---------------------------------------------------------------------------
+
+
+@when("I try to persist the same decision again")
+def step_persist_duplicate(context: Context) -> None:
+    try:
+        context._dp_repo.create(context._dp_root_decision)
+        context._dp_session.commit()
+        context._dp_dup_error = None
+    except DuplicateDecisionError as exc:
+        context._dp_session.rollback()
+        context._dp_dup_error = exc
+
+
+@then("a DuplicateDecisionError should be raised")
+def step_check_duplicate(context: Context) -> None:
+    assert context._dp_dup_error is not None
+    assert isinstance(context._dp_dup_error, DuplicateDecisionError)

robot/decision_persistence.robot

@@ -0,0 +1,65 @@
+*** Settings ***
+Documentation     Smoke tests for decision persistence via DecisionRepository
+Resource          ${CURDIR}/common.resource
+Suite Setup       Setup Test Environment
+Suite Teardown    Cleanup Test Environment
+
+*** Variables ***
+${HELPER_SCRIPT}    robot/helper_decision_persistence.py
+
+*** Test Cases ***
+Create And Retrieve Root Decision
+    [Documentation]    Create a root prompt_definition decision, persist, retrieve
+    [Tags]    database    decision    persistence
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    create-retrieve    cwd=${WORKSPACE}    timeout=30s
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    create-retrieve-ok
+
+Create Child Decision With Parent
+    [Documentation]    Create a child strategy_choice under a root decision
+    [Tags]    database    decision    persistence
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    create-child    cwd=${WORKSPACE}    timeout=30s
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    create-child-ok
+
+JSON Fields Round-Trip
+    [Documentation]    Persist alternatives, context snapshot, artifacts and verify round-trip
+    [Tags]    database    decision    json    persistence
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    json-roundtrip    cwd=${WORKSPACE}    timeout=30s
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    json-roundtrip-ok
+
+Get Decisions By Plan
+    [Documentation]    Retrieve all decisions for a plan in sequence order
+    [Tags]    database    decision    query    persistence
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    get-by-plan    cwd=${WORKSPACE}    timeout=30s
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    get-by-plan-ok
+
+Decision Tree BFS Traversal
+    [Documentation]    Build a 3-level tree and retrieve via BFS
+    [Tags]    database    decision    tree    persistence
+    ${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
+
+Path To Root Traversal
+    [Documentation]    Walk from a leaf decision up to the root
+    [Tags]    database    decision    tree    persistence
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    path-to-root    cwd=${WORKSPACE}    timeout=30s
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    path-to-root-ok
+
+Superseded By Update
+    [Documentation]    Mark a decision as superseded and verify
+    [Tags]    database    decision    correction    persistence
+    ${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
+    [Documentation]    Delete a decision and verify it is gone
+    [Tags]    database    decision    persistence
+    ${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

robot/helper_decision_persistence.py

@@ -0,0 +1,380 @@
+"""Helper script for Robot Framework decision persistence smoke tests.
+
+Usage:
+    python robot/helper_decision_persistence.py <subcommand>
+
+Subcommands:
+    create-retrieve   Create + retrieve a root decision
+    create-child      Create a child under a root
+    json-roundtrip    Verify JSON fields survive persistence
+    get-by-plan       Get all decisions for a plan in order
+    tree-bfs          Build a 3-level tree, retrieve via BFS
+    path-to-root      Walk from leaf to root
+    superseded        Mark a decision as superseded
+    delete            Delete a decision
+"""
+
+from __future__ import annotations
+
+import sys
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+# Ensure src is importable when run from workspace root
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from collections.abc import Callable
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+from ulid import ULID
+
+from cleveragents.domain.models.core.action import (
+    Action,
+    ActionState,
+)
+from cleveragents.domain.models.core.decision import (
+    ArtifactRef,
+    ContextSnapshot,
+    Decision,
+    DecisionType,
+    ResourceRef,
+)
+from cleveragents.domain.models.core.plan import (
+    NamespacedName,
+    Plan,
+    PlanIdentity,
+    PlanPhase,
+    PlanTimestamps,
+    ProcessingState,
+)
+from cleveragents.infrastructure.database.models import Base
+from cleveragents.infrastructure.database.repositories import (
+    ActionRepository,
+    DecisionRepository,
+    LifecyclePlanRepository,
+)
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_PLAN_ID = "01HV000000000000000000RP01"
+
+
+def _setup() -> tuple[Session, Callable[[], Session]]:
+    """Create an in-memory SQLite DB with all tables."""
+    engine = create_engine("sqlite:///:memory:", echo=False)
+    Base.metadata.create_all(engine)
+    sm = sessionmaker(bind=engine)
+    session = sm()
+    factory: Callable[[], Session] = lambda: session  # noqa: E731
+    return session, factory
+
+
+def _create_prerequisites(session: Session, factory: Callable[[], Session]) -> None:
+    """Create the action + plan needed to satisfy FK constraints."""
+    action_repo = ActionRepository(session_factory=factory)
+    action = Action(
+        namespaced_name=NamespacedName.parse("local/robot-decision-action"),
+        description="Robot prerequisite action",
+        definition_of_done="Done",
+        strategy_actor="local/s",
+        execution_actor="local/e",
+        state=ActionState.AVAILABLE,
+        created_at=datetime(2026, 1, 1, tzinfo=UTC),
+        updated_at=datetime(2026, 1, 1, tzinfo=UTC),
+    )
+    action_repo.create(action)
+    session.commit()
+
+    plan_repo = LifecyclePlanRepository(session_factory=factory)
+    now = datetime(2026, 3, 1, tzinfo=UTC)
+    plan = Plan(
+        identity=PlanIdentity(plan_id=_PLAN_ID, attempt=1),
+        namespaced_name=NamespacedName(namespace="local", name="robot-plan"),
+        action_name="local/robot-decision-action",
+        description="Robot decision test plan",
+        definition_of_done="All tests pass",
+        phase=PlanPhase.STRATEGIZE,
+        processing_state=ProcessingState.PROCESSING,
+        strategy_actor="local/s",
+        execution_actor="local/e",
+        timestamps=PlanTimestamps(created_at=now, updated_at=now),
+        created_by="robot-test",
+        tags=[],
+        reusable=True,
+        read_only=False,
+    )
+    plan_repo.create(plan)
+    session.commit()
+
+
+def _make_decision(
+    plan_id: str = _PLAN_ID,
+    parent_decision_id: str | None = None,
+    sequence_number: int = 0,
+    decision_type: DecisionType = DecisionType.PROMPT_DEFINITION,
+    **kwargs: Any,
+) -> Decision:
+    return Decision(
+        plan_id=plan_id,
+        parent_decision_id=parent_decision_id,
+        sequence_number=sequence_number,
+        decision_type=decision_type,
+        question=kwargs.get("question", "What approach?"),
+        chosen_option=kwargs.get("chosen_option", "Build API"),
+        **{k: v for k, v in kwargs.items() if k not in ("question", "chosen_option")},
+    )
+
+
+# ---------------------------------------------------------------------------
+# Subcommands
+# ---------------------------------------------------------------------------
+
+
+def _create_retrieve() -> None:
+    session, factory = _setup()
+    _create_prerequisites(session, factory)
+    repo = DecisionRepository(session_factory=factory)
+
+    d = _make_decision()
+    repo.create(d)
+    session.commit()
+
+    got = repo.get(d.decision_id)
+    assert got is not None, "decision not found"
+    assert got.decision_id == d.decision_id
+    assert got.is_root
+    assert str(got.decision_type) == "prompt_definition"
+    print("create-retrieve-ok")
+
+
+def _create_child() -> None:
+    session, factory = _setup()
+    _create_prerequisites(session, factory)
+    repo = DecisionRepository(session_factory=factory)
+
+    root = _make_decision()
+    repo.create(root)
+    session.commit()
+
+    child = _make_decision(
+        parent_decision_id=root.decision_id,
+        sequence_number=1,
+        decision_type=DecisionType.STRATEGY_CHOICE,
+    )
+    repo.create(child)
+    session.commit()
+
+    got = repo.get(child.decision_id)
+    assert got is not None
+    assert not got.is_root
+    assert got.parent_decision_id == root.decision_id
+    print("create-child-ok")
+
+
+def _json_roundtrip() -> None:
+    session, factory = _setup()
+    _create_prerequisites(session, factory)
+    repo = DecisionRepository(session_factory=factory)
+
+    snap = ContextSnapshot(
+        hot_context_hash="sha256:robot-test",
+        hot_context_ref="store://robot",
+        relevant_resources=[
+            ResourceRef(resource_id=str(ULID()), path="src/main.py"),
+            ResourceRef(resource_id=str(ULID())),
+        ],
+        actor_state_ref="checkpoint://robot",
+    )
+    d = _make_decision(
+        alternatives_considered=["Flask", "Django", "Express"],
+        confidence_score=0.92,
+        context_snapshot=snap,
+        artifacts_produced=[
+            ArtifactRef(artifact_path="src/api.py", artifact_type="file"),
+            ArtifactRef(artifact_path="patches/fix.patch", artifact_type="patch"),
+        ],
+        downstream_decision_ids=[str(ULID())],
+        downstream_plan_ids=[str(ULID()), str(ULID())],
+    )
+    repo.create(d)
+    session.commit()
+
+    got = repo.get(d.decision_id)
+    assert got is not None
+    assert len(got.alternatives_considered) == 3
+    assert got.confidence_score == 0.92
+    assert got.context_snapshot.hot_context_hash == "sha256:robot-test"
+    assert len(got.context_snapshot.relevant_resources) == 2
+    assert len(got.artifacts_produced) == 2
+    assert len(got.downstream_decision_ids) == 1
+    assert len(got.downstream_plan_ids) == 2
+    print("json-roundtrip-ok")
+
+
+def _get_by_plan() -> None:
+    session, factory = _setup()
+    _create_prerequisites(session, factory)
+    repo = DecisionRepository(session_factory=factory)
+
+    first_id: str | None = None
+    for i in range(5):
+        dt = DecisionType.PROMPT_DEFINITION if i == 0 else DecisionType.STRATEGY_CHOICE
+        parent = None if i == 0 else first_id
+        d = _make_decision(
+            parent_decision_id=parent,
+            sequence_number=i,
+            decision_type=dt,
+        )
+        repo.create(d)
+        if i == 0:
+            first_id = d.decision_id
+    session.commit()
+
+    results = repo.get_by_plan(_PLAN_ID)
+    assert len(results) == 5
+    for i in range(len(results) - 1):
+        assert results[i].sequence_number <= results[i + 1].sequence_number
+    print("get-by-plan-ok")
+
+
+def _tree_bfs() -> None:
+    session, factory = _setup()
+    _create_prerequisites(session, factory)
+    repo = DecisionRepository(session_factory=factory)
+
+    root = _make_decision(sequence_number=0)
+    repo.create(root)
+
+    c1 = _make_decision(
+        parent_decision_id=root.decision_id,
+        sequence_number=1,
+        decision_type=DecisionType.STRATEGY_CHOICE,
+    )
+    repo.create(c1)
+
+    c2 = _make_decision(
+        parent_decision_id=root.decision_id,
+        sequence_number=2,
+        decision_type=DecisionType.STRATEGY_CHOICE,
+    )
+    repo.create(c2)
+
+    gc = _make_decision(
+        parent_decision_id=c1.decision_id,
+        sequence_number=3,
+        decision_type=DecisionType.IMPLEMENTATION_CHOICE,
+    )
+    repo.create(gc)
+    session.commit()
+
+    tree = repo.get_tree(root.decision_id)
+    assert len(tree) == 4, f"expected 4, got {len(tree)}"
+    assert tree[0].decision_id == root.decision_id
+    print("tree-bfs-ok")
+
+
+def _path_to_root() -> None:
+    session, factory = _setup()
+    _create_prerequisites(session, factory)
+    repo = DecisionRepository(session_factory=factory)
+
+    root = _make_decision(sequence_number=0)
+    repo.create(root)
+
+    child = _make_decision(
+        parent_decision_id=root.decision_id,
+        sequence_number=1,
+        decision_type=DecisionType.STRATEGY_CHOICE,
+    )
+    repo.create(child)
+
+    leaf = _make_decision(
+        parent_decision_id=child.decision_id,
+        sequence_number=2,
+        decision_type=DecisionType.IMPLEMENTATION_CHOICE,
+    )
+    repo.create(leaf)
+    session.commit()
+
+    path = repo.get_path_to_root(leaf.decision_id)
+    assert len(path) == 3
+    assert path[0].decision_id == leaf.decision_id
+    assert path[-1].decision_id == root.decision_id
+    print("path-to-root-ok")
+
+
+def _superseded() -> None:
+    session, factory = _setup()
+    _create_prerequisites(session, factory)
+    repo = DecisionRepository(session_factory=factory)
+
+    d = _make_decision()
+    repo.create(d)
+    session.commit()
+
+    new_id = str(ULID())
+    repo.update_superseded_by(d.decision_id, new_id)
+    session.commit()
+
+    got = repo.get(d.decision_id)
+    assert got is not None
+    assert got.is_superseded
+    assert got.superseded_by == new_id
+    print("superseded-ok")
+
+
+def _delete() -> None:
+    session, factory = _setup()
+    _create_prerequisites(session, factory)
+    repo = DecisionRepository(session_factory=factory)
+
+    d = _make_decision()
+    repo.create(d)
+    session.commit()
+
+    result = repo.delete(d.decision_id)
+    session.commit()
+    assert result is True
+
+    got = repo.get(d.decision_id)
+    assert got is None
+
+    # Delete non-existent
+    result2 = repo.delete(str(ULID()))
+    assert result2 is False
+
+    print("delete-ok")
+
+
+# ---------------------------------------------------------------------------
+# Dispatcher
+# ---------------------------------------------------------------------------
+
+_COMMANDS = {
+    "create-retrieve": _create_retrieve,
+    "create-child": _create_child,
+    "json-roundtrip": _json_roundtrip,
+    "get-by-plan": _get_by_plan,
+    "tree-bfs": _tree_bfs,
+    "path-to-root": _path_to_root,
+    "superseded": _superseded,
+    "delete": _delete,
+}
+
+
+def main() -> None:
+    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/infrastructure/database/models.py

@@ -19,6 +19,8 @@ Alembic migrations.
 | ``resource_edges``     | ``ResourceEdgeModel``      | Resource DAG edges     |
 | ``ns_projects``        | ``NamespacedProjectModel`` | Namespaced projects    |
 | ``project_resource_links`` | ``ProjectResourceLinkModel`` | Project-resource |
+| ``decisions``          | ``DecisionModel``          | Decision tree nodes    |
+| ``decision_dependencies`` | ``DecisionDependencyModel`` | Decision DAG edges  |
 
 Based on ADR-007 (Repository Pattern) and Phase 0 discovery.
 Includes spec-aligned lifecycle models per Stage A5
@@ -2438,6 +2440,289 @@ class LockModel(Base):  # type: ignore[misc]
     )
 
 
+# ---------------------------------------------------------------------------
+# Decision models (Stage M3.2)
+# ---------------------------------------------------------------------------
+
+
+class DecisionModel(Base):  # type: ignore[misc]
+    """Database model for decision tree nodes.
+
+    Maps to the ``decisions`` table.  Each row represents a single
+    choice point recorded during the Strategize or Execute phase of a
+    plan's lifecycle.
+
+    JSON-as-Text columns
+    --------------------
+    ``alternatives_considered_json``, ``context_snapshot_json``,
+    ``downstream_decision_ids_json``, ``downstream_plan_ids_json``, and
+    ``artifacts_produced_json`` are stored as ``Text`` and parsed with
+    :func:`json.loads` / :func:`json.dumps`.
+    """
+
+    __allow_unmapped__ = True
+    __tablename__ = "decisions"
+
+    # --- Identity ---
+    decision_id = Column(String(26), primary_key=True)
+    plan_id = Column(
+        String(26),
+        ForeignKey("v3_plans.plan_id", ondelete="CASCADE"),
+        nullable=False,
+    )
+    parent_decision_id = Column(
+        String(26),
+        ForeignKey("decisions.decision_id", ondelete="SET NULL"),
+        nullable=True,
+    )
+    sequence_number = Column(Integer, nullable=False)
+
+    # --- Classification ---
+    decision_type = Column(String(30), nullable=False)
+
+    # --- The decision itself ---
+    question = Column(Text, nullable=False)
+    chosen_option = Column(Text, nullable=False)
+    alternatives_considered_json = Column(Text, nullable=True)
+    confidence_score = Column(Float, nullable=True)
+
+    # --- Context snapshot (JSON blob) ---
+    context_snapshot_json = Column(Text, nullable=False)
+
+    # --- Rationale ---
+    rationale = Column(Text, nullable=True, default="")
+    actor_reasoning = Column(Text, nullable=True)
+
+    # --- Downstream impact (JSON arrays) ---
+    downstream_decision_ids_json = Column(Text, nullable=True)
+    downstream_plan_ids_json = Column(Text, nullable=True)
+    artifacts_produced_json = Column(Text, nullable=True)
+
+    # --- Timestamps ---
+    created_at = Column(String(30), nullable=False)
+
+    # --- Correction metadata ---
+    is_correction = Column(Boolean, nullable=False, default=False)
+    corrects_decision_id = Column(
+        String(26),
+        ForeignKey("decisions.decision_id", ondelete="SET NULL"),
+        nullable=True,
+    )
+    correction_reason = Column(Text, nullable=True)
+    superseded_by = Column(
+        String(26),
+        ForeignKey("decisions.decision_id", ondelete="SET NULL"),
+        nullable=True,
+    )
+
+    # --- Relationships ---
+    parent_decision = relationship(
+        "DecisionModel",
+        remote_side="DecisionModel.decision_id",
+        foreign_keys=[parent_decision_id],
+    )
+
+    __table_args__ = (
+        CheckConstraint(
+            "decision_type IN ("
+            "'prompt_definition', 'invariant_enforced', "
+            "'strategy_choice', 'implementation_choice', "
+            "'resource_selection', 'subplan_spawn', "
+            "'subplan_parallel_spawn', 'tool_invocation', "
+            "'error_recovery', 'validation_response', "
+            "'user_intervention')",
+            name="ck_decisions_type",
+        ),
+        CheckConstraint(
+            "confidence_score IS NULL OR "
+            "(confidence_score >= 0.0 AND confidence_score <= 1.0)",
+            name="ck_decisions_confidence",
+        ),
+        Index("ix_decisions_plan", "plan_id"),
+        Index("ix_decisions_parent", "parent_decision_id"),
+        Index("ix_decisions_type", "decision_type"),
+        Index("ix_decisions_created", "created_at"),
+        Index("ix_decisions_plan_seq", "plan_id", "sequence_number"),
+    )
+
+    # --- Domain conversion ---
+
+    def to_domain(self) -> Any:
+        """Convert to a domain :class:`Decision` instance."""
+        from cleveragents.domain.models.core.decision import (
+            ArtifactRef,
+            ContextSnapshot,
+            Decision,
+            DecisionType,
+            ResourceRef,
+        )
+
+        # Deserialise JSON-as-text columns
+        alt_raw = cast(str, self.alternatives_considered_json or "[]")
+        alternatives: list[str] = json.loads(alt_raw)
+
+        snapshot_raw: dict[str, Any] = json.loads(
+            cast(str, self.context_snapshot_json),
+        )
+        relevant_resources = [
+            ResourceRef(**r) for r in snapshot_raw.get("relevant_resources", [])
+        ]
+        context_snapshot = ContextSnapshot(
+            hot_context_hash=snapshot_raw.get("hot_context_hash", ""),
+            hot_context_ref=snapshot_raw.get("hot_context_ref", ""),
+            relevant_resources=relevant_resources,
+            actor_state_ref=snapshot_raw.get("actor_state_ref", ""),
+        )
+
+        dd_raw = cast(str, self.downstream_decision_ids_json or "[]")
+        downstream_decision_ids: list[str] = json.loads(dd_raw)
+
+        dp_raw = cast(str, self.downstream_plan_ids_json or "[]")
+        downstream_plan_ids: list[str] = json.loads(dp_raw)
+
+        artifacts_produced: list[ArtifactRef] = []
+        ap_raw = cast(str, self.artifacts_produced_json or "[]")
+        for a in json.loads(ap_raw):
+            artifacts_produced.append(ArtifactRef(**a))
+
+        # Parse created_at
+        created_at = datetime.fromisoformat(cast(str, self.created_at))
+
+        return Decision(
+            decision_id=cast(str, self.decision_id),
+            plan_id=cast(str, self.plan_id),
+            parent_decision_id=cast(str | None, self.parent_decision_id),
+            sequence_number=cast(int, self.sequence_number),
+            decision_type=DecisionType(cast(str, self.decision_type)),
+            question=cast(str, self.question),
+            chosen_option=cast(str, self.chosen_option),
+            alternatives_considered=alternatives,
+            confidence_score=cast(float | None, self.confidence_score),
+            context_snapshot=context_snapshot,
+            rationale=cast(str, self.rationale or ""),
+            actor_reasoning=cast(str | None, self.actor_reasoning),
+            downstream_decision_ids=downstream_decision_ids,
+            downstream_plan_ids=downstream_plan_ids,
+            artifacts_produced=artifacts_produced,
+            created_at=created_at,
+            is_correction=bool(self.is_correction),
+            corrects_decision_id=cast(str | None, self.corrects_decision_id),
+            correction_reason=cast(str | None, self.correction_reason),
+            superseded_by=cast(str | None, self.superseded_by),
+        )
+
+    @classmethod
+    def from_domain(cls, decision: Any) -> DecisionModel:
+        """Create a model instance from a domain :class:`Decision`."""
+        # Serialise JSON-as-text columns
+        alternatives_json: str | None = None
+        if decision.alternatives_considered:
+            alternatives_json = json.dumps(decision.alternatives_considered)
+
+        # Serialise context snapshot
+        snapshot = decision.context_snapshot
+        snapshot_dict: dict[str, Any] = {
+            "hot_context_hash": snapshot.hot_context_hash,
+            "hot_context_ref": snapshot.hot_context_ref,
+            "relevant_resources": [
+                {"resource_id": r.resource_id, "path": r.path}
+                for r in snapshot.relevant_resources
+            ],
+            "actor_state_ref": snapshot.actor_state_ref,
+        }
+        context_snapshot_json = json.dumps(snapshot_dict)
+
+        downstream_decision_ids_json: str | None = None
+        if decision.downstream_decision_ids:
+            downstream_decision_ids_json = json.dumps(
+                decision.downstream_decision_ids,
+            )
+
+        downstream_plan_ids_json: str | None = None
+        if decision.downstream_plan_ids:
+            downstream_plan_ids_json = json.dumps(decision.downstream_plan_ids)
+
+        artifacts_produced_json: str | None = None
+        if decision.artifacts_produced:
+            artifacts_produced_json = json.dumps(
+                [
+                    {
+                        "artifact_path": a.artifact_path,
+                        "artifact_type": a.artifact_type,
+                    }
+                    for a in decision.artifacts_produced
+                ],
+            )
+
+        # Format created_at as ISO-8601
+        created_at_str = decision.created_at.isoformat()
+
+        return cls(
+            decision_id=decision.decision_id,
+            plan_id=decision.plan_id,
+            parent_decision_id=decision.parent_decision_id,
+            sequence_number=decision.sequence_number,
+            decision_type=(
+                decision.decision_type.value
+                if hasattr(decision.decision_type, "value")
+                else str(decision.decision_type)
+            ),
+            question=decision.question,
+            chosen_option=decision.chosen_option,
+            alternatives_considered_json=alternatives_json,
+            confidence_score=decision.confidence_score,
+            context_snapshot_json=context_snapshot_json,
+            rationale=decision.rationale,
+            actor_reasoning=decision.actor_reasoning,
+            downstream_decision_ids_json=downstream_decision_ids_json,
+            downstream_plan_ids_json=downstream_plan_ids_json,
+            artifacts_produced_json=artifacts_produced_json,
+            created_at=created_at_str,
+            is_correction=decision.is_correction,
+            corrects_decision_id=decision.corrects_decision_id,
+            correction_reason=decision.correction_reason,
+            superseded_by=decision.superseded_by,
+        )
+
+
+class DecisionDependencyModel(Base):  # type: ignore[misc]
+    """Database model for the decision dependency DAG.
+
+    Maps to the ``decision_dependencies`` table.  Each row represents
+    an influence edge from one decision to another (the *source*
+    decision influenced or produced the *target* decision).
+    """
+
+    __allow_unmapped__ = True
+    __tablename__ = "decision_dependencies"
+
+    source_decision_id = Column(
+        String(26),
+        ForeignKey("decisions.decision_id", ondelete="CASCADE"),
+        primary_key=True,
+    )
+    target_decision_id = Column(
+        String(26),
+        ForeignKey("decisions.decision_id", ondelete="CASCADE"),
+        primary_key=True,
+    )
+    relationship_type = Column(
+        String(30),
+        nullable=False,
+        default="influences",
+    )
+    created_at = Column(String(30), nullable=False)
+
+    __table_args__ = (
+        CheckConstraint(
+            "source_decision_id != target_decision_id",
+            name="ck_decision_deps_no_self_loop",
+        ),
+        Index("ix_decision_deps_source", "source_decision_id"),
+        Index("ix_decision_deps_target", "target_decision_id"),
+    )
+
+
 # Database initialization functions
 def init_database(database_url: str = "sqlite:///.cleveragents/db.sqlite") -> Any:
     """Initialize the database.

src/cleveragents/infrastructure/database/repositories.py

@@ -11,6 +11,7 @@ Provides database-backed repositories following the session-factory pattern
 | ``ActionRepository``          | CRUD for actions + arguments/invariants    |
 | ``NamespacedProjectRepository``| CRUD for namespaced projects              |
 | ``ProjectResourceLinkRepository``| Project-resource link management        |
+| ``DecisionRepository``         | CRUD for decision tree nodes             |
 
 ## Session-Factory Pattern
 
@@ -43,6 +44,8 @@ with uow:
 | ``PlanNotFoundError``   | Plan ID not found                             |
 | ``ProjectNotFoundError``| Project not found                             |
 | ``DuplicateLinkError``  | Project-resource link already exists           |
+| ``DuplicateDecisionError``| Decision ID already exists                   |
+| ``DecisionNotFoundError``| Decision ID not found                         |
 
 Based on ADR-007 (Repository Pattern) and ADR-033 (Retry Patterns).
 """
@@ -51,10 +54,11 @@ from __future__ import annotations
 
 import json
 import warnings
+from collections import deque
 from collections.abc import Callable
 from datetime import UTC, datetime
 from pathlib import Path
-from typing import Any, ClassVar, cast
+from typing import TYPE_CHECKING, Any, ClassVar, cast
 
 from sqlalchemy.exc import DatabaseError as SQLAlchemyDatabaseError
 from sqlalchemy.exc import IntegrityError, OperationalError
@@ -81,6 +85,7 @@ from cleveragents.infrastructure.database.models import (
     ChangeModel,
     ContextModel,
     DebugAttemptModel,
+    DecisionModel,
     LifecycleActionModel,
     LifecyclePlanModel,
     NamespacedProjectModel,
@@ -103,6 +108,9 @@ from cleveragents.infrastructure.database.models import (
     ValidationAttachmentModel,
 )
 
+if TYPE_CHECKING:
+    from cleveragents.domain.models.core.decision import Decision
+
 
 class ProjectRepository:
     """Repository for project persistence."""
@@ -4570,3 +4578,375 @@ class SkillRepository:
         except (OperationalError, SQLAlchemyDatabaseError) as exc:
             session.rollback()
             raise DatabaseError(f"Failed to delete skill {name}: {exc}") from exc
+
+
+# ---------------------------------------------------------------------------
+# Decision repository (Stage M3.2)
+# ---------------------------------------------------------------------------
+
+
+class DuplicateDecisionError(DatabaseError):
+    """Raised when creating a decision with an ID that already exists."""
+
+    def __init__(self, decision_id: str) -> None:
+        super().__init__(f"Decision '{decision_id}' already exists")
+        self.decision_id = decision_id
+
+
+class DecisionNotFoundError(DatabaseError):
+    """Raised when a decision cannot be found."""
+
+    def __init__(self, decision_id: str) -> None:
+        super().__init__(f"Decision '{decision_id}' not found")
+        self.decision_id = decision_id
+
+
+class DecisionRepository:
+    """Repository for decision tree node persistence.
+
+    Uses the session-factory pattern (ADR-007).  All mutating methods
+    flush but do **not** commit; the caller or a :class:`UnitOfWork`
+    wrapper is responsible for committing the transaction.
+    """
+
+    def __init__(self, session_factory: Callable[[], Session]) -> None:
+        """Initialise with a callable that returns a SQLAlchemy Session."""
+        self._session_factory = session_factory
+
+    def _session(self) -> Session:
+        """Convenience helper to obtain a session."""
+        return self._session_factory()
+
+    # --- CREATE ------------------------------------------------------------
+
+    @database_retry
+    def create(self, decision: Decision) -> Decision:
+        """Persist a new decision.
+
+        Args:
+            decision: Domain :class:`Decision` instance.
+
+        Returns:
+            The same domain decision (pass-through for fluent API).
+
+        Raises:
+            DuplicateDecisionError: If the ``decision_id`` already exists.
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            db_model = DecisionModel.from_domain(decision)
+            session.add(db_model)
+            session.flush()
+            return decision
+        except IntegrityError as exc:
+            session.rollback()
+            if "UNIQUE" in str(exc).upper() or "unique" in str(exc).lower():
+                raise DuplicateDecisionError(
+                    str(decision.decision_id),
+                ) from exc
+            raise DatabaseError(
+                f"Failed to create decision: {exc}",
+            ) from exc
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(
+                f"Failed to create decision: {exc}",
+            ) from exc
+
+    # --- GET ---------------------------------------------------------------
+
+    @database_retry
+    def get(self, decision_id: str) -> Decision | None:
+        """Retrieve a single decision by ID.
+
+        Args:
+            decision_id: ULID of the decision.
+
+        Returns:
+            Domain :class:`Decision` or ``None`` if not found.
+
+        Raises:
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            row = (
+                session.query(DecisionModel).filter_by(decision_id=decision_id).first()
+            )
+            if row is None:
+                return None
+            return row.to_domain()
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            raise DatabaseError(
+                f"Failed to get decision {decision_id}: {exc}",
+            ) from exc
+
+    # --- GET BY PLAN -------------------------------------------------------
+
+    @database_retry
+    def get_by_plan(self, plan_id: str) -> list[Decision]:
+        """Retrieve all decisions for a plan, ordered by sequence number.
+
+        Args:
+            plan_id: ULID of the plan.
+
+        Returns:
+            List of domain :class:`Decision` instances.
+
+        Raises:
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            rows = (
+                session.query(DecisionModel)
+                .filter_by(plan_id=plan_id)
+                .order_by(DecisionModel.sequence_number)
+                .all()
+            )
+            return [r.to_domain() for r in rows]
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            raise DatabaseError(
+                f"Failed to get decisions for plan {plan_id}: {exc}",
+            ) from exc
+
+    # --- GET TREE ----------------------------------------------------------
+
+    @database_retry
+    def get_tree(self, root_decision_id: str) -> list[Decision]:
+        """Retrieve the full decision tree rooted at *root_decision_id*.
+
+        Performs a breadth-first walk using the ``parent_decision_id``
+        relationship.  Returns the root followed by its descendants in
+        BFS order.
+
+        Args:
+            root_decision_id: ULID of the root decision.
+
+        Returns:
+            List of domain :class:`Decision` instances (BFS order).
+
+        Raises:
+            DecisionNotFoundError: If the root decision does not exist.
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            # Fetch root
+            root_row = (
+                session.query(DecisionModel)
+                .filter_by(decision_id=root_decision_id)
+                .first()
+            )
+            if root_row is None:
+                raise DecisionNotFoundError(root_decision_id)
+
+            # BFS walk
+            result: list[Decision] = [root_row.to_domain()]
+            bfs_queue: deque[str] = deque([root_decision_id])
+            while bfs_queue:
+                parent_id = bfs_queue.popleft()
+                children = (
+                    session.query(DecisionModel)
+                    .filter_by(parent_decision_id=parent_id)
+                    .order_by(DecisionModel.sequence_number)
+                    .all()
+                )
+                for child in children:
+                    result.append(child.to_domain())
+                    bfs_queue.append(cast(str, child.decision_id))
+            return result
+        except DecisionNotFoundError:
+            raise
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            raise DatabaseError(
+                f"Failed to get decision tree {root_decision_id}: {exc}",
+            ) from exc
+
+    # --- GET PATH TO ROOT --------------------------------------------------
+
+    @database_retry
+    def get_path_to_root(self, decision_id: str) -> list[Decision]:
+        """Walk from *decision_id* up to the tree root.
+
+        Returns a list starting with the given decision and ending
+        with the root (``parent_decision_id is None``).
+
+        Args:
+            decision_id: ULID of the starting decision.
+
+        Returns:
+            List of domain :class:`Decision` instances (leaf → root).
+
+        Raises:
+            DecisionNotFoundError: If the starting decision is missing.
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            path: list[Decision] = []
+            current_id: str | None = decision_id
+            visited: set[str] = set()
+            while current_id is not None:
+                if current_id in visited:
+                    break  # safety: avoid cycles
+                visited.add(current_id)
+                row = (
+                    session.query(DecisionModel)
+                    .filter_by(decision_id=current_id)
+                    .first()
+                )
+                if row is None:
+                    if not path:
+                        raise DecisionNotFoundError(decision_id)
+                    break
+                path.append(row.to_domain())
+                current_id = cast(str | None, row.parent_decision_id)
+            return path
+        except DecisionNotFoundError:
+            raise
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            raise DatabaseError(
+                f"Failed to get path to root from {decision_id}: {exc}",
+            ) from exc
+
+    # --- GET SUPERSEDED ----------------------------------------------------
+
+    @database_retry
+    def get_superseded(self, plan_id: str) -> list[Decision]:
+        """Retrieve all superseded decisions for a plan.
+
+        Args:
+            plan_id: ULID of the plan.
+
+        Returns:
+            List of domain :class:`Decision` instances that have been
+            superseded (``superseded_by IS NOT NULL``).
+
+        Raises:
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            rows = (
+                session.query(DecisionModel)
+                .filter(
+                    DecisionModel.plan_id == plan_id,
+                    DecisionModel.superseded_by.isnot(None),
+                )
+                .order_by(DecisionModel.sequence_number)
+                .all()
+            )
+            return [r.to_domain() for r in rows]
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            raise DatabaseError(
+                f"Failed to get superseded decisions for plan {plan_id}: {exc}",
+            ) from exc
+
+    # --- UPDATE SUPERSEDED BY ----------------------------------------------
+
+    @database_retry
+    def update_superseded_by(
+        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:
+            Updated domain :class:`Decision`.
+
+        Raises:
+            DecisionNotFoundError: If *decision_id* does not exist.
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            row = (
+                session.query(DecisionModel).filter_by(decision_id=decision_id).first()
+            )
+            if row is None:
+                raise DecisionNotFoundError(decision_id)
+            row.superseded_by = new_decision_id  # type: ignore[assignment]
+            session.flush()
+            return row.to_domain()
+        except DecisionNotFoundError:
+            raise
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(
+                f"Failed to update superseded_by for {decision_id}: {exc}",
+            ) from exc
+
+    # --- LIST BY TYPE ------------------------------------------------------
+
+    @database_retry
+    def list_by_type(
+        self,
+        plan_id: str,
+        decision_type: str,
+    ) -> list[Decision]:
+        """Retrieve decisions of a specific type for a plan.
+
+        Args:
+            plan_id: ULID of the plan.
+            decision_type: String value of the :class:`DecisionType` enum.
+
+        Returns:
+            List of domain :class:`Decision` instances.
+
+        Raises:
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            rows = (
+                session.query(DecisionModel)
+                .filter(
+                    DecisionModel.plan_id == plan_id,
+                    DecisionModel.decision_type == decision_type,
+                )
+                .order_by(DecisionModel.sequence_number)
+                .all()
+            )
+            return [r.to_domain() for r in rows]
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            raise DatabaseError(
+                f"Failed to list decisions by type {decision_type}: {exc}",
+            ) from exc
+
+    # --- DELETE -------------------------------------------------------------
+
+    @database_retry
+    def delete(self, decision_id: str) -> bool:
+        """Delete a decision by ID.
+
+        Args:
+            decision_id: ULID of the decision to delete.
+
+        Returns:
+            ``True`` if the decision was deleted, ``False`` if not found.
+
+        Raises:
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            row = (
+                session.query(DecisionModel).filter_by(decision_id=decision_id).first()
+            )
+            if row is None:
+                return False
+            session.delete(row)
+            session.flush()
+            return True
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(
+                f"Failed to delete decision {decision_id}: {exc}",
+            ) from exc

src/cleveragents/infrastructure/database/unit_of_work.py

@@ -19,6 +19,7 @@ from cleveragents.infrastructure.database.repositories import (
     ChangeRepository,
     ContextRepository,
     DebugAttemptRepository,
+    DecisionRepository,
     LifecyclePlanRepository,
     PlanRepository,
     ProjectRepository,
@@ -184,6 +185,7 @@ class UnitOfWorkContext:
         self._actors: ActorRepository | None = None
         self._actions: ActionRepository | None = None
         self._lifecycle_plans: LifecyclePlanRepository | None = None
+        self._decisions: DecisionRepository | None = None
 
     def _session_factory(self) -> Session:
         """Return the transaction's session for factory-pattern repositories."""
@@ -259,6 +261,18 @@ class UnitOfWorkContext:
             self._actors = ActorRepository(self._session)
         return self._actors
 
+    @property
+    def decisions(self) -> DecisionRepository:
+        """Get decision repository for this transaction.
+
+        Uses the session-factory pattern required by DecisionRepository.
+        """
+        if self._decisions is None:
+            self._decisions = DecisionRepository(
+                session_factory=self._session_factory,
+            )
+        return self._decisions
+
     def add(self, entity: Any) -> None:
         """Add an entity to the session.