Architecture

# Analyze agent mode
/specfact.01-import legacy-api --repo . --confidence 0.7
# → Enhanced prompts for code understanding
# → Context injection (current file, selection, workspace)
# → Interactive assistance for complex codebases

# Plan agent mode
/specfact.02-plan init legacy-api
# → Guided wizard mode
# → Natural language prompts
# → Context-aware feature extraction

# Sync agent mode
/specfact.06-sync --adapter speckit --repo . --bidirectional
# → Automatic source detection via bridge adapter
# → Conflict resolution assistance
# → Change explanation and preview

from icontract import require, ensure
from beartype import beartype

@require(lambda plan: plan.version == "1.0")
@ensure(lambda result: len(result.features) > 0)
@beartype
def validate_plan(plan: PlanBundle) -> ValidationResult:
    """Validate plan bundle against contracts."""
    return ValidationResult(valid=True)

from hypothesis import given
from hypothesis.strategies import text

@given(text())
def test_plan_key_format(feature_key: str):
    """All feature keys must match FEATURE-\d+ format."""
    if feature_key.startswith("FEATURE-"):
        assert feature_key[8:].isdigit()

import asyncio
from typing import Optional

class EventLoopMonitor:
    """Monitor event loop health."""
    
    def __init__(self, lag_threshold_ms: float = 100.0):
        self.lag_threshold_ms = lag_threshold_ms
    
    async def check_lag(self) -> Optional[float]:
        """Return lag in ms if above threshold."""
        start = asyncio.get_event_loop().time()
        await asyncio.sleep(0)
        lag_ms = (asyncio.get_event_loop().time() - start) * 1000
        return lag_ms if lag_ms > self.lag_threshold_ms else None

from typing import Optional
import time

class BudgetedValidator:
    """Validator with time budget."""
    
    def __init__(self, budget_seconds: int = 120):
        self.budget_seconds = budget_seconds
        self.start_time: Optional[float] = None
    
    def start(self):
        """Start budget timer."""
        self.start_time = time.time()
    
    def check_budget(self) -> bool:
        """Return True if budget exceeded."""
        if self.start_time is None:
            return False
        elapsed = time.time() - self.start_time
        return elapsed > self.budget_seconds

from pydantic import BaseModel, Field
from typing import List

class Idea(BaseModel):
    """High-level idea."""
    title: str
    narrative: str

class Story(BaseModel):
    """User story."""
    key: str = Field(pattern=r"^STORY-\d+$")
    title: str
    acceptance: List[str]

class Feature(BaseModel):
    """Feature with stories."""
    key: str = Field(pattern=r"^FEATURE-\d+$")
    title: str
    outcomes: List[str]
    stories: List[Story]

class PlanBundle(BaseModel):
    """Complete plan bundle."""
    version: str = "1.0"
    idea: Idea
    features: List[Feature]

from pydantic import BaseModel
from typing import List, Optional

class Transition(BaseModel):
    """State machine transition."""
    from_state: str
    on_event: str
    to_state: str
    guard: Optional[str] = None

class ProtocolSpec(BaseModel):
    """FSM protocol specification."""
    states: List[str]
    start: str
    transitions: List[Transition]

from enum import Enum
from pydantic import BaseModel

class DeviationSeverity(str, Enum):
    """Severity levels."""
    LOW = "LOW"
    MEDIUM = "MEDIUM"
    HIGH = "HIGH"
    CRITICAL = "CRITICAL"

class Deviation(BaseModel):
    """Detected deviation."""
    type: str
    severity: DeviationSeverity
    description: str
    location: str
    suggestion: Optional[str] = None

from enum import Enum
from pydantic import BaseModel
from typing import Optional, Dict, List, Any

class ChangeType(str, Enum):
    """Change type for delta specs (tool-agnostic)."""
    ADDED = "added"
    MODIFIED = "modified"
    REMOVED = "removed"

class FeatureDelta(BaseModel):
    """Delta tracking for a feature change (tool-agnostic)."""
    feature_key: str
    change_type: ChangeType
    original_feature: Optional[Feature] = None  # For MODIFIED/REMOVED
    proposed_feature: Optional[Feature] = None  # For ADDED/MODIFIED
    change_rationale: Optional[str] = None
    change_date: Optional[str] = None  # ISO timestamp
    validation_status: Optional[str] = None  # pending, passed, failed
    validation_results: Optional[Dict[str, Any]] = None
    source_tracking: Optional[SourceTracking] = None  # Tool-specific metadata

class ChangeProposal(BaseModel):
    """Change proposal (tool-agnostic, used by OpenSpec and other tools)."""
    name: str  # Change identifier (e.g., 'add-user-feedback')
    title: str
    description: str  # What: Description of the change
    rationale: str  # Why: Rationale and business value
    timeline: Optional[str] = None  # When: Timeline and dependencies
    owner: Optional[str] = None  # Who: Owner and stakeholders
    stakeholders: List[str] = []
    dependencies: List[str] = []
    status: str = "proposed"  # proposed, in-progress, applied, archived
    created_at: str  # ISO timestamp
    applied_at: Optional[str] = None
    archived_at: Optional[str] = None
    source_tracking: Optional[SourceTracking] = None  # Tool-specific metadata

class ChangeTracking(BaseModel):
    """Change tracking for a bundle (tool-agnostic capability)."""
    proposals: Dict[str, ChangeProposal] = {}  # change_name -> ChangeProposal
    feature_deltas: Dict[str, List[FeatureDelta]] = {}  # change_name -> [FeatureDelta]

class ChangeArchive(BaseModel):
    """Archive entry for completed changes (tool-agnostic)."""
    change_name: str
    applied_at: str  # ISO timestamp
    applied_by: Optional[str] = None
    pr_number: Optional[str] = None
    commit_hash: Optional[str] = None
    feature_deltas: List[FeatureDelta] = []
    validation_results: Optional[Dict[str, Any]] = None
    source_tracking: Optional[SourceTracking] = None  # Tool-specific metadata

from abc import ABC, abstractmethod
from pathlib import Path
from specfact_cli.models.bridge import BridgeConfig
from specfact_cli.models.change import ChangeProposal, ChangeTracking

class BridgeAdapter(ABC):
    @abstractmethod
    def detect(self, repo_path: Path, bridge_config: BridgeConfig | None = None) -> bool:
        """Detect if adapter applies to repository."""

    @abstractmethod
    def import_artifact(self, artifact_key: str, artifact_path: Path | dict, project_bundle: Any, bridge_config: BridgeConfig | None = None) -> None:
        """Import artifact from tool format to SpecFact."""

    @abstractmethod
    def export_artifact(self, artifact_key: str, artifact_data: Any, bridge_config: BridgeConfig | None = None) -> Path | dict:
        """Export artifact from SpecFact to tool format."""

    @abstractmethod
    def generate_bridge_config(self, repo_path: Path) -> BridgeConfig:
        """Generate bridge configuration for adapter."""
    
    @abstractmethod
    def get_capabilities(self, repo_path: Path, bridge_config: BridgeConfig | None = None) -> ToolCapabilities:
        """Get adapter capabilities (sync modes, layout, etc.)."""

@abstractmethod
def load_change_tracking(
    self, bundle_dir: Path, bridge_config: BridgeConfig | None = None
) -> ChangeTracking | None:
    """
    Load change tracking from adapter-specific storage location.
    
    Args:
        bundle_dir: Path to bundle directory (.specfact/projects/<bundle-name>/)
        bridge_config: Bridge configuration (may contain external_base_path for cross-repo)
    
    Returns:
        ChangeTracking instance or None if not available
    """

@abstractmethod
def save_change_tracking(
    self, bundle_dir: Path, change_tracking: ChangeTracking, bridge_config: BridgeConfig | None = None
) -> None:
    """
    Save change tracking to adapter-specific storage location.
    
    Args:
        bundle_dir: Path to bundle directory
        change_tracking: ChangeTracking instance to save
        bridge_config: Bridge configuration (may contain external_base_path for cross-repo)
    """

@abstractmethod
def load_change_proposal(
    self, bundle_dir: Path, change_name: str, bridge_config: BridgeConfig | None = None
) -> ChangeProposal | None:
    """
    Load change proposal from adapter-specific storage location.
    
    Args:
        bundle_dir: Path to bundle directory
        change_name: Change identifier (e.g., 'add-user-feedback')
        bridge_config: Bridge configuration (may contain external_base_path for cross-repo)
    
    Returns:
        ChangeProposal instance or None if not found
    """

@abstractmethod
def save_change_proposal(
    self, bundle_dir: Path, proposal: ChangeProposal, bridge_config: BridgeConfig | None = None
) -> None:
    """
    Save change proposal to adapter-specific storage location.
    
    Args:
        bundle_dir: Path to bundle directory
        proposal: ChangeProposal instance to save
        bridge_config: Bridge configuration (may contain external_base_path for cross-repo)
    """

class OpenSpecAdapter(BridgeAdapter):
    def detect(self, repo_path: Path, bridge_config: BridgeConfig | None = None) -> bool:
        # Detects openspec/project.md or openspec/specs/ directory
        base_path = bridge_config.external_base_path if bridge_config and bridge_config.external_base_path else repo_path
        return (base_path / "openspec" / "project.md").exists() or (base_path / "openspec" / "specs").exists()
    
    def get_capabilities(self, repo_path: Path, bridge_config: BridgeConfig | None = None) -> ToolCapabilities:
        # Returns OpenSpec-specific capabilities
        return ToolCapabilities(tool="openspec", layout="openspec", specs_dir="openspec/specs")
    
    def load_change_tracking(self, bundle_dir: Path, bridge_config: BridgeConfig | None = None) -> ChangeTracking | None:
        # Load from openspec/changes/ directory
        base_path = bridge_config.external_base_path if bridge_config and bridge_config.external_base_path else bundle_dir.parent.parent.parent
        changes_dir = base_path / "openspec" / "changes"
        # Parse change proposals and feature deltas
        return ChangeTracking(...)
    
    def import_artifact(self, artifact_key: str, artifact_path: Path, project_bundle: Any, bridge_config: BridgeConfig | None = None) -> None:
        # Supports: specification, project_context, change_proposal, change_spec_delta
        # Parses OpenSpec markdown and updates project bundle
        pass

class SpecKitAdapter(BridgeAdapter):
    def detect(self, repo_path: Path, bridge_config: BridgeConfig | None = None) -> bool:
        # Detects .specify/ directory or specs/ directory (classic/modern layouts)
        base_path = bridge_config.external_base_path if bridge_config and bridge_config.external_base_path else repo_path
        return (base_path / ".specify").exists() or (base_path / "specs").exists() or (base_path / "docs" / "specs").exists()
    
    def get_capabilities(self, repo_path: Path, bridge_config: BridgeConfig | None = None) -> ToolCapabilities:
        # Returns Spec-Kit-specific capabilities (bidirectional sync supported)
        return ToolCapabilities(
            tool="speckit",
            layout="classic" or "modern",
            specs_dir="specs" or "docs/specs",
            supported_sync_modes=["bidirectional", "unidirectional"]
        )
    
    def import_artifact(self, artifact_key: str, artifact_path: Path, project_bundle: Any, bridge_config: BridgeConfig | None = None) -> None:
        # Supports: specification, plan, tasks, constitution
        # Parses Spec-Kit markdown and updates project bundle
        pass
    
    def export_artifact(self, artifact_key: str, artifact_data: Any, bridge_config: BridgeConfig | None = None) -> Path:
        # Supports: specification, plan, tasks, constitution
        # Exports SpecFact models to Spec-Kit markdown format
        pass

class GitHubAdapter(BridgeAdapter):
    def load_change_tracking(self, bundle_dir: Path, bridge_config: BridgeConfig | None = None) -> ChangeTracking | None:
        # GitHub adapter is export-only (OpenSpec → GitHub Issues)
        return None
    
    def save_change_tracking(self, bundle_dir: Path, change_tracking: ChangeTracking, bridge_config: BridgeConfig | None = None) -> None:
        # Export change proposals to GitHub Issues
        pass
    
    def export_artifact(self, artifact_key: str, artifact_data: Any, bridge_config: BridgeConfig | None = None) -> dict:
        # Supports artifact keys: change_proposal, change_status, change_proposal_update, code_change_progress
        if artifact_key == "code_change_progress":
            # Add progress comment to existing GitHub issue based on code changes
            return self._add_progress_comment(artifact_data, ...)