feat(prompts): implement Phase 6 - Incremental Execution (FR-7, FR-8)

Add change detection, structural diff-based impact analysis,
configurable-depth incremental recomputation with circular suppression,
and impact debt tracking.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

markitect/prompts/incremental/__init__.py

@@ -0,0 +1,46 @@
+"""
+Incremental execution for prompt artifacts.
+
+Implements FR-7: Incremental Recomputation
+Implements FR-8: Impact Analysis
+
+- FR-7.1: Change detection with persistent records
+- FR-7.2: Configurable-depth incremental recomputation
+- FR-8.1: Structural diff-based impact analysis
+- FR-8.2: Impact threshold decisions
+"""
+
+from markitect.prompts.incremental.models import (
+    ChangeType,
+    ArtifactChange,
+    ImpactDebt,
+    RecomputeConfig,
+    RecomputeResult,
+)
+from markitect.prompts.incremental.detector import ChangeDetector
+from markitect.prompts.incremental.metrics import (
+    structural_diff_ratio,
+    line_diff_ratio,
+    calculate_change_magnitude,
+)
+from markitect.prompts.incremental.impact import ImpactAnalyzer
+from markitect.prompts.incremental.engine import IncrementalExecutionEngine
+
+__all__ = [
+    # Models
+    "ChangeType",
+    "ArtifactChange",
+    "ImpactDebt",
+    "RecomputeConfig",
+    "RecomputeResult",
+    # Detector
+    "ChangeDetector",
+    # Metrics
+    "structural_diff_ratio",
+    "line_diff_ratio",
+    "calculate_change_magnitude",
+    # Impact
+    "ImpactAnalyzer",
+    # Engine
+    "IncrementalExecutionEngine",
+]

markitect/prompts/incremental/detector.py

@@ -0,0 +1,232 @@
+"""
+Change detector for artifact content changes.
+
+Implements FR-7.1: Change detection with persistent records.
+
+Detects changes by comparing content digests and persists
+ArtifactChange records to the artifact_changes table.
+"""
+
+import sqlite3
+from pathlib import Path
+from typing import List, Optional
+
+from markitect.prompts.models import Artifact, calculate_content_digest
+from markitect.prompts.incremental.models import ArtifactChange, ChangeType
+
+
+# SQL schema for artifact_changes table
+CHANGES_TABLE_SQL = """
+CREATE TABLE IF NOT EXISTS artifact_changes (
+    id TEXT PRIMARY KEY,
+    artifact_id TEXT NOT NULL,
+    old_digest TEXT,
+    new_digest TEXT NOT NULL,
+    change_type TEXT NOT NULL,
+    detected_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+);
+CREATE INDEX IF NOT EXISTS idx_changes_artifact ON artifact_changes(artifact_id);
+CREATE INDEX IF NOT EXISTS idx_changes_type ON artifact_changes(change_type);
+"""
+
+
+class ChangeDetector:
+    """
+    Detects and records artifact content changes.
+
+    Uses digest comparison to detect changes and persists
+    ArtifactChange records to SQLite for auditing and
+    incremental recomputation triggering.
+    """
+
+    def __init__(self, db_path: str):
+        """
+        Initialize with database path and ensure tables exist.
+
+        Args:
+            db_path: Path to SQLite database file
+        """
+        self.db_path = db_path
+        self._initialize_tables()
+
+    def _initialize_tables(self) -> None:
+        """Initialize artifact_changes table if not exists."""
+        db_dir = Path(self.db_path).parent
+        if db_dir and not db_dir.exists():
+            db_dir.mkdir(parents=True, exist_ok=True)
+
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.executescript(CHANGES_TABLE_SQL)
+            conn.commit()
+        finally:
+            conn.close()
+
+    def _get_connection(self) -> sqlite3.Connection:
+        """Get a database connection."""
+        conn = sqlite3.connect(self.db_path)
+        conn.row_factory = sqlite3.Row
+        return conn
+
+    def detect_change(
+        self,
+        artifact: Artifact,
+        new_content: str,
+    ) -> Optional[ArtifactChange]:
+        """
+        Detect if artifact content has changed.
+
+        Compares the artifact's stored digest against the new content digest.
+        If changed, creates and returns an ArtifactChange record (not yet persisted).
+
+        Args:
+            artifact: Existing artifact to check
+            new_content: New content to compare against
+
+        Returns:
+            ArtifactChange if content changed, None otherwise
+        """
+        new_digest = calculate_content_digest(new_content)
+
+        if not artifact.has_changed(new_digest):
+            return None
+
+        return ArtifactChange.create(
+            artifact_id=artifact.id,
+            old_digest=artifact.content_digest,
+            new_digest=new_digest,
+            change_type=ChangeType.MODIFIED,
+        )
+
+    def detect_creation(
+        self,
+        artifact_id: str,
+        content: str,
+    ) -> ArtifactChange:
+        """
+        Record creation of a new artifact.
+
+        Args:
+            artifact_id: ID of newly created artifact
+            content: Content of new artifact
+
+        Returns:
+            ArtifactChange record for the creation
+        """
+        new_digest = calculate_content_digest(content)
+        return ArtifactChange.create(
+            artifact_id=artifact_id,
+            old_digest=None,
+            new_digest=new_digest,
+            change_type=ChangeType.CREATED,
+        )
+
+    def detect_deletion(
+        self,
+        artifact: Artifact,
+    ) -> ArtifactChange:
+        """
+        Record deletion of an artifact.
+
+        Args:
+            artifact: Artifact being deleted
+
+        Returns:
+            ArtifactChange record for the deletion
+        """
+        return ArtifactChange.create(
+            artifact_id=artifact.id,
+            old_digest=artifact.content_digest,
+            new_digest=artifact.content_digest,
+            change_type=ChangeType.DELETED,
+        )
+
+    def record_change(self, change: ArtifactChange) -> ArtifactChange:
+        """
+        Persist an ArtifactChange record to the database.
+
+        Args:
+            change: Change record to persist
+
+        Returns:
+            Persisted change record
+        """
+        conn = self._get_connection()
+        try:
+            conn.execute(
+                """
+                INSERT INTO artifact_changes (
+                    id, artifact_id, old_digest, new_digest,
+                    change_type, detected_at
+                ) VALUES (?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    change.id,
+                    change.artifact_id,
+                    change.old_digest,
+                    change.new_digest,
+                    change.change_type.value,
+                    change.detected_at.isoformat(),
+                ),
+            )
+            conn.commit()
+            return change
+        finally:
+            conn.close()
+
+    def get_changes_for_artifact(self, artifact_id: str) -> List[ArtifactChange]:
+        """
+        Get all recorded changes for an artifact.
+
+        Args:
+            artifact_id: Artifact identifier
+
+        Returns:
+            List of change records, ordered by detection time
+        """
+        conn = self._get_connection()
+        try:
+            cursor = conn.execute(
+                """
+                SELECT * FROM artifact_changes
+                WHERE artifact_id = ?
+                ORDER BY detected_at
+                """,
+                (artifact_id,),
+            )
+            return [self._row_to_change(row) for row in cursor.fetchall()]
+        finally:
+            conn.close()
+
+    def get_changes_by_type(self, change_type: ChangeType) -> List[ArtifactChange]:
+        """
+        Get all recorded changes of a specific type.
+
+        Args:
+            change_type: Type of change to filter by
+
+        Returns:
+            List of change records
+        """
+        conn = self._get_connection()
+        try:
+            cursor = conn.execute(
+                "SELECT * FROM artifact_changes WHERE change_type = ?",
+                (change_type.value,),
+            )
+            return [self._row_to_change(row) for row in cursor.fetchall()]
+        finally:
+            conn.close()
+
+    def _row_to_change(self, row: sqlite3.Row) -> ArtifactChange:
+        """Convert database row to ArtifactChange instance."""
+        from datetime import datetime
+
+        return ArtifactChange(
+            id=row["id"],
+            artifact_id=row["artifact_id"],
+            old_digest=row["old_digest"],
+            new_digest=row["new_digest"],
+            change_type=ChangeType(row["change_type"]),
+            detected_at=datetime.fromisoformat(row["detected_at"]),
+        )

markitect/prompts/incremental/engine.py

@@ -0,0 +1,327 @@
+"""
+Incremental execution engine.
+
+Implements FR-7: Incremental Recomputation
+Implements FR-8: Impact Analysis
+
+Orchestrates: find dependents → check cycles → assess impact →
+execute or suppress → record debt.
+"""
+
+import sqlite3
+from collections import deque
+from pathlib import Path
+from typing import Callable, List, Optional, Set
+
+from markitect.prompts.dependencies.queries import DependencyQueryService
+from markitect.prompts.execution.models import PromptRun
+from markitect.prompts.incremental.impact import ImpactAnalyzer
+from markitect.prompts.incremental.models import (
+    ArtifactChange,
+    ImpactDebt,
+    RecomputeConfig,
+    RecomputeResult,
+)
+
+
+# SQL schema for impact_debt table
+DEBT_TABLE_SQL = """
+CREATE TABLE IF NOT EXISTS impact_debt (
+    id TEXT PRIMARY KEY,
+    artifact_id TEXT NOT NULL,
+    dependent_run_id TEXT NOT NULL,
+    change_magnitude REAL NOT NULL,
+    suppression_reason TEXT NOT NULL,
+    recorded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+);
+CREATE INDEX IF NOT EXISTS idx_debt_artifact ON impact_debt(artifact_id);
+CREATE INDEX IF NOT EXISTS idx_debt_run ON impact_debt(dependent_run_id);
+"""
+
+
+class IncrementalExecutionEngine:
+    """
+    Engine for incremental recomputation of dependent artifacts.
+
+    Orchestrates the full recompute flow:
+    1. Find dependents at configured depth (BFS)
+    2. Check for circular dependencies
+    3. Assess impact magnitude
+    4. Execute or suppress recomputation
+    5. Record impact debt for suppressed items
+    """
+
+    def __init__(
+        self,
+        db_path: str,
+        query_service: DependencyQueryService,
+        impact_analyzer: Optional[ImpactAnalyzer] = None,
+    ):
+        """
+        Initialize engine with dependencies.
+
+        Args:
+            db_path: Path to SQLite database for debt persistence
+            query_service: Service for dependency graph queries
+            impact_analyzer: Impact analyzer (created if not provided)
+        """
+        self.db_path = db_path
+        self.query_service = query_service
+        self.impact_analyzer = impact_analyzer or ImpactAnalyzer()
+        self._initialize_tables()
+
+    def _initialize_tables(self) -> None:
+        """Initialize impact_debt table if not exists."""
+        db_dir = Path(self.db_path).parent
+        if db_dir and not db_dir.exists():
+            db_dir.mkdir(parents=True, exist_ok=True)
+
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.executescript(DEBT_TABLE_SQL)
+            conn.commit()
+        finally:
+            conn.close()
+
+    def _get_connection(self) -> sqlite3.Connection:
+        """Get a database connection."""
+        conn = sqlite3.connect(self.db_path)
+        conn.row_factory = sqlite3.Row
+        return conn
+
+    def find_dependents_at_depth(
+        self,
+        artifact_id: str,
+        max_depth: int = 1,
+    ) -> Set[str]:
+        """
+        Find dependents up to a configured BFS depth.
+
+        Implements FR-7.2: Configurable depth control.
+
+        Args:
+            artifact_id: Artifact to find dependents of
+            max_depth: Maximum BFS layers to traverse
+
+        Returns:
+            Set of dependent artifact IDs within depth limit
+        """
+        visited: Set[str] = set()
+        queue: deque[tuple] = deque([(artifact_id, 0)])
+
+        while queue:
+            current, depth = queue.popleft()
+            if depth >= max_depth:
+                continue
+
+            direct_dependents = self.query_service.find_dependents(current)
+            for dep in direct_dependents:
+                if dep not in visited:
+                    visited.add(dep)
+                    queue.append((dep, depth + 1))
+
+        return visited
+
+    def recompute(
+        self,
+        change: ArtifactChange,
+        config: Optional[RecomputeConfig] = None,
+        execution_callback: Optional[Callable[[str], Optional[PromptRun]]] = None,
+        old_content: Optional[str] = None,
+        new_content: Optional[str] = None,
+    ) -> RecomputeResult:
+        """
+        Perform incremental recomputation for a detected change.
+
+        Orchestrates the full recompute flow:
+        1. Find dependents at configured depth
+        2. For each dependent, check cycles, assess impact
+        3. Execute or suppress and record debt
+
+        Args:
+            change: Detected artifact change
+            config: Recompute configuration
+            execution_callback: Callable that takes a run_id and re-executes.
+                If None, records what would be recomputed without executing.
+            old_content: Old content for magnitude calculation
+            new_content: New content for magnitude calculation
+
+        Returns:
+            RecomputeResult summarizing the recomputation
+        """
+        config = config or RecomputeConfig()
+
+        # Calculate change magnitude
+        magnitude = self.impact_analyzer.calculate_magnitude(
+            old_content, new_content
+        )
+
+        # Find dependents within configured depth
+        dependents = self.find_dependents_at_depth(
+            change.artifact_id, config.max_depth
+        )
+
+        result = RecomputeResult(
+            changed_artifact_id=change.artifact_id,
+            change=change,
+            total_dependents=len(dependents),
+        )
+
+        recompute_count = 0
+
+        for dependent_id in dependents:
+            # Budget check
+            if recompute_count >= config.max_recomputes:
+                debt = ImpactDebt.create(
+                    artifact_id=change.artifact_id,
+                    dependent_run_id=dependent_id,
+                    change_magnitude=magnitude,
+                    suppression_reason="budget_exhausted",
+                )
+                self._record_debt(debt)
+                result.suppressed.append(debt)
+                result.suppressed_count += 1
+                continue
+
+            # Circular dependency check
+            if config.suppress_circular:
+                if self.query_service.would_create_cycle(
+                    dependent_id, change.artifact_id
+                ):
+                    debt = ImpactDebt.create(
+                        artifact_id=change.artifact_id,
+                        dependent_run_id=dependent_id,
+                        change_magnitude=magnitude,
+                        suppression_reason="circular_dependency",
+                    )
+                    self._record_debt(debt)
+                    result.suppressed.append(debt)
+                    result.suppressed_count += 1
+                    continue
+
+            # Impact threshold check
+            if not self.impact_analyzer.should_recompute(magnitude, config):
+                debt = ImpactDebt.create(
+                    artifact_id=change.artifact_id,
+                    dependent_run_id=dependent_id,
+                    change_magnitude=magnitude,
+                    suppression_reason="below_threshold",
+                )
+                self._record_debt(debt)
+                result.suppressed.append(debt)
+                result.suppressed_count += 1
+                continue
+
+            # Execute recomputation
+            if execution_callback is not None:
+                run = execution_callback(dependent_id)
+                if run is not None:
+                    result.executed_run_ids.append(run.id)
+            else:
+                # Dry-run mode: just record the ID
+                result.executed_run_ids.append(dependent_id)
+
+            result.recomputed_count += 1
+            recompute_count += 1
+
+        return result
+
+    def _record_debt(self, debt: ImpactDebt) -> ImpactDebt:
+        """
+        Persist an ImpactDebt record to the database.
+
+        Args:
+            debt: Debt record to persist
+
+        Returns:
+            Persisted debt record
+        """
+        conn = self._get_connection()
+        try:
+            conn.execute(
+                """
+                INSERT INTO impact_debt (
+                    id, artifact_id, dependent_run_id,
+                    change_magnitude, suppression_reason, recorded_at
+                ) VALUES (?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    debt.id,
+                    debt.artifact_id,
+                    debt.dependent_run_id,
+                    debt.change_magnitude,
+                    debt.suppression_reason,
+                    debt.recorded_at.isoformat(),
+                ),
+            )
+            conn.commit()
+            return debt
+        finally:
+            conn.close()
+
+    def get_debt_for_artifact(self, artifact_id: str) -> List[ImpactDebt]:
+        """
+        Get all impact debt records for a given artifact.
+
+        Args:
+            artifact_id: Artifact identifier
+
+        Returns:
+            List of ImpactDebt records
+        """
+        conn = self._get_connection()
+        try:
+            cursor = conn.execute(
+                "SELECT * FROM impact_debt WHERE artifact_id = ?",
+                (artifact_id,),
+            )
+            return [self._row_to_debt(row) for row in cursor.fetchall()]
+        finally:
+            conn.close()
+
+    def get_debt_for_run(self, run_id: str) -> List[ImpactDebt]:
+        """
+        Get all impact debt records for a given run.
+
+        Args:
+            run_id: Run identifier
+
+        Returns:
+            List of ImpactDebt records
+        """
+        conn = self._get_connection()
+        try:
+            cursor = conn.execute(
+                "SELECT * FROM impact_debt WHERE dependent_run_id = ?",
+                (run_id,),
+            )
+            return [self._row_to_debt(row) for row in cursor.fetchall()]
+        finally:
+            conn.close()
+
+    def get_all_debt(self) -> List[ImpactDebt]:
+        """
+        Get all impact debt records.
+
+        Returns:
+            List of all ImpactDebt records
+        """
+        conn = self._get_connection()
+        try:
+            cursor = conn.execute("SELECT * FROM impact_debt")
+            return [self._row_to_debt(row) for row in cursor.fetchall()]
+        finally:
+            conn.close()
+
+    def _row_to_debt(self, row: sqlite3.Row) -> ImpactDebt:
+        """Convert database row to ImpactDebt instance."""
+        from datetime import datetime
+
+        return ImpactDebt(
+            id=row["id"],
+            artifact_id=row["artifact_id"],
+            dependent_run_id=row["dependent_run_id"],
+            change_magnitude=row["change_magnitude"],
+            suppression_reason=row["suppression_reason"],
+            recorded_at=datetime.fromisoformat(row["recorded_at"]),
+        )

markitect/prompts/incremental/impact.py

@@ -0,0 +1,57 @@
+"""
+Impact analyzer for assessing change significance.
+
+Implements FR-8: Impact Analysis
+Stateless analyzer that computes change magnitude and decides
+whether recomputation is warranted based on configurable thresholds.
+"""
+
+from typing import Optional
+
+from markitect.prompts.incremental.metrics import calculate_change_magnitude
+from markitect.prompts.incremental.models import RecomputeConfig
+
+
+class ImpactAnalyzer:
+    """
+    Stateless impact analyzer for artifact changes.
+
+    Computes change magnitude using diff metrics and compares against
+    configurable thresholds to determine whether recomputation is needed.
+    """
+
+    def calculate_magnitude(
+        self,
+        old_content: Optional[str],
+        new_content: Optional[str],
+        method: str = "structural",
+    ) -> float:
+        """
+        Calculate the magnitude of a change.
+
+        Args:
+            old_content: Previous content (None for creation)
+            new_content: Current content (None for deletion)
+            method: Diff method ("structural" or "line")
+
+        Returns:
+            Change magnitude from 0.0 to 1.0
+        """
+        return calculate_change_magnitude(old_content, new_content, method)
+
+    def should_recompute(
+        self,
+        magnitude: float,
+        config: RecomputeConfig,
+    ) -> bool:
+        """
+        Determine whether a dependent should be recomputed based on magnitude.
+
+        Args:
+            magnitude: Change magnitude (0.0-1.0)
+            config: Recompute configuration with threshold
+
+        Returns:
+            True if magnitude meets or exceeds threshold
+        """
+        return magnitude >= config.impact_threshold

markitect/prompts/incremental/metrics.py

@@ -0,0 +1,95 @@
+"""
+Pure diff functions for impact analysis.
+
+Implements FR-8.2: Structural diff-based impact measurement.
+
+Uses difflib.SequenceMatcher for computing change magnitude between
+old and new content versions.
+"""
+
+import difflib
+from typing import Optional
+
+
+def structural_diff_ratio(old_content: str, new_content: str) -> float:
+    """
+    Calculate structural similarity ratio between old and new content.
+
+    Uses SequenceMatcher to compute a ratio of matching blocks.
+    Returns the *change* ratio (1.0 - similarity), so higher values
+    mean more change.
+
+    Args:
+        old_content: Previous content
+        new_content: Current content
+
+    Returns:
+        Change ratio from 0.0 (identical) to 1.0 (completely different)
+    """
+    if old_content == new_content:
+        return 0.0
+    if not old_content and not new_content:
+        return 0.0
+    if not old_content or not new_content:
+        return 1.0
+
+    matcher = difflib.SequenceMatcher(None, old_content, new_content)
+    similarity = matcher.ratio()
+    return 1.0 - similarity
+
+
+def line_diff_ratio(old_content: str, new_content: str) -> float:
+    """
+    Calculate line-level change ratio between old and new content.
+
+    Splits content into lines and computes SequenceMatcher ratio
+    at the line level. Returns the change ratio.
+
+    Args:
+        old_content: Previous content
+        new_content: Current content
+
+    Returns:
+        Change ratio from 0.0 (identical) to 1.0 (completely different)
+    """
+    if old_content == new_content:
+        return 0.0
+    if not old_content and not new_content:
+        return 0.0
+    if not old_content or not new_content:
+        return 1.0
+
+    old_lines = old_content.splitlines()
+    new_lines = new_content.splitlines()
+    matcher = difflib.SequenceMatcher(None, old_lines, new_lines)
+    similarity = matcher.ratio()
+    return 1.0 - similarity
+
+
+def calculate_change_magnitude(
+    old_content: Optional[str],
+    new_content: Optional[str],
+    method: str = "structural",
+) -> float:
+    """
+    Calculate the magnitude of a change between two content versions.
+
+    This is the primary entry point for impact measurement (FR-8.2).
+
+    Args:
+        old_content: Previous content (None for created artifacts)
+        new_content: Current content (None for deleted artifacts)
+        method: Diff method to use ("structural" or "line")
+
+    Returns:
+        Change magnitude from 0.0 (no change) to 1.0 (complete change)
+    """
+    # Handle None cases (creation/deletion)
+    if old_content is None and new_content is None:
+        return 0.0
+    if old_content is None or new_content is None:
+        return 1.0
+
+    if method == "line":
+        return line_diff_ratio(old_content, new_content)
+    return structural_diff_ratio(old_content, new_content)

markitect/prompts/incremental/models.py

@@ -0,0 +1,241 @@
+"""
+Data models for incremental execution.
+
+Implements FR-7: Incremental Recomputation
+Implements FR-8: Impact Analysis
+
+Provides ChangeType enum, ArtifactChange, ImpactDebt, RecomputeConfig,
+and RecomputeResult dataclasses.
+"""
+
+import uuid
+from dataclasses import dataclass, field
+from datetime import datetime
+from enum import Enum
+from typing import Any, Dict, List, Optional
+
+
+class ChangeType(Enum):
+    """Type of artifact change detected."""
+    CREATED = "created"
+    MODIFIED = "modified"
+    DELETED = "deleted"
+
+
+@dataclass
+class ArtifactChange:
+    """
+    Record of a detected artifact change.
+
+    Attributes:
+        id: Unique change identifier
+        artifact_id: ID of changed artifact
+        old_digest: Previous content digest (None for created)
+        new_digest: New content digest
+        change_type: Type of change
+        detected_at: When change was detected
+    """
+    id: str
+    artifact_id: str
+    old_digest: Optional[str]
+    new_digest: str
+    change_type: ChangeType
+    detected_at: datetime = field(default_factory=datetime.utcnow)
+
+    @classmethod
+    def create(
+        cls,
+        artifact_id: str,
+        old_digest: Optional[str],
+        new_digest: str,
+        change_type: ChangeType,
+    ) -> "ArtifactChange":
+        """
+        Create a new ArtifactChange record.
+
+        Args:
+            artifact_id: ID of changed artifact
+            old_digest: Previous digest (None for created)
+            new_digest: New digest
+            change_type: Type of change
+
+        Returns:
+            New ArtifactChange instance
+        """
+        return cls(
+            id=str(uuid.uuid4()),
+            artifact_id=artifact_id,
+            old_digest=old_digest,
+            new_digest=new_digest,
+            change_type=change_type,
+        )
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dictionary for serialization."""
+        return {
+            "id": self.id,
+            "artifact_id": self.artifact_id,
+            "old_digest": self.old_digest,
+            "new_digest": self.new_digest,
+            "change_type": self.change_type.value,
+            "detected_at": self.detected_at.isoformat(),
+        }
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "ArtifactChange":
+        """Create from dictionary."""
+        return cls(
+            id=data["id"],
+            artifact_id=data["artifact_id"],
+            old_digest=data.get("old_digest"),
+            new_digest=data["new_digest"],
+            change_type=ChangeType(data["change_type"]),
+            detected_at=datetime.fromisoformat(data["detected_at"]),
+        )
+
+
+@dataclass
+class ImpactDebt:
+    """
+    Record of a suppressed recomputation.
+
+    When a dependent is not recomputed (due to circular dependency,
+    below-threshold impact, or budget exhaustion), an ImpactDebt
+    record is created to track the outstanding work.
+
+    Attributes:
+        id: Unique debt identifier
+        artifact_id: ID of the changed artifact that triggered this
+        dependent_run_id: Run ID of the dependent that was not recomputed
+        change_magnitude: Measured impact magnitude (0.0-1.0)
+        suppression_reason: Why recomputation was suppressed
+        recorded_at: When debt was recorded
+    """
+    id: str
+    artifact_id: str
+    dependent_run_id: str
+    change_magnitude: float
+    suppression_reason: str
+    recorded_at: datetime = field(default_factory=datetime.utcnow)
+
+    @classmethod
+    def create(
+        cls,
+        artifact_id: str,
+        dependent_run_id: str,
+        change_magnitude: float,
+        suppression_reason: str,
+    ) -> "ImpactDebt":
+        """
+        Create a new ImpactDebt record.
+
+        Args:
+            artifact_id: Changed artifact ID
+            dependent_run_id: Dependent run ID
+            change_magnitude: Impact magnitude (0.0-1.0)
+            suppression_reason: Reason for suppression
+
+        Returns:
+            New ImpactDebt instance
+        """
+        return cls(
+            id=str(uuid.uuid4()),
+            artifact_id=artifact_id,
+            dependent_run_id=dependent_run_id,
+            change_magnitude=change_magnitude,
+            suppression_reason=suppression_reason,
+        )
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dictionary for serialization."""
+        return {
+            "id": self.id,
+            "artifact_id": self.artifact_id,
+            "dependent_run_id": self.dependent_run_id,
+            "change_magnitude": self.change_magnitude,
+            "suppression_reason": self.suppression_reason,
+            "recorded_at": self.recorded_at.isoformat(),
+        }
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "ImpactDebt":
+        """Create from dictionary."""
+        return cls(
+            id=data["id"],
+            artifact_id=data["artifact_id"],
+            dependent_run_id=data["dependent_run_id"],
+            change_magnitude=data["change_magnitude"],
+            suppression_reason=data["suppression_reason"],
+            recorded_at=datetime.fromisoformat(data["recorded_at"]),
+        )
+
+
+@dataclass
+class RecomputeConfig:
+    """
+    Configuration for incremental recomputation.
+
+    Attributes:
+        max_depth: Maximum BFS depth for finding dependents (FR-7.2)
+        impact_threshold: Minimum magnitude to trigger recompute (FR-8.2)
+        max_recomputes: Budget limit on recomputations per change
+        suppress_circular: Whether to suppress circular dependencies
+    """
+    max_depth: int = 1
+    impact_threshold: float = 0.0
+    max_recomputes: int = 100
+    suppress_circular: bool = True
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dictionary for serialization."""
+        return {
+            "max_depth": self.max_depth,
+            "impact_threshold": self.impact_threshold,
+            "max_recomputes": self.max_recomputes,
+            "suppress_circular": self.suppress_circular,
+        }
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "RecomputeConfig":
+        """Create from dictionary."""
+        return cls(
+            max_depth=data.get("max_depth", 1),
+            impact_threshold=data.get("impact_threshold", 0.0),
+            max_recomputes=data.get("max_recomputes", 100),
+            suppress_circular=data.get("suppress_circular", True),
+        )
+
+
+@dataclass
+class RecomputeResult:
+    """
+    Result of an incremental recomputation triggered by a change.
+
+    Attributes:
+        changed_artifact_id: ID of the artifact that changed
+        change: The detected change record
+        executed_run_ids: Run IDs that were recomputed
+        suppressed: List of suppressed recomputation debt records
+        total_dependents: Total number of dependents found
+        recomputed_count: Number actually recomputed
+        suppressed_count: Number suppressed
+    """
+    changed_artifact_id: str
+    change: ArtifactChange
+    executed_run_ids: List[str] = field(default_factory=list)
+    suppressed: List[ImpactDebt] = field(default_factory=list)
+    total_dependents: int = 0
+    recomputed_count: int = 0
+    suppressed_count: int = 0
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dictionary for serialization."""
+        return {
+            "changed_artifact_id": self.changed_artifact_id,
+            "change": self.change.to_dict(),
+            "executed_run_ids": self.executed_run_ids,
+            "suppressed": [d.to_dict() for d in self.suppressed],
+            "total_dependents": self.total_dependents,
+            "recomputed_count": self.recomputed_count,
+            "suppressed_count": self.suppressed_count,
+        }

migrations/prompts/005_create_changes_and_debt.sql

@@ -0,0 +1,25 @@
+-- Phase 6: Incremental Execution tables
+-- artifact_changes: tracks detected changes to artifacts
+-- impact_debt: tracks suppressed recomputations
+
+CREATE TABLE IF NOT EXISTS artifact_changes (
+    id TEXT PRIMARY KEY,
+    artifact_id TEXT NOT NULL,
+    old_digest TEXT,
+    new_digest TEXT NOT NULL,
+    change_type TEXT NOT NULL,
+    detected_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+);
+CREATE INDEX IF NOT EXISTS idx_changes_artifact ON artifact_changes(artifact_id);
+CREATE INDEX IF NOT EXISTS idx_changes_type ON artifact_changes(change_type);
+
+CREATE TABLE IF NOT EXISTS impact_debt (
+    id TEXT PRIMARY KEY,
+    artifact_id TEXT NOT NULL,
+    dependent_run_id TEXT NOT NULL,
+    change_magnitude REAL NOT NULL,
+    suppression_reason TEXT NOT NULL,
+    recorded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+);
+CREATE INDEX IF NOT EXISTS idx_debt_artifact ON impact_debt(artifact_id);
+CREATE INDEX IF NOT EXISTS idx_debt_run ON impact_debt(dependent_run_id);

tests/integration/prompts/test_circular_suppression.py

@@ -0,0 +1,213 @@
+"""
+Integration tests for circular dependency suppression.
+
+Tests circular dependency handling with real DB, debt recording,
+and various cycle topologies.
+"""
+
+import pytest
+import tempfile
+from pathlib import Path
+
+from markitect.prompts.models import Artifact, ArtifactType
+from markitect.prompts.repositories.sqlite import SQLiteArtifactRepository
+from markitect.prompts.dependencies.models import DependencyEdge, EdgeType
+from markitect.prompts.dependencies.repository import SQLiteDependencyRepository
+from markitect.prompts.dependencies.queries import DependencyQueryService
+from markitect.prompts.incremental.detector import ChangeDetector
+from markitect.prompts.incremental.engine import IncrementalExecutionEngine
+from markitect.prompts.incremental.models import RecomputeConfig, ChangeType
+
+
+@pytest.fixture
+def temp_db():
+    """Create temporary database for testing."""
+    with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
+        db_path = f.name
+    yield db_path
+    Path(db_path).unlink(missing_ok=True)
+
+
+@pytest.fixture
+def artifact_repo(temp_db):
+    """Create artifact repository."""
+    return SQLiteArtifactRepository(temp_db)
+
+
+@pytest.fixture
+def dep_repo(temp_db):
+    """Create dependency repository."""
+    return SQLiteDependencyRepository(temp_db)
+
+
+@pytest.fixture
+def query_service(dep_repo):
+    """Create DependencyQueryService."""
+    return DependencyQueryService(dep_repo)
+
+
+@pytest.fixture
+def detector(temp_db):
+    """Create ChangeDetector."""
+    return ChangeDetector(temp_db)
+
+
+@pytest.fixture
+def engine(temp_db, query_service):
+    """Create IncrementalExecutionEngine."""
+    return IncrementalExecutionEngine(temp_db, query_service)
+
+
+def _create_artifact(repo, space_id, name, content="content"):
+    """Helper to create and persist an artifact."""
+    artifact = Artifact.create(
+        space_id=space_id,
+        name=name,
+        content=content,
+        artifact_type=ArtifactType.CONTENT,
+    )
+    return repo.create(artifact)
+
+
+def _create_edge(repo, src, tgt, run_id="run-1"):
+    """Helper to create and persist a dependency edge."""
+    edge = DependencyEdge.create(
+        source_artifact_id=src,
+        target_artifact_id=tgt,
+        run_id=run_id,
+        edge_type=EdgeType.REQUIRES,
+    )
+    return repo.create(edge)
+
+
+class TestDirectCircularSuppression:
+    """Tests for direct circular dependency (A <-> B) suppression."""
+
+    def test_mutual_dependency_suppressed(
+        self, artifact_repo, dep_repo, detector, engine
+    ):
+        """Test mutual dependency suppresses recompute and records debt."""
+        art_a = _create_artifact(artifact_repo, "space-1", "a", "content-a")
+        art_b = _create_artifact(artifact_repo, "space-1", "b", "content-b")
+
+        # A -> B and B -> A (circular)
+        _create_edge(dep_repo, art_a.id, art_b.id)
+        _create_edge(dep_repo, art_b.id, art_a.id)
+
+        # Detect change in B
+        change = detector.detect_change(art_b, "content-b-modified")
+        assert change is not None
+        detector.record_change(change)
+
+        # Recompute: A depends on B, but A -> B creates cycle
+        result = engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=1, suppress_circular=True),
+            old_content="content-b",
+            new_content="content-b-modified",
+        )
+
+        assert result.total_dependents == 1
+        assert result.suppressed_count == 1
+        assert result.recomputed_count == 0
+        assert result.suppressed[0].suppression_reason == "circular_dependency"
+
+    def test_debt_persisted_for_circular(
+        self, artifact_repo, dep_repo, detector, engine
+    ):
+        """Test that circular suppression debt is persisted in DB."""
+        art_a = _create_artifact(artifact_repo, "space-1", "a", "a-v1")
+        art_b = _create_artifact(artifact_repo, "space-1", "b", "b-v1")
+
+        _create_edge(dep_repo, art_a.id, art_b.id)
+        _create_edge(dep_repo, art_b.id, art_a.id)
+
+        change = detector.detect_change(art_b, "b-v2")
+        detector.record_change(change)
+
+        engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=1, suppress_circular=True),
+            old_content="b-v1",
+            new_content="b-v2",
+        )
+
+        # Verify debt persisted
+        debt = engine.get_debt_for_artifact(art_b.id)
+        assert len(debt) == 1
+        assert debt[0].suppression_reason == "circular_dependency"
+        assert debt[0].dependent_run_id == art_a.id
+
+
+class TestThreeNodeCycleSuppression:
+    """Tests for three-node circular dependency suppression."""
+
+    def test_three_node_cycle(
+        self, artifact_repo, dep_repo, detector, engine
+    ):
+        """Test 3-node cycle: A -> B -> C -> A."""
+        art_a = _create_artifact(artifact_repo, "space-1", "a", "a")
+        art_b = _create_artifact(artifact_repo, "space-1", "b", "b")
+        art_c = _create_artifact(artifact_repo, "space-1", "c", "c")
+
+        # A -> B -> C -> A
+        _create_edge(dep_repo, art_a.id, art_b.id)
+        _create_edge(dep_repo, art_b.id, art_c.id)
+        _create_edge(dep_repo, art_c.id, art_a.id)
+
+        # Change C, dependent at depth 1 is B
+        change = detector.detect_change(art_c, "c-modified")
+        detector.record_change(change)
+
+        result = engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=1, suppress_circular=True),
+            old_content="c",
+            new_content="c-modified",
+        )
+
+        # B depends on C. would_create_cycle(B, C) checks if C can reach B.
+        # C -> A -> B: yes, C can reach B. So B is suppressed.
+        assert result.total_dependents == 1
+        assert result.suppressed_count == 1
+        assert result.suppressed[0].suppression_reason == "circular_dependency"
+
+
+class TestMixedCircularAndNormal:
+    """Tests with a mix of circular and normal dependencies."""
+
+    def test_some_suppressed_some_recomputed(
+        self, artifact_repo, dep_repo, detector, engine
+    ):
+        """Test graph with both circular and normal dependents."""
+        art_a = _create_artifact(artifact_repo, "space-1", "a", "a")
+        art_b = _create_artifact(artifact_repo, "space-1", "b", "b")
+        art_c = _create_artifact(artifact_repo, "space-1", "c", "c")
+
+        # B -> A (normal), A -> B (creates cycle with B -> A)
+        # C -> A (normal, no cycle)
+        _create_edge(dep_repo, art_b.id, art_a.id)
+        _create_edge(dep_repo, art_a.id, art_b.id)
+        _create_edge(dep_repo, art_c.id, art_a.id)
+
+        # Change A: dependents are B and C
+        change = detector.detect_change(art_a, "a-modified")
+        detector.record_change(change)
+
+        result = engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=1, suppress_circular=True),
+            old_content="a",
+            new_content="a-modified",
+        )
+
+        assert result.total_dependents == 2
+
+        # B has circular dep with A → suppressed
+        # C has no circular dep with A → recomputed
+        circular_debts = [d for d in result.suppressed if d.suppression_reason == "circular_dependency"]
+        assert len(circular_debts) == 1
+        assert circular_debts[0].dependent_run_id == art_b.id
+
+        assert result.recomputed_count == 1
+        assert art_c.id in result.executed_run_ids

tests/integration/prompts/test_impact_debt.py

@@ -0,0 +1,289 @@
+"""
+Integration tests for impact debt tracking.
+
+Tests below-threshold suppression, budget exhaustion, and debt querying
+with a real SQLite database.
+"""
+
+import pytest
+import tempfile
+from pathlib import Path
+
+from markitect.prompts.models import Artifact, ArtifactType
+from markitect.prompts.repositories.sqlite import SQLiteArtifactRepository
+from markitect.prompts.dependencies.models import DependencyEdge, EdgeType
+from markitect.prompts.dependencies.repository import SQLiteDependencyRepository
+from markitect.prompts.dependencies.queries import DependencyQueryService
+from markitect.prompts.incremental.detector import ChangeDetector
+from markitect.prompts.incremental.engine import IncrementalExecutionEngine
+from markitect.prompts.incremental.models import RecomputeConfig
+
+
+@pytest.fixture
+def temp_db():
+    """Create temporary database for testing."""
+    with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
+        db_path = f.name
+    yield db_path
+    Path(db_path).unlink(missing_ok=True)
+
+
+@pytest.fixture
+def artifact_repo(temp_db):
+    """Create artifact repository."""
+    return SQLiteArtifactRepository(temp_db)
+
+
+@pytest.fixture
+def dep_repo(temp_db):
+    """Create dependency repository."""
+    return SQLiteDependencyRepository(temp_db)
+
+
+@pytest.fixture
+def query_service(dep_repo):
+    """Create DependencyQueryService."""
+    return DependencyQueryService(dep_repo)
+
+
+@pytest.fixture
+def detector(temp_db):
+    """Create ChangeDetector."""
+    return ChangeDetector(temp_db)
+
+
+@pytest.fixture
+def engine(temp_db, query_service):
+    """Create IncrementalExecutionEngine."""
+    return IncrementalExecutionEngine(temp_db, query_service)
+
+
+def _create_artifact(repo, space_id, name, content="content"):
+    """Helper to create and persist an artifact."""
+    artifact = Artifact.create(
+        space_id=space_id,
+        name=name,
+        content=content,
+        artifact_type=ArtifactType.CONTENT,
+    )
+    return repo.create(artifact)
+
+
+def _create_edge(repo, src, tgt, run_id="run-1"):
+    """Helper to create and persist a dependency edge."""
+    edge = DependencyEdge.create(
+        source_artifact_id=src,
+        target_artifact_id=tgt,
+        run_id=run_id,
+        edge_type=EdgeType.REQUIRES,
+    )
+    return repo.create(edge)
+
+
+class TestBelowThresholdSuppression:
+    """Tests for impact debt from below-threshold suppression."""
+
+    def test_small_change_creates_debt(
+        self, artifact_repo, dep_repo, detector, engine
+    ):
+        """Test small change below threshold creates impact debt."""
+        lib = _create_artifact(artifact_repo, "space-1", "lib", "hello world")
+        app = _create_artifact(artifact_repo, "space-1", "app", "uses lib")
+
+        _create_edge(dep_repo, app.id, lib.id)
+
+        change = detector.detect_change(lib, "hello World")  # tiny change
+        assert change is not None
+        detector.record_change(change)
+
+        result = engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=1, impact_threshold=0.5),
+            old_content="hello world",
+            new_content="hello World",
+        )
+
+        assert result.suppressed_count == 1
+        assert result.recomputed_count == 0
+        assert result.suppressed[0].suppression_reason == "below_threshold"
+
+    def test_debt_records_magnitude(
+        self, artifact_repo, dep_repo, detector, engine
+    ):
+        """Test debt records include change magnitude."""
+        lib = _create_artifact(artifact_repo, "space-1", "lib", "content A")
+        app = _create_artifact(artifact_repo, "space-1", "app", "uses lib")
+
+        _create_edge(dep_repo, app.id, lib.id)
+
+        change = detector.detect_change(lib, "content B")
+        detector.record_change(change)
+
+        engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=1, impact_threshold=0.9),
+            old_content="content A",
+            new_content="content B",
+        )
+
+        debt = engine.get_debt_for_artifact(lib.id)
+        assert len(debt) == 1
+        assert debt[0].change_magnitude > 0.0
+        assert debt[0].change_magnitude < 1.0
+
+    def test_large_change_no_debt(
+        self, artifact_repo, dep_repo, detector, engine
+    ):
+        """Test large change above threshold creates no debt."""
+        lib = _create_artifact(artifact_repo, "space-1", "lib", "old content here")
+        app = _create_artifact(artifact_repo, "space-1", "app", "uses lib")
+
+        _create_edge(dep_repo, app.id, lib.id)
+
+        change = detector.detect_change(lib, "completely new different content xyz")
+        detector.record_change(change)
+
+        result = engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=1, impact_threshold=0.1),
+            old_content="old content here",
+            new_content="completely new different content xyz",
+        )
+
+        assert result.recomputed_count == 1
+        assert result.suppressed_count == 0
+
+        debt = engine.get_debt_for_artifact(lib.id)
+        assert len(debt) == 0
+
+
+class TestBudgetExhaustion:
+    """Tests for impact debt from budget exhaustion."""
+
+    def test_budget_creates_debt_for_excess(
+        self, artifact_repo, dep_repo, detector, engine
+    ):
+        """Test budget exhaustion creates debt for overflow dependents."""
+        lib = _create_artifact(artifact_repo, "space-1", "lib", "lib v1")
+
+        # Create 5 apps depending on lib
+        apps = []
+        for i in range(5):
+            app = _create_artifact(artifact_repo, "space-1", f"app-{i}", f"app-{i}")
+            _create_edge(dep_repo, app.id, lib.id, run_id=f"run-{i}")
+            apps.append(app)
+
+        change = detector.detect_change(lib, "lib v2")
+        detector.record_change(change)
+
+        result = engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=1, max_recomputes=2),
+            old_content="lib v1",
+            new_content="lib v2",
+        )
+
+        assert result.total_dependents == 5
+        assert result.recomputed_count == 2
+        assert result.suppressed_count == 3
+
+        budget_debt = [d for d in result.suppressed if d.suppression_reason == "budget_exhausted"]
+        assert len(budget_debt) == 3
+
+    def test_budget_debt_queryable(
+        self, artifact_repo, dep_repo, detector, engine
+    ):
+        """Test budget-exhaustion debt is queryable from DB."""
+        lib = _create_artifact(artifact_repo, "space-1", "lib", "lib v1")
+
+        for i in range(3):
+            app = _create_artifact(artifact_repo, "space-1", f"app-{i}", f"app-{i}")
+            _create_edge(dep_repo, app.id, lib.id, run_id=f"run-{i}")
+
+        change = detector.detect_change(lib, "lib v2")
+        detector.record_change(change)
+
+        engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=1, max_recomputes=1),
+            old_content="lib v1",
+            new_content="lib v2",
+        )
+
+        all_debt = engine.get_all_debt()
+        budget_debt = [d for d in all_debt if d.suppression_reason == "budget_exhausted"]
+        assert len(budget_debt) == 2
+
+
+class TestDebtQuerying:
+    """Tests for querying impact debt records."""
+
+    def test_query_by_artifact(
+        self, artifact_repo, dep_repo, detector, engine
+    ):
+        """Test querying debt by artifact ID."""
+        lib_a = _create_artifact(artifact_repo, "space-1", "lib-a", "a-v1")
+        lib_b = _create_artifact(artifact_repo, "space-1", "lib-b", "b-v1")
+        app = _create_artifact(artifact_repo, "space-1", "app", "app")
+
+        _create_edge(dep_repo, app.id, lib_a.id)
+        _create_edge(dep_repo, app.id, lib_b.id)
+
+        # Suppress change to lib_a
+        change_a = detector.detect_change(lib_a, "a-v2")
+        detector.record_change(change_a)
+        engine.recompute(
+            change_a,
+            config=RecomputeConfig(max_depth=1, impact_threshold=0.99),
+            old_content="a-v1",
+            new_content="a-v2",
+        )
+
+        # Suppress change to lib_b
+        change_b = detector.detect_change(lib_b, "b-v2")
+        detector.record_change(change_b)
+        engine.recompute(
+            change_b,
+            config=RecomputeConfig(max_depth=1, impact_threshold=0.99),
+            old_content="b-v1",
+            new_content="b-v2",
+        )
+
+        # Query by artifact
+        debt_a = engine.get_debt_for_artifact(lib_a.id)
+        assert len(debt_a) == 1
+
+        debt_b = engine.get_debt_for_artifact(lib_b.id)
+        assert len(debt_b) == 1
+
+        # Total debt
+        all_debt = engine.get_all_debt()
+        assert len(all_debt) == 2
+
+    def test_query_by_run(
+        self, artifact_repo, dep_repo, detector, engine
+    ):
+        """Test querying debt by dependent run ID."""
+        lib = _create_artifact(artifact_repo, "space-1", "lib", "lib-v1")
+        app = _create_artifact(artifact_repo, "space-1", "app", "app")
+
+        _create_edge(dep_repo, app.id, lib.id)
+
+        change = detector.detect_change(lib, "lib-v2")
+        detector.record_change(change)
+        engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=1, impact_threshold=0.99),
+            old_content="lib-v1",
+            new_content="lib-v2",
+        )
+
+        debt = engine.get_debt_for_run(app.id)
+        assert len(debt) == 1
+        assert debt[0].dependent_run_id == app.id
+
+    def test_no_debt_returns_empty(self, engine):
+        """Test querying debt when none exists returns empty list."""
+        assert engine.get_debt_for_artifact("nonexistent") == []
+        assert engine.get_debt_for_run("nonexistent") == []
+        assert engine.get_all_debt() == []

tests/integration/prompts/test_incremental_recompute.py

@@ -0,0 +1,229 @@
+"""
+Integration test for full incremental recompute workflow.
+
+Tests: change artifact → detect → find dependents → recompute
+with a real SQLite database.
+"""
+
+import pytest
+import tempfile
+from pathlib import Path
+
+from markitect.prompts.models import Artifact, ArtifactType, calculate_content_digest
+from markitect.prompts.repositories.sqlite import SQLiteArtifactRepository
+from markitect.prompts.dependencies.models import DependencyEdge, EdgeType
+from markitect.prompts.dependencies.repository import SQLiteDependencyRepository
+from markitect.prompts.dependencies.queries import DependencyQueryService
+from markitect.prompts.incremental.detector import ChangeDetector
+from markitect.prompts.incremental.engine import IncrementalExecutionEngine
+from markitect.prompts.incremental.models import RecomputeConfig
+
+
+@pytest.fixture
+def temp_db():
+    """Create temporary database for testing."""
+    with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
+        db_path = f.name
+    yield db_path
+    Path(db_path).unlink(missing_ok=True)
+
+
+@pytest.fixture
+def artifact_repo(temp_db):
+    """Create artifact repository."""
+    return SQLiteArtifactRepository(temp_db)
+
+
+@pytest.fixture
+def dep_repo(temp_db):
+    """Create dependency repository (shares same DB)."""
+    return SQLiteDependencyRepository(temp_db)
+
+
+@pytest.fixture
+def query_service(dep_repo):
+    """Create DependencyQueryService."""
+    return DependencyQueryService(dep_repo)
+
+
+@pytest.fixture
+def detector(temp_db):
+    """Create ChangeDetector."""
+    return ChangeDetector(temp_db)
+
+
+@pytest.fixture
+def engine(temp_db, query_service):
+    """Create IncrementalExecutionEngine."""
+    return IncrementalExecutionEngine(temp_db, query_service)
+
+
+def _create_artifact(repo, space_id, name, content):
+    """Helper to create and persist an artifact."""
+    artifact = Artifact.create(
+        space_id=space_id,
+        name=name,
+        content=content,
+        artifact_type=ArtifactType.CONTENT,
+    )
+    return repo.create(artifact)
+
+
+def _create_edge(repo, src, tgt, run_id="run-1"):
+    """Helper to create and persist a dependency edge."""
+    edge = DependencyEdge.create(
+        source_artifact_id=src,
+        target_artifact_id=tgt,
+        run_id=run_id,
+        edge_type=EdgeType.REQUIRES,
+    )
+    return repo.create(edge)
+
+
+class TestFullRecomputeWorkflow:
+    """Full end-to-end incremental recompute workflow."""
+
+    def test_change_detect_and_recompute(
+        self, artifact_repo, dep_repo, query_service, detector, engine
+    ):
+        """Test complete flow: create artifacts, detect change, recompute dependents."""
+        # Step 1: Create artifacts
+        lib = _create_artifact(artifact_repo, "space-1", "lib", "library v1")
+        app = _create_artifact(artifact_repo, "space-1", "app", "app using lib")
+
+        # Step 2: Establish dependency (app depends on lib)
+        _create_edge(dep_repo, app.id, lib.id)
+
+        # Step 3: Detect a change in lib
+        change = detector.detect_change(lib, "library v2")
+        assert change is not None
+        detector.record_change(change)
+
+        # Step 4: Recompute dependents
+        executed_ids = []
+
+        def callback(dep_id):
+            from markitect.prompts.execution.models import PromptRun
+            run = PromptRun.create(
+                template_id=dep_id,
+                input_bundle_hash="recompute-hash",
+            )
+            run.mark_complete()
+            executed_ids.append(dep_id)
+            return run
+
+        result = engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=1),
+            execution_callback=callback,
+            old_content="library v1",
+            new_content="library v2",
+        )
+
+        # Verify
+        assert result.total_dependents == 1
+        assert result.recomputed_count == 1
+        assert result.suppressed_count == 0
+        assert app.id in executed_ids
+
+    def test_multi_level_recompute(
+        self, artifact_repo, dep_repo, query_service, detector, engine
+    ):
+        """Test recompute propagates through multiple dependency levels."""
+        # core -> utils -> app
+        core = _create_artifact(artifact_repo, "space-1", "core", "core v1")
+        utils = _create_artifact(artifact_repo, "space-1", "utils", "utils v1")
+        app = _create_artifact(artifact_repo, "space-1", "app", "app v1")
+
+        _create_edge(dep_repo, utils.id, core.id)
+        _create_edge(dep_repo, app.id, utils.id)
+
+        # Change core
+        change = detector.detect_change(core, "core v2")
+        assert change is not None
+        detector.record_change(change)
+
+        # Recompute with depth 2
+        result = engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=2),
+            old_content="core v1",
+            new_content="core v2",
+        )
+
+        assert result.total_dependents == 2
+        assert result.recomputed_count == 2
+        assert set(result.executed_run_ids) == {utils.id, app.id}
+
+    def test_no_change_no_recompute(
+        self, artifact_repo, dep_repo, detector, engine
+    ):
+        """Test that no change means no recompute."""
+        lib = _create_artifact(artifact_repo, "space-1", "lib", "unchanged")
+        app = _create_artifact(artifact_repo, "space-1", "app", "app")
+
+        _create_edge(dep_repo, app.id, lib.id)
+
+        # Same content → no change
+        change = detector.detect_change(lib, "unchanged")
+        assert change is None
+
+    def test_change_record_persisted(
+        self, artifact_repo, detector
+    ):
+        """Test change records are persisted across detector instances."""
+        lib = _create_artifact(artifact_repo, "space-1", "lib", "v1")
+
+        change = detector.detect_change(lib, "v2")
+        assert change is not None
+        detector.record_change(change)
+
+        # Verify persisted
+        changes = detector.get_changes_for_artifact(lib.id)
+        assert len(changes) == 1
+        assert changes[0].id == change.id
+
+
+class TestMultipleArtifactChanges:
+    """Tests for handling changes to multiple artifacts."""
+
+    def test_independent_changes(
+        self, artifact_repo, dep_repo, detector, engine
+    ):
+        """Test independent artifact changes trigger separate recomputes."""
+        lib_a = _create_artifact(artifact_repo, "space-1", "lib-a", "lib-a v1")
+        lib_b = _create_artifact(artifact_repo, "space-1", "lib-b", "lib-b v1")
+        app = _create_artifact(artifact_repo, "space-1", "app", "app v1")
+
+        _create_edge(dep_repo, app.id, lib_a.id)
+        _create_edge(dep_repo, app.id, lib_b.id)
+
+        # Change lib_a
+        change_a = detector.detect_change(lib_a, "lib-a v2")
+        assert change_a is not None
+        detector.record_change(change_a)
+
+        result_a = engine.recompute(
+            change_a,
+            config=RecomputeConfig(max_depth=1),
+            old_content="lib-a v1",
+            new_content="lib-a v2",
+        )
+
+        assert result_a.total_dependents == 1
+        assert result_a.recomputed_count == 1
+
+        # Change lib_b
+        change_b = detector.detect_change(lib_b, "lib-b v2")
+        assert change_b is not None
+        detector.record_change(change_b)
+
+        result_b = engine.recompute(
+            change_b,
+            config=RecomputeConfig(max_depth=1),
+            old_content="lib-b v1",
+            new_content="lib-b v2",
+        )
+
+        assert result_b.total_dependents == 1
+        assert result_b.recomputed_count == 1

tests/unit/prompts/test_change_detector.py

@@ -0,0 +1,166 @@
+"""
+Unit tests for ChangeDetector.
+
+Tests change detection, recording, change types, and no-change cases.
+"""
+
+import pytest
+import tempfile
+from pathlib import Path
+
+from markitect.prompts.models import Artifact, ArtifactType, calculate_content_digest
+from markitect.prompts.incremental.detector import ChangeDetector
+from markitect.prompts.incremental.models import ChangeType
+
+
+@pytest.fixture
+def temp_db():
+    """Create temporary database for testing."""
+    with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
+        db_path = f.name
+    yield db_path
+    Path(db_path).unlink(missing_ok=True)
+
+
+@pytest.fixture
+def detector(temp_db):
+    """Create ChangeDetector instance."""
+    return ChangeDetector(temp_db)
+
+
+def _make_artifact(content="original content"):
+    """Helper to create an in-memory artifact."""
+    return Artifact.create(
+        space_id="space-1",
+        name="test-artifact",
+        content=content,
+        artifact_type=ArtifactType.CONTENT,
+    )
+
+
+class TestDetectChange:
+    """Tests for detecting content changes."""
+
+    def test_detect_modification(self, detector):
+        """Test detecting a content modification."""
+        artifact = _make_artifact("original content")
+        change = detector.detect_change(artifact, "modified content")
+
+        assert change is not None
+        assert change.artifact_id == artifact.id
+        assert change.old_digest == artifact.content_digest
+        assert change.new_digest == calculate_content_digest("modified content")
+        assert change.change_type == ChangeType.MODIFIED
+
+    def test_no_change_returns_none(self, detector):
+        """Test that identical content returns None."""
+        artifact = _make_artifact("same content")
+        change = detector.detect_change(artifact, "same content")
+
+        assert change is None
+
+    def test_detect_whitespace_change(self, detector):
+        """Test detecting whitespace-only changes."""
+        artifact = _make_artifact("content")
+        change = detector.detect_change(artifact, "content ")
+
+        assert change is not None
+        assert change.change_type == ChangeType.MODIFIED
+
+    def test_detect_empty_to_content(self, detector):
+        """Test detecting change from empty to content."""
+        artifact = _make_artifact("")
+        change = detector.detect_change(artifact, "new content")
+
+        assert change is not None
+        assert change.change_type == ChangeType.MODIFIED
+
+
+class TestDetectCreation:
+    """Tests for recording artifact creation."""
+
+    def test_detect_creation(self, detector):
+        """Test creation change record."""
+        change = detector.detect_creation("artifact-123", "new content")
+
+        assert change.artifact_id == "artifact-123"
+        assert change.old_digest is None
+        assert change.new_digest == calculate_content_digest("new content")
+        assert change.change_type == ChangeType.CREATED
+
+    def test_creation_has_unique_id(self, detector):
+        """Test that each creation gets a unique ID."""
+        change1 = detector.detect_creation("art-1", "content")
+        change2 = detector.detect_creation("art-2", "content")
+
+        assert change1.id != change2.id
+
+
+class TestDetectDeletion:
+    """Tests for recording artifact deletion."""
+
+    def test_detect_deletion(self, detector):
+        """Test deletion change record."""
+        artifact = _make_artifact("content to delete")
+        change = detector.detect_deletion(artifact)
+
+        assert change.artifact_id == artifact.id
+        assert change.old_digest == artifact.content_digest
+        assert change.change_type == ChangeType.DELETED
+
+
+class TestRecordChange:
+    """Tests for persisting change records."""
+
+    def test_record_and_retrieve(self, detector):
+        """Test recording a change and retrieving it."""
+        artifact = _make_artifact("original")
+        change = detector.detect_change(artifact, "modified")
+        assert change is not None
+
+        detector.record_change(change)
+        changes = detector.get_changes_for_artifact(artifact.id)
+
+        assert len(changes) == 1
+        assert changes[0].id == change.id
+        assert changes[0].artifact_id == artifact.id
+        assert changes[0].change_type == ChangeType.MODIFIED
+
+    def test_record_multiple_changes(self, detector):
+        """Test recording multiple changes for same artifact."""
+        artifact = _make_artifact("v1")
+
+        change1 = detector.detect_change(artifact, "v2")
+        detector.record_change(change1)
+
+        # Simulate artifact update
+        artifact.update_content("v2")
+        change2 = detector.detect_change(artifact, "v3")
+        detector.record_change(change2)
+
+        changes = detector.get_changes_for_artifact(artifact.id)
+        assert len(changes) == 2
+
+    def test_get_changes_by_type(self, detector):
+        """Test filtering changes by type."""
+        # Record a creation
+        creation = detector.detect_creation("art-new", "content")
+        detector.record_change(creation)
+
+        # Record a modification
+        artifact = _make_artifact("old")
+        modification = detector.detect_change(artifact, "new")
+        detector.record_change(modification)
+
+        created_changes = detector.get_changes_by_type(ChangeType.CREATED)
+        assert len(created_changes) == 1
+        assert created_changes[0].change_type == ChangeType.CREATED
+
+        modified_changes = detector.get_changes_by_type(ChangeType.MODIFIED)
+        assert len(modified_changes) == 1
+        assert modified_changes[0].change_type == ChangeType.MODIFIED
+
+    def test_no_changes_returns_empty(self, detector):
+        """Test querying changes for artifact with none recorded."""
+        changes = detector.get_changes_for_artifact("nonexistent")
+        assert changes == []

tests/unit/prompts/test_impact_analyzer.py

@@ -0,0 +1,162 @@
+"""
+Unit tests for ImpactAnalyzer and metrics functions.
+
+Tests diff ratios, magnitude scoring, and threshold decisions.
+"""
+
+import pytest
+
+from markitect.prompts.incremental.metrics import (
+    structural_diff_ratio,
+    line_diff_ratio,
+    calculate_change_magnitude,
+)
+from markitect.prompts.incremental.impact import ImpactAnalyzer
+from markitect.prompts.incremental.models import RecomputeConfig
+
+
+class TestStructuralDiffRatio:
+    """Tests for structural_diff_ratio."""
+
+    def test_identical_content(self):
+        """Test identical content returns 0.0."""
+        assert structural_diff_ratio("hello", "hello") == 0.0
+
+    def test_completely_different(self):
+        """Test completely different content returns high ratio."""
+        ratio = structural_diff_ratio("aaa", "zzz")
+        assert ratio > 0.5
+
+    def test_empty_strings(self):
+        """Test both empty returns 0.0."""
+        assert structural_diff_ratio("", "") == 0.0
+
+    def test_one_empty(self):
+        """Test one empty returns 1.0."""
+        assert structural_diff_ratio("", "content") == 1.0
+        assert structural_diff_ratio("content", "") == 1.0
+
+    def test_small_change(self):
+        """Test small change returns low ratio."""
+        old = "The quick brown fox jumps over the lazy dog"
+        new = "The quick brown fox leaps over the lazy dog"
+        ratio = structural_diff_ratio(old, new)
+        assert 0.0 < ratio < 0.5
+
+    def test_returns_float(self):
+        """Test return value is float between 0 and 1."""
+        ratio = structural_diff_ratio("abc", "abd")
+        assert isinstance(ratio, float)
+        assert 0.0 <= ratio <= 1.0
+
+
+class TestLineDiffRatio:
+    """Tests for line_diff_ratio."""
+
+    def test_identical_lines(self):
+        """Test identical multi-line content returns 0.0."""
+        content = "line1\nline2\nline3"
+        assert line_diff_ratio(content, content) == 0.0
+
+    def test_one_line_changed(self):
+        """Test changing one line of several."""
+        old = "line1\nline2\nline3"
+        new = "line1\nmodified\nline3"
+        ratio = line_diff_ratio(old, new)
+        assert 0.0 < ratio < 1.0
+
+    def test_all_lines_changed(self):
+        """Test all lines changed returns high ratio."""
+        old = "aaa\nbbb\nccc"
+        new = "xxx\nyyy\nzzz"
+        ratio = line_diff_ratio(old, new)
+        assert ratio > 0.5
+
+    def test_empty_strings(self):
+        """Test both empty returns 0.0."""
+        assert line_diff_ratio("", "") == 0.0
+
+    def test_one_empty(self):
+        """Test one empty returns 1.0."""
+        assert line_diff_ratio("", "content") == 1.0
+        assert line_diff_ratio("content", "") == 1.0
+
+
+class TestCalculateChangeMagnitude:
+    """Tests for calculate_change_magnitude."""
+
+    def test_none_old_content(self):
+        """Test None old_content (creation) returns 1.0."""
+        assert calculate_change_magnitude(None, "new content") == 1.0
+
+    def test_none_new_content(self):
+        """Test None new_content (deletion) returns 1.0."""
+        assert calculate_change_magnitude("old content", None) == 1.0
+
+    def test_both_none(self):
+        """Test both None returns 0.0."""
+        assert calculate_change_magnitude(None, None) == 0.0
+
+    def test_structural_method(self):
+        """Test structural method (default)."""
+        result = calculate_change_magnitude("abc", "abd", method="structural")
+        assert 0.0 < result < 1.0
+
+    def test_line_method(self):
+        """Test line method."""
+        result = calculate_change_magnitude("abc\ndef", "abc\nxyz", method="line")
+        assert 0.0 < result < 1.0
+
+    def test_identical_content(self):
+        """Test identical content returns 0.0."""
+        assert calculate_change_magnitude("same", "same") == 0.0
+
+
+class TestImpactAnalyzer:
+    """Tests for ImpactAnalyzer class."""
+
+    @pytest.fixture
+    def analyzer(self):
+        """Create ImpactAnalyzer instance."""
+        return ImpactAnalyzer()
+
+    def test_calculate_magnitude(self, analyzer):
+        """Test magnitude calculation delegates to metrics."""
+        result = analyzer.calculate_magnitude("old", "new")
+        assert isinstance(result, float)
+        assert 0.0 <= result <= 1.0
+
+    def test_calculate_magnitude_creation(self, analyzer):
+        """Test magnitude for creation."""
+        assert analyzer.calculate_magnitude(None, "new") == 1.0
+
+    def test_calculate_magnitude_identical(self, analyzer):
+        """Test magnitude for identical content."""
+        assert analyzer.calculate_magnitude("same", "same") == 0.0
+
+    def test_should_recompute_above_threshold(self, analyzer):
+        """Test recompute when magnitude exceeds threshold."""
+        config = RecomputeConfig(impact_threshold=0.3)
+        assert analyzer.should_recompute(0.5, config) is True
+
+    def test_should_recompute_at_threshold(self, analyzer):
+        """Test recompute when magnitude equals threshold."""
+        config = RecomputeConfig(impact_threshold=0.5)
+        assert analyzer.should_recompute(0.5, config) is True
+
+    def test_should_not_recompute_below_threshold(self, analyzer):
+        """Test no recompute when magnitude below threshold."""
+        config = RecomputeConfig(impact_threshold=0.5)
+        assert analyzer.should_recompute(0.3, config) is False
+
+    def test_zero_threshold_always_recomputes(self, analyzer):
+        """Test zero threshold means any change triggers recompute."""
+        config = RecomputeConfig(impact_threshold=0.0)
+        assert analyzer.should_recompute(0.0, config) is True
+        assert analyzer.should_recompute(0.01, config) is True
+
+    def test_high_threshold_only_major_changes(self, analyzer):
+        """Test high threshold only triggers on major changes."""
+        config = RecomputeConfig(impact_threshold=0.9)
+        assert analyzer.should_recompute(0.5, config) is False
+        assert analyzer.should_recompute(0.95, config) is True

tests/unit/prompts/test_incremental_engine.py

@@ -0,0 +1,364 @@
+"""
+Unit tests for IncrementalExecutionEngine.
+
+Tests recompute flow, depth control, circular suppression, and budget limits.
+"""
+
+import pytest
+import tempfile
+from pathlib import Path
+from unittest.mock import MagicMock
+
+from markitect.prompts.dependencies.models import DependencyEdge, EdgeType
+from markitect.prompts.dependencies.repository import SQLiteDependencyRepository
+from markitect.prompts.dependencies.queries import DependencyQueryService
+from markitect.prompts.execution.models import PromptRun, RunConfig, RunStatus
+from markitect.prompts.incremental.engine import IncrementalExecutionEngine
+from markitect.prompts.incremental.models import (
+    ArtifactChange,
+    ChangeType,
+    ImpactDebt,
+    RecomputeConfig,
+)
+
+
+@pytest.fixture
+def temp_db():
+    """Create temporary database for testing."""
+    with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
+        db_path = f.name
+    yield db_path
+    Path(db_path).unlink(missing_ok=True)
+
+
+@pytest.fixture
+def dep_repo(temp_db):
+    """Create dependency repository."""
+    return SQLiteDependencyRepository(temp_db)
+
+
+@pytest.fixture
+def query_service(dep_repo):
+    """Create DependencyQueryService."""
+    return DependencyQueryService(dep_repo)
+
+
+@pytest.fixture
+def engine(temp_db, query_service):
+    """Create IncrementalExecutionEngine."""
+    return IncrementalExecutionEngine(temp_db, query_service)
+
+
+def _create_edge(repo, src, tgt, run_id="run-1", edge_type=EdgeType.REQUIRES):
+    """Helper to create and persist a dependency edge."""
+    edge = DependencyEdge.create(
+        source_artifact_id=src,
+        target_artifact_id=tgt,
+        run_id=run_id,
+        edge_type=edge_type,
+    )
+    return repo.create(edge)
+
+
+def _make_change(artifact_id="art-1"):
+    """Helper to create a test ArtifactChange."""
+    return ArtifactChange.create(
+        artifact_id=artifact_id,
+        old_digest="old-digest",
+        new_digest="new-digest",
+        change_type=ChangeType.MODIFIED,
+    )
+
+
+class TestFindDependentsAtDepth:
+    """Tests for BFS depth-controlled dependent finding."""
+
+    def test_depth_1_direct_only(self, dep_repo, engine):
+        """Test depth=1 finds only direct dependents."""
+        # A -> B -> C (A depends on B, B depends on C)
+        _create_edge(dep_repo, "A", "B")
+        _create_edge(dep_repo, "B", "C")
+
+        # Dependents of C at depth 1: only B
+        dependents = engine.find_dependents_at_depth("C", max_depth=1)
+        assert dependents == {"B"}
+
+    def test_depth_2_transitive(self, dep_repo, engine):
+        """Test depth=2 finds two levels of dependents."""
+        # A -> B -> C
+        _create_edge(dep_repo, "A", "B")
+        _create_edge(dep_repo, "B", "C")
+
+        # Dependents of C at depth 2: B and A
+        dependents = engine.find_dependents_at_depth("C", max_depth=2)
+        assert dependents == {"A", "B"}
+
+    def test_depth_0_returns_empty(self, dep_repo, engine):
+        """Test depth=0 returns no dependents."""
+        _create_edge(dep_repo, "A", "B")
+
+        dependents = engine.find_dependents_at_depth("B", max_depth=0)
+        assert dependents == set()
+
+    def test_no_dependents(self, engine):
+        """Test artifact with no dependents."""
+        dependents = engine.find_dependents_at_depth("isolated", max_depth=5)
+        assert dependents == set()
+
+    def test_diamond_dependents(self, dep_repo, engine):
+        """Test diamond-shaped dependency graph."""
+        # A -> C, B -> C, D -> A, D -> B
+        _create_edge(dep_repo, "A", "C")
+        _create_edge(dep_repo, "B", "C")
+        _create_edge(dep_repo, "D", "A")
+        _create_edge(dep_repo, "D", "B")
+
+        dependents = engine.find_dependents_at_depth("C", max_depth=2)
+        assert dependents == {"A", "B", "D"}
+
+
+class TestRecompute:
+    """Tests for the recompute orchestration flow."""
+
+    def test_basic_recompute(self, dep_repo, engine):
+        """Test basic recompute with execution callback."""
+        _create_edge(dep_repo, "A", "B")
+
+        change = _make_change("B")
+        mock_run = PromptRun.create(
+            template_id="template-1",
+            input_bundle_hash="hash-1",
+        )
+
+        def callback(run_id):
+            return mock_run
+
+        result = engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=1),
+            execution_callback=callback,
+            old_content="old",
+            new_content="new",
+        )
+
+        assert result.changed_artifact_id == "B"
+        assert result.total_dependents == 1
+        assert result.recomputed_count == 1
+        assert result.suppressed_count == 0
+        assert len(result.executed_run_ids) == 1
+
+    def test_dry_run_no_callback(self, dep_repo, engine):
+        """Test recompute without callback records what would be recomputed."""
+        _create_edge(dep_repo, "A", "B")
+
+        change = _make_change("B")
+        result = engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=1),
+            old_content="old",
+            new_content="new",
+        )
+
+        assert result.recomputed_count == 1
+        assert result.executed_run_ids == ["A"]
+
+    def test_no_dependents(self, engine):
+        """Test recompute with no dependents."""
+        change = _make_change("isolated")
+        result = engine.recompute(change)
+
+        assert result.total_dependents == 0
+        assert result.recomputed_count == 0
+        assert result.suppressed_count == 0
+
+    def test_depth_control(self, dep_repo, engine):
+        """Test depth limiting controls recompute scope."""
+        # A -> B -> C
+        _create_edge(dep_repo, "A", "B")
+        _create_edge(dep_repo, "B", "C")
+
+        change = _make_change("C")
+
+        # Depth 1: only B
+        result1 = engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=1),
+            old_content="old",
+            new_content="new",
+        )
+        assert result1.total_dependents == 1
+        assert result1.recomputed_count == 1
+
+        # Depth 2: B and A
+        result2 = engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=2),
+            old_content="old",
+            new_content="new",
+        )
+        assert result2.total_dependents == 2
+        assert result2.recomputed_count == 2
+
+
+class TestBudgetLimits:
+    """Tests for recompute budget exhaustion."""
+
+    def test_budget_exhaustion(self, dep_repo, engine):
+        """Test budget limit suppresses excess recomputes."""
+        # Create 5 dependents of C
+        for i in range(5):
+            _create_edge(dep_repo, f"dep-{i}", "C")
+
+        change = _make_change("C")
+        result = engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=1, max_recomputes=3),
+            old_content="old",
+            new_content="new",
+        )
+
+        assert result.total_dependents == 5
+        assert result.recomputed_count == 3
+        assert result.suppressed_count == 2
+        assert all(
+            d.suppression_reason == "budget_exhausted"
+            for d in result.suppressed
+        )
+
+    def test_budget_zero_suppresses_all(self, dep_repo, engine):
+        """Test zero budget suppresses all recomputes."""
+        _create_edge(dep_repo, "A", "B")
+
+        change = _make_change("B")
+        result = engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=1, max_recomputes=0),
+            old_content="old",
+            new_content="new",
+        )
+
+        assert result.recomputed_count == 0
+        assert result.suppressed_count == 1
+
+
+class TestCircularSuppression:
+    """Tests for circular dependency suppression."""
+
+    def test_circular_dependency_suppressed(self, dep_repo, engine):
+        """Test circular dependency is suppressed."""
+        # A -> B and B -> A (circular)
+        _create_edge(dep_repo, "A", "B")
+        _create_edge(dep_repo, "B", "A")
+
+        change = _make_change("B")
+        result = engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=1, suppress_circular=True),
+            old_content="old",
+            new_content="new",
+        )
+
+        assert result.total_dependents == 1  # A is a dependent of B
+        # A depends on B, and B depends on A — would_create_cycle(A, B) is True
+        assert result.suppressed_count == 1
+        assert result.suppressed[0].suppression_reason == "circular_dependency"
+
+    def test_circular_suppression_disabled(self, dep_repo, engine):
+        """Test circular suppression can be disabled."""
+        _create_edge(dep_repo, "A", "B")
+        _create_edge(dep_repo, "B", "A")
+
+        change = _make_change("B")
+        result = engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=1, suppress_circular=False),
+            old_content="old",
+            new_content="new",
+        )
+
+        # With suppression disabled, circular deps are still recomputed
+        assert result.recomputed_count == 1
+        assert result.suppressed_count == 0
+
+
+class TestThresholdSuppression:
+    """Tests for impact threshold suppression."""
+
+    def test_below_threshold_suppressed(self, dep_repo, engine):
+        """Test below-threshold changes are suppressed."""
+        _create_edge(dep_repo, "A", "B")
+
+        change = _make_change("B")
+        # High threshold, small change
+        result = engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=1, impact_threshold=0.9),
+            old_content="hello world",
+            new_content="hello World",  # small change
+        )
+
+        assert result.suppressed_count == 1
+        assert result.suppressed[0].suppression_reason == "below_threshold"
+
+    def test_above_threshold_recomputed(self, dep_repo, engine):
+        """Test above-threshold changes trigger recompute."""
+        _create_edge(dep_repo, "A", "B")
+
+        change = _make_change("B")
+        result = engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=1, impact_threshold=0.1),
+            old_content="completely old content here",
+            new_content="entirely new different stuff",
+        )
+
+        assert result.recomputed_count == 1
+        assert result.suppressed_count == 0
+
+
+class TestDebtPersistence:
+    """Tests for impact debt persistence."""
+
+    def test_debt_recorded_in_db(self, dep_repo, engine):
+        """Test suppressed recomputes are persisted as debt."""
+        _create_edge(dep_repo, "A", "B")
+        _create_edge(dep_repo, "B", "A")
+
+        change = _make_change("B")
+        engine.recompute(
+            change,
+            config=RecomputeConfig(max_depth=1, suppress_circular=True),
+            old_content="old",
+            new_content="new",
+        )
+
+        debt = engine.get_debt_for_artifact("B")
+        assert len(debt) == 1
+        assert debt[0].suppression_reason == "circular_dependency"
+
+    def test_get_all_debt(self, dep_repo, engine):
+        """Test retrieving all debt records."""
+        # Create two separate suppressed recomputes
+        _create_edge(dep_repo, "A", "B")
+        _create_edge(dep_repo, "B", "A")
+        _create_edge(dep_repo, "C", "D")
+        _create_edge(dep_repo, "D", "C")
+
+        change1 = _make_change("B")
+        engine.recompute(
+            change1,
+            config=RecomputeConfig(max_depth=1, suppress_circular=True),
+            old_content="old",
+            new_content="new",
+        )
+
+        change2 = _make_change("D")
+        engine.recompute(
+            change2,
+            config=RecomputeConfig(max_depth=1, suppress_circular=True),
+            old_content="old",
+            new_content="new",
+        )
+
+        all_debt = engine.get_all_debt()
+        assert len(all_debt) == 2