From 878bad4848c30c2d0b6803c0b10fe4b86c1e0b08 Mon Sep 17 00:00:00 2001 From: khyari hamza Date: Wed, 25 Feb 2026 13:11:42 +0000 Subject: [PATCH 1/9] feat(service): add decision recording and snapshot store --- benchmarks/decision_recording_bench.py | 169 ++++ docs/reference/decision_service.md | 148 ++++ features/decision_recording.feature | 232 +++++ features/steps/decision_recording_steps.py | 818 ++++++++++++++++++ robot/decision_recording.robot | 58 ++ robot/helper_decision_recording.py | 216 +++++ .../application/services/__init__.py | 6 + .../application/services/decision_service.py | 711 ++++++++++++--- 8 files changed, 2241 insertions(+), 117 deletions(-) create mode 100644 benchmarks/decision_recording_bench.py create mode 100644 docs/reference/decision_service.md create mode 100644 features/decision_recording.feature create mode 100644 features/steps/decision_recording_steps.py create mode 100644 robot/decision_recording.robot create mode 100644 robot/helper_decision_recording.py diff --git a/benchmarks/decision_recording_bench.py b/benchmarks/decision_recording_bench.py new file mode 100644 index 000000000..634bfe125 --- /dev/null +++ b/benchmarks/decision_recording_bench.py @@ -0,0 +1,169 @@ +"""ASV benchmarks for DecisionService recording throughput. + +Measures record_decision, list_decisions, get_tree, and snapshot operations. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +try: + from cleveragents.application.services.decision_service import ( + DecisionService, + SnapshotStore, + ) + from cleveragents.domain.models.core.decision import ( + ContextSnapshot, + DecisionType, + ) +except ModuleNotFoundError: + sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + from cleveragents.application.services.decision_service import ( + DecisionService, + SnapshotStore, + ) + from cleveragents.domain.models.core.decision import ( + ContextSnapshot, + DecisionType, + ) + + +_PLAN_ID = "01HV00000000000000BENCH002" + + +class TimeRecordDecision: + """Benchmark decision recording throughput.""" + + timeout = 60 + + def setup(self): + self.svc = DecisionService() + self._plan_counter = 0 + + def time_record_single(self): + self._plan_counter += 1 + plan_id = f"01HV00000000000000BN{self._plan_counter:06d}" + self.svc.record_decision( + plan_id=plan_id, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Benchmark question", + chosen_option="Benchmark choice", + ) + + +class TimeRecordBatch: + """Benchmark recording 50 decisions for one plan.""" + + timeout = 60 + + def time_record_50(self): + svc = DecisionService() + root = svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.PROMPT_DEFINITION, + question="Root", + chosen_option="Root choice", + ) + for _ in range(49): + svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Child", + chosen_option="Child choice", + parent_decision_id=root.decision_id, + ) + + +class TimeListDecisions: + """Benchmark list_decisions with 50 pre-seeded decisions.""" + + timeout = 60 + + def setup(self): + self.svc = DecisionService() + root = self.svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.PROMPT_DEFINITION, + question="Root", + chosen_option="Root choice", + ) + for _ in range(49): + self.svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Child", + chosen_option="Child choice", + parent_decision_id=root.decision_id, + ) + + def time_list_all(self): + self.svc.list_decisions(_PLAN_ID) + + def time_list_by_type(self): + self.svc.list_by_type(_PLAN_ID, DecisionType.STRATEGY_CHOICE) + + +class TimeGetTree: + """Benchmark get_tree with a 3-level tree (1 + 3 + 6 = 10 nodes).""" + + timeout = 60 + + def setup(self): + self.svc = DecisionService() + root = self.svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.PROMPT_DEFINITION, + question="Root", + chosen_option="Root choice", + ) + for _ in range(3): + child = self.svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Child", + chosen_option="Child choice", + parent_decision_id=root.decision_id, + ) + for _ in range(2): + self.svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.IMPLEMENTATION_CHOICE, + question="Grandchild", + chosen_option="GC choice", + parent_decision_id=child.decision_id, + ) + + def time_get_tree(self): + self.svc.get_tree(_PLAN_ID) + + +class TimeSnapshotStore: + """Benchmark SnapshotStore operations.""" + + timeout = 60 + + def setup(self): + self.store = SnapshotStore() + self._counter = 0 + # Pre-seed 100 snapshots + for i in range(100): + snap = ContextSnapshot( + hot_context_hash=f"sha256:bench{i:04d}", + hot_context_ref=f"ref-{i}", + ) + self.store.store(f"DEC_{i:06d}", snap) + + def time_store_snapshot(self): + self._counter += 1 + snap = ContextSnapshot( + hot_context_hash=f"sha256:new{self._counter:06d}", + hot_context_ref=f"ref-new-{self._counter}", + ) + self.store.store(f"NEW_{self._counter:06d}", snap) + + def time_get_by_hash(self): + self.store.get_by_hash("sha256:bench0050") + + def time_get_snapshot(self): + self.store.get("DEC_000050") diff --git a/docs/reference/decision_service.md b/docs/reference/decision_service.md new file mode 100644 index 000000000..2a27f889a --- /dev/null +++ b/docs/reference/decision_service.md @@ -0,0 +1,148 @@ +# Decision Service Reference + +## Overview + +`DecisionService` provides the application-layer interface for recording +decisions during plan execution, retrieving decision histories, and +managing context snapshots. It lives in +`src/cleveragents/application/services/decision_service.py`. + +## Dual-Mode Persistence + +| Mode | UnitOfWork | Storage | +|------------|------------|------------------------------------------------| +| In-memory | `None` | Internal dicts (`_decisions`, `_plan_decisions`) | +| Persisted | provided | DB via `DecisionRepository` + in-memory cache | + +When a `UnitOfWork` is wired, mutations are written to the database +first, then the in-memory cache is updated (write-through). + +## Constructor + +```python +DecisionService( + settings: Settings | None = None, + unit_of_work: UnitOfWork | None = None, +) +``` + +- **settings** – Application settings (optional for in-memory mode). +- **unit_of_work** – When provided, decisions are persisted via + `DecisionRepository`. + +## Recording + +### `record_decision()` + +```python +svc.record_decision( + plan_id="01HV...", + decision_type=DecisionType.STRATEGY_CHOICE, # or a string + question="Which approach?", + chosen_option="Build a REST API", + parent_decision_id=None, # optional + alternatives_considered=None, # list[str] + confidence_score=None, # 0.0-1.0 + rationale="", # human-readable + actor_reasoning=None, # raw LLM trace + context_snapshot=None, # auto-captured if omitted + artifacts_produced=None, # list[ArtifactRef] + is_correction=False, + corrects_decision_id=None, + correction_reason=None, +) +``` + +- Auto-assigns a monotonically increasing **sequence number** per plan. +- Auto-generates a **ULID** for `decision_id`. +- Auto-captures a **context snapshot** (SHA-256 hash) when no explicit + snapshot is provided. +- Stores the snapshot in the built-in `SnapshotStore`. + +**Raises:** + +- `ValidationError` – required fields missing or empty. +- `DuplicateDecisionError` – decision ID already exists. + +## Retrieval + +| Method | Description | +|---------------------------------------|--------------------------------------------| +| `get_decision(decision_id)` | Single decision by ULID | +| `list_decisions(plan_id)` | All decisions for a plan, ordered by seq | +| `list_by_type(plan_id, decision_type)`| Filtered by type, ordered by seq | + +All raise `DecisionNotFoundError` when the target decision does not +exist (where applicable). + +## Tree Operations + +| Method | Description | +|-------------------------------|------------------------------------------| +| `get_tree(plan_id)` | BFS from root(s), level by level | +| `get_path_to_root(decision_id)` | Walk from a decision up to the root | + +## Superseded Decisions + +| Method | Description | +|-----------------------------------------------|---------------------------------------| +| `mark_superseded(decision_id, new_decision_id)` | Mark a decision as superseded | +| `get_superseded(plan_id)` | List all superseded decisions | + +## Delete + +```python +svc.delete_decision(decision_id) # -> True +``` + +Removes from both persistence and the snapshot store. Raises +`DecisionNotFoundError` if not found. + +## Snapshot Store + +`SnapshotStore` manages `ContextSnapshot` objects keyed by decision ID +with hash-based deduplication. + +| Method | Description | +|---------------------------------|----------------------------------------------| +| `get_snapshot(decision_id)` | Retrieve snapshot for a decision | +| `get_snapshots_for_plan(plan_id)` | All snapshots for a plan's decisions | +| `snapshots.get_by_hash(hash)` | Decision IDs sharing the same context hash | +| `snapshots.store(id, snap)` | Store a snapshot (internal) | +| `snapshots.remove(id)` | Remove a snapshot | + +## Statistics + +| Method | Description | +|-------------------------------|----------------------------------| +| `count_decisions(plan_id)` | Number of decisions for a plan | +| `get_next_sequence(plan_id)` | Next available sequence number | + +## Custom Exceptions + +| Exception | Base | Description | +|--------------------------|--------------------------|-------------------------------------| +| `DuplicateDecisionError` | `BusinessRuleViolation` | Decision ID already exists | +| `DecisionNotFoundError` | `ResourceNotFoundError` | Decision not found | +| `SequenceConflictError` | `BusinessRuleViolation` | Sequence number already used | + +## Running Tests + +```bash +# Behave BDD tests +nox -s unit_tests -- features/decision_recording.feature + +# Robot Framework integration smoke tests +nox -s integration_tests + +# ASV benchmarks +nox -s benchmark +``` + +## Related + +- Domain model: `src/cleveragents/domain/models/core/decision.py` +- Persistence: `src/cleveragents/infrastructure/database/repositories.py` + (`DecisionRepository`) +- Database schema: `docs/reference/database_schema.md` +- ADR: `docs/adr/ADR-033-decision-recording-protocol.md` diff --git a/features/decision_recording.feature b/features/decision_recording.feature new file mode 100644 index 000000000..518571703 --- /dev/null +++ b/features/decision_recording.feature @@ -0,0 +1,232 @@ +Feature: Decision recording and snapshot store + As a developer + I want a DecisionService that records decisions with auto-sequencing and snapshots + So that plan decision trees are tracked with full context for replay + + Background: + Given a decision service + + # --- Recording --- + + Scenario: Record a root decision with auto-sequencing + When I record a prompt_definition decision for plan "P1" with question "What is the plan prompt?" + Then the dsvc decision should be recorded successfully + And the dsvc decision sequence number should be 0 + And the dsvc decision type should be "prompt_definition" + And the dsvc decision should be a root decision + And the dsvc decision should have a valid ULID as decision_id + And the dsvc decision context snapshot hash should not be empty + + Scenario: Record multiple decisions with monotonic sequencing + When I record a prompt_definition decision for plan "P1" with question "Plan prompt" + And I record a strategy_choice decision for plan "P1" with question "Which approach?" + And I record an implementation_choice decision for plan "P1" with question "How to implement?" + Then the dsvc plan "P1" should have 3 decisions + And the dsvc decisions should have sequence numbers 0 1 2 + + Scenario: Record a decision with explicit context snapshot + Given a context snapshot with hash "sha256:abc123" and ref "s3://bucket/ctx" + When I record a strategy_choice decision for plan "P1" with the explicit snapshot + Then the dsvc decision context snapshot hash should be "sha256:abc123" + And the dsvc decision context snapshot ref should be "s3://bucket/ctx" + + Scenario: Record a decision with confidence score + When I record a strategy_choice decision for plan "P1" with confidence 0.85 + Then the dsvc decision confidence score should be 0.85 + + Scenario: Record a correction decision + When I record a prompt_definition decision for plan "P1" with question "Original" + And I record a correction decision for plan "P1" correcting the first decision + Then the dsvc correction decision is_correction should be true + And the dsvc correction decision corrects_decision_id should match the first decision + + Scenario: Record a decision with alternatives + When I record a strategy_choice decision for plan "P1" with alternatives "Option A" "Option B" "Option C" + Then the dsvc decision should have 3 alternatives considered + + Scenario: Record a decision with artifacts + When I record an implementation_choice decision for plan "P1" with artifacts + Then the dsvc decision should have 2 artifacts produced + + Scenario: Recording with empty plan_id raises ValidationError + When I try to record a decision with empty plan_id + Then a dsvc validation error should be raised + And the dsvc error should mention "plan_id" + + Scenario: Recording with empty question raises ValidationError + When I try to record a decision with empty question + Then a dsvc validation error should be raised + And the dsvc error should mention "question" + + Scenario: Recording with empty chosen_option raises ValidationError + When I try to record a decision with empty chosen_option + Then a dsvc validation error should be raised + And the dsvc error should mention "chosen_option" + + Scenario: Auto-capture generates a snapshot hash + When I record a strategy_choice decision for plan "P1" with question "Test auto-capture" + Then the dsvc decision context snapshot hash should start with "sha256:" + + # --- Retrieval --- + + Scenario: Get a decision by ID + When I record a prompt_definition decision for plan "P1" with question "Test get" + And I get the decision by its ID + Then the dsvc retrieved decision should match the recorded decision + + Scenario: Get a non-existent decision raises DecisionNotFoundError + When I try to get decision "01NONEXISTENT00000000000000" + Then a dsvc decision not found error should be raised + + Scenario: List decisions for a plan returns all decisions ordered + When I record a prompt_definition decision for plan "P1" with question "First" + And I record a strategy_choice decision for plan "P1" with question "Second" + And I record an implementation_choice decision for plan "P1" with question "Third" + And I list decisions for plan "P1" + Then the dsvc decision list should have 3 entries + And the dsvc decision list should be ordered by sequence number + + Scenario: List decisions for empty plan returns empty list + When I list decisions for plan "P_EMPTY" + Then the dsvc decision list should have 0 entries + + Scenario: List decisions by type filters correctly + When I record a prompt_definition decision for plan "P1" with question "Root" + And I record a strategy_choice decision for plan "P1" with question "Strategy 1" + And I record a strategy_choice decision for plan "P1" with question "Strategy 2" + And I record an implementation_choice decision for plan "P1" with question "Impl" + And I filter decisions for plan "P1" by type "strategy_choice" + Then the dsvc decision list should have 2 entries + + # --- Tree operations --- + + Scenario: Get tree returns BFS order from root + When I record a prompt_definition decision for plan "P1" with question "Root" + And I record a strategy_choice decision for plan "P1" with parent as child "Child 1" + And I record a strategy_choice decision for plan "P1" with parent as child "Child 2" + And I get the tree for plan "P1" + Then the dsvc tree should have 3 decisions + And the dsvc first tree decision should be a root decision + + Scenario: Get tree for empty plan returns empty list + When I get the tree for plan "P_EMPTY" + Then the dsvc tree should have 0 decisions + + Scenario: Get path to root navigates upward + When I record a prompt_definition decision for plan "P1" with question "Root" + And I record a strategy_choice decision for plan "P1" with parent as child "Level 1" + And I record an implementation_choice decision for plan "P1" with second parent as child "Level 2" + And I get the path to root from the last decision + Then the dsvc path should have 3 decisions + And the dsvc first path decision should be the last recorded decision + And the dsvc last path decision should be the root + + Scenario: Get path to root for non-existent decision raises error + When I try to get path to root for decision "01NONEXISTENT00000000000000" + Then a dsvc decision not found error should be raised + + # --- Superseded decisions --- + + Scenario: Mark a decision as superseded + When I record a prompt_definition decision for plan "P1" with question "Original" + And I record a strategy_choice decision for plan "P1" with question "Replacement" + And I mark the first decision as superseded by the second + Then the dsvc first decision should be superseded + And the dsvc first decision superseded_by should match the second decision + + Scenario: Get superseded decisions for a plan + When I record a prompt_definition decision for plan "P1" with question "Original" + And I record a strategy_choice decision for plan "P1" with question "Replacement" + And I mark the first decision as superseded by the second + And I get superseded decisions for plan "P1" + Then the dsvc superseded list should have 1 entry + + Scenario: Mark non-existent decision raises DecisionNotFoundError + When I try to mark decision "01NONEXISTENT00000000000000" as superseded + Then a dsvc decision not found error should be raised + + # --- Delete --- + + Scenario: Delete a decision + When I record a prompt_definition decision for plan "P1" with question "To delete" + And I delete the recorded decision + Then the dsvc decision should be deleted + And the dsvc plan "P1" should have 0 decisions + + Scenario: Delete non-existent decision raises DecisionNotFoundError + When I try to delete decision "01NONEXISTENT00000000000000" + Then a dsvc decision not found error should be raised + + # --- Snapshot store --- + + Scenario: Snapshot is stored when decision is recorded + When I record a strategy_choice decision for plan "P1" with question "Snapshot test" + And I get the snapshot for the recorded decision + Then the dsvc snapshot should not be None + + Scenario: Get snapshots for plan returns all snapshots + When I record a prompt_definition decision for plan "P1" with question "First" + And I record a strategy_choice decision for plan "P1" with question "Second" + And I get snapshots for plan "P1" + Then the dsvc snapshots dict should have 2 entries + + Scenario: Snapshot hash deduplication index works + Given a context snapshot with hash "sha256:samehash" and ref "ref1" + When I record a strategy_choice decision for plan "P1" with the explicit snapshot + And I record a strategy_choice decision for plan "P2" with the same explicit snapshot + And I query the snapshot store by hash "sha256:samehash" + Then the dsvc hash query should return 2 decision IDs + + Scenario: Snapshot is removed when decision is deleted + When I record a prompt_definition decision for plan "P1" with question "Ephemeral" + And I delete the recorded decision + And I get the snapshot for the deleted decision + Then the dsvc snapshot should be None + + # --- Statistics --- + + Scenario: Count decisions for a plan + When I record a prompt_definition decision for plan "P1" with question "One" + And I record a strategy_choice decision for plan "P1" with question "Two" + Then the dsvc decision count for plan "P1" should be 2 + + Scenario: Get next sequence returns the expected value + Then the dsvc next sequence for plan "P_NEW" should be 0 + When I record a prompt_definition decision for plan "P_NEW" with question "First" + Then the dsvc next sequence for plan "P_NEW" should be 1 + + # --- DuplicateDecisionError --- + + Scenario: DuplicateDecisionError contains decision_id + Given a DuplicateDecisionError for decision "DUP123" + Then the dsvc duplicate error decision_id should be "DUP123" + And the dsvc duplicate error message should contain "DUP123" + + # --- SequenceConflictError --- + + Scenario: SequenceConflictError contains plan_id and sequence + Given a SequenceConflictError for plan "P1" sequence 5 + Then the dsvc sequence error plan_id should be "P1" + And the dsvc sequence error sequence_number should be 5 + And the dsvc sequence error message should contain "P1" + + # --- SnapshotStore standalone --- + + Scenario: SnapshotStore remove returns False for unknown decision + Given a standalone snapshot store + When I remove snapshot for decision "UNKNOWN" + Then the dsvc snapshot remove result should be False + + Scenario: SnapshotStore get returns None for unknown decision + Given a standalone snapshot store + When I get snapshot for decision "UNKNOWN" + Then the dsvc standalone snapshot should be None + + Scenario: SnapshotStore get_by_hash returns empty for unknown hash + Given a standalone snapshot store + When I query the standalone store by hash "sha256:nope" + Then the dsvc standalone hash query should return 0 decision IDs + + Scenario: DecisionService string type coercion works + When I record a decision for plan "P1" with string type "strategy_choice" + Then the dsvc decision type should be "strategy_choice" diff --git a/features/steps/decision_recording_steps.py b/features/steps/decision_recording_steps.py new file mode 100644 index 000000000..871e1e2f4 --- /dev/null +++ b/features/steps/decision_recording_steps.py @@ -0,0 +1,818 @@ +"""Step definitions for decision_recording.feature. + +All ``Then`` step texts are prefixed with ``dsvc`` (decision-service) to +avoid collisions with the existing ``decision_model_steps.py`` which +defines similar assertion patterns for the domain model layer. +""" + +from __future__ import annotations + +import re + +from behave import given, then, when +from behave.runner import Context +from ulid import ULID + +from cleveragents.application.services.decision_service import ( + DecisionNotFoundError, + DecisionService, + DuplicateDecisionError, + SequenceConflictError, + SnapshotStore, +) +from cleveragents.core.exceptions import ValidationError +from cleveragents.domain.models.core.decision import ( + ArtifactRef, + ContextSnapshot, + DecisionType, +) + + +def _resolve_plan_id(context: Context, symbolic_id: str) -> str: + """Map a human-readable plan name to a stable ULID. + + The Decision model requires plan_id to be a valid 26-char ULID. + Feature files use friendly names like ``"P1"``; this helper + lazily generates and caches a ULID for each symbolic name. + """ + registry: dict[str, str] = getattr(context, "_plan_id_registry", {}) + if symbolic_id not in registry: + registry[symbolic_id] = str(ULID()) + context._plan_id_registry = registry + return registry[symbolic_id] + + +# --------------------------------------------------------------------------- +# Background +# --------------------------------------------------------------------------- + + +@given("a decision service") +def step_decision_service(context: Context) -> None: + context.decision_service = DecisionService() + context.recorded_decisions = [] + context.decision_error = None + context.decision_result = None + context.decision_list = None + context.tree_result = None + context.path_result = None + context.superseded_result = None + context.snapshot_result = None + context.snapshots_dict = None + context.hash_query_result = None + context.delete_result = None + context.explicit_snapshot = None + context._plan_id_registry = {} + + +# --------------------------------------------------------------------------- +# Given +# --------------------------------------------------------------------------- + + +@given('a context snapshot with hash "{ctx_hash}" and ref "{ctx_ref}"') +def step_explicit_snapshot(context: Context, ctx_hash: str, ctx_ref: str) -> None: + context.explicit_snapshot = ContextSnapshot( + hot_context_hash=ctx_hash, + hot_context_ref=ctx_ref, + ) + + +@given('a DuplicateDecisionError for decision "{decision_id}"') +def step_duplicate_error(context: Context, decision_id: str) -> None: + context.decision_error = DuplicateDecisionError(decision_id) + + +@given('a SequenceConflictError for plan "{plan_id}" sequence {seq:d}') +def step_sequence_error(context: Context, plan_id: str, seq: int) -> None: + context.decision_error = SequenceConflictError(plan_id, seq) + + +@given("a standalone snapshot store") +def step_standalone_snapshot_store(context: Context) -> None: + context.standalone_store = SnapshotStore() + + +# --------------------------------------------------------------------------- +# When — Recording +# --------------------------------------------------------------------------- + + +@when( + 'I record a prompt_definition decision for plan "{plan_id}" ' + 'with question "{question}"' +) +def step_record_root(context: Context, plan_id: str, question: str) -> None: + svc = context.decision_service + d = svc.record_decision( + plan_id=_resolve_plan_id(context, plan_id), + decision_type=DecisionType.PROMPT_DEFINITION, + question=question, + chosen_option=question, + ) + context.decision_result = d + context.recorded_decisions.append(d) + + +@when( + 'I record a strategy_choice decision for plan "{plan_id}" ' + 'with question "{question}"' +) +def step_record_strategy(context: Context, plan_id: str, question: str) -> None: + svc = context.decision_service + parent_id = None + if context.recorded_decisions: + parent_id = context.recorded_decisions[0].decision_id + d = svc.record_decision( + plan_id=_resolve_plan_id(context, plan_id), + decision_type=DecisionType.STRATEGY_CHOICE, + question=question, + chosen_option=f"Chosen: {question}", + parent_decision_id=parent_id, + ) + context.decision_result = d + context.recorded_decisions.append(d) + + +@when( + 'I record an implementation_choice decision for plan "{plan_id}" ' + 'with question "{question}"' +) +def step_record_impl(context: Context, plan_id: str, question: str) -> None: + svc = context.decision_service + parent_id = None + if context.recorded_decisions: + parent_id = context.recorded_decisions[0].decision_id + d = svc.record_decision( + plan_id=_resolve_plan_id(context, plan_id), + decision_type=DecisionType.IMPLEMENTATION_CHOICE, + question=question, + chosen_option=f"Chosen: {question}", + parent_decision_id=parent_id, + ) + context.decision_result = d + context.recorded_decisions.append(d) + + +@when( + 'I record a strategy_choice decision for plan "{plan_id}" ' + "with the explicit snapshot" +) +def step_record_with_snapshot(context: Context, plan_id: str) -> None: + svc = context.decision_service + parent_id = None + if context.recorded_decisions: + parent_id = context.recorded_decisions[0].decision_id + d = svc.record_decision( + plan_id=_resolve_plan_id(context, plan_id), + decision_type=DecisionType.STRATEGY_CHOICE, + question="Snapshot question", + chosen_option="Snapshot choice", + parent_decision_id=parent_id, + context_snapshot=context.explicit_snapshot, + ) + context.decision_result = d + context.recorded_decisions.append(d) + + +@when( + 'I record a strategy_choice decision for plan "{plan_id}" ' + "with the same explicit snapshot" +) +def step_record_with_same_snapshot(context: Context, plan_id: str) -> None: + svc = context.decision_service + d = svc.record_decision( + plan_id=_resolve_plan_id(context, plan_id), + decision_type=DecisionType.STRATEGY_CHOICE, + question="Another snapshot question", + chosen_option="Another choice", + context_snapshot=context.explicit_snapshot, + ) + context.decision_result = d + context.recorded_decisions.append(d) + + +@when( + 'I record a strategy_choice decision for plan "{plan_id}" with confidence {score:g}' +) +def step_record_with_confidence(context: Context, plan_id: str, score: float) -> None: + svc = context.decision_service + d = svc.record_decision( + plan_id=_resolve_plan_id(context, plan_id), + decision_type=DecisionType.STRATEGY_CHOICE, + question="Confidence test", + chosen_option="Chosen", + confidence_score=score, + ) + context.decision_result = d + context.recorded_decisions.append(d) + + +@when( + 'I record a correction decision for plan "{plan_id}" correcting the first decision' +) +def step_record_correction(context: Context, plan_id: str) -> None: + svc = context.decision_service + first = context.recorded_decisions[0] + d = svc.record_decision( + plan_id=_resolve_plan_id(context, plan_id), + decision_type=DecisionType.STRATEGY_CHOICE, + question="Correction question", + chosen_option="Corrected choice", + parent_decision_id=first.parent_decision_id, + is_correction=True, + corrects_decision_id=first.decision_id, + correction_reason="Fixed mistake", + ) + context.decision_result = d + context.recorded_decisions.append(d) + + +@when( + 'I record a strategy_choice decision for plan "{plan_id}" ' + 'with alternatives "{a1}" "{a2}" "{a3}"' +) +def step_record_with_alternatives( + context: Context, plan_id: str, a1: str, a2: str, a3: str +) -> None: + svc = context.decision_service + d = svc.record_decision( + plan_id=_resolve_plan_id(context, plan_id), + decision_type=DecisionType.STRATEGY_CHOICE, + question="Alternatives test", + chosen_option="Chosen", + alternatives_considered=[a1, a2, a3], + ) + context.decision_result = d + context.recorded_decisions.append(d) + + +@when('I record an implementation_choice decision for plan "{plan_id}" with artifacts') +def step_record_with_artifacts(context: Context, plan_id: str) -> None: + svc = context.decision_service + d = svc.record_decision( + plan_id=_resolve_plan_id(context, plan_id), + decision_type=DecisionType.IMPLEMENTATION_CHOICE, + question="Artifact test", + chosen_option="Chosen", + artifacts_produced=[ + ArtifactRef(artifact_path="src/main.py", artifact_type="file"), + ArtifactRef(artifact_path="tests/test_main.py", artifact_type="file"), + ], + ) + context.decision_result = d + context.recorded_decisions.append(d) + + +@when('I record a decision for plan "{plan_id}" with string type "{dtype}"') +def step_record_string_type(context: Context, plan_id: str, dtype: str) -> None: + svc = context.decision_service + d = svc.record_decision( + plan_id=_resolve_plan_id(context, plan_id), + decision_type=dtype, + question="String type test", + chosen_option="Chosen", + ) + context.decision_result = d + context.recorded_decisions.append(d) + + +# --- Recording errors --- + + +@when("I try to record a decision with empty plan_id") +def step_try_empty_plan_id(context: Context) -> None: + try: + context.decision_service.record_decision( + plan_id="", + decision_type=DecisionType.STRATEGY_CHOICE, + question="Q", + chosen_option="A", + ) + context.decision_error = None + except ValidationError as exc: + context.decision_error = exc + + +@when("I try to record a decision with empty question") +def step_try_empty_question(context: Context) -> None: + try: + context.decision_service.record_decision( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV", + decision_type=DecisionType.STRATEGY_CHOICE, + question="", + chosen_option="A", + ) + context.decision_error = None + except ValidationError as exc: + context.decision_error = exc + + +@when("I try to record a decision with empty chosen_option") +def step_try_empty_chosen(context: Context) -> None: + try: + context.decision_service.record_decision( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV", + decision_type=DecisionType.STRATEGY_CHOICE, + question="Q", + chosen_option="", + ) + context.decision_error = None + except ValidationError as exc: + context.decision_error = exc + + +# --------------------------------------------------------------------------- +# When — Retrieval +# --------------------------------------------------------------------------- + + +@when("I get the decision by its ID") +def step_get_by_id(context: Context) -> None: + d = context.decision_result + context.decision_result = context.decision_service.get_decision(d.decision_id) + + +@when('I try to get decision "{decision_id}"') +def step_try_get(context: Context, decision_id: str) -> None: + try: + context.decision_service.get_decision(decision_id) + context.decision_error = None + except DecisionNotFoundError as exc: + context.decision_error = exc + + +@when('I list decisions for plan "{plan_id}"') +def step_list_decisions(context: Context, plan_id: str) -> None: + context.decision_list = context.decision_service.list_decisions( + _resolve_plan_id(context, plan_id) + ) + + +@when('I filter decisions for plan "{plan_id}" by type "{dtype}"') +def step_list_by_type(context: Context, plan_id: str, dtype: str) -> None: + context.decision_list = context.decision_service.list_by_type( + _resolve_plan_id(context, plan_id), dtype + ) + + +# --------------------------------------------------------------------------- +# When — Tree operations +# --------------------------------------------------------------------------- + + +@when( + 'I record a strategy_choice decision for plan "{plan_id}" ' + 'with parent as child "{question}"' +) +def step_record_child(context: Context, plan_id: str, question: str) -> None: + svc = context.decision_service + parent = context.recorded_decisions[0] + d = svc.record_decision( + plan_id=_resolve_plan_id(context, plan_id), + decision_type=DecisionType.STRATEGY_CHOICE, + question=question, + chosen_option=f"Chosen: {question}", + parent_decision_id=parent.decision_id, + ) + context.decision_result = d + context.recorded_decisions.append(d) + + +@when( + 'I record an implementation_choice decision for plan "{plan_id}" ' + 'with second parent as child "{question}"' +) +def step_record_grandchild(context: Context, plan_id: str, question: str) -> None: + svc = context.decision_service + # Use the second recorded decision (index 1) as parent + parent = context.recorded_decisions[1] + d = svc.record_decision( + plan_id=_resolve_plan_id(context, plan_id), + decision_type=DecisionType.IMPLEMENTATION_CHOICE, + question=question, + chosen_option=f"Chosen: {question}", + parent_decision_id=parent.decision_id, + ) + context.decision_result = d + context.recorded_decisions.append(d) + + +@when('I get the tree for plan "{plan_id}"') +def step_get_tree(context: Context, plan_id: str) -> None: + context.tree_result = context.decision_service.get_tree( + _resolve_plan_id(context, plan_id) + ) + + +@when("I get the path to root from the last decision") +def step_get_path(context: Context) -> None: + last = context.recorded_decisions[-1] + context.path_result = context.decision_service.get_path_to_root(last.decision_id) + + +@when('I try to get path to root for decision "{decision_id}"') +def step_try_path(context: Context, decision_id: str) -> None: + try: + context.decision_service.get_path_to_root(decision_id) + context.decision_error = None + except DecisionNotFoundError as exc: + context.decision_error = exc + + +# --------------------------------------------------------------------------- +# When — Superseded +# --------------------------------------------------------------------------- + + +@when("I mark the first decision as superseded by the second") +def step_mark_superseded(context: Context) -> None: + first = context.recorded_decisions[0] + second = context.recorded_decisions[1] + context.decision_result = context.decision_service.mark_superseded( + first.decision_id, second.decision_id + ) + # Update stored reference + context.recorded_decisions[0] = context.decision_result + + +@when('I get superseded decisions for plan "{plan_id}"') +def step_get_superseded(context: Context, plan_id: str) -> None: + context.superseded_result = context.decision_service.get_superseded( + _resolve_plan_id(context, plan_id) + ) + + +@when('I try to mark decision "{decision_id}" as superseded') +def step_try_supersede(context: Context, decision_id: str) -> None: + try: + context.decision_service.mark_superseded( + decision_id, "01FAKE0000000000000000000" + ) + context.decision_error = None + except DecisionNotFoundError as exc: + context.decision_error = exc + + +# --------------------------------------------------------------------------- +# When — Delete +# --------------------------------------------------------------------------- + + +@when("I delete the recorded decision") +def step_delete(context: Context) -> None: + d = context.decision_result + context.delete_result = context.decision_service.delete_decision(d.decision_id) + context.deleted_decision_id = d.decision_id + + +@when('I try to delete decision "{decision_id}"') +def step_try_delete(context: Context, decision_id: str) -> None: + try: + context.decision_service.delete_decision(decision_id) + context.decision_error = None + except DecisionNotFoundError as exc: + context.decision_error = exc + + +# --------------------------------------------------------------------------- +# When — Snapshots +# --------------------------------------------------------------------------- + + +@when("I get the snapshot for the recorded decision") +def step_get_snapshot(context: Context) -> None: + d = context.decision_result + context.snapshot_result = context.decision_service.get_snapshot(d.decision_id) + + +@when("I get the snapshot for the deleted decision") +def step_get_snapshot_deleted(context: Context) -> None: + context.snapshot_result = context.decision_service.get_snapshot( + context.deleted_decision_id + ) + + +@when('I get snapshots for plan "{plan_id}"') +def step_get_plan_snapshots(context: Context, plan_id: str) -> None: + context.snapshots_dict = context.decision_service.get_snapshots_for_plan( + _resolve_plan_id(context, plan_id) + ) + + +@when('I query the snapshot store by hash "{ctx_hash}"') +def step_query_hash(context: Context, ctx_hash: str) -> None: + context.hash_query_result = context.decision_service.snapshots.get_by_hash(ctx_hash) + + +# --- Standalone snapshot store --- + + +@when('I remove snapshot for decision "{decision_id}"') +def step_standalone_remove(context: Context, decision_id: str) -> None: + context.snapshot_remove_result = context.standalone_store.remove(decision_id) + + +@when('I get snapshot for decision "{decision_id}"') +def step_standalone_get(context: Context, decision_id: str) -> None: + context.standalone_snapshot = context.standalone_store.get(decision_id) + + +@when('I query the standalone store by hash "{ctx_hash}"') +def step_standalone_hash(context: Context, ctx_hash: str) -> None: + context.standalone_hash_result = context.standalone_store.get_by_hash(ctx_hash) + + +# --------------------------------------------------------------------------- +# Then — Recording assertions (dsvc-prefixed) +# --------------------------------------------------------------------------- + + +@then("the dsvc decision should be recorded successfully") +def step_recorded(context: Context) -> None: + assert context.decision_result is not None + + +@then("the dsvc decision sequence number should be {seq:d}") +def step_seq_number(context: Context, seq: int) -> None: + assert context.decision_result.sequence_number == seq + + +@then('the dsvc decision type should be "{dtype}"') +def step_decision_type(context: Context, dtype: str) -> None: + assert str(context.decision_result.decision_type) == dtype + + +@then("the dsvc decision should be a root decision") +def step_is_root(context: Context) -> None: + assert context.decision_result.is_root + + +@then("the dsvc decision should have a valid ULID as decision_id") +def step_valid_ulid(context: Context) -> None: + ulid_re = re.compile(r"^[0-9A-HJKMNP-TV-Z]{26}$") + assert ulid_re.match(context.decision_result.decision_id) + + +@then("the dsvc decision context snapshot hash should not be empty") +def step_snapshot_hash_not_empty(context: Context) -> None: + assert context.decision_result.context_snapshot.hot_context_hash != "" + + +@then('the dsvc plan "{plan_id}" should have {count:d} decisions') +def step_plan_count(context: Context, plan_id: str, count: int) -> None: + assert ( + context.decision_service.count_decisions(_resolve_plan_id(context, plan_id)) + == count + ) + + +@then("the dsvc decisions should have sequence numbers {nums}") +def step_seq_numbers(context: Context, nums: str) -> None: + expected = [int(n) for n in nums.split()] + actual = [d.sequence_number for d in context.recorded_decisions] + assert actual == expected, f"Expected {expected}, got {actual}" + + +@then('the dsvc decision context snapshot hash should be "{expected}"') +def step_snapshot_hash_exact(context: Context, expected: str) -> None: + assert context.decision_result.context_snapshot.hot_context_hash == expected + + +@then('the dsvc decision context snapshot ref should be "{expected}"') +def step_snapshot_ref_exact(context: Context, expected: str) -> None: + assert context.decision_result.context_snapshot.hot_context_ref == expected + + +@then("the dsvc decision confidence score should be {score:g}") +def step_confidence(context: Context, score: float) -> None: + assert context.decision_result.confidence_score == score + + +@then("the dsvc correction decision is_correction should be true") +def step_is_correction(context: Context) -> None: + assert context.decision_result.is_correction + + +@then( + "the dsvc correction decision corrects_decision_id should match the first decision" +) +def step_corrects_first(context: Context) -> None: + first = context.recorded_decisions[0] + assert context.decision_result.corrects_decision_id == first.decision_id + + +@then("the dsvc decision should have {count:d} alternatives considered") +def step_alternatives_count(context: Context, count: int) -> None: + assert len(context.decision_result.alternatives_considered) == count + + +@then("the dsvc decision should have {count:d} artifacts produced") +def step_artifacts_count(context: Context, count: int) -> None: + assert len(context.decision_result.artifacts_produced) == count + + +@then('the dsvc decision context snapshot hash should start with "{prefix}"') +def step_snapshot_hash_prefix(context: Context, prefix: str) -> None: + assert context.decision_result.context_snapshot.hot_context_hash.startswith(prefix) + + +# --------------------------------------------------------------------------- +# Then — Validation errors (dsvc-prefixed) +# --------------------------------------------------------------------------- + + +@then("a dsvc validation error should be raised") +def step_validation_error(context: Context) -> None: + assert context.decision_error is not None + assert isinstance(context.decision_error, (ValidationError, ValueError)) + + +@then('the dsvc error should mention "{text}"') +def step_error_mention(context: Context, text: str) -> None: + assert text in str(context.decision_error) + + +@then("a dsvc decision not found error should be raised") +def step_not_found_error(context: Context) -> None: + assert context.decision_error is not None + assert isinstance(context.decision_error, DecisionNotFoundError) + + +# --------------------------------------------------------------------------- +# Then — Retrieval assertions (dsvc-prefixed) +# --------------------------------------------------------------------------- + + +@then("the dsvc retrieved decision should match the recorded decision") +def step_match_recorded(context: Context) -> None: + recorded = context.recorded_decisions[-1] + assert context.decision_result.decision_id == recorded.decision_id + + +@then("the dsvc decision list should have {count:d} entries") +def step_list_count(context: Context, count: int) -> None: + assert len(context.decision_list) == count + + +@then("the dsvc decision list should be ordered by sequence number") +def step_list_ordered(context: Context) -> None: + seqs = [d.sequence_number for d in context.decision_list] + assert seqs == sorted(seqs) + + +# --------------------------------------------------------------------------- +# Then — Tree assertions (dsvc-prefixed) +# --------------------------------------------------------------------------- + + +@then("the dsvc tree should have {count:d} decisions") +def step_tree_count(context: Context, count: int) -> None: + assert len(context.tree_result) == count + + +@then("the dsvc first tree decision should be a root decision") +def step_tree_root(context: Context) -> None: + assert context.tree_result[0].is_root + + +@then("the dsvc path should have {count:d} decisions") +def step_path_count(context: Context, count: int) -> None: + assert len(context.path_result) == count + + +@then("the dsvc first path decision should be the last recorded decision") +def step_path_first(context: Context) -> None: + last = context.recorded_decisions[-1] + assert context.path_result[0].decision_id == last.decision_id + + +@then("the dsvc last path decision should be the root") +def step_path_last_root(context: Context) -> None: + assert context.path_result[-1].is_root + + +# --------------------------------------------------------------------------- +# Then — Superseded assertions (dsvc-prefixed) +# --------------------------------------------------------------------------- + + +@then("the dsvc first decision should be superseded") +def step_first_superseded(context: Context) -> None: + assert context.recorded_decisions[0].is_superseded + + +@then("the dsvc first decision superseded_by should match the second decision") +def step_superseded_by(context: Context) -> None: + first = context.recorded_decisions[0] + second = context.recorded_decisions[1] + assert first.superseded_by == second.decision_id + + +@then("the dsvc superseded list should have {count:d} entry") +def step_superseded_count(context: Context, count: int) -> None: + assert len(context.superseded_result) == count + + +# --------------------------------------------------------------------------- +# Then — Delete assertions (dsvc-prefixed) +# --------------------------------------------------------------------------- + + +@then("the dsvc decision should be deleted") +def step_deleted(context: Context) -> None: + assert context.delete_result is True + + +# --------------------------------------------------------------------------- +# Then — Snapshot assertions (dsvc-prefixed) +# --------------------------------------------------------------------------- + + +@then("the dsvc snapshot should not be None") +def step_snapshot_not_none(context: Context) -> None: + assert context.snapshot_result is not None + + +@then("the dsvc snapshot should be None") +def step_snapshot_none(context: Context) -> None: + assert context.snapshot_result is None + + +@then("the dsvc snapshots dict should have {count:d} entries") +def step_snapshots_dict_count(context: Context, count: int) -> None: + assert len(context.snapshots_dict) == count + + +@then("the dsvc hash query should return {count:d} decision IDs") +def step_hash_count(context: Context, count: int) -> None: + assert len(context.hash_query_result) == count + + +# --------------------------------------------------------------------------- +# Then — Statistics assertions (dsvc-prefixed) +# --------------------------------------------------------------------------- + + +@then('the dsvc decision count for plan "{plan_id}" should be {count:d}') +def step_count(context: Context, plan_id: str, count: int) -> None: + assert ( + context.decision_service.count_decisions(_resolve_plan_id(context, plan_id)) + == count + ) + + +@then('the dsvc next sequence for plan "{plan_id}" should be {seq:d}') +def step_next_seq(context: Context, plan_id: str, seq: int) -> None: + assert ( + context.decision_service.get_next_sequence(_resolve_plan_id(context, plan_id)) + == seq + ) + + +# --------------------------------------------------------------------------- +# Then — Error object assertions (dsvc-prefixed) +# --------------------------------------------------------------------------- + + +@then('the dsvc duplicate error decision_id should be "{decision_id}"') +def step_dup_id(context: Context, decision_id: str) -> None: + assert context.decision_error.decision_id == decision_id + + +@then('the dsvc duplicate error message should contain "{text}"') +def step_dup_msg(context: Context, text: str) -> None: + assert text in str(context.decision_error) + + +@then('the dsvc sequence error plan_id should be "{plan_id}"') +def step_seq_err_plan(context: Context, plan_id: str) -> None: + assert context.decision_error.plan_id == plan_id + + +@then("the dsvc sequence error sequence_number should be {seq:d}") +def step_seq_err_num(context: Context, seq: int) -> None: + assert context.decision_error.sequence_number == seq + + +@then('the dsvc sequence error message should contain "{text}"') +def step_seq_err_msg(context: Context, text: str) -> None: + assert text in str(context.decision_error) + + +# --- Standalone snapshot store (dsvc-prefixed) --- + + +@then("the dsvc snapshot remove result should be False") +def step_standalone_remove_false(context: Context) -> None: + assert context.snapshot_remove_result is False + + +@then("the dsvc standalone snapshot should be None") +def step_standalone_none(context: Context) -> None: + assert context.standalone_snapshot is None + + +@then("the dsvc standalone hash query should return {count:d} decision IDs") +def step_standalone_hash_count(context: Context, count: int) -> None: + assert len(context.standalone_hash_result) == count diff --git a/robot/decision_recording.robot b/robot/decision_recording.robot new file mode 100644 index 000000000..8876c2999 --- /dev/null +++ b/robot/decision_recording.robot @@ -0,0 +1,58 @@ +*** Settings *** +Documentation Smoke tests for DecisionService recording and snapshot store +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER_SCRIPT} robot/helper_decision_recording.py + +*** Test Cases *** +Record And Retrieve Decision + [Documentation] Record a decision via DecisionService and retrieve by ID + [Tags] service decision recording + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} record-retrieve cwd=${WORKSPACE} timeout=30s + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} record-retrieve-ok + +Record Multiple With Sequencing + [Documentation] Record 3 decisions and verify monotonic sequence numbers + [Tags] service decision sequencing + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} sequencing cwd=${WORKSPACE} timeout=30s + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} sequencing-ok + +Snapshot Auto-Capture + [Documentation] Record a decision and verify context snapshot is auto-captured + [Tags] service decision snapshot + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} snapshot-capture cwd=${WORKSPACE} timeout=30s + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} snapshot-capture-ok + +Decision Tree BFS Via Service + [Documentation] Build a 3-node tree and verify BFS order via get_tree + [Tags] service decision tree + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} tree-bfs cwd=${WORKSPACE} timeout=30s + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} tree-bfs-ok + +Mark Decision Superseded Via Service + [Documentation] Mark a decision as superseded via the service layer + [Tags] service decision superseded + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} superseded cwd=${WORKSPACE} timeout=30s + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} superseded-ok + +Delete Decision Via Service + [Documentation] Delete a decision and verify snapshot is also removed + [Tags] service decision delete + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} delete cwd=${WORKSPACE} timeout=30s + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} delete-ok + +Snapshot Hash Deduplication + [Documentation] Verify hash-based deduplication in SnapshotStore + [Tags] service decision snapshot dedup + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} snapshot-dedup cwd=${WORKSPACE} timeout=30s + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} snapshot-dedup-ok diff --git a/robot/helper_decision_recording.py b/robot/helper_decision_recording.py new file mode 100644 index 000000000..9a2c70e38 --- /dev/null +++ b/robot/helper_decision_recording.py @@ -0,0 +1,216 @@ +"""Helper script for Robot Framework decision recording smoke tests. + +Usage: + python robot/helper_decision_recording.py + +Subcommands: + record-retrieve Record + retrieve a decision via DecisionService + sequencing Verify monotonic sequencing + snapshot-capture Verify auto-captured context snapshot + tree-bfs Build a tree and verify BFS via get_tree + superseded Mark a decision as superseded via service + delete Delete a decision and verify snapshot removal + snapshot-dedup Verify hash-based deduplication in SnapshotStore +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +# Ensure src is importable when run from workspace root +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from cleveragents.application.services.decision_service import ( + DecisionService, + SnapshotStore, +) +from cleveragents.domain.models.core.decision import ( + ContextSnapshot, + DecisionType, +) + +_PLAN_ID = "01HV000000000000000000RS01" + + +# --------------------------------------------------------------------------- +# Subcommands +# --------------------------------------------------------------------------- + + +def _record_retrieve(): + svc = DecisionService() + d = svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.PROMPT_DEFINITION, + question="What approach?", + chosen_option="Build API", + ) + got = svc.get_decision(d.decision_id) + assert got.decision_id == d.decision_id + assert got.is_root + assert str(got.decision_type) == "prompt_definition" + print("record-retrieve-ok") + + +def _sequencing(): + svc = DecisionService() + d0 = svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.PROMPT_DEFINITION, + question="First", + chosen_option="Chosen first", + ) + d1 = svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Second", + chosen_option="Chosen second", + parent_decision_id=d0.decision_id, + ) + d2 = svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.IMPLEMENTATION_CHOICE, + question="Third", + chosen_option="Chosen third", + parent_decision_id=d0.decision_id, + ) + assert d0.sequence_number == 0 + assert d1.sequence_number == 1 + assert d2.sequence_number == 2 + assert svc.count_decisions(_PLAN_ID) == 3 + print("sequencing-ok") + + +def _snapshot_capture(): + svc = DecisionService() + d = svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Snapshot test", + chosen_option="Chosen", + ) + snap = svc.get_snapshot(d.decision_id) + assert snap is not None + assert snap.hot_context_hash.startswith("sha256:") + print("snapshot-capture-ok") + + +def _tree_bfs(): + svc = DecisionService() + root = svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.PROMPT_DEFINITION, + question="Root", + chosen_option="Root choice", + ) + svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Child 1", + chosen_option="C1", + parent_decision_id=root.decision_id, + ) + svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Child 2", + chosen_option="C2", + parent_decision_id=root.decision_id, + ) + tree = svc.get_tree(_PLAN_ID) + assert len(tree) == 3, f"expected 3, got {len(tree)}" + assert tree[0].is_root + print("tree-bfs-ok") + + +def _superseded(): + svc = DecisionService() + d1 = svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.PROMPT_DEFINITION, + question="Original", + chosen_option="Orig", + ) + d2 = svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Replacement", + chosen_option="New", + parent_decision_id=d1.decision_id, + ) + updated = svc.mark_superseded(d1.decision_id, d2.decision_id) + assert updated.is_superseded + assert updated.superseded_by == d2.decision_id + superseded = svc.get_superseded(_PLAN_ID) + assert len(superseded) == 1 + print("superseded-ok") + + +def _delete(): + svc = DecisionService() + d = svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.PROMPT_DEFINITION, + question="Ephemeral", + chosen_option="Gone", + ) + did = d.decision_id + assert svc.get_snapshot(did) is not None + result = svc.delete_decision(did) + assert result is True + assert svc.get_snapshot(did) is None + assert svc.count_decisions(_PLAN_ID) == 0 + print("delete-ok") + + +def _snapshot_dedup(): + store = SnapshotStore() + snap = ContextSnapshot( + hot_context_hash="sha256:samehash", + hot_context_ref="ref1", + ) + store.store("DEC_A", snap) + store.store("DEC_B", snap) + + ids = store.get_by_hash("sha256:samehash") + assert len(ids) == 2, f"expected 2, got {len(ids)}" + assert "DEC_A" in ids + assert "DEC_B" in ids + + # Remove one + store.remove("DEC_A") + ids2 = store.get_by_hash("sha256:samehash") + assert len(ids2) == 1 + assert "DEC_B" in ids2 + + print("snapshot-dedup-ok") + + +# --------------------------------------------------------------------------- +# Dispatcher +# --------------------------------------------------------------------------- + +_COMMANDS = { + "record-retrieve": _record_retrieve, + "sequencing": _sequencing, + "snapshot-capture": _snapshot_capture, + "tree-bfs": _tree_bfs, + "superseded": _superseded, + "delete": _delete, + "snapshot-dedup": _snapshot_dedup, +} + + +def main(): + if len(sys.argv) < 2: + raise SystemExit(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>") + command = sys.argv[1] + handler = _COMMANDS.get(command) + if handler is None: + raise SystemExit(f"Unknown command: {command}") + handler() + + +if __name__ == "__main__": + main() diff --git a/src/cleveragents/application/services/__init__.py b/src/cleveragents/application/services/__init__.py index de3faaf46..52c2cc8eb 100644 --- a/src/cleveragents/application/services/__init__.py +++ b/src/cleveragents/application/services/__init__.py @@ -16,7 +16,10 @@ from cleveragents.application.services.correction_service import ( CorrectionService, ) from cleveragents.application.services.decision_service import ( + DecisionNotFoundError, DecisionService, + DuplicateDecisionError, + SnapshotStore, ) from cleveragents.application.services.invariant_service import ( InvariantService, @@ -117,9 +120,11 @@ __all__ = [ "ConfigService", "ContextFragment", "CorrectionService", + "DecisionNotFoundError", "DecisionService", "DefaultValidationRunner", "DependencyCycleRule", + "DuplicateDecisionError", "DuplicateImportRule", "FileMergeOutcome", "InvariantService", @@ -142,6 +147,7 @@ __all__ = [ "SemanticValidationSeverity", "SkeletonCompressorService", "SkillRegistryService", + "SnapshotStore", "SpawnEntry", "SpawnMetadata", "SpawnResult", diff --git a/src/cleveragents/application/services/decision_service.py b/src/cleveragents/application/services/decision_service.py index 59fd951fd..e3f3f2618 100644 --- a/src/cleveragents/application/services/decision_service.py +++ b/src/cleveragents/application/services/decision_service.py @@ -1,210 +1,687 @@ -"""Application-layer service for decision tree operations. +"""Decision recording and snapshot store service. -``DecisionService`` wraps the :class:`DecisionRepository` behind a -thin application-layer façade, adding structured logging, transaction -management via :class:`UnitOfWork`, and a consistent API surface for -callers (e.g. ``PlanLifecycleService``). +``DecisionService`` provides a clean interface for recording decisions +during plan execution, retrieving decision histories, and managing +context snapshots. + +Dual-mode persistence +--------------------- + +| Mode | UnitOfWork | Storage | +|------------|------------|-----------------------------------------| +| In-memory | ``None`` | ``self._decisions`` / ``self._snapshots`` dicts | +| Persisted | provided | DB via ``DecisionRepository`` + in-memory cache | + +In persisted mode the in-memory dicts act as a write-through cache: +mutations are written to DB first, then the cache is updated. Based on: - - ADR-007 (Repository Pattern / Unit of Work) - - ADR-033 (Decision Recording Protocol) - - Forgejo issue #173 (Wire decision services into DI) + - Forgejo issue #172 + - docs/specification.md L18345-L18521 + - docs/adr/ADR-033-decision-recording-protocol.md """ from __future__ import annotations -from typing import TYPE_CHECKING +import hashlib +import json +from collections import deque +from typing import TYPE_CHECKING, Any import structlog -from cleveragents.domain.models.core.decision import Decision +from cleveragents.core.exceptions import ( + BusinessRuleViolation, + ResourceNotFoundError, + ValidationError, +) +from cleveragents.domain.models.core.decision import ( + ContextSnapshot, + Decision, + DecisionType, +) if TYPE_CHECKING: + from cleveragents.config.settings import Settings from cleveragents.infrastructure.database.unit_of_work import UnitOfWork logger = structlog.get_logger(__name__) +# --------------------------------------------------------------------------- +# Custom exceptions +# --------------------------------------------------------------------------- + + +class DuplicateDecisionError(BusinessRuleViolation): + """Raised when a decision with the same ID already exists.""" + + def __init__(self, decision_id: str) -> None: + super().__init__(f"Decision '{decision_id}' already exists") + self.decision_id = decision_id + + +class DecisionNotFoundError(ResourceNotFoundError): + """Raised when a decision cannot be found.""" + + def __init__(self, decision_id: str) -> None: + super().__init__( + resource_type="decision", + resource_id=decision_id, + ) + + +class SequenceConflictError(BusinessRuleViolation): + """Raised when a sequence number is already in use for a plan.""" + + def __init__(self, plan_id: str, sequence_number: int) -> None: + super().__init__( + f"Sequence number {sequence_number} already exists for plan '{plan_id}'" + ) + self.plan_id = plan_id + self.sequence_number = sequence_number + + +# --------------------------------------------------------------------------- +# Snapshot store +# --------------------------------------------------------------------------- + + +class SnapshotStore: + """In-memory store for context snapshots keyed by decision_id. + + Snapshots are stored alongside decisions but this dedicated store + provides hash-based deduplication and retrieval helpers. + """ + + def __init__(self) -> None: + self._snapshots: dict[str, ContextSnapshot] = {} + self._hash_index: dict[str, list[str]] = {} + + def store(self, decision_id: str, snapshot: ContextSnapshot) -> None: + """Store a snapshot for a given decision. + + Args: + decision_id: ULID of the decision. + snapshot: The context snapshot to store. + """ + self._snapshots[decision_id] = snapshot + h = snapshot.hot_context_hash + if h: + self._hash_index.setdefault(h, []).append(decision_id) + + def get(self, decision_id: str) -> ContextSnapshot | None: + """Retrieve a snapshot by decision ID. + + Returns: + The snapshot, or ``None`` if not found. + """ + return self._snapshots.get(decision_id) + + def get_by_hash(self, context_hash: str) -> list[str]: + """Return decision IDs sharing the same context hash. + + Args: + context_hash: The hot_context_hash to look up. + + Returns: + List of decision IDs with matching hash. + """ + return list(self._hash_index.get(context_hash, [])) + + def remove(self, decision_id: str) -> bool: + """Remove a snapshot. + + Returns: + ``True`` if a snapshot was removed, ``False`` otherwise. + """ + snapshot = self._snapshots.pop(decision_id, None) + if snapshot is None: + return False + h = snapshot.hot_context_hash + if h and h in self._hash_index: + ids = self._hash_index[h] + if decision_id in ids: + ids.remove(decision_id) + if not ids: + del self._hash_index[h] + return True + + def list_for_plan(self, decisions: list[Decision]) -> dict[str, ContextSnapshot]: + """Return snapshots for a list of decisions. + + Args: + decisions: Decisions to look up snapshots for. + + Returns: + Dict mapping decision_id → snapshot for found entries. + """ + result: dict[str, ContextSnapshot] = {} + for d in decisions: + snap = self._snapshots.get(d.decision_id) + if snap is not None: + result[d.decision_id] = snap + return result + + +# --------------------------------------------------------------------------- +# Decision service +# --------------------------------------------------------------------------- + + class DecisionService: - """Coordinate decision persistence through the Unit of Work. + """Service for recording decisions and managing context snapshots. - All public methods open a transaction, delegate to - :class:`DecisionRepository`, and commit. Callers do not need to - manage sessions directly. - - Args: - settings: Application settings (unused currently but kept for - consistency with the service constructor convention). - unit_of_work: A :class:`UnitOfWork` instance used to open - transactional scopes. + Provides record, list, tree, and path-to-root helpers for the + decision subsystem. Supports dual-mode persistence: in-memory + when no UnitOfWork is provided, or DB-backed via + ``DecisionRepository`` when a UnitOfWork is wired. """ def __init__( self, - settings: object, - unit_of_work: UnitOfWork, + settings: Settings | None = None, + unit_of_work: UnitOfWork | None = None, ) -> None: + """Initialize the decision service. + + Args: + settings: Application settings (optional for in-memory mode). + unit_of_work: Unit of Work for database transactions. + When provided, decisions are persisted via + ``DecisionRepository``. When ``None``, the service + falls back to in-memory storage. + """ self.settings = settings self.unit_of_work = unit_of_work self._logger = logger.bind(service="decision") + # In-memory storage (used as cache in persisted mode, + # or as primary storage when no UoW) + self._decisions: dict[str, Decision] = {} + self._plan_decisions: dict[str, list[str]] = {} + self._plan_sequence: dict[str, int] = {} + + # Snapshot store + self.snapshots = SnapshotStore() + + @property + def _persisted(self) -> bool: + """Return True when a UnitOfWork is wired for persistence.""" + return self.unit_of_work is not None + # ------------------------------------------------------------------ - # record_decision + # Recording # ------------------------------------------------------------------ - def record_decision(self, decision: Decision) -> Decision: - """Persist a new decision. + def record_decision( + self, + plan_id: str, + decision_type: DecisionType | str, + question: str, + chosen_option: str, + *, + parent_decision_id: str | None = None, + alternatives_considered: list[str] | None = None, + confidence_score: float | None = None, + rationale: str = "", + actor_reasoning: str | None = None, + context_snapshot: ContextSnapshot | None = None, + artifacts_produced: list[Any] | None = None, + is_correction: bool = False, + corrects_decision_id: str | None = None, + correction_reason: str | None = None, + ) -> Decision: + """Record a new decision in the plan's decision tree. - Opens a UnitOfWork transaction, creates the decision via the - repository, and commits. + Automatically assigns a monotonically increasing sequence number, + generates a decision ID (ULID), and captures a context snapshot. Args: - decision: The :class:`Decision` domain object to persist. + plan_id: ULID of the plan. + decision_type: Type of decision being recorded. + question: What question was being answered. + chosen_option: The option that was chosen. + parent_decision_id: Optional parent in the decision tree. + alternatives_considered: Other options evaluated. + confidence_score: Confidence in the decision (0.0-1.0). + rationale: Human-readable rationale. + actor_reasoning: Raw LLM reasoning trace. + context_snapshot: Snapshot of context at decision time. + artifacts_produced: Artifacts created as side-effects. + is_correction: Whether this corrects another decision. + corrects_decision_id: ULID of the decision being corrected. + correction_reason: Why the correction was made. Returns: - The same :class:`Decision` instance (pass-through). + The recorded :class:`Decision`. Raises: + ValidationError: If required fields are missing or invalid. DuplicateDecisionError: If a decision with the same ID exists. - DatabaseError: On transient or unexpected DB errors. """ - self._logger.info( - "recording_decision", - decision_id=decision.decision_id, - plan_id=decision.plan_id, - decision_type=str(decision.decision_type), - sequence_number=decision.sequence_number, + if not plan_id or not plan_id.strip(): + raise ValidationError("plan_id must not be empty") + if not question or not question.strip(): + raise ValidationError("question must not be empty") + if not chosen_option or not chosen_option.strip(): + raise ValidationError("chosen_option must not be empty") + + # Coerce string to enum if needed + if isinstance(decision_type, str): + decision_type = DecisionType(decision_type) + + # Auto-assign sequence number + seq = self._next_sequence(plan_id) + + # Build context snapshot + snapshot = context_snapshot or ContextSnapshot() + if not snapshot.hot_context_hash: + # Generate a hash from the question + chosen_option as a + # minimal snapshot when no explicit hash is provided + snapshot = self._auto_capture_snapshot(question, chosen_option, snapshot) + + decision = Decision( + plan_id=plan_id, + decision_type=decision_type, + sequence_number=seq, + question=question, + chosen_option=chosen_option, + parent_decision_id=parent_decision_id, + alternatives_considered=alternatives_considered or [], + confidence_score=confidence_score, + rationale=rationale, + actor_reasoning=actor_reasoning, + context_snapshot=snapshot, + artifacts_produced=artifacts_produced or [], + is_correction=is_correction, + corrects_decision_id=corrects_decision_id, + correction_reason=correction_reason, ) - with self.unit_of_work.transaction() as ctx: - ctx.decisions.create(decision) - self._logger.debug( - "decision_recorded", + + self._store_decision(decision) + + self._logger.info( + "decision.recorded", decision_id=decision.decision_id, + plan_id=plan_id, + decision_type=str(decision_type), + sequence=seq, ) return decision # ------------------------------------------------------------------ - # get_decision + # Retrieval # ------------------------------------------------------------------ - def get_decision(self, decision_id: str) -> Decision | None: - """Retrieve a single decision by its ULID. + def get_decision(self, decision_id: str) -> Decision: + """Retrieve a single decision by ID. Args: decision_id: ULID of the decision. Returns: - The :class:`Decision` or ``None`` if not found. + The :class:`Decision`. + + Raises: + DecisionNotFoundError: If the decision does not exist. """ - self._logger.debug("getting_decision", decision_id=decision_id) - with self.unit_of_work.transaction() as ctx: - return ctx.decisions.get(decision_id) + if self._persisted and self.unit_of_work is not None: + with self.unit_of_work.transaction() as ctx: + result = ctx.decisions.get(decision_id) + if result is None: + raise DecisionNotFoundError(decision_id) + return result + decision = self._decisions.get(decision_id) + if decision is None: + raise DecisionNotFoundError(decision_id) + return decision - # ------------------------------------------------------------------ - # get_decisions_for_plan - # ------------------------------------------------------------------ - - def get_decisions_for_plan(self, plan_id: str) -> list[Decision]: - """Return all decisions for a plan ordered by sequence number. + def list_decisions(self, plan_id: str) -> list[Decision]: + """List all decisions for a plan, ordered by sequence number. Args: plan_id: ULID of the plan. Returns: - List of :class:`Decision` instances. + List of decisions sorted by sequence_number. """ - self._logger.debug("getting_decisions_for_plan", plan_id=plan_id) - with self.unit_of_work.transaction() as ctx: - return ctx.decisions.get_by_plan(plan_id) + if self._persisted and self.unit_of_work is not None: + with self.unit_of_work.transaction() as ctx: + return ctx.decisions.get_by_plan(plan_id) - # ------------------------------------------------------------------ - # get_decision_tree - # ------------------------------------------------------------------ + ids = self._plan_decisions.get(plan_id, []) + decisions = [self._decisions[did] for did in ids if did in self._decisions] + return sorted(decisions, key=lambda d: d.sequence_number) - def get_decision_tree(self, root_id: str) -> list[Decision]: - """Retrieve the full decision tree rooted at *root_id* (BFS order). + def list_by_type( + self, plan_id: str, decision_type: DecisionType | str + ) -> list[Decision]: + """List decisions for a plan filtered by type. Args: - root_id: ULID of the root decision. + plan_id: ULID of the plan. + decision_type: The type to filter by. Returns: - List of :class:`Decision` instances (BFS order). - - Raises: - DecisionNotFoundError: If the root decision does not exist. + Filtered list sorted by sequence_number. """ - self._logger.debug("getting_decision_tree", root_id=root_id) - with self.unit_of_work.transaction() as ctx: - return ctx.decisions.get_tree(root_id) + if isinstance(decision_type, str): + decision_type = DecisionType(decision_type) + + if self._persisted and self.unit_of_work is not None: + with self.unit_of_work.transaction() as ctx: + return ctx.decisions.list_by_type(plan_id, str(decision_type)) + + all_decisions = self.list_decisions(plan_id) + return [d for d in all_decisions if d.decision_type == decision_type] # ------------------------------------------------------------------ - # get_path_to_root + # Tree operations # ------------------------------------------------------------------ + def get_tree(self, plan_id: str) -> list[Decision]: + """Get the full decision tree for a plan via BFS from root. + + Returns decisions in BFS order (root first, then children + level by level). + + Args: + plan_id: ULID of the plan. + + Returns: + List of decisions in BFS order. + """ + all_decisions = self.list_decisions(plan_id) + if not all_decisions: + return [] + + # Find root(s) — decisions with no parent + roots = [d for d in all_decisions if d.is_root] + if not roots: + return all_decisions + + # Build adjacency: parent_id → children + children_map: dict[str | None, list[Decision]] = {} + for d in all_decisions: + children_map.setdefault(d.parent_decision_id, []).append(d) + + # BFS from roots + result: list[Decision] = [] + queue: deque[Decision] = deque(roots) + visited: set[str] = set() + + while queue: + node = queue.popleft() + if node.decision_id in visited: + continue + visited.add(node.decision_id) + result.append(node) + children = children_map.get(node.decision_id, []) + for child in sorted(children, key=lambda c: c.sequence_number): + if child.decision_id not in visited: + queue.append(child) + + return result + def get_path_to_root(self, decision_id: str) -> list[Decision]: - """Walk from *decision_id* up to the root (leaf → root order). + """Navigate from a decision up to the tree root. + + Returns the path from the given decision to the root, + inclusive of both endpoints. Args: decision_id: ULID of the starting decision. Returns: - List of :class:`Decision` instances (leaf first). + List from the given decision up to the root. Raises: - DecisionNotFoundError: If the starting decision is missing. + DecisionNotFoundError: If the starting decision does not exist. """ - self._logger.debug("getting_path_to_root", decision_id=decision_id) - with self.unit_of_work.transaction() as ctx: - return ctx.decisions.get_path_to_root(decision_id) + if self._persisted and self.unit_of_work is not None: + with self.unit_of_work.transaction() as ctx: + return ctx.decisions.get_path_to_root(decision_id) - # ------------------------------------------------------------------ - # mark_superseded - # ------------------------------------------------------------------ + current = self._decisions.get(decision_id) + if current is None: + raise DecisionNotFoundError(decision_id) - def mark_superseded( - self, - decision_id: str, - new_decision_id: str, - ) -> Decision: - """Mark a decision as superseded by a new one. + path: list[Decision] = [current] + visited: set[str] = {current.decision_id} - Args: - decision_id: ULID of the decision to mark. - new_decision_id: ULID of the replacement decision. + while current.parent_decision_id is not None: + parent = self._decisions.get(current.parent_decision_id) + if parent is None or parent.decision_id in visited: + break + visited.add(parent.decision_id) + path.append(parent) + current = parent - Returns: - Updated :class:`Decision`. + return path - Raises: - DecisionNotFoundError: If *decision_id* does not exist. - """ - self._logger.info( - "marking_superseded", - decision_id=decision_id, - new_decision_id=new_decision_id, - ) - with self.unit_of_work.transaction() as ctx: - return ctx.decisions.update_superseded_by(decision_id, new_decision_id) - - # ------------------------------------------------------------------ - # list_by_type - # ------------------------------------------------------------------ - - def list_by_type(self, plan_id: str, decision_type: str) -> list[Decision]: - """List decisions of a given type for a plan. + def get_superseded(self, plan_id: str) -> list[Decision]: + """Get all superseded decisions for a plan. Args: plan_id: ULID of the plan. - decision_type: String value of a :class:`DecisionType` enum - member (e.g. ``"strategy_choice"``). Returns: - List of :class:`Decision` instances. + List of decisions that have been superseded. """ - self._logger.debug( - "listing_by_type", - plan_id=plan_id, - decision_type=decision_type, + if self._persisted and self.unit_of_work is not None: + with self.unit_of_work.transaction() as ctx: + return ctx.decisions.get_superseded(plan_id) + + all_decisions = self.list_decisions(plan_id) + return [d for d in all_decisions if d.is_superseded] + + # ------------------------------------------------------------------ + # Mutations + # ------------------------------------------------------------------ + + def mark_superseded(self, decision_id: str, new_decision_id: str) -> Decision: + """Mark a decision as superseded by another. + + Args: + decision_id: ULID of the decision to supersede. + new_decision_id: ULID of the replacement decision. + + Returns: + The updated :class:`Decision` with superseded_by set. + + Raises: + DecisionNotFoundError: If the decision does not exist. + """ + if self._persisted and self.unit_of_work is not None: + with self.unit_of_work.transaction() as ctx: + result = ctx.decisions.update_superseded_by( + decision_id, new_decision_id + ) + # Update cache + if decision_id in self._decisions: + self._decisions[decision_id] = result + self._logger.info( + "decision.superseded", + decision_id=decision_id, + superseded_by=new_decision_id, + ) + return result + + original = self._decisions.get(decision_id) + if original is None: + raise DecisionNotFoundError(decision_id) + + updated = original.with_superseded_by(new_decision_id) + self._decisions[decision_id] = updated + + self._logger.info( + "decision.superseded", + decision_id=decision_id, + superseded_by=new_decision_id, ) - with self.unit_of_work.transaction() as ctx: - return ctx.decisions.list_by_type(plan_id, decision_type) + return updated + + def delete_decision(self, decision_id: str) -> bool: + """Delete a decision. + + Args: + decision_id: ULID of the decision to delete. + + Returns: + ``True`` if the decision was deleted. + + Raises: + DecisionNotFoundError: If the decision does not exist. + """ + if self._persisted and self.unit_of_work is not None: + with self.unit_of_work.transaction() as ctx: + ctx.decisions.delete(decision_id) + # Remove from cache + decision = self._decisions.pop(decision_id, None) + if decision: + plan_ids = self._plan_decisions.get(decision.plan_id, []) + if decision_id in plan_ids: + plan_ids.remove(decision_id) + self.snapshots.remove(decision_id) + self._logger.info("decision.deleted", decision_id=decision_id) + return True + + decision = self._decisions.pop(decision_id, None) + if decision is None: + raise DecisionNotFoundError(decision_id) + + plan_ids = self._plan_decisions.get(decision.plan_id, []) + if decision_id in plan_ids: + plan_ids.remove(decision_id) + + self.snapshots.remove(decision_id) + self._logger.info("decision.deleted", decision_id=decision_id) + return True + + # ------------------------------------------------------------------ + # Snapshot helpers + # ------------------------------------------------------------------ + + def get_snapshot(self, decision_id: str) -> ContextSnapshot | None: + """Retrieve the context snapshot for a decision. + + Args: + decision_id: ULID of the decision. + + Returns: + The snapshot, or ``None`` if not stored separately. + """ + return self.snapshots.get(decision_id) + + def get_snapshots_for_plan(self, plan_id: str) -> dict[str, ContextSnapshot]: + """Retrieve all snapshots for a plan's decisions. + + Args: + plan_id: ULID of the plan. + + Returns: + Dict mapping decision_id → ContextSnapshot. + """ + decisions = self.list_decisions(plan_id) + return self.snapshots.list_for_plan(decisions) + + # ------------------------------------------------------------------ + # Statistics + # ------------------------------------------------------------------ + + def count_decisions(self, plan_id: str) -> int: + """Return the number of decisions recorded for a plan. + + Args: + plan_id: ULID of the plan. + + Returns: + Decision count. + """ + return len(self.list_decisions(plan_id)) + + def get_next_sequence(self, plan_id: str) -> int: + """Return the next available sequence number for a plan. + + This is useful for callers that need to know the next + sequence number before recording. + + Args: + plan_id: ULID of the plan. + + Returns: + The next sequence number. + """ + return self._plan_sequence.get(plan_id, 0) + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _next_sequence(self, plan_id: str) -> int: + """Generate the next monotonically increasing sequence number.""" + seq = self._plan_sequence.get(plan_id, 0) + self._plan_sequence[plan_id] = seq + 1 + return seq + + def _store_decision(self, decision: Decision) -> None: + """Store a decision in both persistence and cache.""" + decision_id = decision.decision_id + plan_id = decision.plan_id + + if decision_id in self._decisions: + raise DuplicateDecisionError(decision_id) + + # Persist to DB if available + if self._persisted and self.unit_of_work is not None: + with self.unit_of_work.transaction() as ctx: + ctx.decisions.create(decision) + + # Update in-memory cache + self._decisions[decision_id] = decision + self._plan_decisions.setdefault(plan_id, []).append(decision_id) + + # Store snapshot + self.snapshots.store(decision_id, decision.context_snapshot) + + @staticmethod + def _auto_capture_snapshot( + question: str, + chosen_option: str, + base_snapshot: ContextSnapshot, + ) -> ContextSnapshot: + """Generate a minimal context snapshot hash. + + When no explicit ``hot_context_hash`` is provided, we compute + a SHA-256 hash from the question + chosen option as a minimal + content-addressable identifier. + + Args: + question: The decision question. + chosen_option: The chosen option. + base_snapshot: The base snapshot to augment. + + Returns: + A new :class:`ContextSnapshot` with the hash populated. + """ + content = json.dumps( + {"question": question, "chosen_option": chosen_option}, + sort_keys=True, + ) + context_hash = hashlib.sha256(content.encode()).hexdigest()[:16] + return ContextSnapshot( + hot_context_hash=f"sha256:{context_hash}", + hot_context_ref=base_snapshot.hot_context_ref, + relevant_resources=list(base_snapshot.relevant_resources), + actor_state_ref=base_snapshot.actor_state_ref, + ) + + +__all__ = [ + "DecisionNotFoundError", + "DecisionService", + "DuplicateDecisionError", + "SequenceConflictError", + "SnapshotStore", +] -- 2.52.0 From 4e871e70af3786bf557ec227a03038789c645ee3 Mon Sep 17 00:00:00 2001 From: Jeffrey Phillips Freeman Date: Wed, 25 Feb 2026 18:13:13 +0000 Subject: [PATCH 2/9] docs: add CHANGELOG entry for decision recording and snapshot store --- CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0d684b9a0..bb7349a64 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -191,6 +191,8 @@ - Added `ReadOnlyViolationError` to `ChangeSetCapture` to prevent write artifacts on read-only plans. - Added CLI fail-fast guards on `plan execute` and `plan apply` for read-only plans. +- Added `DecisionService` with record/list/tree helpers and `SnapshotStore` for hash-based + deduplication of context snapshots during plan execution. - Expanded CONTRIBUTING.md with detailed guidance on the issue creation process, label system, ticket lifecycle, pull request requirements, and review/merge process. - Added commit scope, quality, and message format guidelines to CONTRIBUTING.md. -- 2.52.0 From db81c4cfd6a057fd010736486b0bd5a87ad2bcc1 Mon Sep 17 00:00:00 2001 From: khyari hamza Date: Thu, 26 Feb 2026 00:28:41 +0000 Subject: [PATCH 3/9] test(service): add edge-case scenarios for coverage Add 11 new Behave scenarios covering SnapshotStore hash-less storage, hash-index cleanup on remove, multi-entry removal, list_for_plan with missing snapshots, whitespace-only validation, no-root tree traversal, and list-by-type string coercion. Brings scenario count from 37 to 48. ISSUES CLOSED: #172 --- features/decision_recording.feature | 73 ++++++++++++ features/steps/decision_recording_steps.py | 132 +++++++++++++++++++++ 2 files changed, 205 insertions(+) diff --git a/features/decision_recording.feature b/features/decision_recording.feature index 518571703..3b02c2313 100644 --- a/features/decision_recording.feature +++ b/features/decision_recording.feature @@ -230,3 +230,76 @@ Feature: Decision recording and snapshot store Scenario: DecisionService string type coercion works When I record a decision for plan "P1" with string type "strategy_choice" Then the dsvc decision type should be "strategy_choice" + + # --- Edge-case coverage --- + + Scenario: Snapshot stored without a hash skips hash index + Given a standalone snapshot store + And a context snapshot with no hash + When I store the hashless snapshot for decision "D1" + And I get standalone snapshot for decision "D1" + Then the dsvc standalone retrieved snapshot should not be None + And the dsvc standalone retrieved snapshot hash should be empty + + Scenario: Snapshot store remove cleans up hash index + Given a standalone snapshot store + And a context snapshot with hash "sha256:removeme" and ref "ref" + When I store the explicit snapshot for decision "D_REM" + And I remove snapshot for decision "D_REM" + Then the dsvc snapshot remove result should be True + And the dsvc standalone hash query for "sha256:removeme" should return 0 decision IDs + + Scenario: Snapshot store remove last entry deletes hash bucket + Given a standalone snapshot store + And a context snapshot with hash "sha256:onlyone" and ref "ref" + When I store the explicit snapshot for decision "D_ONLY" + And I remove snapshot for decision "D_ONLY" + Then the dsvc snapshot remove result should be True + And the dsvc standalone hash query for "sha256:onlyone" should return 0 decision IDs + + Scenario: Snapshot store remove snapshot with no hash + Given a standalone snapshot store + And a context snapshot with no hash + When I store the hashless snapshot for decision "D_NOHASH" + And I remove snapshot for decision "D_NOHASH" + Then the dsvc snapshot remove result should be True + + Scenario: Snapshot store remove with multiple entries keeps remaining + Given a standalone snapshot store + And a context snapshot with hash "sha256:shared" and ref "ref" + When I store the explicit snapshot for decision "D_KEEP1" + And I store the explicit snapshot for decision "D_KEEP2" + And I remove snapshot for decision "D_KEEP1" + Then the dsvc snapshot remove result should be True + And the dsvc standalone hash query for "sha256:shared" should return 1 decision IDs + + Scenario: Get tree when no decision is a root returns all decisions + When I record a decision for plan "P1" with a forced non-root parent + And I get the tree for plan "P1" + Then the dsvc tree should have 1 decisions + + Scenario: List by type with string coercion filters correctly + When I record a prompt_definition decision for plan "P1" with question "Type1" + And I record a strategy_choice decision for plan "P1" with question "Type2" + And I filter decisions for plan "P1" by type "prompt_definition" + Then the dsvc decision list should have 1 entries + + Scenario: Recording with whitespace-only plan_id raises ValidationError + When I try to record a decision with whitespace plan_id + Then a dsvc validation error should be raised + And the dsvc error should mention "plan_id" + + Scenario: Recording with whitespace-only question raises ValidationError + When I try to record a decision with whitespace question + Then a dsvc validation error should be raised + And the dsvc error should mention "question" + + Scenario: Recording with whitespace-only chosen_option raises ValidationError + When I try to record a decision with whitespace chosen_option + Then a dsvc validation error should be raised + And the dsvc error should mention "chosen_option" + + Scenario: SnapshotStore list_for_plan skips missing snapshots + Given a standalone snapshot store + When I list snapshots for decisions with missing entries + Then the dsvc standalone plan snapshots should have 0 entries diff --git a/features/steps/decision_recording_steps.py b/features/steps/decision_recording_steps.py index 871e1e2f4..b50b98f42 100644 --- a/features/steps/decision_recording_steps.py +++ b/features/steps/decision_recording_steps.py @@ -24,6 +24,7 @@ from cleveragents.core.exceptions import ValidationError from cleveragents.domain.models.core.decision import ( ArtifactRef, ContextSnapshot, + Decision, DecisionType, ) @@ -816,3 +817,134 @@ def step_standalone_none(context: Context) -> None: @then("the dsvc standalone hash query should return {count:d} decision IDs") def step_standalone_hash_count(context: Context, count: int) -> None: assert len(context.standalone_hash_result) == count + + +# --------------------------------------------------------------------------- +# Edge-case coverage steps +# --------------------------------------------------------------------------- + + +@given("a context snapshot with no hash") +def step_snapshot_no_hash(context: Context) -> None: + context.explicit_snapshot = ContextSnapshot( + hot_context_hash="", + hot_context_ref="", + ) + + +@when('I store the hashless snapshot for decision "{decision_id}"') +def step_store_hashless(context: Context, decision_id: str) -> None: + context.standalone_store.store(decision_id, context.explicit_snapshot) + + +@when('I get standalone snapshot for decision "{decision_id}"') +def step_get_standalone_snapshot(context: Context, decision_id: str) -> None: + context.standalone_retrieved = context.standalone_store.get(decision_id) + + +@when('I store the explicit snapshot for decision "{decision_id}"') +def step_store_explicit(context: Context, decision_id: str) -> None: + context.standalone_store.store(decision_id, context.explicit_snapshot) + + +@then("the dsvc standalone retrieved snapshot should not be None") +def step_standalone_retrieved_not_none(context: Context) -> None: + assert context.standalone_retrieved is not None + + +@then("the dsvc standalone retrieved snapshot hash should be empty") +def step_standalone_retrieved_hash_empty(context: Context) -> None: + assert context.standalone_retrieved.hot_context_hash == "" + + +@then("the dsvc snapshot remove result should be True") +def step_standalone_remove_true(context: Context) -> None: + assert context.snapshot_remove_result is True + + +@then( + 'the dsvc standalone hash query for "{ctx_hash}" ' + "should return {count:d} decision IDs" +) +def step_standalone_hash_query_count( + context: Context, ctx_hash: str, count: int +) -> None: + result = context.standalone_store.get_by_hash(ctx_hash) + assert len(result) == count, f"Expected {count}, got {len(result)}" + + +@when('I record a decision for plan "{plan_id}" with a forced non-root parent') +def step_record_forced_non_root(context: Context, plan_id: str) -> None: + """Record a decision with a fake parent_decision_id so it's not a root.""" + svc = context.decision_service + d = svc.record_decision( + plan_id=_resolve_plan_id(context, plan_id), + decision_type=DecisionType.STRATEGY_CHOICE, + question="Non-root decision", + chosen_option="Chosen", + parent_decision_id=str(ULID()), + ) + context.decision_result = d + context.recorded_decisions.append(d) + + +@when("I try to record a decision with whitespace plan_id") +def step_try_whitespace_plan_id(context: Context) -> None: + try: + context.decision_service.record_decision( + plan_id=" ", + decision_type=DecisionType.STRATEGY_CHOICE, + question="Q", + chosen_option="A", + ) + context.decision_error = None + except ValidationError as exc: + context.decision_error = exc + + +@when("I try to record a decision with whitespace question") +def step_try_whitespace_question(context: Context) -> None: + try: + context.decision_service.record_decision( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV", + decision_type=DecisionType.STRATEGY_CHOICE, + question=" ", + chosen_option="A", + ) + context.decision_error = None + except ValidationError as exc: + context.decision_error = exc + + +@when("I try to record a decision with whitespace chosen_option") +def step_try_whitespace_chosen(context: Context) -> None: + try: + context.decision_service.record_decision( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV", + decision_type=DecisionType.STRATEGY_CHOICE, + question="Q", + chosen_option=" ", + ) + context.decision_error = None + except ValidationError as exc: + context.decision_error = exc + + +@when("I list snapshots for decisions with missing entries") +def step_list_snapshots_missing(context: Context) -> None: + """Call list_for_plan with a Decision whose snapshot is not in the store.""" + fake_decision = Decision( + plan_id=str(ULID()), + decision_type=DecisionType.STRATEGY_CHOICE, + sequence_number=0, + question="ghost", + chosen_option="ghost", + ) + context.standalone_plan_snapshots = context.standalone_store.list_for_plan( + [fake_decision] + ) + + +@then("the dsvc standalone plan snapshots should have {count:d} entries") +def step_standalone_plan_snapshots_count(context: Context, count: int) -> None: + assert len(context.standalone_plan_snapshots) == count -- 2.52.0 From e179f3426659de73e57730c76c799aca368536be Mon Sep 17 00:00:00 2001 From: khyari hamza Date: Thu, 26 Feb 2026 03:08:00 +0000 Subject: [PATCH 4/9] test(service): add persisted-mode integration scenarios for coverage Add 9 Behave scenarios exercising all database-backed code paths in DecisionService: record, get, list, list_by_type, get_path_to_root, get_superseded, mark_superseded, delete, and duplicate detection. File-level coverage rises from 81% to 96%. ISSUES CLOSED: #172 --- features/decision_recording.feature | 67 ++++++++++++++++++++++ features/steps/decision_recording_steps.py | 62 ++++++++++++++++++++ 2 files changed, 129 insertions(+) diff --git a/features/decision_recording.feature b/features/decision_recording.feature index 3b02c2313..5505989a9 100644 --- a/features/decision_recording.feature +++ b/features/decision_recording.feature @@ -303,3 +303,70 @@ Feature: Decision recording and snapshot store Given a standalone snapshot store When I list snapshots for decisions with missing entries Then the dsvc standalone plan snapshots should have 0 entries + + # --- Persisted-mode (database-backed) --- + + Scenario: Persisted record and get decision round-trips through database + Given a persisted decision service + When I record a prompt_definition decision for plan "P1" with question "Persisted root" + And I get the decision by its ID + Then the dsvc retrieved decision should match the recorded decision + + Scenario: Persisted get non-existent decision raises DecisionNotFoundError + Given a persisted decision service + When I try to get decision "01NONEXISTENT00000000000000" + Then a dsvc decision not found error should be raised + + Scenario: Persisted list decisions returns ordered results from database + Given a persisted decision service + When I record a prompt_definition decision for plan "P1" with question "First persisted" + And I record a strategy_choice decision for plan "P1" with question "Second persisted" + And I list decisions for plan "P1" + Then the dsvc decision list should have 2 entries + And the dsvc decision list should be ordered by sequence number + + Scenario: Persisted list by type filters through database + Given a persisted decision service + When I record a prompt_definition decision for plan "P1" with question "Root" + And I record a strategy_choice decision for plan "P1" with question "Strategy" + And I record a strategy_choice decision for plan "P1" with question "Strategy 2" + And I filter decisions for plan "P1" by type "strategy_choice" + Then the dsvc decision list should have 2 entries + + Scenario: Persisted get path to root navigates through database + Given a persisted decision service + When I record a prompt_definition decision for plan "P1" with question "Root" + And I record a strategy_choice decision for plan "P1" with parent as child "Level 1" + And I record an implementation_choice decision for plan "P1" with second parent as child "Level 2" + And I get the path to root from the last decision + Then the dsvc path should have 3 decisions + And the dsvc last path decision should be the root + + Scenario: Persisted get superseded returns superseded decisions from database + Given a persisted decision service + When I record a prompt_definition decision for plan "P1" with question "Original" + And I record a strategy_choice decision for plan "P1" with question "Replacement" + And I mark the first decision as superseded by the second + And I get superseded decisions for plan "P1" + Then the dsvc superseded list should have 1 entry + + Scenario: Persisted mark superseded updates database and cache + Given a persisted decision service + When I record a prompt_definition decision for plan "P1" with question "Old" + And I record a strategy_choice decision for plan "P1" with question "New" + And I mark the first decision as superseded by the second + Then the dsvc first decision should be superseded + And the dsvc first decision superseded_by should match the second decision + + Scenario: Persisted delete removes decision from database + Given a persisted decision service + When I record a prompt_definition decision for plan "P1" with question "Ephemeral" + And I delete the recorded decision + Then the dsvc decision should be deleted + And the dsvc plan "P1" should have 0 decisions + + Scenario: Persisted duplicate decision raises DuplicateDecisionError + Given a persisted decision service + When I record a prompt_definition decision for plan "P1" with question "Original" + And I try to store a duplicate of the recorded decision + Then a dsvc duplicate decision error should be raised diff --git a/features/steps/decision_recording_steps.py b/features/steps/decision_recording_steps.py index b50b98f42..76064a3fb 100644 --- a/features/steps/decision_recording_steps.py +++ b/features/steps/decision_recording_steps.py @@ -7,7 +7,9 @@ defines similar assertion patterns for the domain model layer. from __future__ import annotations +import os import re +import tempfile from behave import given, then, when from behave.runner import Context @@ -948,3 +950,63 @@ def step_list_snapshots_missing(context: Context) -> None: @then("the dsvc standalone plan snapshots should have {count:d} entries") def step_standalone_plan_snapshots_count(context: Context, count: int) -> None: assert len(context.standalone_plan_snapshots) == count + + +# --------------------------------------------------------------------------- +# Persisted-mode (database-backed) steps +# --------------------------------------------------------------------------- + + +@given("a persisted decision service") +def step_persisted_decision_service(context: Context) -> None: + """Create a DecisionService backed by a real SQLite database via UnitOfWork.""" + from cleveragents.infrastructure.database.unit_of_work import UnitOfWork + + db_path = tempfile.mktemp(suffix=".db", prefix="dsvc_persisted_") + context._dsvc_db_path = db_path + uow = UnitOfWork(f"sqlite:///{db_path}") + uow.init_database() + + context.decision_service = DecisionService(unit_of_work=uow) + context.recorded_decisions = [] + context.decision_error = None + context.decision_result = None + context.decision_list = None + context.tree_result = None + context.path_result = None + context.superseded_result = None + context.snapshot_result = None + context.snapshots_dict = None + context.hash_query_result = None + context.delete_result = None + context.explicit_snapshot = None + context._plan_id_registry = {} + + # Register cleanup to remove temp DB file + def _cleanup_db() -> None: + for suffix in ("", "-journal", "-wal", "-shm"): + try: + os.unlink(db_path + suffix) + except OSError: + pass + + if not hasattr(context, "_cleanup_handlers"): + context._cleanup_handlers = [] + context._cleanup_handlers.append(_cleanup_db) + + +@when("I try to store a duplicate of the recorded decision") +def step_try_store_duplicate(context: Context) -> None: + """Attempt to store a decision with the same ID that already exists.""" + try: + decision = context.decision_result + context.decision_service._store_decision(decision) + context.decision_error = None + except DuplicateDecisionError as exc: + context.decision_error = exc + + +@then("a dsvc duplicate decision error should be raised") +def step_duplicate_error_raised(context: Context) -> None: + assert context.decision_error is not None + assert isinstance(context.decision_error, DuplicateDecisionError) -- 2.52.0 From 4cbf985a17a3922e13bdec95c6bac33d1396e60c Mon Sep 17 00:00:00 2001 From: khyari hamza Date: Thu, 26 Feb 2026 03:15:57 +0000 Subject: [PATCH 5/9] style(service): use contextlib.suppress per ruff SIM105 ISSUES CLOSED: #172 --- features/steps/decision_recording_steps.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/features/steps/decision_recording_steps.py b/features/steps/decision_recording_steps.py index 76064a3fb..e036daa08 100644 --- a/features/steps/decision_recording_steps.py +++ b/features/steps/decision_recording_steps.py @@ -7,6 +7,7 @@ defines similar assertion patterns for the domain model layer. from __future__ import annotations +import contextlib import os import re import tempfile @@ -985,10 +986,8 @@ def step_persisted_decision_service(context: Context) -> None: # Register cleanup to remove temp DB file def _cleanup_db() -> None: for suffix in ("", "-journal", "-wal", "-shm"): - try: + with contextlib.suppress(OSError): os.unlink(db_path + suffix) - except OSError: - pass if not hasattr(context, "_cleanup_handlers"): context._cleanup_handlers = [] -- 2.52.0 From d5802a4878edaac98c25b326a79293555a494024 Mon Sep 17 00:00:00 2001 From: khyari hamza Date: Fri, 27 Feb 2026 12:44:22 +0000 Subject: [PATCH 6/9] fix(service): use DecisionService kwargs API in plan lifecycle caller Align _record_decision_safe() to call record_decision() with keyword arguments directly instead of manually constructing a Decision model. The service now handles sequence numbering and model construction internally. ISSUES CLOSED: #172 --- .../application/services/plan_lifecycle_service.py | 11 +++-------- 1 file changed, 3 insertions(+), 8 deletions(-) diff --git a/src/cleveragents/application/services/plan_lifecycle_service.py b/src/cleveragents/application/services/plan_lifecycle_service.py index 194bc0893..f130d7440 100644 --- a/src/cleveragents/application/services/plan_lifecycle_service.py +++ b/src/cleveragents/application/services/plan_lifecycle_service.py @@ -203,19 +203,14 @@ class PlanLifecycleService: if self.decision_service is None: return - from cleveragents.domain.models.core.decision import Decision as DecisionModel - from cleveragents.domain.models.core.decision import DecisionType as DT - try: - decision = DecisionModel( + self.decision_service.record_decision( plan_id=plan_id, - parent_decision_id=parent_decision_id, - sequence_number=self._next_seq(plan_id), - decision_type=DT(decision_type), + decision_type=decision_type, question=question, chosen_option=chosen_option, + parent_decision_id=parent_decision_id, ) - self.decision_service.record_decision(decision) except Exception: self._logger.warning( "decision_recording_failed", -- 2.52.0 From cebf71ad8eaddcba1e2e873943d35c0e668dd890 Mon Sep 17 00:00:00 2001 From: khyari hamza Date: Fri, 27 Feb 2026 13:10:58 +0000 Subject: [PATCH 7/9] fix(service): use list_decisions in robot helper The Robot helper called the non-existent get_decisions_for_plan method on DecisionService. The correct method is list_decisions. ISSUES CLOSED: #172 --- robot/helper_decision_di.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/robot/helper_decision_di.py b/robot/helper_decision_di.py index 8b76cd93f..de5ca06d1 100644 --- a/robot/helper_decision_di.py +++ b/robot/helper_decision_di.py @@ -108,7 +108,7 @@ def _record_integration() -> None: # Start strategize -> should record a strategy_choice decision lifecycle_svc.start_strategize(plan_id) - decisions = decision_svc.get_decisions_for_plan(plan_id) + decisions = decision_svc.list_decisions(plan_id) strategy_decisions = [ d for d in decisions if str(d.decision_type) == "strategy_choice" ] @@ -128,7 +128,7 @@ def _record_integration() -> None: # Start execute -> should record an implementation_choice decision lifecycle_svc.start_execute(plan_id) - decisions = decision_svc.get_decisions_for_plan(plan_id) + decisions = decision_svc.list_decisions(plan_id) impl_decisions = [ d for d in decisions if str(d.decision_type) == "implementation_choice" ] -- 2.52.0 From 67c63f4c58287bfd987b1e17b7848dc57d67562a Mon Sep 17 00:00:00 2001 From: khyari hamza Date: Fri, 27 Feb 2026 13:43:40 +0000 Subject: [PATCH 8/9] fix(service): align test and doc refs with DecisionService API Behave steps, benchmarks, vulture whitelist, and docs referenced renamed methods (get_decisions_for_plan, get_decision_tree). Updated to use the actual API names (list_decisions, get_tree) and the kwargs record_decision signature. ISSUES CLOSED: #172 --- benchmarks/decision_di_bench.py | 60 +++++++++------------- docs/reference/di.md | 12 ++--- features/steps/decision_di_wiring_steps.py | 4 +- vulture_whitelist.py | 4 +- 4 files changed, 33 insertions(+), 47 deletions(-) diff --git a/benchmarks/decision_di_bench.py b/benchmarks/decision_di_bench.py index b7c0d51cd..deaa9dd2b 100644 --- a/benchmarks/decision_di_bench.py +++ b/benchmarks/decision_di_bench.py @@ -106,21 +106,6 @@ def _seed_prerequisites(uow: UnitOfWork) -> None: plan_repo.create(plan) -def _make_decision( - seq: int = 0, - parent_id: str | None = None, - dtype: DecisionType = DecisionType.PROMPT_DEFINITION, -) -> Decision: - return Decision( - plan_id=_PLAN_ID, - parent_decision_id=parent_id, - sequence_number=seq, - decision_type=dtype, - question="Benchmark question?", - chosen_option="Benchmark option", - ) - - class TimeDecisionServiceResolution: """Benchmark DecisionService resolution from the DI container.""" @@ -160,11 +145,12 @@ class TimeDecisionRecord: def time_decision_record(self) -> None: self._seq += 1 - d = _make_decision( - seq=self._seq, - dtype=DecisionType.STRATEGY_CHOICE, + self.svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Benchmark question?", + chosen_option="Benchmark option", ) - self.svc.record_decision(d) class TimeDecisionTreeRetrieval: @@ -178,27 +164,29 @@ class TimeDecisionTreeRetrieval: self.svc = DecisionService(settings=MagicMock(), unit_of_work=self.uow) # Build a small tree - root = _make_decision(seq=0) - self.svc.record_decision(root) - self.root_id = root.decision_id + root = self.svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.PROMPT_DEFINITION, + question="Benchmark question?", + chosen_option="Benchmark option", + ) - seq = 1 for _ in range(3): - child = _make_decision( - seq=seq, - parent_id=root.decision_id, - dtype=DecisionType.STRATEGY_CHOICE, + child = self.svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Benchmark question?", + chosen_option="Benchmark option", + parent_decision_id=root.decision_id, ) - self.svc.record_decision(child) for _ in range(2): - seq += 1 - gc = _make_decision( - seq=seq, - parent_id=child.decision_id, - dtype=DecisionType.IMPLEMENTATION_CHOICE, + self.svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.IMPLEMENTATION_CHOICE, + question="Benchmark question?", + chosen_option="Benchmark option", + parent_decision_id=child.decision_id, ) - self.svc.record_decision(gc) - seq += 1 def time_decision_tree_retrieval(self) -> None: - self.svc.get_decision_tree(self.root_id) + self.svc.get_tree(_PLAN_ID) diff --git a/docs/reference/di.md b/docs/reference/di.md index bfd7e2ae2..e53bbdba0 100644 --- a/docs/reference/di.md +++ b/docs/reference/di.md @@ -51,8 +51,8 @@ decision_service = providers.Factory( |-------------------------|--------------------------------------------| | `record_decision()` | Persist a new decision | | `get_decision()` | Retrieve a decision by ID | -| `get_decisions_for_plan()` | Get all decisions for a plan | -| `get_decision_tree()` | BFS traversal of the decision tree | +| `list_decisions()` | Get all decisions for a plan | +| `get_tree()` | BFS traversal of the decision tree | | `get_path_to_root()` | Walk from a decision up to the root | | `mark_superseded()` | Mark a decision as superseded | | `list_by_type()` | List decisions by type for a plan | @@ -92,17 +92,15 @@ decision_svc = container.decision_service() lifecycle_svc = container.plan_lifecycle_service() # Record a decision manually -from cleveragents.domain.models.core.decision import Decision, DecisionType +from cleveragents.domain.models.core.decision import DecisionType -decision = Decision( +decision = decision_svc.record_decision( plan_id="01HV...", - sequence_number=0, decision_type=DecisionType.STRATEGY_CHOICE, question="Which approach?", chosen_option="Approach A", ) -decision_svc.record_decision(decision) # Retrieve decisions -decisions = decision_svc.get_decisions_for_plan("01HV...") +decisions = decision_svc.list_decisions("01HV...") ``` diff --git a/features/steps/decision_di_wiring_steps.py b/features/steps/decision_di_wiring_steps.py index 15c55e2f8..7c9965444 100644 --- a/features/steps/decision_di_wiring_steps.py +++ b/features/steps/decision_di_wiring_steps.py @@ -143,7 +143,7 @@ def step_start_strategize(context: Context) -> None: @then("decdi- a strategy_choice decision should be recorded for the plan") def step_check_strategize_decision(context: Context) -> None: - decisions = context.decision_svc.get_decisions_for_plan(context.plan_id) + decisions = context.decision_svc.list_decisions(context.plan_id) strategy_decisions = [ d for d in decisions if str(d.decision_type) == "strategy_choice" ] @@ -192,7 +192,7 @@ def step_start_execute(context: Context) -> None: @then("decdi- an implementation_choice decision should be recorded for the plan") def step_check_execute_decision(context: Context) -> None: - decisions = context.decision_svc.get_decisions_for_plan(context.plan_id) + decisions = context.decision_svc.list_decisions(context.plan_id) impl_decisions = [ d for d in decisions if str(d.decision_type) == "implementation_choice" ] diff --git a/vulture_whitelist.py b/vulture_whitelist.py index 4d34c2972..3ae790b83 100644 --- a/vulture_whitelist.py +++ b/vulture_whitelist.py @@ -359,8 +359,8 @@ decision_service # noqa: B018, F821 plan_lifecycle_service # noqa: B018, F821 record_decision # noqa: B018, F821 get_decision # noqa: B018, F821 -get_decisions_for_plan # noqa: B018, F821 -get_decision_tree # noqa: B018, F821 +list_decisions # noqa: B018, F821 +get_tree # noqa: B018, F821 get_path_to_root # noqa: B018, F821 mark_superseded # noqa: B018, F821 list_by_type # noqa: B018, F821 -- 2.52.0 From 0e36755db9f285d53a316164c66b93f87f39bb5a Mon Sep 17 00:00:00 2001 From: khyari hamza Date: Tue, 3 Mar 2026 03:17:19 +0000 Subject: [PATCH 9/9] fix(service): address review findings for decision service - Rehydrate sequence counter from DB on restart (BUG-1) - Add uniqueness guard raising SequenceConflictError (BUG-2) - Fix delete_decision consistency between persisted/in-memory (BUG-3) - Validate new_decision_id exists in mark_superseded (BUG-4) - Include orphaned subtrees in get_tree output (BUG-5) - Remove dead _decision_seq/_next_seq from plan_lifecycle (BUG-7) - Export SequenceConflictError from services __init__ (SPEC-2) - Replace list[Any] with list[ArtifactRef] typing (SPEC-4) - Add actor_reasoning max_length validation (SEC-1) - Use SELECT COUNT(*) in count_decisions (PERF-1) - Replace MagicMock with create_autospec(Settings) (TEST-3) - Eliminate UnitOfWork.__new__() anti-pattern (TEST-4) - Use exact assertion counts in DI tests (TEST-5) - Add restart rehydration, BFS order, invalid type, and confidence boundary test scenarios (TEST-1/2/6/7) - Fix decision_service_coverage.feature to match actual API ISSUES CLOSED: #172 --- benchmarks/decision_di_bench.py | 32 ++-- features/decision_recording.feature | 33 +++- features/decision_service_coverage.feature | 76 +++------- features/steps/decision_di_wiring_steps.py | 52 +++---- features/steps/decision_recording_steps.py | 98 +++++++++++- .../steps/decision_service_coverage_steps.py | 142 +++++++++++------- .../steps/plan_explain_cli_coverage_steps.py | 8 +- robot/helper_decision_di.py | 28 +--- robot/helper_decision_recording.py | 33 +++- .../application/services/__init__.py | 2 + .../application/services/decision_service.py | 119 ++++++++++++--- .../services/plan_lifecycle_service.py | 9 -- src/cleveragents/cli/commands/plan.py | 2 +- .../infrastructure/database/repositories.py | 63 ++++++++ 14 files changed, 482 insertions(+), 215 deletions(-) diff --git a/benchmarks/decision_di_bench.py b/benchmarks/decision_di_bench.py index deaa9dd2b..5cdf9574d 100644 --- a/benchmarks/decision_di_bench.py +++ b/benchmarks/decision_di_bench.py @@ -11,10 +11,11 @@ from __future__ import annotations import sys from datetime import UTC, datetime from pathlib import Path -from unittest.mock import MagicMock +from unittest.mock import create_autospec try: from cleveragents.application.services.decision_service import DecisionService + from cleveragents.config.settings import Settings from cleveragents.domain.models.core.action import Action, ActionState from cleveragents.domain.models.core.decision import Decision, DecisionType from cleveragents.domain.models.core.plan import ( @@ -25,11 +26,11 @@ try: PlanTimestamps, ProcessingState, ) - from cleveragents.infrastructure.database.models import Base from cleveragents.infrastructure.database.unit_of_work import UnitOfWork except ModuleNotFoundError: sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) from cleveragents.application.services.decision_service import DecisionService + from cleveragents.config.settings import Settings from cleveragents.domain.models.core.action import Action, ActionState from cleveragents.domain.models.core.decision import Decision, DecisionType from cleveragents.domain.models.core.plan import ( @@ -40,31 +41,16 @@ except ModuleNotFoundError: PlanTimestamps, ProcessingState, ) - from cleveragents.infrastructure.database.models import Base from cleveragents.infrastructure.database.unit_of_work import UnitOfWork -from sqlalchemy import create_engine -from sqlalchemy.orm import sessionmaker _PLAN_ID = "01HV00000000000000DIBENCH1" def _make_uow() -> UnitOfWork: """Create a UoW backed by an in-memory SQLite database.""" - engine = create_engine("sqlite:///:memory:", echo=False) - Base.metadata.create_all(engine) - - uow = UnitOfWork.__new__(UnitOfWork) - uow.database_url = "sqlite:///:memory:" - uow._engine = engine - uow._session_factory = sessionmaker( - bind=engine, - expire_on_commit=False, - autoflush=False, - autocommit=False, - ) - uow._database_initialized = True - uow._prompt_for_migration = None + uow = UnitOfWork("sqlite:///:memory:") + uow.init_database() return uow @@ -140,7 +126,9 @@ class TimeDecisionRecord: def setup(self) -> None: self.uow = _make_uow() _seed_prerequisites(self.uow) - self.svc = DecisionService(settings=MagicMock(), unit_of_work=self.uow) + mock_settings = create_autospec(Settings, instance=True) + mock_settings.database_url = "sqlite:///:memory:" + self.svc = DecisionService(settings=mock_settings, unit_of_work=self.uow) self._seq = 100 def time_decision_record(self) -> None: @@ -161,7 +149,9 @@ class TimeDecisionTreeRetrieval: def setup(self) -> None: self.uow = _make_uow() _seed_prerequisites(self.uow) - self.svc = DecisionService(settings=MagicMock(), unit_of_work=self.uow) + mock_settings = create_autospec(Settings, instance=True) + mock_settings.database_url = "sqlite:///:memory:" + self.svc = DecisionService(settings=mock_settings, unit_of_work=self.uow) # Build a small tree root = self.svc.record_decision( diff --git a/features/decision_recording.feature b/features/decision_recording.feature index 5505989a9..b984862b8 100644 --- a/features/decision_recording.feature +++ b/features/decision_recording.feature @@ -63,6 +63,25 @@ Feature: Decision recording and snapshot store Then a dsvc validation error should be raised And the dsvc error should mention "chosen_option" + Scenario: Recording with invalid decision_type raises ValueError + When I try to record a decision with invalid type "nonexistent_type" + Then a dsvc value error should be raised + And the dsvc error should mention "nonexistent_type" + + Scenario: Recording with confidence_score at boundaries + When I record a strategy_choice decision for plan "P1" with confidence 0.0 + Then the dsvc decision confidence score should be 0.0 + When I record a strategy_choice decision for plan "P1" with confidence 1.0 + Then the dsvc decision confidence score should be 1.0 + + Scenario: Recording with negative confidence_score raises ValidationError + When I try to record a decision with confidence -0.1 + Then a dsvc validation error should be raised + + Scenario: Recording with confidence_score above 1.0 raises ValidationError + When I try to record a decision with confidence 1.1 + Then a dsvc validation error should be raised + Scenario: Auto-capture generates a snapshot hash When I record a strategy_choice decision for plan "P1" with question "Test auto-capture" Then the dsvc decision context snapshot hash should start with "sha256:" @@ -104,9 +123,11 @@ Feature: Decision recording and snapshot store When I record a prompt_definition decision for plan "P1" with question "Root" And I record a strategy_choice decision for plan "P1" with parent as child "Child 1" And I record a strategy_choice decision for plan "P1" with parent as child "Child 2" + And I record an implementation_choice decision for plan "P1" with second parent as child "Grandchild" And I get the tree for plan "P1" - Then the dsvc tree should have 3 decisions + Then the dsvc tree should have 4 decisions And the dsvc first tree decision should be a root decision + And the dsvc tree should be in BFS level order Scenario: Get tree for empty plan returns empty list When I get the tree for plan "P_EMPTY" @@ -370,3 +391,13 @@ Feature: Decision recording and snapshot store When I record a prompt_definition decision for plan "P1" with question "Original" And I try to store a duplicate of the recorded decision Then a dsvc duplicate decision error should be raised + + Scenario: Persisted sequence resumes after service restart + Given a persisted decision service + When I record a prompt_definition decision for plan "P1" with question "Before restart" + And I record a strategy_choice decision for plan "P1" with question "Also before restart" + Then the dsvc decisions should have sequence numbers 0 1 + When I recreate the decision service with the same database + And I record a strategy_choice decision for plan "P1" with question "After restart" + Then the dsvc decision sequence number should be 2 + And the dsvc next sequence for plan "P1" should be 3 diff --git a/features/decision_service_coverage.feature b/features/decision_service_coverage.feature index e56eb7c39..5b84fdeb4 100644 --- a/features/decision_service_coverage.feature +++ b/features/decision_service_coverage.feature @@ -22,18 +22,15 @@ Feature: DecisionService application-layer coverage Scenario: dsvc- record_decision persists and returns the decision Given dsvc- a DecisionService with a mocked UnitOfWork - And dsvc- a sample root Decision object - When dsvc- I call record_decision with the sample decision + When dsvc- I call record_decision with plan_id and required args Then dsvc- the UoW transaction should have been entered - And dsvc- ctx.decisions.create should have been called with the decision - And dsvc- the returned decision should be the same object + And dsvc- ctx.decisions.create should have been called once + And dsvc- the returned decision should have the correct plan_id - Scenario: dsvc- record_decision logs info and debug messages + Scenario: dsvc- record_decision logs info on success Given dsvc- a DecisionService with a mocked UnitOfWork and captured logger - And dsvc- a sample root Decision object - When dsvc- I call record_decision with the sample decision - Then dsvc- the logger should have recorded an info call with "recording_decision" - And dsvc- the logger should have recorded a debug call with "decision_recorded" + When dsvc- I call record_decision with plan_id and required args + Then dsvc- the logger should have recorded an info call with "decision.recorded" # ------------------------------------------------------------------ # get_decision @@ -46,57 +43,38 @@ Feature: DecisionService application-layer coverage Then dsvc- ctx.decisions.get should have been called with the ID And dsvc- the returned value should be the expected Decision - Scenario: dsvc- get_decision returns None when not found + Scenario: dsvc- get_decision raises DecisionNotFoundError when not found Given dsvc- a DecisionService with a mocked UnitOfWork And dsvc- the mock repo get method returns None - When dsvc- I call get_decision with an unknown ID - Then dsvc- the returned value should be None - - Scenario: dsvc- get_decision logs a debug message - Given dsvc- a DecisionService with a mocked UnitOfWork and captured logger - And dsvc- the mock repo get method returns a Decision - When dsvc- I call get_decision with a known ID - Then dsvc- the logger should have recorded a debug call with "getting_decision" + When dsvc- I call get_decision with an unknown ID expecting not-found + Then dsvc- a DecisionNotFoundError should have been raised # ------------------------------------------------------------------ - # get_decisions_for_plan + # list_decisions # ------------------------------------------------------------------ - Scenario: dsvc- get_decisions_for_plan returns list of decisions + Scenario: dsvc- list_decisions returns list of decisions Given dsvc- a DecisionService with a mocked UnitOfWork And dsvc- the mock repo get_by_plan method returns 3 decisions - When dsvc- I call get_decisions_for_plan with a plan ID + When dsvc- I call list_decisions with a plan ID Then dsvc- ctx.decisions.get_by_plan should have been called with the plan ID And dsvc- the returned list should have 3 decisions - Scenario: dsvc- get_decisions_for_plan returns empty list when none exist + Scenario: dsvc- list_decisions returns empty list when none exist Given dsvc- a DecisionService with a mocked UnitOfWork And dsvc- the mock repo get_by_plan method returns 0 decisions - When dsvc- I call get_decisions_for_plan with a plan ID + When dsvc- I call list_decisions with a plan ID Then dsvc- the returned list should have 0 decisions - Scenario: dsvc- get_decisions_for_plan logs a debug message - Given dsvc- a DecisionService with a mocked UnitOfWork and captured logger - And dsvc- the mock repo get_by_plan method returns 3 decisions - When dsvc- I call get_decisions_for_plan with a plan ID - Then dsvc- the logger should have recorded a debug call with "getting_decisions_for_plan" - # ------------------------------------------------------------------ - # get_decision_tree + # get_tree # ------------------------------------------------------------------ - Scenario: dsvc- get_decision_tree returns BFS-ordered list + Scenario: dsvc- get_tree returns BFS-ordered list Given dsvc- a DecisionService with a mocked UnitOfWork - And dsvc- the mock repo get_tree method returns 4 decisions - When dsvc- I call get_decision_tree with a root ID - Then dsvc- ctx.decisions.get_tree should have been called with the root ID - And dsvc- the returned tree list should have 4 decisions - - Scenario: dsvc- get_decision_tree logs a debug message - Given dsvc- a DecisionService with a mocked UnitOfWork and captured logger - And dsvc- the mock repo get_tree method returns 4 decisions - When dsvc- I call get_decision_tree with a root ID - Then dsvc- the logger should have recorded a debug call with "getting_decision_tree" + And dsvc- the mock repo get_by_plan method returns 4 decisions as a tree + When dsvc- I call get_tree with a plan ID + Then dsvc- the returned tree list should have 4 decisions # ------------------------------------------------------------------ # get_path_to_root @@ -109,18 +87,13 @@ Feature: DecisionService application-layer coverage Then dsvc- ctx.decisions.get_path_to_root should have been called with the leaf ID And dsvc- the returned path list should have 3 decisions - Scenario: dsvc- get_path_to_root logs a debug message - Given dsvc- a DecisionService with a mocked UnitOfWork and captured logger - And dsvc- the mock repo get_path_to_root method returns 3 decisions - When dsvc- I call get_path_to_root with a leaf ID - Then dsvc- the logger should have recorded a debug call with "getting_path_to_root" - # ------------------------------------------------------------------ # mark_superseded # ------------------------------------------------------------------ Scenario: dsvc- mark_superseded updates and returns the decision Given dsvc- a DecisionService with a mocked UnitOfWork + And dsvc- the mock repo has the replacement decision in cache And dsvc- the mock repo update_superseded_by method returns a superseded Decision When dsvc- I call mark_superseded with old and new IDs Then dsvc- ctx.decisions.update_superseded_by should have been called with both IDs @@ -128,9 +101,10 @@ Feature: DecisionService application-layer coverage Scenario: dsvc- mark_superseded logs info with both IDs Given dsvc- a DecisionService with a mocked UnitOfWork and captured logger + And dsvc- the mock repo has the replacement decision in cache And dsvc- the mock repo update_superseded_by method returns a superseded Decision When dsvc- I call mark_superseded with old and new IDs - Then dsvc- the logger should have recorded an info call with "marking_superseded" + Then dsvc- the logger should have recorded an info call with "decision.superseded" # ------------------------------------------------------------------ # list_by_type @@ -148,9 +122,3 @@ Feature: DecisionService application-layer coverage And dsvc- the mock repo list_by_type method returns 0 decisions When dsvc- I call list_by_type with plan ID and type "error_recovery" Then dsvc- the returned type list should have 0 decisions - - Scenario: dsvc- list_by_type logs a debug message - Given dsvc- a DecisionService with a mocked UnitOfWork and captured logger - And dsvc- the mock repo list_by_type method returns 2 decisions - When dsvc- I call list_by_type with plan ID and type "strategy_choice" - Then dsvc- the logger should have recorded a debug call with "listing_by_type" diff --git a/features/steps/decision_di_wiring_steps.py b/features/steps/decision_di_wiring_steps.py index 7c9965444..7cb48c8cf 100644 --- a/features/steps/decision_di_wiring_steps.py +++ b/features/steps/decision_di_wiring_steps.py @@ -10,16 +10,33 @@ existing step definitions. from __future__ import annotations import os -from unittest.mock import MagicMock +from typing import TYPE_CHECKING +from unittest.mock import create_autospec from behave import given, then, when from behave.runner import Context +if TYPE_CHECKING: + from cleveragents.infrastructure.database.unit_of_work import UnitOfWork + # ------------------------------------------------------------------- # Helpers # ------------------------------------------------------------------- +def _make_test_uow() -> UnitOfWork: + """Create a UnitOfWork backed by an in-memory SQLite for testing. + + Uses ``UnitOfWork.__init__`` + ``init_database()`` rather than the + ``__new__()`` anti-pattern to stay in sync with UoW internals. + """ + from cleveragents.infrastructure.database.unit_of_work import UnitOfWork + + uow = UnitOfWork("sqlite:///:memory:") + uow.init_database() + return uow + + def _ensure_test_db_env(context: Context) -> None: """Set up a test database URL if not already set.""" if not hasattr(context, "_decdi_env_saved"): @@ -81,34 +98,15 @@ def step_check_decision_service(context: Context) -> None: @given("decdi- a PlanLifecycleService with a DecisionService") def step_create_lifecycle_with_decision(context: Context) -> None: """Build an in-memory PlanLifecycleService wired with DecisionService.""" - from sqlalchemy import create_engine - from sqlalchemy.orm import sessionmaker - from cleveragents.application.services.decision_service import DecisionService from cleveragents.application.services.plan_lifecycle_service import ( PlanLifecycleService, ) - from cleveragents.infrastructure.database.models import Base - from cleveragents.infrastructure.database.unit_of_work import UnitOfWork + from cleveragents.config.settings import Settings - # Build a fresh in-memory DB - engine = create_engine("sqlite:///:memory:", echo=False) - Base.metadata.create_all(engine) + uow = _make_test_uow() - # Create a UoW that skips migration - uow = UnitOfWork.__new__(UnitOfWork) - uow.database_url = "sqlite:///:memory:" - uow._engine = engine - uow._session_factory = sessionmaker( - bind=engine, - expire_on_commit=False, - autoflush=False, - autocommit=False, - ) - uow._database_initialized = True - uow._prompt_for_migration = None - - mock_settings = MagicMock() + mock_settings = create_autospec(Settings, instance=True) mock_settings.database_url = "sqlite:///:memory:" decision_svc = DecisionService(settings=mock_settings, unit_of_work=uow) @@ -147,8 +145,8 @@ def step_check_strategize_decision(context: Context) -> None: strategy_decisions = [ d for d in decisions if str(d.decision_type) == "strategy_choice" ] - assert len(strategy_decisions) >= 1, ( - f"Expected at least 1 strategy_choice decision, got {len(strategy_decisions)}" + assert len(strategy_decisions) == 1, ( + f"Expected exactly 1 strategy_choice decision, got {len(strategy_decisions)}" ) @@ -196,8 +194,8 @@ def step_check_execute_decision(context: Context) -> None: impl_decisions = [ d for d in decisions if str(d.decision_type) == "implementation_choice" ] - assert len(impl_decisions) >= 1, ( - f"Expected at least 1 implementation_choice decision, got {len(impl_decisions)}" + assert len(impl_decisions) == 1, ( + f"Expected exactly 1 implementation_choice decision, got {len(impl_decisions)}" ) diff --git a/features/steps/decision_recording_steps.py b/features/steps/decision_recording_steps.py index e036daa08..da55104c8 100644 --- a/features/steps/decision_recording_steps.py +++ b/features/steps/decision_recording_steps.py @@ -326,6 +326,37 @@ def step_try_empty_chosen(context: Context) -> None: context.decision_error = exc +@when('I try to record a decision with invalid type "{dtype}"') +def step_try_invalid_type(context: Context, dtype: str) -> None: + try: + context.decision_service.record_decision( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV", + decision_type=dtype, + question="Q", + chosen_option="A", + ) + context.decision_error = None + except (ValueError, ValidationError) as exc: + context.decision_error = exc + + +@when("I try to record a decision with confidence {score:g}") +def step_try_invalid_confidence(context: Context, score: float) -> None: + import pydantic + + try: + context.decision_service.record_decision( + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV", + decision_type=DecisionType.STRATEGY_CHOICE, + question="Q", + chosen_option="A", + confidence_score=score, + ) + context.decision_error = None + except (ValidationError, pydantic.ValidationError) as exc: + context.decision_error = exc + + # --------------------------------------------------------------------------- # When — Retrieval # --------------------------------------------------------------------------- @@ -628,8 +659,12 @@ def step_snapshot_hash_prefix(context: Context, prefix: str) -> None: @then("a dsvc validation error should be raised") def step_validation_error(context: Context) -> None: + import pydantic + assert context.decision_error is not None - assert isinstance(context.decision_error, (ValidationError, ValueError)) + assert isinstance( + context.decision_error, (ValidationError, ValueError, pydantic.ValidationError) + ) @then('the dsvc error should mention "{text}"') @@ -637,6 +672,12 @@ def step_error_mention(context: Context, text: str) -> None: assert text in str(context.decision_error) +@then("a dsvc value error should be raised") +def step_value_error(context: Context) -> None: + assert context.decision_error is not None + assert isinstance(context.decision_error, ValueError) + + @then("a dsvc decision not found error should be raised") def step_not_found_error(context: Context) -> None: assert context.decision_error is not None @@ -680,6 +721,40 @@ def step_tree_root(context: Context) -> None: assert context.tree_result[0].is_root +@then("the dsvc tree should be in BFS level order") +def step_tree_bfs_order(context: Context) -> None: + """Verify that all nodes at depth *d* appear before any node at depth *d+1*. + + We compute each node's depth by walking ``parent_decision_id`` links, + then assert the depth sequence is non-decreasing. + """ + tree: list[Decision] = context.tree_result + if len(tree) <= 1: + return # trivially ordered + + # Build a lookup: decision_id -> Decision + by_id = {d.decision_id: d for d in tree} + + def _depth(node: Decision) -> int: + depth = 0 + current = node + while current.parent_decision_id is not None: + parent = by_id.get(current.parent_decision_id) + if parent is None: + break + depth += 1 + current = parent + return depth + + depths = [_depth(d) for d in tree] + for i in range(1, len(depths)): + assert depths[i] >= depths[i - 1], ( + f"BFS order violated at index {i}: " + f"depth {depths[i]} follows depth {depths[i - 1]}. " + f"Depths: {depths}" + ) + + @then("the dsvc path should have {count:d} decisions") def step_path_count(context: Context, count: int) -> None: assert len(context.path_result) == count @@ -994,6 +1069,27 @@ def step_persisted_decision_service(context: Context) -> None: context._cleanup_handlers.append(_cleanup_db) +@when("I recreate the decision service with the same database") +def step_recreate_service_same_db(context: Context) -> None: + """Destroy the current DecisionService and create a fresh one against the same DB. + + This simulates a service restart: the in-memory sequence counters + are lost and must be rehydrated from the database. + """ + from cleveragents.infrastructure.database.unit_of_work import UnitOfWork + + db_path = context._dsvc_db_path + uow = UnitOfWork(f"sqlite:///{db_path}") + uow.init_database() + + context.decision_service = DecisionService(unit_of_work=uow) + # Preserve _plan_id_registry so symbolic IDs ("P1") still resolve + # to the same ULIDs used before the restart. + context.recorded_decisions = [] + context.decision_error = None + context.decision_result = None + + @when("I try to store a duplicate of the recorded decision") def step_try_store_duplicate(context: Context) -> None: """Attempt to store a decision with the same ID that already exists.""" diff --git a/features/steps/decision_service_coverage_steps.py b/features/steps/decision_service_coverage_steps.py index cd5422756..d9eda5b15 100644 --- a/features/steps/decision_service_coverage_steps.py +++ b/features/steps/decision_service_coverage_steps.py @@ -18,7 +18,10 @@ from behave import given, then, when from behave.runner import Context from ulid import ULID -from cleveragents.application.services.decision_service import DecisionService +from cleveragents.application.services.decision_service import ( + DecisionNotFoundError, + DecisionService, +) from cleveragents.domain.models.core.decision import ( Decision, DecisionType, @@ -126,7 +129,6 @@ def step_dsvc_check_uow(context: Context) -> None: @then("dsvc- the service should have a bound structlog logger") def step_dsvc_check_logger(context: Context) -> None: - # The _logger is a BoundLogger with service="decision" assert hasattr(context.dsvc_service, "_logger"), "Service missing _logger attribute" @@ -174,37 +176,34 @@ def step_dsvc_service_with_logger(context: Context) -> None: # --------------------------------------------------------------------------- -@given("dsvc- a sample root Decision object") -def step_dsvc_sample_decision(context: Context) -> None: - context.dsvc_decision = _make_decision(plan_id=_PLAN_ID) - - -@when("dsvc- I call record_decision with the sample decision") +@when("dsvc- I call record_decision with plan_id and required args") def step_dsvc_call_record(context: Context) -> None: context.dsvc_result = context.dsvc_service.record_decision( - context.dsvc_decision, + plan_id=_PLAN_ID, + decision_type=DecisionType.PROMPT_DEFINITION, + question="Which approach?", + chosen_option="REST API", ) @then("dsvc- the UoW transaction should have been entered") def step_dsvc_txn_entered(context: Context) -> None: - # The fact that ctx.decisions.create was called proves we entered - # the transaction context manager successfully. We verify that - # the mock context's decisions repo was accessed. assert context.dsvc_decisions.create.called, ( "Transaction context was not entered — create was never called" ) -@then("dsvc- ctx.decisions.create should have been called with the decision") +@then("dsvc- ctx.decisions.create should have been called once") def step_dsvc_create_called(context: Context) -> None: - context.dsvc_decisions.create.assert_called_once_with(context.dsvc_decision) + assert context.dsvc_decisions.create.call_count == 1, ( + f"Expected create to be called once, got {context.dsvc_decisions.create.call_count}" + ) -@then("dsvc- the returned decision should be the same object") -def step_dsvc_record_returns_same(context: Context) -> None: - assert context.dsvc_result is context.dsvc_decision, ( - "record_decision should return the same Decision instance" +@then("dsvc- the returned decision should have the correct plan_id") +def step_dsvc_record_returns_correct_plan(context: Context) -> None: + assert context.dsvc_result.plan_id == _PLAN_ID, ( + f"Expected plan_id '{_PLAN_ID}', got '{context.dsvc_result.plan_id}'" ) @@ -220,13 +219,6 @@ def step_dsvc_logger_info(context: Context, event: str) -> None: assert event in events, f"Expected info event '{event}' in {events}" -@then('dsvc- the logger should have recorded a debug call with "{event}"') -def step_dsvc_logger_debug(context: Context, event: str) -> None: - debug_calls = context.dsvc_logger.debug.call_args_list - events = [c.args[0] if c.args else c.kwargs.get("event", "") for c in debug_calls] - assert event in events, f"Expected debug event '{event}' in {events}" - - # --------------------------------------------------------------------------- # get_decision # --------------------------------------------------------------------------- @@ -251,9 +243,13 @@ def step_dsvc_call_get(context: Context) -> None: context.dsvc_result = context.dsvc_service.get_decision(_DECISION_ID) -@when("dsvc- I call get_decision with an unknown ID") +@when("dsvc- I call get_decision with an unknown ID expecting not-found") def step_dsvc_call_get_unknown(context: Context) -> None: - context.dsvc_result = context.dsvc_service.get_decision("NONEXISTENT_ID_12345678") + try: + context.dsvc_service.get_decision("NONEXISTENT_ID_12345678") + context.dsvc_error = None + except DecisionNotFoundError as exc: + context.dsvc_error = exc @then("dsvc- ctx.decisions.get should have been called with the ID") @@ -268,13 +264,14 @@ def step_dsvc_get_returns_expected(context: Context) -> None: ) -@then("dsvc- the returned value should be None") -def step_dsvc_result_is_none(context: Context) -> None: - assert context.dsvc_result is None, f"Expected None, got {context.dsvc_result!r}" +@then("dsvc- a DecisionNotFoundError should have been raised") +def step_dsvc_not_found_raised(context: Context) -> None: + assert context.dsvc_error is not None, "Expected DecisionNotFoundError but got None" + assert isinstance(context.dsvc_error, DecisionNotFoundError) # --------------------------------------------------------------------------- -# get_decisions_for_plan +# list_decisions # --------------------------------------------------------------------------- @@ -297,9 +294,9 @@ def step_dsvc_mock_get_by_plan(context: Context, count: int) -> None: context.dsvc_expected_count = count -@when("dsvc- I call get_decisions_for_plan with a plan ID") -def step_dsvc_call_get_for_plan(context: Context) -> None: - context.dsvc_result = context.dsvc_service.get_decisions_for_plan(_PLAN_ID) +@when("dsvc- I call list_decisions with a plan ID") +def step_dsvc_call_list_decisions(context: Context) -> None: + context.dsvc_result = context.dsvc_service.list_decisions(_PLAN_ID) @then("dsvc- ctx.decisions.get_by_plan should have been called with the plan ID") @@ -315,34 +312,48 @@ def step_dsvc_list_count(context: Context, count: int) -> None: # --------------------------------------------------------------------------- -# get_decision_tree +# get_tree # --------------------------------------------------------------------------- - -@given("dsvc- the mock repo get_tree method returns {count:d} decisions") -def step_dsvc_mock_get_tree(context: Context, count: int) -> None: - root = _make_decision(decision_id=_ROOT_ID, plan_id=_PLAN_ID) - decisions = [root] + [ - _make_decision( - plan_id=_PLAN_ID, - sequence_number=i, - decision_type=DecisionType.STRATEGY_CHOICE, - parent_decision_id=_ROOT_ID, - ) - for i in range(1, count) - ] - context.dsvc_decisions.get_tree.return_value = decisions - context.dsvc_expected_tree_count = count +_CHILD1_ID = str(ULID()) +_CHILD2_ID = str(ULID()) +_GRANDCHILD_ID = str(ULID()) -@when("dsvc- I call get_decision_tree with a root ID") +@given("dsvc- the mock repo get_by_plan method returns 4 decisions as a tree") +def step_dsvc_mock_get_tree(context: Context) -> None: + root = _make_decision( + decision_id=_ROOT_ID, + plan_id=_PLAN_ID, + sequence_number=0, + ) + child1 = _make_decision( + decision_id=_CHILD1_ID, + plan_id=_PLAN_ID, + sequence_number=1, + decision_type=DecisionType.STRATEGY_CHOICE, + parent_decision_id=_ROOT_ID, + ) + child2 = _make_decision( + decision_id=_CHILD2_ID, + plan_id=_PLAN_ID, + sequence_number=2, + decision_type=DecisionType.STRATEGY_CHOICE, + parent_decision_id=_ROOT_ID, + ) + grandchild = _make_decision( + decision_id=_GRANDCHILD_ID, + plan_id=_PLAN_ID, + sequence_number=3, + decision_type=DecisionType.IMPLEMENTATION_CHOICE, + parent_decision_id=_CHILD1_ID, + ) + context.dsvc_decisions.get_by_plan.return_value = [root, child1, child2, grandchild] + + +@when("dsvc- I call get_tree with a plan ID") def step_dsvc_call_get_tree(context: Context) -> None: - context.dsvc_result = context.dsvc_service.get_decision_tree(_ROOT_ID) - - -@then("dsvc- ctx.decisions.get_tree should have been called with the root ID") -def step_dsvc_get_tree_called(context: Context) -> None: - context.dsvc_decisions.get_tree.assert_called_once_with(_ROOT_ID) + context.dsvc_result = context.dsvc_service.get_tree(_PLAN_ID) @then("dsvc- the returned tree list should have {count:d} decisions") @@ -398,6 +409,23 @@ def step_dsvc_path_count(context: Context, count: int) -> None: # --------------------------------------------------------------------------- +@given("dsvc- the mock repo has the replacement decision in cache") +def step_dsvc_seed_replacement_in_cache(context: Context) -> None: + """Seed the replacement decision into the service's in-memory cache. + + ``mark_superseded`` now validates that ``new_decision_id`` exists + before proceeding. We add the replacement to ``_decisions`` so + the validation lookup succeeds. + """ + replacement = _make_decision( + decision_id=_NEW_DECISION_ID, + plan_id=_PLAN_ID, + sequence_number=99, + decision_type=DecisionType.STRATEGY_CHOICE, + ) + context.dsvc_service._decisions[_NEW_DECISION_ID] = replacement + + @given("dsvc- the mock repo update_superseded_by method returns a superseded Decision") def step_dsvc_mock_superseded(context: Context) -> None: context.dsvc_superseded_decision = _make_decision( diff --git a/features/steps/plan_explain_cli_coverage_steps.py b/features/steps/plan_explain_cli_coverage_steps.py index 11107d0db..e71223ce9 100644 --- a/features/steps/plan_explain_cli_coverage_steps.py +++ b/features/steps/plan_explain_cli_coverage_steps.py @@ -217,7 +217,7 @@ def step_pec_mock_tree_decisions(context: Context) -> None: context.pec_plan_id = str(ULID()) decisions = _make_tree_decisions() svc = MagicMock() - svc.get_decisions_for_plan.return_value = decisions + svc.list_decisions.return_value = decisions context.pec_container = _mock_container_with_decision_svc(svc) @@ -226,7 +226,7 @@ def step_pec_mock_deep_decisions(context: Context) -> None: context.pec_plan_id = str(ULID()) decisions = _make_deep_decisions() svc = MagicMock() - svc.get_decisions_for_plan.return_value = decisions + svc.list_decisions.return_value = decisions context.pec_container = _mock_container_with_decision_svc(svc) @@ -260,7 +260,7 @@ def step_pec_mock_superseded_decisions(context: Context) -> None: ), ] svc = MagicMock() - svc.get_decisions_for_plan.return_value = decisions + svc.list_decisions.return_value = decisions context.pec_container = _mock_container_with_decision_svc(svc) @@ -268,7 +268,7 @@ def step_pec_mock_superseded_decisions(context: Context) -> None: def step_pec_mock_empty_decisions(context: Context) -> None: context.pec_plan_id = str(ULID()) svc = MagicMock() - svc.get_decisions_for_plan.return_value = [] + svc.list_decisions.return_value = [] context.pec_container = _mock_container_with_decision_svc(svc) diff --git a/robot/helper_decision_di.py b/robot/helper_decision_di.py index de5ca06d1..b711c2610 100644 --- a/robot/helper_decision_di.py +++ b/robot/helper_decision_di.py @@ -13,19 +13,15 @@ from __future__ import annotations import os import sys from pathlib import Path -from unittest.mock import MagicMock +from unittest.mock import create_autospec # Ensure src is importable when run from workspace root sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) -from sqlalchemy import create_engine # noqa: I001 -from sqlalchemy.orm import sessionmaker - -from cleveragents.application.services.decision_service import DecisionService +from cleveragents.application.services.decision_service import DecisionService # noqa: I001 from cleveragents.application.services.plan_lifecycle_service import ( PlanLifecycleService, ) -from cleveragents.infrastructure.database.models import Base from cleveragents.infrastructure.database.unit_of_work import UnitOfWork @@ -36,20 +32,8 @@ from cleveragents.infrastructure.database.unit_of_work import UnitOfWork def _make_uow() -> UnitOfWork: """Create a UoW backed by an in-memory SQLite database.""" - engine = create_engine("sqlite:///:memory:", echo=False) - Base.metadata.create_all(engine) - - uow = UnitOfWork.__new__(UnitOfWork) - uow.database_url = "sqlite:///:memory:" - uow._engine = engine - uow._session_factory = sessionmaker( - bind=engine, - expire_on_commit=False, - autoflush=False, - autocommit=False, - ) - uow._database_initialized = True - uow._prompt_for_migration = None + uow = UnitOfWork("sqlite:///:memory:") + uow.init_database() return uow @@ -84,7 +68,9 @@ def _resolve_service() -> None: def _record_integration() -> None: """Verify decision recording during plan lifecycle transitions.""" uow = _make_uow() - mock_settings = MagicMock() + from cleveragents.config.settings import Settings + + mock_settings = create_autospec(Settings, instance=True) mock_settings.database_url = "sqlite:///:memory:" decision_svc = DecisionService(settings=mock_settings, unit_of_work=uow) diff --git a/robot/helper_decision_recording.py b/robot/helper_decision_recording.py index 9a2c70e38..10fc2cb8f 100644 --- a/robot/helper_decision_recording.py +++ b/robot/helper_decision_recording.py @@ -104,7 +104,7 @@ def _tree_bfs(): question="Root", chosen_option="Root choice", ) - svc.record_decision( + child1 = svc.record_decision( plan_id=_PLAN_ID, decision_type=DecisionType.STRATEGY_CHOICE, question="Child 1", @@ -118,9 +118,38 @@ def _tree_bfs(): chosen_option="C2", parent_decision_id=root.decision_id, ) + svc.record_decision( + plan_id=_PLAN_ID, + decision_type=DecisionType.IMPLEMENTATION_CHOICE, + question="Grandchild", + chosen_option="GC", + parent_decision_id=child1.decision_id, + ) tree = svc.get_tree(_PLAN_ID) - assert len(tree) == 3, f"expected 3, got {len(tree)}" + assert len(tree) == 4, f"expected 4, got {len(tree)}" assert tree[0].is_root + + # Verify BFS level order: compute depths and check non-decreasing. + by_id = {d.decision_id: d for d in tree} + + def _depth(node): + depth = 0 + current = node + while current.parent_decision_id is not None: + parent = by_id.get(current.parent_decision_id) + if parent is None: + break + depth += 1 + current = parent + return depth + + depths = [_depth(d) for d in tree] + for i in range(1, len(depths)): + assert depths[i] >= depths[i - 1], ( + f"BFS order violated at index {i}: " + f"depth {depths[i]} follows depth {depths[i - 1]}. " + f"Depths: {depths}" + ) print("tree-bfs-ok") diff --git a/src/cleveragents/application/services/__init__.py b/src/cleveragents/application/services/__init__.py index 52c2cc8eb..5ec0076fe 100644 --- a/src/cleveragents/application/services/__init__.py +++ b/src/cleveragents/application/services/__init__.py @@ -19,6 +19,7 @@ from cleveragents.application.services.decision_service import ( DecisionNotFoundError, DecisionService, DuplicateDecisionError, + SequenceConflictError, SnapshotStore, ) from cleveragents.application.services.invariant_service import ( @@ -145,6 +146,7 @@ __all__ = [ "SemanticValidationRule", "SemanticValidationService", "SemanticValidationSeverity", + "SequenceConflictError", "SkeletonCompressorService", "SkillRegistryService", "SnapshotStore", diff --git a/src/cleveragents/application/services/decision_service.py b/src/cleveragents/application/services/decision_service.py index e3f3f2618..d2785a953 100644 --- a/src/cleveragents/application/services/decision_service.py +++ b/src/cleveragents/application/services/decision_service.py @@ -26,7 +26,7 @@ from __future__ import annotations import hashlib import json from collections import deque -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING import structlog @@ -36,6 +36,7 @@ from cleveragents.core.exceptions import ( ValidationError, ) from cleveragents.domain.models.core.decision import ( + ArtifactRef, ContextSnapshot, Decision, DecisionType, @@ -47,6 +48,15 @@ if TYPE_CHECKING: logger = structlog.get_logger(__name__) +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +#: Maximum length for ``actor_reasoning`` (raw LLM trace). Prevents +#: unbounded storage from a misbehaving model while remaining generous +#: enough for typical chain-of-thought traces. +_MAX_ACTOR_REASONING: int = 100_000 + # --------------------------------------------------------------------------- # Custom exceptions @@ -201,6 +211,7 @@ class DecisionService: self._decisions: dict[str, Decision] = {} self._plan_decisions: dict[str, list[str]] = {} self._plan_sequence: dict[str, int] = {} + self._sequence_initialised: set[str] = set() # Snapshot store self.snapshots = SnapshotStore() @@ -227,7 +238,7 @@ class DecisionService: rationale: str = "", actor_reasoning: str | None = None, context_snapshot: ContextSnapshot | None = None, - artifacts_produced: list[Any] | None = None, + artifacts_produced: list[ArtifactRef] | None = None, is_correction: bool = False, corrects_decision_id: str | None = None, correction_reason: str | None = None, @@ -266,6 +277,10 @@ class DecisionService: raise ValidationError("question must not be empty") if not chosen_option or not chosen_option.strip(): raise ValidationError("chosen_option must not be empty") + if actor_reasoning is not None and len(actor_reasoning) > _MAX_ACTOR_REASONING: + raise ValidationError( + f"actor_reasoning exceeds maximum length of {_MAX_ACTOR_REASONING}" + ) # Coerce string to enum if needed if isinstance(decision_type, str): @@ -422,6 +437,13 @@ class DecisionService: if child.decision_id not in visited: queue.append(child) + # Include orphaned subtrees (decisions unreachable from any root). + # These can arise when a parent decision is deleted. They are + # appended in sequence-number order after the BFS-reachable nodes. + if len(visited) < len(all_decisions): + orphans = [d for d in all_decisions if d.decision_id not in visited] + result.extend(orphans) + return result def get_path_to_root(self, decision_id: str) -> list[Decision]: @@ -491,8 +513,17 @@ class DecisionService: The updated :class:`Decision` with superseded_by set. Raises: - DecisionNotFoundError: If the decision does not exist. + DecisionNotFoundError: If the decision or the replacement + decision does not exist. """ + # Validate the replacement decision exists. + replacement = self._decisions.get(new_decision_id) + if replacement is None and self._persisted and self.unit_of_work is not None: + with self.unit_of_work.transaction() as ctx: + replacement = ctx.decisions.get(new_decision_id) + if replacement is None: + raise DecisionNotFoundError(new_decision_id) + if self._persisted and self.unit_of_work is not None: with self.unit_of_work.transaction() as ctx: result = ctx.decisions.update_superseded_by( @@ -534,23 +565,22 @@ class DecisionService: Raises: DecisionNotFoundError: If the decision does not exist. """ - if self._persisted and self.unit_of_work is not None: + # Check cache first; in persisted mode also consult DB. + decision = self._decisions.get(decision_id) + if decision is None and self._persisted and self.unit_of_work is not None: with self.unit_of_work.transaction() as ctx: - ctx.decisions.delete(decision_id) - # Remove from cache - decision = self._decisions.pop(decision_id, None) - if decision: - plan_ids = self._plan_decisions.get(decision.plan_id, []) - if decision_id in plan_ids: - plan_ids.remove(decision_id) - self.snapshots.remove(decision_id) - self._logger.info("decision.deleted", decision_id=decision_id) - return True + decision = ctx.decisions.get(decision_id) - decision = self._decisions.pop(decision_id, None) if decision is None: raise DecisionNotFoundError(decision_id) + # Delete from DB if persisted + if self._persisted and self.unit_of_work is not None: + with self.unit_of_work.transaction() as ctx: + ctx.decisions.delete(decision_id) + + # Remove from cache + self._decisions.pop(decision_id, None) plan_ids = self._plan_decisions.get(decision.plan_id, []) if decision_id in plan_ids: plan_ids.remove(decision_id) @@ -593,19 +623,25 @@ class DecisionService: def count_decisions(self, plan_id: str) -> int: """Return the number of decisions recorded for a plan. + In persisted mode uses ``SELECT COUNT(*)`` for efficiency. + Args: plan_id: ULID of the plan. Returns: Decision count. """ + if self._persisted and self.unit_of_work is not None: + with self.unit_of_work.transaction() as ctx: + return ctx.decisions.count(plan_id) return len(self.list_decisions(plan_id)) def get_next_sequence(self, plan_id: str) -> int: """Return the next available sequence number for a plan. This is useful for callers that need to know the next - sequence number before recording. + sequence number before recording. Triggers DB rehydration + on the first call for a given *plan_id*. Args: plan_id: ULID of the plan. @@ -613,6 +649,9 @@ class DecisionService: Returns: The next sequence number. """ + if plan_id not in self._sequence_initialised: + self._rehydrate_sequence(plan_id) + self._sequence_initialised.add(plan_id) return self._plan_sequence.get(plan_id, 0) # ------------------------------------------------------------------ @@ -620,11 +659,57 @@ class DecisionService: # ------------------------------------------------------------------ def _next_sequence(self, plan_id: str) -> int: - """Generate the next monotonically increasing sequence number.""" + """Generate the next monotonically increasing sequence number. + + On the first call for a given *plan_id* the counter is seeded from + the database (``MAX(sequence_number) + 1``) when a + :class:`UnitOfWork` is available. Subsequent calls for the same + plan use the in-memory counter. + + Raises: + SequenceConflictError: If the generated sequence number is + already present in the in-memory cache (indicates a + concurrency or logic bug). + """ + if plan_id not in self._sequence_initialised: + self._rehydrate_sequence(plan_id) + self._sequence_initialised.add(plan_id) + seq = self._plan_sequence.get(plan_id, 0) + + # Uniqueness guard: ensure no existing decision already holds + # this sequence number for the plan. + for did in self._plan_decisions.get(plan_id, []): + existing = self._decisions.get(did) + if existing is not None and existing.sequence_number == seq: + raise SequenceConflictError(plan_id, seq) + self._plan_sequence[plan_id] = seq + 1 return seq + def _rehydrate_sequence(self, plan_id: str) -> None: + """Seed the in-memory sequence counter from the database. + + When a :class:`UnitOfWork` is wired, queries + ``MAX(sequence_number)`` for *plan_id* so that new decisions + continue the monotonic sequence after a service restart. When + no persistence layer is available the counter starts at ``0`` + (or stays at its current value if already set). + """ + if plan_id in self._plan_sequence: + # Already has a counter — nothing to rehydrate. + return + + if self._persisted and self.unit_of_work is not None: + with self.unit_of_work.transaction() as ctx: + max_seq = ctx.decisions.get_max_sequence_number(plan_id) + if max_seq is not None: + self._plan_sequence[plan_id] = max_seq + 1 + return + + # No DB or no existing rows — start from 0. + self._plan_sequence.setdefault(plan_id, 0) + def _store_decision(self, decision: Decision) -> None: """Store a decision in both persistence and cache.""" decision_id = decision.decision_id diff --git a/src/cleveragents/application/services/plan_lifecycle_service.py b/src/cleveragents/application/services/plan_lifecycle_service.py index f130d7440..235601051 100644 --- a/src/cleveragents/application/services/plan_lifecycle_service.py +++ b/src/cleveragents/application/services/plan_lifecycle_service.py @@ -173,20 +173,11 @@ class PlanLifecycleService: self._actions: dict[str, Action] = {} self._plans: dict[str, Plan] = {} - # Decision sequence counter per plan (monotonic within a plan) - self._decision_seq: dict[str, int] = {} - @property def _persisted(self) -> bool: """Return True when a UnitOfWork is wired for persistence.""" return self.unit_of_work is not None - def _next_seq(self, plan_id: str) -> int: - """Return the next decision sequence number for *plan_id*.""" - seq = self._decision_seq.get(plan_id, -1) + 1 - self._decision_seq[plan_id] = seq - return seq - def _try_record_decision( self, plan_id: str, diff --git a/src/cleveragents/cli/commands/plan.py b/src/cleveragents/cli/commands/plan.py index 6a8d89355..c80986666 100644 --- a/src/cleveragents/cli/commands/plan.py +++ b/src/cleveragents/cli/commands/plan.py @@ -2870,7 +2870,7 @@ def tree_decisions_cmd( container = get_container() svc: DecisionService = container.resolve(_DS) - decisions = svc.get_decisions_for_plan(plan_id) + decisions = svc.list_decisions(plan_id) if not decisions: console.print(f"No decisions found for plan '{plan_id}'.") return diff --git a/src/cleveragents/infrastructure/database/repositories.py b/src/cleveragents/infrastructure/database/repositories.py index 2fe1e24df..35226e32e 100644 --- a/src/cleveragents/infrastructure/database/repositories.py +++ b/src/cleveragents/infrastructure/database/repositories.py @@ -65,6 +65,7 @@ if TYPE_CHECKING: from cleveragents.domain.models.core.checkpoint import Checkpoint import structlog +from sqlalchemy import func as sa_func from sqlalchemy.exc import DatabaseError as SQLAlchemyDatabaseError from sqlalchemy.exc import IntegrityError, OperationalError from sqlalchemy.orm import Session @@ -5172,6 +5173,68 @@ class DecisionRepository: f"Failed to delete decision {decision_id}: {exc}", ) from exc + # --- SEQUENCE ---------------------------------------------------------- + + @database_retry + def get_max_sequence_number(self, plan_id: str) -> int | None: + """Return the highest sequence number for *plan_id*, or ``None``. + + Uses the composite index ``ix_decisions_plan_seq`` for an + efficient ``MAX()`` aggregate. + + Args: + plan_id: ULID of the plan. + + Returns: + The maximum ``sequence_number`` stored for the plan, or + ``None`` when no decisions exist for the plan. + + Raises: + DatabaseError: On transient or unexpected DB errors. + """ + session = self._session() + try: + result: int | None = ( + session.query(sa_func.max(DecisionModel.sequence_number)) + .filter(DecisionModel.plan_id == plan_id) + .scalar() + ) + return result + except (OperationalError, SQLAlchemyDatabaseError) as exc: + raise DatabaseError( + f"Failed to get max sequence number for plan {plan_id}: {exc}", + ) from exc + + # --- COUNT ------------------------------------------------------------- + + @database_retry + def count(self, plan_id: str) -> int: + """Return the number of decisions for *plan_id*. + + Uses ``COUNT(*)`` for efficiency instead of loading all rows. + + Args: + plan_id: ULID of the plan. + + Returns: + The decision count (``0`` when no decisions exist). + + Raises: + DatabaseError: On transient or unexpected DB errors. + """ + session = self._session() + try: + result: int = ( + session.query(sa_func.count(DecisionModel.decision_id)) + .filter(DecisionModel.plan_id == plan_id) + .scalar() + ) or 0 + return result + except (OperationalError, SQLAlchemyDatabaseError) as exc: + raise DatabaseError( + f"Failed to count decisions for plan {plan_id}: {exc}", + ) from exc + # --------------------------------------------------------------------------- # Checkpoint Repository (Stage M6 - checkpointing and rollback) -- 2.52.0