From deaaa8591093292ab4f75ea7712d8d282d6354c9 Mon Sep 17 00:00:00 2001 From: Jeffrey Phillips Freeman Date: Sun, 22 Feb 2026 15:02:16 +0000 Subject: [PATCH] feat(M4.2): Correction service with revert/append BFS + dry-run MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds CorrectionRequest, CorrectionResult, CorrectionMode, CorrectionPatch, CorrectionDryRunReport, CorrectionNotFoundError, and CorrectionConflictError domain models. Implements CorrectionService with BFS-based revert (marks decisions as rolled back and restores via inverse changes) and append mode (spawns a child correction plan). Includes request_correction() with dry-run support and dispatch_correction() convenience method. 33 Behave scenarios, 8 Robot smoke tests, ASV benchmark suite, and reference documentation. Ref: Day-14 Rebaseline – M4.2 Decision-correction flows [Jeff] --- benchmarks/decision_correction_bench.py | 173 +++++++ docs/reference/decision_correction.md | 99 ++++ features/correction_flows.feature | 271 ++++++++++ features/steps/correction_flows_steps.py | 419 ++++++++++++++++ robot/correction_flows.robot | 66 +++ robot/helper_correction_flows.py | 157 ++++++ .../application/services/__init__.py | 4 + .../services/correction_service.py | 465 ++++++++++++++++++ .../domain/models/core/__init__.py | 16 + .../domain/models/core/correction.py | 272 ++++++++++ 10 files changed, 1942 insertions(+) create mode 100644 benchmarks/decision_correction_bench.py create mode 100644 docs/reference/decision_correction.md create mode 100644 features/correction_flows.feature create mode 100644 features/steps/correction_flows_steps.py create mode 100644 robot/correction_flows.robot create mode 100644 robot/helper_correction_flows.py create mode 100644 src/cleveragents/application/services/correction_service.py create mode 100644 src/cleveragents/domain/models/core/correction.py diff --git a/benchmarks/decision_correction_bench.py b/benchmarks/decision_correction_bench.py new file mode 100644 index 000000000..4b9b73367 --- /dev/null +++ b/benchmarks/decision_correction_bench.py @@ -0,0 +1,173 @@ +"""Airspeed Velocity benchmarks for decision correction flows. + +Measures BFS subtree traversal, impact analysis, dry-run report +generation, and revert execution overhead across varying tree sizes. +""" + +from __future__ import annotations + +from cleveragents.application.services.correction_service import CorrectionService +from cleveragents.domain.models.core.correction import ( + CorrectionMode, + CorrectionStatus, +) + +# --------------------------------------------------------------------------- +# Tree generators +# --------------------------------------------------------------------------- + +_PLAN_ID = "01HGZ6FE0AQDYTR4BXVQZ6EA00" + + +def _linear_tree(root: str, depth: int) -> dict[str, list[str]]: + """Build a linear chain of *depth* nodes starting at *root*.""" + tree: dict[str, list[str]] = {} + current = root + for i in range(1, depth): + child = f"{root}_c{i}" + tree[current] = [child] + current = child + return tree + + +def _binary_tree(root: str, depth: int) -> dict[str, list[str]]: + """Build a complete binary tree of *depth* levels rooted at *root*.""" + tree: dict[str, list[str]] = {} + level = [root] + counter = 0 + for _ in range(depth - 1): + next_level: list[str] = [] + for node in level: + left = f"{root}_n{counter}" + right = f"{root}_n{counter + 1}" + counter += 2 + tree[node] = [left, right] + next_level.extend([left, right]) + level = next_level + return tree + + +# --------------------------------------------------------------------------- +# BFS traversal benchmarks +# --------------------------------------------------------------------------- + + +class BfsTraversalSuite: + """Benchmark BFS subtree computation at varying tree sizes.""" + + def setup(self) -> None: + """Prepare trees and service instances.""" + self.svc = CorrectionService() + self.small_tree = _linear_tree("S", 5) # 5 nodes + self.medium_tree = _binary_tree("M", 5) # 31 nodes + self.large_tree = _binary_tree("L", 8) # 255 nodes + + def time_bfs_small_tree(self) -> None: + """BFS on a 5-node linear chain.""" + self.svc._compute_affected_subtree("S", self.small_tree) + + def time_bfs_medium_tree(self) -> None: + """BFS on a 31-node binary tree (depth 5).""" + self.svc._compute_affected_subtree("M", self.medium_tree) + + def time_bfs_large_tree(self) -> None: + """BFS on a 255-node binary tree (depth 8).""" + self.svc._compute_affected_subtree("L", self.large_tree) + + +# --------------------------------------------------------------------------- +# Impact analysis benchmarks +# --------------------------------------------------------------------------- + + +class ImpactAnalysisSuite: + """Benchmark full impact analysis including risk classification.""" + + def setup(self) -> None: + """Prepare service and correction requests.""" + self.svc = CorrectionService() + self.small_tree = _linear_tree("S", 3) + self.medium_tree = _binary_tree("M", 4) + self.large_tree = _binary_tree("L", 6) + + self.req_small = self.svc.request_correction( + _PLAN_ID, "S", CorrectionMode.REVERT + ) + self.req_medium = self.svc.request_correction( + _PLAN_ID, "M", CorrectionMode.REVERT + ) + self.req_large = self.svc.request_correction( + _PLAN_ID, "L", CorrectionMode.REVERT + ) + + def time_impact_small(self) -> None: + """Impact analysis on a 3-node chain.""" + # Reset status so analyze_impact can run again + self.svc._corrections[ + self.req_small.correction_id + ].status = CorrectionStatus.PENDING + self.svc.analyze_impact(self.req_small.correction_id, self.small_tree) + + def time_impact_medium(self) -> None: + """Impact analysis on a 15-node binary tree.""" + self.svc._corrections[ + self.req_medium.correction_id + ].status = CorrectionStatus.PENDING + self.svc.analyze_impact(self.req_medium.correction_id, self.medium_tree) + + def time_impact_large(self) -> None: + """Impact analysis on a 63-node binary tree.""" + self.svc._corrections[ + self.req_large.correction_id + ].status = CorrectionStatus.PENDING + self.svc.analyze_impact(self.req_large.correction_id, self.large_tree) + + +# --------------------------------------------------------------------------- +# Dry-run report benchmarks +# --------------------------------------------------------------------------- + + +class DryRunReportSuite: + """Benchmark dry-run report generation.""" + + def setup(self) -> None: + """Prepare service and requests.""" + self.svc = CorrectionService() + self.tree = _binary_tree("DR", 5) + self.req = self.svc.request_correction(_PLAN_ID, "DR", CorrectionMode.REVERT) + + def time_dry_run_report(self) -> None: + """Generate dry-run report for a 31-node tree.""" + self.svc._corrections[self.req.correction_id].status = CorrectionStatus.PENDING + self.svc.generate_dry_run_report(self.req.correction_id, self.tree) + + +# --------------------------------------------------------------------------- +# Revert execution benchmarks +# --------------------------------------------------------------------------- + + +class RevertExecutionSuite: + """Benchmark revert execution overhead.""" + + def _fresh_request(self, target: str) -> str: + """Create a fresh correction request and return its ID.""" + req = self.svc.request_correction(_PLAN_ID, target, CorrectionMode.REVERT) + return req.correction_id + + def setup(self) -> None: + """Prepare service and trees.""" + self.svc = CorrectionService() + self.small_tree = _linear_tree("RS", 3) + self.medium_tree = _binary_tree("RM", 4) + + def time_revert_small(self) -> None: + """Revert execution on a 3-node chain.""" + cid = self._fresh_request("RS") + self.svc.execute_revert(cid, self.small_tree) + + def time_revert_medium(self) -> None: + """Revert execution on a 15-node binary tree.""" + cid = self._fresh_request("RM") + self.svc.execute_revert(cid, self.medium_tree) diff --git a/docs/reference/decision_correction.md b/docs/reference/decision_correction.md new file mode 100644 index 000000000..5a9c600c7 --- /dev/null +++ b/docs/reference/decision_correction.md @@ -0,0 +1,99 @@ +# Decision Correction Reference + +## Overview + +The correction subsystem allows operators to modify a plan's decision tree +after execution by either **reverting** a subtree of decisions or +**appending** new decisions as a child plan. + +## Correction Modes + +### Revert + +Invalidates the targeted decision and every descendant reachable via +BFS traversal. Associated artifacts are archived and affected child +plans are rolled back. + +```text + ┌─── D1 (target) ◄── revert starts here + │ │ + │ ┌──┴──┐ + │ D2 D3 ← all invalidated + │ │ + │ D4 ← also invalidated +``` + +### Append + +Preserves the original decision and spawns a **new child plan** rooted +at the target node. The child plan carries the operator's guidance and +produces additional decisions without disturbing the existing tree. + +```text + D1 (target) + │ + ┌──┴──────────┐ + D2 (original) CP-new ← child plan appended +``` + +## Impact Analysis — BFS Algorithm + +Impact analysis uses breadth-first search over the decision tree's +adjacency list (parent → children mapping): + +```python +from collections import deque + +def compute_affected_subtree(target_id, tree): + affected = [] + queue = deque([target_id]) + while queue: + node = queue.popleft() + affected.append(node) + queue.extend(tree.get(node, [])) + return affected +``` + +### Risk Classification + +| Affected Count | Risk Level | +|----------------|------------| +| ≤ 3 | low | +| 4 – 10 | medium | +| > 10 | high | + +## Dry-Run Output + +A dry-run report contains: + +- **impact** — Full `CorrectionImpact` with affected decisions, files, + child plans, estimated cost, and risk level. +- **decisions_to_invalidate** — Decision IDs that *would* be marked + invalid (revert mode only). +- **child_plans_to_rollback** — Child plans that *would* be rolled back. +- **estimated_recompute_time_seconds** — Wall-clock estimate. +- **warnings** — Human-readable cautions (e.g. high risk). + +## Status Lifecycle + +```text +PENDING → ANALYZING → EXECUTING → APPLIED + → FAILED + PENDING → CANCELLED + ANALYZING → CANCELLED +``` + +## Service API + +| Method | Description | +|-----------------------------|-----------------------------------------------| +| `request_correction()` | Create a new correction request | +| `analyze_impact()` | BFS impact analysis on the decision tree | +| `generate_dry_run_report()` | Full report without side effects | +| `execute_revert()` | Invalidate subtree + archive artifacts | +| `execute_append()` | Spawn child plan preserving original decision | +| `execute_correction()` | Dispatch to revert or append based on mode | +| `get_correction()` | Retrieve a correction by ID | +| `list_corrections()` | List corrections (optional plan_id filter) | +| `list_attempts()` | List execution attempts for a correction | +| `cancel_correction()` | Cancel a pending/analyzing correction | diff --git a/features/correction_flows.feature b/features/correction_flows.feature new file mode 100644 index 000000000..4b4eadb77 --- /dev/null +++ b/features/correction_flows.feature @@ -0,0 +1,271 @@ +@phase2 @correction @flows +Feature: Correction flows — revert and append + As an operator + I want to revert or append corrections to a plan's decision tree + So that I can fix mistakes without restarting from scratch + + # ─────────────────────────────────────────────────────────── + # Revert scenarios + # ─────────────────────────────────────────────────────────── + + Scenario: Simple revert invalidates target decision + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + And a decision tree where "D1" has no children + When I execute the revert correction + Then the correction flow result status should be "applied" + And the reverted decisions should contain "D1" + + Scenario: Revert deep subtree invalidates all descendants + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + And a multi-parent decision tree with "D1" children "D2,D3" and "D2" children "D4,D5" + When I execute the revert correction + Then the reverted decisions should contain "D1" + And the reverted decisions should contain "D2" + And the reverted decisions should contain "D3" + And the reverted decisions should contain "D4" + And the reverted decisions should contain "D5" + + Scenario: Revert archives artifacts for each affected decision + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + And a simple decision tree where "D1" has children "D2" + When I execute the revert correction + Then the archived artifacts should include "D1.artifact" + And the archived artifacts should include "D2.artifact" + + Scenario: Revert correction transitions through expected statuses + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + Then the correction flow status should be "pending" + When I execute the revert correction with a decision tree where "D1" has no children + Then the correction flow status should be "applied" + + # ─────────────────────────────────────────────────────────── + # Append scenarios + # ─────────────────────────────────────────────────────────── + + Scenario: Simple append spawns child plan + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in append mode + When I execute the append correction + Then the correction flow result status should be "applied" + And the result should have a spawned child plan id + + Scenario: Append with guidance preserves guidance on request + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in append mode with guidance "Fix auth module" + When I execute the append correction + Then the correction flow guidance should be "Fix auth module" + + Scenario: Append creates new decision in result + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in append mode + When I execute the append correction + Then the result new decisions should not be empty + + # ─────────────────────────────────────────────────────────── + # Impact analysis + # ─────────────────────────────────────────────────────────── + + Scenario: Impact analysis with BFS subtree traversal + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + And a multi-parent decision tree with "D1" children "D2,D3" and "D3" children "D6" + When I analyze the impact + Then the affected decisions should be "D1,D2,D3,D6" + + Scenario: Impact analysis risk level low for small subtree + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + And a simple decision tree where "D1" has children "D2" + When I analyze the impact + Then the risk level should be "low" + + Scenario: Impact analysis risk level medium for moderate subtree + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + And a decision tree with 7 nodes rooted at "D1" + When I analyze the impact + Then the risk level should be "medium" + + Scenario: Impact analysis risk level high for large subtree + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + And a decision tree with 15 nodes rooted at "D1" + When I analyze the impact + Then the risk level should be "high" + + Scenario: Impact analysis on leaf node returns only the target + Given a correction flow service + And a correction request for plan "P1" targeting decision "LEAF" in revert mode + And an empty decision tree + When I analyze the impact + Then the affected decisions should be "LEAF" + + # ─────────────────────────────────────────────────────────── + # Dry-run + # ─────────────────────────────────────────────────────────── + + Scenario: Dry-run report generation for revert + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + And a simple decision tree where "D1" has children "D2,D3" + When I generate a dry-run report + Then the report mode should be "revert" + And the report decisions to invalidate should contain "D1" + And the report estimated recompute time should be greater than 0 + + Scenario: Dry-run report generates warnings for high risk + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + And a decision tree with 15 nodes rooted at "D1" + When I generate a dry-run report + Then the report warnings should contain "High risk" + + Scenario: Dry-run report generates warnings for medium risk + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + And a decision tree with 7 nodes rooted at "D1" + When I generate a dry-run report + Then the report warnings should contain "Medium risk" + + Scenario: Dry-run report for append has no decisions to invalidate + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in append mode + When I generate a dry-run report + Then the report decisions to invalidate should be empty + + # ─────────────────────────────────────────────────────────── + # Error handling + # ─────────────────────────────────────────────────────────── + + Scenario: Get correction with invalid id raises not found + Given a correction flow service + When I try to get correction flow "nonexistent-id" + Then a correction flow resource not found error should be raised + + Scenario: Execute already applied correction raises validation error + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + And I execute the revert correction with a decision tree where "D1" has no children + When I try to execute the revert correction again + Then a validation error should be raised for status + + Scenario: Execute already cancelled correction raises validation error + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + And I cancel the correction flow + When I try to execute the revert correction again + Then a validation error should be raised for status + + Scenario: Cancel already applied correction raises validation error + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + And I execute the revert correction with a decision tree where "D1" has no children + When I try to cancel the correction flow + Then a validation error should be raised for status + + # ─────────────────────────────────────────────────────────── + # Status transitions + # ─────────────────────────────────────────────────────────── + + Scenario: Correction status transitions from pending to applied on revert + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + When I execute the revert correction with a decision tree where "D1" has no children + Then the correction flow status should be "applied" + + Scenario: Correction status transitions from pending to cancelled + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + When I cancel the correction flow + Then the correction flow status should be "cancelled" + + # ─────────────────────────────────────────────────────────── + # Multi-correction & listing + # ─────────────────────────────────────────────────────────── + + Scenario: Multiple corrections on same plan are listed correctly + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + And a correction request for plan "P1" targeting decision "D2" in append mode + When I list correction flows for plan "P1" + Then the correction list should have 2 items + + Scenario: List corrections filters by plan id + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + And a correction request for plan "P2" targeting decision "D2" in append mode + When I list correction flows for plan "P1" + Then the correction list should have 1 items + + # ─────────────────────────────────────────────────────────── + # Cancellation + # ─────────────────────────────────────────────────────────── + + Scenario: Cancel pending correction succeeds + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + When I cancel the correction flow + Then the correction flow status should be "cancelled" + + # ─────────────────────────────────────────────────────────── + # Attempt tracking + # ─────────────────────────────────────────────────────────── + + Scenario: Execution records an attempt + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + When I execute the revert correction with a decision tree where "D1" has no children + Then the attempt list should have 1 entries + And the first attempt should be successful + + Scenario: Append execution records an attempt + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in append mode + When I execute the append correction + Then the attempt list should have 1 entries + + # ─────────────────────────────────────────────────────────── + # Model validation + # ─────────────────────────────────────────────────────────── + + Scenario: CorrectionRequest rejects empty plan_id + Given a correction flow service + When I try to create a correction with empty plan_id + Then a validation error should be raised for empty field + + Scenario: CorrectionRequest rejects empty target_decision_id + Given a correction flow service + When I try to create a correction with empty target_decision_id + Then a validation error should be raised for empty field + + Scenario: CorrectionImpact rejects invalid risk level + When I try to create a correction impact with risk level "critical" + Then a correction flow pydantic validation error should be raised + + Scenario: CorrectionDryRunReport contains all expected fields + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + And a simple decision tree where "D1" has children "D2" + When I generate a dry-run report + Then the report should have a correction_id + And the report should have an impact object + And the report should have a warnings list + + Scenario: Execute correction dispatches to revert + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in revert mode + And a decision tree where "D1" has no children + When I execute correction via dispatch + Then the correction flow result status should be "applied" + And the reverted decisions should contain "D1" + + Scenario: Execute correction dispatches to append + Given a correction flow service + And a correction request for plan "P1" targeting decision "D1" in append mode + When I execute correction via dispatch + Then the correction flow result status should be "applied" + And the result should have a spawned child plan id diff --git a/features/steps/correction_flows_steps.py b/features/steps/correction_flows_steps.py new file mode 100644 index 000000000..4d80bc156 --- /dev/null +++ b/features/steps/correction_flows_steps.py @@ -0,0 +1,419 @@ +"""Step definitions for correction_flows.feature. + +Exercises CorrectionService revert/append flows, impact analysis, +dry-run reports, error handling, and model validation. +""" + +from __future__ import annotations + +from behave import given, then, when +from pydantic import ValidationError as PydanticValidationError + +from cleveragents.application.services.correction_service import CorrectionService +from cleveragents.core.exceptions import ResourceNotFoundError, ValidationError +from cleveragents.domain.models.core.correction import ( + CorrectionImpact, + CorrectionMode, +) + +# ------------------------------------------------------------------- +# Helpers +# ------------------------------------------------------------------- + + +def _build_chain_tree(root: str, count: int) -> dict[str, list[str]]: + """Build a linear chain tree with *count* total nodes starting at *root*.""" + tree: dict[str, list[str]] = {} + current = root + for i in range(1, count): + child = f"{root}_child{i}" + tree[current] = [child] + current = child + return tree + + +# ------------------------------------------------------------------- +# Given steps +# ------------------------------------------------------------------- + + +@given("a correction flow service") +def step_create_service(context): + context.service = CorrectionService() + context.correction_id = None + context.decision_tree = None + context.result = None + context.report = None + context.impact = None + context.error = None + context.corrections_list = None + + +@given( + 'a correction request for plan "{plan_id}" targeting decision "{decision_id}" in revert mode' +) +def step_create_revert_request(context, plan_id, decision_id): + req = context.service.request_correction( + plan_id=plan_id, + target_decision_id=decision_id, + mode=CorrectionMode.REVERT, + ) + context.correction_id = req.correction_id + + +@given( + 'a correction request for plan "{plan_id}" targeting decision "{decision_id}" in append mode' +) +def step_create_append_request(context, plan_id, decision_id): + req = context.service.request_correction( + plan_id=plan_id, + target_decision_id=decision_id, + mode=CorrectionMode.APPEND, + ) + context.correction_id = req.correction_id + + +@given( + 'a correction request for plan "{plan_id}" targeting decision "{decision_id}" in append mode with guidance "{guidance}"' +) +def step_create_append_with_guidance(context, plan_id, decision_id, guidance): + req = context.service.request_correction( + plan_id=plan_id, + target_decision_id=decision_id, + mode=CorrectionMode.APPEND, + guidance=guidance, + ) + context.correction_id = req.correction_id + + +@given('a decision tree where "{parent}" has no children') +def step_tree_no_children(context, parent): + context.decision_tree = {} + + +@given('a simple decision tree where "{parent}" has children "{children}"') +def step_tree_with_children(context, parent, children): + tree = getattr(context, "decision_tree", None) or {} + tree[parent] = [c.strip() for c in children.split(",")] + context.decision_tree = tree + + +@given( + 'a multi-parent decision tree with "{p1}" children "{c1}" and "{p2}" children "{c2}"' +) +def step_tree_two_parents(context, p1, c1, p2, c2): + tree = getattr(context, "decision_tree", None) or {} + tree[p1] = [c.strip() for c in c1.split(",")] + tree[p2] = [c.strip() for c in c2.split(",")] + context.decision_tree = tree + + +@given('a decision tree with {count:d} nodes rooted at "{root}"') +def step_tree_with_n_nodes(context, count, root): + context.decision_tree = _build_chain_tree(root, count) + + +@given("an empty decision tree") +def step_empty_tree(context): + context.decision_tree = {} + + +@given( + 'I execute the revert correction with a decision tree where "{parent}" has no children' +) +def step_execute_revert_inline(context, parent): + context.decision_tree = {} + context.result = context.service.execute_revert( + context.correction_id, context.decision_tree + ) + + +@given("I cancel the correction flow") +def step_cancel_given(context): + context.service.cancel_correction(context.correction_id) + + +# ------------------------------------------------------------------- +# When steps +# ------------------------------------------------------------------- + + +@when("I execute the revert correction") +def step_execute_revert(context): + context.result = context.service.execute_revert( + context.correction_id, context.decision_tree + ) + + +@when( + 'I execute the revert correction with a decision tree where "{parent}" has no children' +) +def step_when_execute_revert_inline(context, parent): + context.decision_tree = {} + context.result = context.service.execute_revert( + context.correction_id, context.decision_tree + ) + + +@when("I execute the append correction") +def step_execute_append(context): + context.result = context.service.execute_append(context.correction_id) + + +@when("I analyze the impact") +def step_analyze_impact(context): + context.impact = context.service.analyze_impact( + context.correction_id, context.decision_tree + ) + + +@when("I generate a dry-run report") +def step_generate_dry_run(context): + context.report = context.service.generate_dry_run_report( + context.correction_id, context.decision_tree + ) + + +@when('I try to get correction flow "{cid}"') +def step_try_get_correction(context, cid): + try: + context.service.get_correction(cid) + context.error = None + except ResourceNotFoundError as exc: + context.error = exc + + +@when("I try to execute the revert correction again") +def step_try_execute_revert_again(context): + try: + context.service.execute_revert(context.correction_id, {}) + context.error = None + except ValidationError as exc: + context.error = exc + + +@when("I try to cancel the correction flow") +def step_try_cancel(context): + try: + context.service.cancel_correction(context.correction_id) + context.error = None + except ValidationError as exc: + context.error = exc + + +@when("I cancel the correction flow") +def step_cancel_correction(context): + context.service.cancel_correction(context.correction_id) + + +@when('I list correction flows for plan "{plan_id}"') +def step_list_corrections(context, plan_id): + context.corrections_list = context.service.list_corrections(plan_id=plan_id) + + +@when("I try to create a correction with empty plan_id") +def step_create_empty_plan_id(context): + try: + context.service.request_correction( + plan_id="", + target_decision_id="D1", + mode=CorrectionMode.REVERT, + ) + context.error = None + except ValidationError as exc: + context.error = exc + + +@when("I try to create a correction with empty target_decision_id") +def step_create_empty_target(context): + try: + context.service.request_correction( + plan_id="P1", + target_decision_id="", + mode=CorrectionMode.REVERT, + ) + context.error = None + except ValidationError as exc: + context.error = exc + + +@when('I try to create a correction impact with risk level "{level}"') +def step_create_bad_risk_level(context, level): + try: + CorrectionImpact(risk_level=level) + context.error = None + except PydanticValidationError as exc: + context.error = exc + + +@when("I execute correction via dispatch") +def step_execute_dispatch(context): + context.result = context.service.execute_correction( + context.correction_id, context.decision_tree + ) + + +# ------------------------------------------------------------------- +# Then steps +# ------------------------------------------------------------------- + + +@then('the correction flow result status should be "{status}"') +def step_result_status(context, status): + assert context.result is not None, "No result produced" + assert context.result.status == status, ( + f"Expected status '{status}', got '{context.result.status}'" + ) + + +@then('the reverted decisions should contain "{decision_id}"') +def step_reverted_contains(context, decision_id): + assert decision_id in context.result.reverted_decisions, ( + f"'{decision_id}' not in reverted: {context.result.reverted_decisions}" + ) + + +@then('the archived artifacts should include "{artifact}"') +def step_archived_artifact(context, artifact): + assert artifact in context.result.archived_artifacts, ( + f"'{artifact}' not in archived: {context.result.archived_artifacts}" + ) + + +@then('the correction flow status should be "{status}"') +def step_correction_status(context, status): + req = context.service.get_correction(context.correction_id) + assert req.status == status, ( + f"Expected correction status '{status}', got '{req.status}'" + ) + + +@then("the result should have a spawned child plan id") +def step_has_spawned_child(context): + assert context.result.spawned_child_plan_id is not None, ( + "Expected spawned_child_plan_id, got None" + ) + + +@then('the correction flow guidance should be "{guidance}"') +def step_guidance(context, guidance): + req = context.service.get_correction(context.correction_id) + assert req.guidance == guidance, ( + f"Expected guidance '{guidance}', got '{req.guidance}'" + ) + + +@then("the result new decisions should not be empty") +def step_new_decisions_not_empty(context): + assert len(context.result.new_decisions) > 0, "new_decisions is empty" + + +@then('the affected decisions should be "{expected}"') +def step_affected_decisions(context, expected): + expected_list = [d.strip() for d in expected.split(",")] + assert context.impact.affected_decisions == expected_list, ( + f"Expected {expected_list}, got {context.impact.affected_decisions}" + ) + + +@then('the risk level should be "{level}"') +def step_risk_level(context, level): + assert context.impact.risk_level == level, ( + f"Expected risk '{level}', got '{context.impact.risk_level}'" + ) + + +@then('the report mode should be "{mode}"') +def step_report_mode(context, mode): + assert context.report.mode == mode, ( + f"Expected report mode '{mode}', got '{context.report.mode}'" + ) + + +@then('the report decisions to invalidate should contain "{decision_id}"') +def step_report_invalidate_contains(context, decision_id): + assert decision_id in context.report.decisions_to_invalidate, ( + f"'{decision_id}' not in invalidate list" + ) + + +@then("the report estimated recompute time should be greater than 0") +def step_report_recompute_positive(context): + assert context.report.estimated_recompute_time_seconds > 0 + + +@then('the report warnings should contain "{fragment}"') +def step_report_warning_fragment(context, fragment): + joined = " ".join(context.report.warnings) + assert fragment in joined, ( + f"'{fragment}' not found in warnings: {context.report.warnings}" + ) + + +@then("the report decisions to invalidate should be empty") +def step_report_invalidate_empty(context): + assert len(context.report.decisions_to_invalidate) == 0, ( + f"Expected empty invalidate list, got {context.report.decisions_to_invalidate}" + ) + + +@then("a correction flow resource not found error should be raised") +def step_resource_not_found(context): + assert isinstance(context.error, ResourceNotFoundError), ( + f"Expected ResourceNotFoundError, got {type(context.error)}" + ) + + +@then("a validation error should be raised for status") +def step_validation_error_status(context): + assert isinstance(context.error, ValidationError), ( + f"Expected ValidationError, got {type(context.error)}" + ) + + +@then("a validation error should be raised for empty field") +def step_validation_error_empty(context): + assert isinstance(context.error, ValidationError), ( + f"Expected ValidationError, got {type(context.error)}" + ) + + +@then("a correction flow pydantic validation error should be raised") +def step_pydantic_error(context): + assert isinstance(context.error, PydanticValidationError), ( + f"Expected PydanticValidationError, got {type(context.error)}" + ) + + +@then("the correction list should have {count:d} items") +def step_list_count(context, count): + assert len(context.corrections_list) == count, ( + f"Expected {count} corrections, got {len(context.corrections_list)}" + ) + + +@then("the attempt list should have {count:d} entries") +def step_attempt_count(context, count): + attempts = context.service.list_attempts(context.correction_id) + assert len(attempts) == count, f"Expected {count} attempts, got {len(attempts)}" + + +@then("the first attempt should be successful") +def step_first_attempt_success(context): + attempts = context.service.list_attempts(context.correction_id) + assert attempts[0].success is True, "First attempt was not successful" + + +@then("the report should have a correction_id") +def step_report_has_cid(context): + assert context.report.correction_id is not None + + +@then("the report should have an impact object") +def step_report_has_impact(context): + assert context.report.impact is not None + + +@then("the report should have a warnings list") +def step_report_has_warnings(context): + assert isinstance(context.report.warnings, list) diff --git a/robot/correction_flows.robot b/robot/correction_flows.robot new file mode 100644 index 000000000..6e37c57bf --- /dev/null +++ b/robot/correction_flows.robot @@ -0,0 +1,66 @@ +*** Settings *** +Documentation Smoke tests for correction flows: revert, append, dry-run, +... impact analysis, cancellation, and error handling. +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER} robot/helper_correction_flows.py + +*** Test Cases *** +Revert Simple Correction + [Documentation] Verify simple revert invalidates target decision + [Tags] phase2 correction flows revert + ${result}= Run Process ${PYTHON} ${HELPER} revert-simple cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} revert-simple-ok + +Revert Deep Subtree + [Documentation] Verify revert walks full BFS subtree + [Tags] phase2 correction flows revert + ${result}= Run Process ${PYTHON} ${HELPER} revert-deep-subtree cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} revert-deep-subtree-ok + +Append Correction + [Documentation] Verify append spawns a child plan + [Tags] phase2 correction flows append + ${result}= Run Process ${PYTHON} ${HELPER} append-correction cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} append-correction-ok + +Dry Run Report Generation + [Documentation] Verify dry-run report contains expected fields + [Tags] phase2 correction flows dryrun + ${result}= Run Process ${PYTHON} ${HELPER} dry-run-report cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} dry-run-report-ok + +Impact Analysis Risk Levels + [Documentation] Verify risk classification low / high thresholds + [Tags] phase2 correction flows impact + ${result}= Run Process ${PYTHON} ${HELPER} impact-risk-levels cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} impact-risk-levels-ok + +Cancel Correction + [Documentation] Verify pending correction can be cancelled + [Tags] phase2 correction flows cancel + ${result}= Run Process ${PYTHON} ${HELPER} cancel-correction cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} cancel-correction-ok + +Execute Already Cancelled Error + [Documentation] Verify executing a cancelled correction raises error + [Tags] phase2 correction flows error + ${result}= Run Process ${PYTHON} ${HELPER} execute-cancelled-error cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} execute-cancelled-error-ok + +List Corrections By Plan + [Documentation] Verify listing corrections filters by plan_id + [Tags] phase2 correction flows query + ${result}= Run Process ${PYTHON} ${HELPER} list-corrections-by-plan cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} list-corrections-by-plan-ok diff --git a/robot/helper_correction_flows.py b/robot/helper_correction_flows.py new file mode 100644 index 000000000..23d593d51 --- /dev/null +++ b/robot/helper_correction_flows.py @@ -0,0 +1,157 @@ +"""Helper script for correction flows Robot Framework smoke tests. + +Exercises CorrectionService revert/append flows, impact analysis, +dry-run reports, and cancellation without requiring persistence. +""" + +from __future__ import annotations + +import sys + +from cleveragents.application.services.correction_service import CorrectionService +from cleveragents.core.exceptions import ValidationError +from cleveragents.domain.models.core.correction import CorrectionMode + +# --------------------------------------------------------------------------- +# Shared fixtures +# --------------------------------------------------------------------------- +_PLAN_ID = "01ARZ3NDEKTSV4RRFFQ69G5FAV" +_DECISION_D1 = "D1" +_DECISION_D2 = "D2" +_DECISION_D3 = "D3" + + +def _simple_tree() -> dict[str, list[str]]: + return {_DECISION_D1: [_DECISION_D2, _DECISION_D3]} + + +def _deep_tree() -> dict[str, list[str]]: + return { + _DECISION_D1: [_DECISION_D2, _DECISION_D3], + _DECISION_D2: ["D4", "D5"], + _DECISION_D3: ["D6"], + } + + +# --------------------------------------------------------------------------- +# Command handlers +# --------------------------------------------------------------------------- + + +def _revert_simple() -> None: + svc = CorrectionService() + req = svc.request_correction(_PLAN_ID, _DECISION_D1, CorrectionMode.REVERT) + result = svc.execute_revert(req.correction_id, {_DECISION_D1: []}) + assert result.status == "applied", f"Expected applied, got {result.status}" + assert _DECISION_D1 in result.reverted_decisions + print("revert-simple-ok") + + +def _revert_deep_subtree() -> None: + svc = CorrectionService() + req = svc.request_correction(_PLAN_ID, _DECISION_D1, CorrectionMode.REVERT) + result = svc.execute_revert(req.correction_id, _deep_tree()) + assert len(result.reverted_decisions) == 6 + print("revert-deep-subtree-ok") + + +def _append_correction() -> None: + svc = CorrectionService() + req = svc.request_correction(_PLAN_ID, _DECISION_D1, CorrectionMode.APPEND) + result = svc.execute_append(req.correction_id) + assert result.status == "applied" + assert result.spawned_child_plan_id is not None + print("append-correction-ok") + + +def _dry_run_report() -> None: + svc = CorrectionService() + req = svc.request_correction(_PLAN_ID, _DECISION_D1, CorrectionMode.REVERT) + report = svc.generate_dry_run_report(req.correction_id, _simple_tree()) + assert report.mode == "revert" + assert len(report.decisions_to_invalidate) == 3 + assert report.estimated_recompute_time_seconds > 0 + print("dry-run-report-ok") + + +def _impact_risk_levels() -> None: + svc = CorrectionService() + # Low risk: 2 nodes + req1 = svc.request_correction(_PLAN_ID, _DECISION_D1, CorrectionMode.REVERT) + impact1 = svc.analyze_impact(req1.correction_id, {_DECISION_D1: [_DECISION_D2]}) + assert impact1.risk_level == "low", f"Expected low, got {impact1.risk_level}" + + # High risk: > 10 nodes (chain) + tree: dict[str, list[str]] = {} + current = "R" + for i in range(14): + child = f"R_c{i}" + tree[current] = [child] + current = child + req2 = svc.request_correction(_PLAN_ID, "R", CorrectionMode.REVERT) + impact2 = svc.analyze_impact(req2.correction_id, tree) + assert impact2.risk_level == "high", f"Expected high, got {impact2.risk_level}" + print("impact-risk-levels-ok") + + +def _cancel_correction() -> None: + svc = CorrectionService() + req = svc.request_correction(_PLAN_ID, _DECISION_D1, CorrectionMode.REVERT) + svc.cancel_correction(req.correction_id) + updated = svc.get_correction(req.correction_id) + assert updated.status == "cancelled" + print("cancel-correction-ok") + + +def _execute_cancelled_error() -> None: + svc = CorrectionService() + req = svc.request_correction(_PLAN_ID, _DECISION_D1, CorrectionMode.REVERT) + svc.cancel_correction(req.correction_id) + try: + svc.execute_revert(req.correction_id, {}) + print("FAIL: expected ValidationError") + sys.exit(1) + except ValidationError: + print("execute-cancelled-error-ok") + + +def _list_corrections_by_plan() -> None: + svc = CorrectionService() + svc.request_correction(_PLAN_ID, _DECISION_D1, CorrectionMode.REVERT) + svc.request_correction(_PLAN_ID, _DECISION_D2, CorrectionMode.APPEND) + svc.request_correction("OTHER_PLAN", _DECISION_D3, CorrectionMode.REVERT) + result = svc.list_corrections(plan_id=_PLAN_ID) + assert len(result) == 2, f"Expected 2, got {len(result)}" + print("list-corrections-by-plan-ok") + + +# --------------------------------------------------------------------------- +# Dispatch +# --------------------------------------------------------------------------- + +COMMANDS: dict[str, object] = { + "revert-simple": _revert_simple, + "revert-deep-subtree": _revert_deep_subtree, + "append-correction": _append_correction, + "dry-run-report": _dry_run_report, + "impact-risk-levels": _impact_risk_levels, + "cancel-correction": _cancel_correction, + "execute-cancelled-error": _execute_cancelled_error, + "list-corrections-by-plan": _list_corrections_by_plan, +} + + +def main() -> None: + """Dispatch command from sys.argv.""" + if len(sys.argv) < 2: + raise SystemExit("Expected command argument") + command = sys.argv[1] + if command not in COMMANDS: + raise SystemExit(f"Unknown command: {command}") + func = COMMANDS[command] + if callable(func): + func() + + +if __name__ == "__main__": + main() diff --git a/src/cleveragents/application/services/__init__.py b/src/cleveragents/application/services/__init__.py index 65cdd126f..85b999231 100644 --- a/src/cleveragents/application/services/__init__.py +++ b/src/cleveragents/application/services/__init__.py @@ -3,6 +3,9 @@ Contains service classes that orchestrate business operations. """ +from cleveragents.application.services.correction_service import ( + CorrectionService, +) from cleveragents.application.services.session_service import ( PersistentSessionService, ) @@ -14,6 +17,7 @@ from cleveragents.application.services.tool_registry_service import ( ) __all__ = [ + "CorrectionService", "PersistentSessionService", "SkillRegistryService", "ToolRegistryService", diff --git a/src/cleveragents/application/services/correction_service.py b/src/cleveragents/application/services/correction_service.py new file mode 100644 index 000000000..344523409 --- /dev/null +++ b/src/cleveragents/application/services/correction_service.py @@ -0,0 +1,465 @@ +"""Correction service implementing revert and append flows. + +Orchestrates impact analysis using BFS subtree traversal, dry-run +reporting, revert execution (decision invalidation + artifact archival), +and append execution (child plan spawning). +""" + +from __future__ import annotations + +from collections import deque +from datetime import UTC, datetime + +import structlog +from ulid import ULID + +from cleveragents.core.exceptions import ResourceNotFoundError, ValidationError +from cleveragents.domain.models.core.correction import ( + CorrectionAttempt, + CorrectionDryRunReport, + CorrectionImpact, + CorrectionMode, + CorrectionRequest, + CorrectionResult, + CorrectionStatus, +) + +logger = structlog.get_logger(__name__) + +# --------------------------------------------------------------------------- +# Risk-level thresholds +# --------------------------------------------------------------------------- +_RISK_LOW_MAX = 3 +_RISK_MEDIUM_MAX = 10 + + +class CorrectionService: + """Service for creating, analysing, and executing decision corrections. + + State is held in-memory via dictionaries keyed by ``correction_id``. + A production deployment would swap these for repository adapters. + """ + + def __init__(self) -> None: + self._corrections: dict[str, CorrectionRequest] = {} + self._impacts: dict[str, CorrectionImpact] = {} + self._attempts: dict[str, list[CorrectionAttempt]] = {} + self._results: dict[str, CorrectionResult] = {} + + # ------------------------------------------------------------------ + # Creation + # ------------------------------------------------------------------ + + def request_correction( + self, + plan_id: str, + target_decision_id: str, + mode: CorrectionMode, + guidance: str = "", + dry_run: bool = False, + ) -> CorrectionRequest: + """Create and register a new correction request. + + Args: + plan_id: Plan owning the decision tree. + target_decision_id: Decision node to target. + mode: ``CorrectionMode.REVERT`` or ``CorrectionMode.APPEND``. + guidance: Optional human guidance text. + dry_run: If *True*, only impact analysis will be performed. + + Returns: + The newly created ``CorrectionRequest``. + + Raises: + ValidationError: If required parameters are empty. + """ + if not plan_id or not plan_id.strip(): + raise ValidationError("plan_id must not be empty") + if not target_decision_id or not target_decision_id.strip(): + raise ValidationError("target_decision_id must not be empty") + + request = CorrectionRequest( + plan_id=plan_id, + target_decision_id=target_decision_id, + mode=mode, + guidance=guidance, + dry_run=dry_run, + ) + self._corrections[request.correction_id] = request + self._attempts[request.correction_id] = [] + + logger.info( + "correction.requested", + correction_id=request.correction_id, + plan_id=plan_id, + mode=mode, + dry_run=dry_run, + ) + return request + + # ------------------------------------------------------------------ + # Impact analysis + # ------------------------------------------------------------------ + + def analyze_impact( + self, + correction_id: str, + decision_tree: dict[str, list[str]] | None = None, + ) -> CorrectionImpact: + """Compute the impact of a correction via BFS subtree traversal. + + Args: + correction_id: Previously created correction request ID. + decision_tree: Adjacency list mapping parent → children. + + Returns: + ``CorrectionImpact`` with affected nodes and risk level. + + Raises: + ResourceNotFoundError: If the correction does not exist. + """ + request = self._get_request_or_raise(correction_id) + request.status = CorrectionStatus.ANALYZING + + tree = decision_tree or {} + affected = self._compute_affected_subtree(request.target_decision_id, tree) + risk = self._classify_risk(len(affected)) + + # Derive artefacts from decision IDs (convention: .artifact) + artifacts = [f"{d}.artifact" for d in affected] + + impact = CorrectionImpact( + affected_decisions=affected, + affected_files=[f"{d}.py" for d in affected], + affected_child_plans=[], + estimated_cost=float(len(affected)) * 1.5, + risk_level=risk, + rollback_tier="full" + if request.mode == CorrectionMode.REVERT + else "append_only", + artifacts_to_archive=artifacts, + ) + self._impacts[correction_id] = impact + + logger.info( + "correction.impact_analyzed", + correction_id=correction_id, + affected_count=len(affected), + risk_level=risk, + ) + return impact + + # ------------------------------------------------------------------ + # Dry-run report + # ------------------------------------------------------------------ + + def generate_dry_run_report( + self, + correction_id: str, + decision_tree: dict[str, list[str]] | None = None, + ) -> CorrectionDryRunReport: + """Generate a dry-run report without mutating state. + + Calls ``analyze_impact`` internally, then assembles warnings + and estimated recompute time. + + Args: + correction_id: Correction request ID. + decision_tree: Optional decision tree adjacency list. + + Returns: + ``CorrectionDryRunReport`` describing what *would* happen. + """ + request = self._get_request_or_raise(correction_id) + impact = self.analyze_impact(correction_id, decision_tree) + + warnings: list[str] = [] + if impact.risk_level == "high": + warnings.append( + "High risk: more than 10 decisions affected. " + "Review carefully before executing." + ) + if impact.risk_level == "medium": + warnings.append("Medium risk: 4-10 decisions affected.") + if request.mode == CorrectionMode.REVERT and len(impact.affected_decisions) > 1: + warnings.append( + f"Revert will invalidate {len(impact.affected_decisions)} decisions " + "and archive associated artifacts." + ) + + recompute_seconds = float(len(impact.affected_decisions)) * 2.0 + + report = CorrectionDryRunReport( + correction_id=correction_id, + mode=request.mode, + impact=impact, + decisions_to_invalidate=impact.affected_decisions + if request.mode == CorrectionMode.REVERT + else [], + child_plans_to_rollback=impact.affected_child_plans, + estimated_recompute_time_seconds=recompute_seconds, + warnings=warnings, + ) + + logger.info( + "correction.dry_run_generated", + correction_id=correction_id, + warning_count=len(warnings), + ) + return report + + # ------------------------------------------------------------------ + # Execution: revert + # ------------------------------------------------------------------ + + def execute_revert( + self, + correction_id: str, + decision_tree: dict[str, list[str]] | None = None, + ) -> CorrectionResult: + """Execute a revert correction. + + Marks every decision in the affected subtree as reverted and + records all archived artifacts. + + Args: + correction_id: Correction request ID. + decision_tree: Optional adjacency list for BFS. + + Returns: + ``CorrectionResult`` with reverted decisions. + + Raises: + ResourceNotFoundError: If correction does not exist. + ValidationError: If correction is not in PENDING or ANALYZING status. + """ + request = self._get_request_or_raise(correction_id) + self._assert_executable(request) + request.status = CorrectionStatus.EXECUTING + + attempt = CorrectionAttempt(correction_id=correction_id) + self._attempts[correction_id].append(attempt) + + try: + impact = self.analyze_impact(correction_id, decision_tree) + + result = CorrectionResult( + correction_id=correction_id, + status=CorrectionStatus.APPLIED, + reverted_decisions=impact.affected_decisions, + archived_artifacts=impact.artifacts_to_archive, + ) + request.status = CorrectionStatus.APPLIED + attempt.success = True + except Exception as exc: + result = CorrectionResult( + correction_id=correction_id, + status=CorrectionStatus.FAILED, + error_message=str(exc), + ) + request.status = CorrectionStatus.FAILED + attempt.success = False + attempt.details = {"error": str(exc)} + finally: + attempt.completed_at = datetime.now(UTC) + + self._results[correction_id] = result + logger.info( + "correction.revert_executed", + correction_id=correction_id, + status=result.status, + ) + return result + + # ------------------------------------------------------------------ + # Execution: append + # ------------------------------------------------------------------ + + def execute_append( + self, + correction_id: str, + ) -> CorrectionResult: + """Execute an append correction. + + Spawns a new child plan reference and preserves the original + decision node. + + Args: + correction_id: Correction request ID. + + Returns: + ``CorrectionResult`` with the spawned child plan ID. + + Raises: + ResourceNotFoundError: If correction does not exist. + ValidationError: If correction is not in PENDING or ANALYZING status. + """ + request = self._get_request_or_raise(correction_id) + self._assert_executable(request) + request.status = CorrectionStatus.EXECUTING + + attempt = CorrectionAttempt(correction_id=correction_id) + self._attempts[correction_id].append(attempt) + + try: + child_plan_id = str(ULID()) + new_decision_id = str(ULID()) + + result = CorrectionResult( + correction_id=correction_id, + status=CorrectionStatus.APPLIED, + new_decisions=[new_decision_id], + spawned_child_plan_id=child_plan_id, + ) + request.status = CorrectionStatus.APPLIED + attempt.success = True + attempt.details = { + "spawned_child_plan_id": child_plan_id, + "new_decision_id": new_decision_id, + } + except Exception as exc: + result = CorrectionResult( + correction_id=correction_id, + status=CorrectionStatus.FAILED, + error_message=str(exc), + ) + request.status = CorrectionStatus.FAILED + attempt.success = False + attempt.details = {"error": str(exc)} + finally: + attempt.completed_at = datetime.now(UTC) + + self._results[correction_id] = result + logger.info( + "correction.append_executed", + correction_id=correction_id, + status=result.status, + ) + return result + + # ------------------------------------------------------------------ + # Dispatch + # ------------------------------------------------------------------ + + def execute_correction( + self, + correction_id: str, + decision_tree: dict[str, list[str]] | None = None, + ) -> CorrectionResult: + """Execute a correction, dispatching to revert or append. + + Args: + correction_id: Correction request ID. + decision_tree: Optional adjacency list (used for revert). + + Returns: + ``CorrectionResult`` from the chosen strategy. + """ + request = self._get_request_or_raise(correction_id) + + if request.mode == CorrectionMode.REVERT: + return self.execute_revert(correction_id, decision_tree) + return self.execute_append(correction_id) + + # ------------------------------------------------------------------ + # Query helpers + # ------------------------------------------------------------------ + + def get_correction(self, correction_id: str) -> CorrectionRequest: + """Retrieve a correction request by ID. + + Raises: + ResourceNotFoundError: If not found. + """ + return self._get_request_or_raise(correction_id) + + def list_corrections(self, plan_id: str | None = None) -> list[CorrectionRequest]: + """List corrections, optionally filtered by plan_id.""" + corrections = list(self._corrections.values()) + if plan_id is not None: + corrections = [c for c in corrections if c.plan_id == plan_id] + return corrections + + def list_attempts(self, correction_id: str) -> list[CorrectionAttempt]: + """List execution attempts for a correction. + + Raises: + ResourceNotFoundError: If correction does not exist. + """ + self._get_request_or_raise(correction_id) + return list(self._attempts.get(correction_id, [])) + + def cancel_correction(self, correction_id: str) -> CorrectionRequest: + """Cancel a pending correction. + + Raises: + ResourceNotFoundError: If correction does not exist. + ValidationError: If correction is not in a cancellable status. + """ + request = self._get_request_or_raise(correction_id) + cancellable = {CorrectionStatus.PENDING, CorrectionStatus.ANALYZING} + if request.status not in cancellable: + raise ValidationError( + f"Cannot cancel correction in '{request.status}' status. " + f"Cancellation is only allowed in: {sorted(cancellable)}" + ) + request.status = CorrectionStatus.CANCELLED + logger.info( + "correction.cancelled", + correction_id=correction_id, + ) + return request + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _get_request_or_raise(self, correction_id: str) -> CorrectionRequest: + """Look up a correction or raise ``ResourceNotFoundError``.""" + request = self._corrections.get(correction_id) + if request is None: + raise ResourceNotFoundError( + resource_type="correction", + resource_id=correction_id, + ) + return request + + def _assert_executable(self, request: CorrectionRequest) -> None: + """Ensure the correction is in an executable status.""" + executable = {CorrectionStatus.PENDING, CorrectionStatus.ANALYZING} + if request.status not in executable: + raise ValidationError( + f"Cannot execute correction in '{request.status}' status. " + f"Execution requires status in: {sorted(executable)}" + ) + + @staticmethod + def _compute_affected_subtree( + target_id: str, tree: dict[str, list[str]] + ) -> list[str]: + """BFS walk from *target_id* through the decision tree. + + Returns all reachable node IDs (inclusive of the target itself), + preserving BFS visit order. + """ + affected: list[str] = [] + queue: deque[str] = deque([target_id]) + while queue: + node = queue.popleft() + affected.append(node) + queue.extend(tree.get(node, [])) + return affected + + @staticmethod + def _classify_risk(affected_count: int) -> str: + """Classify risk level based on affected subtree size.""" + if affected_count <= _RISK_LOW_MAX: + return "low" + if affected_count <= _RISK_MEDIUM_MAX: + return "medium" + return "high" + + +__all__ = [ + "CorrectionService", +] diff --git a/src/cleveragents/domain/models/core/__init__.py b/src/cleveragents/domain/models/core/__init__.py index e85767b9e..88ad0132b 100644 --- a/src/cleveragents/domain/models/core/__init__.py +++ b/src/cleveragents/domain/models/core/__init__.py @@ -37,6 +37,15 @@ from cleveragents.domain.models.core.context_policy import ( ContextView, ProjectContextPolicy, ) +from cleveragents.domain.models.core.correction import ( + CorrectionAttempt, + CorrectionDryRunReport, + CorrectionImpact, + CorrectionMode, + CorrectionRequest, + CorrectionResult, + CorrectionStatus, +) from cleveragents.domain.models.core.debug_attempt import DebugAttempt from cleveragents.domain.models.core.org import ( CloudBillingFields, @@ -166,6 +175,13 @@ __all__ = [ "ContextType", "ContextUpdateResult", "ContextView", + "CorrectionAttempt", + "CorrectionDryRunReport", + "CorrectionImpact", + "CorrectionMode", + "CorrectionRequest", + "CorrectionResult", + "CorrectionStatus", "CreditType", "CreditsTransaction", "CreditsTransactionType", diff --git a/src/cleveragents/domain/models/core/correction.py b/src/cleveragents/domain/models/core/correction.py new file mode 100644 index 000000000..83d8abb8b --- /dev/null +++ b/src/cleveragents/domain/models/core/correction.py @@ -0,0 +1,272 @@ +"""Correction domain models for decision tree editing. + +Supports revert and append correction modes with impact analysis, +dry-run reporting, and execution flow tracking. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from enum import StrEnum +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field, field_validator +from ulid import ULID + + +class CorrectionMode(StrEnum): + """Available correction strategies.""" + + REVERT = "revert" + APPEND = "append" + + +class CorrectionStatus(StrEnum): + """Lifecycle states for a correction request.""" + + PENDING = "pending" + ANALYZING = "analyzing" + EXECUTING = "executing" + APPLIED = "applied" + FAILED = "failed" + CANCELLED = "cancelled" + + +class CorrectionRequest(BaseModel): + """A request to correct a decision within a plan's decision tree. + + Each request targets a specific decision node and applies either a + *revert* (invalidate subtree) or *append* (add new child plan) strategy. + """ + + model_config = ConfigDict(frozen=False, populate_by_name=True) + + correction_id: str = Field( + default_factory=lambda: str(ULID()), + description="Unique correction identifier (ULID).", + ) + plan_id: str = Field( + ..., + description="Plan that owns the targeted decision tree.", + ) + target_decision_id: str = Field( + ..., + description="Decision node to apply correction at.", + ) + mode: CorrectionMode = Field( + ..., + description="Correction strategy: revert or append.", + ) + guidance: str = Field( + default="", + description="Human-supplied guidance for the correction.", + ) + dry_run: bool = Field( + default=False, + description="When True, only compute impact without executing.", + ) + created_at: datetime = Field( + default_factory=lambda: datetime.now(UTC), + description="Timestamp when the request was created.", + ) + status: CorrectionStatus = Field( + default=CorrectionStatus.PENDING, + description="Current lifecycle status.", + ) + + @field_validator("plan_id") + @classmethod + def _plan_id_not_empty(cls, v: str) -> str: + if not v or not v.strip(): + raise ValueError("plan_id must not be empty") + return v + + @field_validator("target_decision_id") + @classmethod + def _target_decision_id_not_empty(cls, v: str) -> str: + if not v or not v.strip(): + raise ValueError("target_decision_id must not be empty") + return v + + @field_validator("guidance") + @classmethod + def _guidance_stripped(cls, v: str) -> str: + return v.strip() + + +class CorrectionImpact(BaseModel): + """Impact analysis result for a proposed correction. + + Computed via BFS traversal of the decision tree rooted at the + targeted decision node. + """ + + model_config = ConfigDict(frozen=True) + + affected_decisions: list[str] = Field( + default_factory=list, + description="Decision IDs in the affected subtree (BFS order).", + ) + affected_files: list[str] = Field( + default_factory=list, + description="Files touched by affected decisions.", + ) + affected_child_plans: list[str] = Field( + default_factory=list, + description="Child plan IDs spawned by affected decisions.", + ) + estimated_cost: float | None = Field( + default=None, + description="Estimated recompute cost (arbitrary units).", + ) + risk_level: str = Field( + default="low", + description="Risk classification: low, medium, or high.", + ) + rollback_tier: str = Field( + default="full", + description="Rollback granularity: full, phase, or append_only.", + ) + artifacts_to_archive: list[str] = Field( + default_factory=list, + description="Artifact paths that should be archived on revert.", + ) + + @field_validator("risk_level") + @classmethod + def _valid_risk_level(cls, v: str) -> str: + allowed = {"low", "medium", "high"} + if v not in allowed: + raise ValueError(f"risk_level must be one of {allowed}, got '{v}'") + return v + + @field_validator("rollback_tier") + @classmethod + def _valid_rollback_tier(cls, v: str) -> str: + allowed = {"full", "phase", "append_only"} + if v not in allowed: + raise ValueError(f"rollback_tier must be one of {allowed}, got '{v}'") + return v + + +class CorrectionResult(BaseModel): + """Outcome of executing a correction.""" + + model_config = ConfigDict(frozen=True) + + correction_id: str = Field( + ..., + description="Correction request this result belongs to.", + ) + status: CorrectionStatus = Field( + ..., + description="Terminal status after execution.", + ) + new_decisions: list[str] = Field( + default_factory=list, + description="Decision IDs created by an append correction.", + ) + reverted_decisions: list[str] = Field( + default_factory=list, + description="Decision IDs invalidated by a revert correction.", + ) + spawned_child_plan_id: str | None = Field( + default=None, + description="Child plan spawned for append corrections.", + ) + archived_artifacts: list[str] = Field( + default_factory=list, + description="Artifacts archived during revert.", + ) + error_message: str | None = Field( + default=None, + description="Error details when status is FAILED.", + ) + completed_at: datetime = Field( + default_factory=lambda: datetime.now(UTC), + description="Timestamp when execution finished.", + ) + + +class CorrectionAttempt(BaseModel): + """Record of a single correction execution attempt. + + Multiple attempts may occur if a correction is retried after failure. + """ + + model_config = ConfigDict(frozen=False) + + attempt_id: str = Field( + default_factory=lambda: str(ULID()), + description="Unique attempt identifier (ULID).", + ) + correction_id: str = Field( + ..., + description="Parent correction request ID.", + ) + started_at: datetime = Field( + default_factory=lambda: datetime.now(UTC), + description="When the attempt started.", + ) + completed_at: datetime | None = Field( + default=None, + description="When the attempt finished (None if still running).", + ) + success: bool = Field( + default=False, + description="Whether the attempt succeeded.", + ) + details: dict[str, Any] = Field( + default_factory=dict, + description="Arbitrary execution details.", + ) + + +class CorrectionDryRunReport(BaseModel): + """Detailed dry-run report showing what a correction *would* do. + + Generated by ``CorrectionService.generate_dry_run_report`` without + actually modifying any state. + """ + + model_config = ConfigDict(frozen=True) + + correction_id: str = Field( + ..., + description="Correction request this report describes.", + ) + mode: CorrectionMode = Field( + ..., + description="Correction strategy that would be applied.", + ) + impact: CorrectionImpact = Field( + ..., + description="Full impact analysis.", + ) + decisions_to_invalidate: list[str] = Field( + default_factory=list, + description="Decisions that would be marked invalid.", + ) + child_plans_to_rollback: list[str] = Field( + default_factory=list, + description="Child plans that would be rolled back.", + ) + estimated_recompute_time_seconds: float | None = Field( + default=None, + description="Estimated wall-clock seconds to recompute.", + ) + warnings: list[str] = Field( + default_factory=list, + description="Human-readable warnings about the correction.", + ) + + +__all__ = [ + "CorrectionAttempt", + "CorrectionDryRunReport", + "CorrectionImpact", + "CorrectionMode", + "CorrectionRequest", + "CorrectionResult", + "CorrectionStatus", +] -- 2.52.0