""" Core domain models for Prompt Dependency Resolution. This module provides foundational data models for addressable artifacts, references, and content hashing for change detection. """ import hashlib import uuid from dataclasses import dataclass, field from datetime import datetime from typing import Dict, Any, Optional from enum import Enum class ArtifactType(Enum): """Type classification for artifacts.""" CONTENT = "content" # Regular content artifact TEMPLATE = "template" # Prompt template GENERATED = "generated" # LLM-generated artifact SCHEMA = "schema" # Validation schema CONFIG = "config" # Configuration artifact @dataclass class ArtifactMetadata: """ Extensible metadata for artifacts. Attributes: description: Human-readable description tags: List of tags for categorization author: Author identifier version: Version string (optional) custom: Dictionary for custom metadata fields """ description: Optional[str] = None tags: list[str] = field(default_factory=list) author: Optional[str] = None version: Optional[str] = None custom: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: """Convert metadata to dictionary for serialization.""" return { "description": self.description, "tags": self.tags, "author": self.author, "version": self.version, "custom": self.custom, } @classmethod def from_dict(cls, data: Dict[str, Any]) -> "ArtifactMetadata": """Create metadata from dictionary.""" return cls( description=data.get("description"), tags=data.get("tags", []), author=data.get("author"), version=data.get("version"), custom=data.get("custom", {}), ) @dataclass class Artifact: """ Core artifact entity with content-based addressing. Implements FR-1: InformationSpace Addressability - Unique persistent identifier - Content digest for change detection - Cross-space references via space_id Attributes: id: Unique identifier (UUID) space_id: ID of containing InformationSpace name: Human-readable name (unique within space) artifact_type: Classification of artifact content_digest: SHA-256 hash of content content_size: Size of content in bytes metadata: Extensible metadata created_at: Creation timestamp updated_at: Last modification timestamp """ id: str space_id: str name: str artifact_type: ArtifactType content_digest: str content_size: int = 0 metadata: ArtifactMetadata = field(default_factory=ArtifactMetadata) created_at: datetime = field(default_factory=datetime.utcnow) updated_at: datetime = field(default_factory=datetime.utcnow) @classmethod def create( cls, space_id: str, name: str, content: str, artifact_type: ArtifactType = ArtifactType.CONTENT, metadata: Optional[ArtifactMetadata] = None, ) -> "Artifact": """ Create a new artifact with automatic ID and digest generation. Args: space_id: ID of containing space name: Artifact name content: Artifact content artifact_type: Type classification metadata: Optional metadata Returns: New Artifact instance with computed digest """ artifact_id = str(uuid.uuid4()) content_digest = calculate_content_digest(content) content_size = len(content.encode('utf-8')) return cls( id=artifact_id, space_id=space_id, name=name, artifact_type=artifact_type, content_digest=content_digest, content_size=content_size, metadata=metadata or ArtifactMetadata(), ) def update_content(self, new_content: str) -> None: """ Update artifact content and recalculate digest. Args: new_content: New content string """ self.content_digest = calculate_content_digest(new_content) self.content_size = len(new_content.encode('utf-8')) self.updated_at = datetime.utcnow() def has_changed(self, current_digest: str) -> bool: """ Check if content has changed by comparing digests. Args: current_digest: Digest to compare against Returns: True if digests differ """ return self.content_digest != current_digest def to_dict(self) -> Dict[str, Any]: """Convert artifact to dictionary for serialization.""" return { "id": self.id, "space_id": self.space_id, "name": self.name, "artifact_type": self.artifact_type.value, "content_digest": self.content_digest, "content_size": self.content_size, "metadata": self.metadata.to_dict(), "created_at": self.created_at.isoformat(), "updated_at": self.updated_at.isoformat(), } @classmethod def from_dict(cls, data: Dict[str, Any]) -> "Artifact": """Create artifact from dictionary.""" return cls( id=data["id"], space_id=data["space_id"], name=data["name"], artifact_type=ArtifactType(data["artifact_type"]), content_digest=data["content_digest"], content_size=data.get("content_size", 0), metadata=ArtifactMetadata.from_dict(data.get("metadata", {})), created_at=datetime.fromisoformat(data["created_at"]), updated_at=datetime.fromisoformat(data["updated_at"]), ) @dataclass class ArtifactReference: """ Reference to an artifact, possibly in another space. Implements FR-1.3: Cross-space artifact references Attributes: name: Artifact name to resolve space_id: Optional specific space ID (if not provided, uses resolution order) version: Optional version constraint """ name: str space_id: Optional[str] = None version: Optional[str] = None def __str__(self) -> str: """String representation for display and debugging.""" if self.space_id: ref = f"{self.space_id}:{self.name}" else: ref = self.name if self.version: ref = f"{ref}@{self.version}" return ref @classmethod def parse(cls, reference_str: str) -> "ArtifactReference": """ Parse a reference string into components. Formats: - "artifact-name" - "space-id:artifact-name" - "artifact-name@version" - "space-id:artifact-name@version" Args: reference_str: Reference string to parse Returns: Parsed ArtifactReference Raises: ValueError: If reference string is malformed """ # Split version if present if '@' in reference_str: ref_part, version = reference_str.rsplit('@', 1) else: ref_part, version = reference_str, None # Split space if present if ':' in ref_part: space_id, name = ref_part.split(':', 1) else: space_id, name = None, ref_part if not name: raise ValueError(f"Invalid artifact reference: {reference_str}") return cls(name=name, space_id=space_id, version=version) def calculate_content_digest(content: str) -> str: """ Calculate SHA-256 digest of content. Implements FR-1.2: Content digest computation Args: content: Content string to hash Returns: Hexadecimal SHA-256 digest """ return hashlib.sha256(content.encode('utf-8')).hexdigest() def calculate_bundle_digest(components: Dict[str, str]) -> str: """ Calculate digest of multiple components (for InputBundleHash). Args: components: Dictionary of component name -> digest Returns: Hexadecimal SHA-256 digest of sorted components """ # Sort components for deterministic hashing sorted_items = sorted(components.items()) bundle_str = ':'.join(f"{k}={v}" for k, v in sorted_items) return hashlib.sha256(bundle_str.encode('utf-8')).hexdigest()