coulomb/markitect-main
2
0
Fork 0
Code Issues 61 Pull Requests Actions Projects 12 Releases 1 Wiki Activity

feat(prompts): implement Phase 3 - Resolver Engine (FR-3)

Browse Source 
Implement deterministic multi-space resolution with configurable search order.

Core Features:
- ResolutionContext and ResolutionResult for tracking resolution state
- MultiSpaceResolutionStrategy implementing FR-3.1 search order:
  1. Local InformationSpace
  2. Explicitly included InformationSpaces
  3. Default InformationSpace
  4. Team/Shared InformationSpace
- PromptResolver with macro resolution logic
- ContextCompiler for assembling resolved prompts
- ResolutionConfig for configurable resolution behavior

Resolution Behavior:
- Required macros fail if not found (FR-3.2)
- Optional macros resolve to empty (FR-3.3)
- Generate macros detected for deferred execution (FR-3.4)
- Deterministic search order with duplicate removal
- Partial compilation support for debugging

Tests (31 passing):
- 14 strategy tests (search order, duplicates, priority)
- 9 resolver tests (required, optional, generate, multi-space)
- 8 compiler tests (substitution, dependencies, digests)

Implements:
- FR-3.1: Deterministic resolution order
- FR-3.2: Required macro validation
- FR-3.3: Optional macro fallback
- FR-3.4: Generate macro detection
- FR-3.5: Max generation depth configuration

Files Created:
- markitect/prompts/resolver/models.py
- markitect/prompts/resolver/strategy.py
- markitect/prompts/resolver/resolver.py
- markitect/prompts/resolver/compiler.py
- migrations/prompts/002_create_resolution_config.sql
- tests/unit/prompts/test_resolution_strategy.py
- tests/unit/prompts/test_prompt_resolver.py
- tests/unit/prompts/test_context_compiler.py

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Bernd Worsch
2026-02-08 22:45:46 +01:00
parent e6840fe696
commit 5f463e5b20
9 changed files with 1503 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
markitect/prompts/resolver/__init__.py Normal file
View File

@@ -0,0 +1,31 @@
"""
Resolution engine for Prompt Dependency Resolution.
This package provides the core resolution logic for resolving ContentMacros
across multiple InformationSpaces with deterministic search order.
"""
from markitect.prompts.resolver.models import (
ResolutionContext,
ResolutionResult,
ResolutionError,
ResolvedMacro,
)
from markitect.prompts.resolver.strategy import (
ResolutionStrategy,
MultiSpaceResolutionStrategy,
)
from markitect.prompts.resolver.resolver import PromptResolver
from markitect.prompts.resolver.compiler import ContextCompiler, CompiledPrompt
__all__ = [
"ResolutionContext",
"ResolutionResult",
"ResolutionError",
"ResolvedMacro",
"ResolutionStrategy",
"MultiSpaceResolutionStrategy",
"PromptResolver",
"ContextCompiler",
"CompiledPrompt",
]

202
markitect/prompts/resolver/compiler.py Normal file
View File

@@ -0,0 +1,202 @@
"""
Context compiler for assembling resolved prompts.
Compiles resolved macros into final prompt context.
"""
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional
from markitect.prompts.templates.models import PromptTemplate
from markitect.prompts.resolver.models import ResolutionResult
from markitect.prompts.models import calculate_content_digest
@dataclass
class CompiledPrompt:
"""
Compiled prompt ready for execution.
Contains the final assembled prompt with all macros resolved.
Attributes:
template_id: Source template ID
template_name: Source template name
content: Compiled prompt content with macros substituted
content_digest: SHA-256 digest of compiled content
resolution_result: Original resolution result
dependency_digests: Map of artifact name -> content digest
compiled_at: Compilation timestamp
metadata: Additional metadata
"""
template_id: str
template_name: str
content: str
content_digest: str
resolution_result: ResolutionResult
dependency_digests: Dict[str, str] = field(default_factory=dict)
compiled_at: datetime = field(default_factory=datetime.utcnow)
metadata: Dict[str, str] = field(default_factory=dict)
def to_dict(self) -> dict:
"""Convert to dictionary."""
return {
"template_id": self.template_id,
"template_name": self.template_name,
"content_digest": self.content_digest,
"content_length": len(self.content),
"dependency_count": len(self.dependency_digests),
"compiled_at": self.compiled_at.isoformat(),
"resolution_status": self.resolution_result.status.value,
}
class ContextCompiler:
"""
Compiler for assembling resolved context into final prompt.
Takes resolution results and produces CompiledPrompt with all
macros substituted.
"""
def compile(
self,
template: PromptTemplate,
template_content: str,
resolution_result: ResolutionResult,
) -> CompiledPrompt:
"""
Compile template with resolved macros into final prompt.
Args:
template: Source template
template_content: Original template content
resolution_result: Resolution result with resolved macros
Returns:
CompiledPrompt with macros substituted
Raises:
ValueError: If resolution failed
"""
if not resolution_result.success:
raise ValueError(
f"Cannot compile template '{template.name}': "
f"Resolution failed with unresolved required macros"
)
# Start with original template content
compiled_content = template_content
# Track dependency digests
dependency_digests = {}
# Substitute each resolved macro
for resolved in resolution_result.context.resolved_macros:
if resolved.resolved and resolved.artifact:
# Replace macro with resolved content
compiled_content = compiled_content.replace(
resolved.macro.raw_text,
resolved.content,
)
# Track dependency
dependency_digests[resolved.artifact.name] = resolved.artifact.content_digest
# Substitute unresolved optional macros with empty string
for macro in resolution_result.context.unresolved_optional:
compiled_content = compiled_content.replace(macro.raw_text, "")
# Calculate digest of compiled content
content_digest = calculate_content_digest(compiled_content)
return CompiledPrompt(
template_id=template.id,
template_name=template.name,
content=compiled_content,
content_digest=content_digest,
resolution_result=resolution_result,
dependency_digests=dependency_digests,
)
def compile_partial(
self,
template: PromptTemplate,
template_content: str,
resolution_result: ResolutionResult,
placeholder: str = "[UNRESOLVED]",
) -> CompiledPrompt:
"""
Compile template even with unresolved required macros.
Useful for debugging or preview. Unresolved required macros
are replaced with placeholder text.
Args:
template: Source template
template_content: Original template content
resolution_result: Resolution result (may have failures)
placeholder: Text to use for unresolved macros
Returns:
CompiledPrompt with partial resolution
"""
# Start with original template content
compiled_content = template_content
# Track dependency digests
dependency_digests = {}
# Substitute resolved macros
for resolved in resolution_result.context.resolved_macros:
if resolved.resolved and resolved.artifact:
compiled_content = compiled_content.replace(
resolved.macro.raw_text,
resolved.content,
)
dependency_digests[resolved.artifact.name] = resolved.artifact.content_digest
# Substitute unresolved required with placeholder
for macro in resolution_result.context.unresolved_required:
placeholder_text = f"{placeholder}:{macro.target}"
compiled_content = compiled_content.replace(
macro.raw_text,
placeholder_text,
)
# Substitute unresolved optional with empty
for macro in resolution_result.context.unresolved_optional:
compiled_content = compiled_content.replace(macro.raw_text, "")
content_digest = calculate_content_digest(compiled_content)
return CompiledPrompt(
template_id=template.id,
template_name=template.name,
content=compiled_content,
content_digest=content_digest,
resolution_result=resolution_result,
dependency_digests=dependency_digests,
metadata={"partial": "true", "placeholder": placeholder},
)
def get_compilation_info(self, compiled: CompiledPrompt) -> dict:
"""
Get information about compilation.
Args:
compiled: Compiled prompt
Returns:
Dictionary with compilation metadata
"""
return {
"template_id": compiled.template_id,
"template_name": compiled.template_name,
"content_length": len(compiled.content),
"content_digest": compiled.content_digest,
"dependencies": list(compiled.dependency_digests.keys()),
"dependency_count": len(compiled.dependency_digests),
"compiled_at": compiled.compiled_at.isoformat(),
"is_partial": compiled.metadata.get("partial") == "true",
}

183
markitect/prompts/resolver/models.py Normal file
View File

@@ -0,0 +1,183 @@
"""
Models for resolution engine.
Defines resolution context, results, and error types for macro resolution.
"""
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Any
from enum import Enum
from markitect.prompts.templates.models import ContentMacro, MacroKind
from markitect.prompts.models import Artifact
class ResolutionStatus(Enum):
"""Status of resolution operation."""
SUCCESS = "success"
PARTIAL = "partial" # Some optional macros missing
FAILED = "failed" # Required macros missing
class ResolutionError(Exception):
"""Raised when macro resolution fails."""
def __init__(self, message: str, macro: Optional[ContentMacro] = None):
super().__init__(message)
self.macro = macro
@dataclass
class ResolvedMacro:
"""
Result of resolving a single macro.
Attributes:
macro: Original macro
artifact: Resolved artifact (None if not found)
resolved: Whether artifact was found
space_id: Space where artifact was found
content: Resolved content (empty string if not found for optional)
"""
macro: ContentMacro
artifact: Optional[Artifact] = None
resolved: bool = False
space_id: Optional[str] = None
content: str = ""
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary."""
return {
"macro": self.macro.to_dict(),
"artifact_id": self.artifact.id if self.artifact else None,
"artifact_name": self.artifact.name if self.artifact else None,
"resolved": self.resolved,
"space_id": self.space_id,
"content_preview": self.content[:100] + "..." if len(self.content) > 100 else self.content,
}
@dataclass
class ResolutionContext:
"""
Context for macro resolution.
Tracks resolution state, search order, and resolved artifacts.
Attributes:
template_id: ID of template being resolved
space_id: Primary space ID
search_order: Ordered list of space IDs to search
resolved_macros: List of resolved macros
unresolved_required: List of unresolved required macros
unresolved_optional: List of unresolved optional macros
generator_macros: List of generate macros (deferred)
errors: List of resolution errors
metadata: Additional context metadata
"""
template_id: str
space_id: str
search_order: List[str] = field(default_factory=list)
resolved_macros: List[ResolvedMacro] = field(default_factory=list)
unresolved_required: List[ContentMacro] = field(default_factory=list)
unresolved_optional: List[ContentMacro] = field(default_factory=list)
generator_macros: List[ContentMacro] = field(default_factory=list)
errors: List[str] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
def add_resolved(self, resolved: ResolvedMacro) -> None:
"""Add a resolved macro to context."""
self.resolved_macros.append(resolved)
def add_unresolved_required(self, macro: ContentMacro) -> None:
"""Record an unresolved required macro."""
self.unresolved_required.append(macro)
self.errors.append(
f"Required macro '{macro.target}' not found (line {macro.line_number})"
)
def add_unresolved_optional(self, macro: ContentMacro) -> None:
"""Record an unresolved optional macro."""
self.unresolved_optional.append(macro)
def add_generator(self, macro: ContentMacro) -> None:
"""Record a generator macro for deferred execution."""
self.generator_macros.append(macro)
def has_errors(self) -> bool:
"""Check if any resolution errors occurred."""
return len(self.errors) > 0
def get_status(self) -> ResolutionStatus:
"""Get overall resolution status."""
if self.unresolved_required:
return ResolutionStatus.FAILED
elif self.unresolved_optional:
return ResolutionStatus.PARTIAL
else:
return ResolutionStatus.SUCCESS
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary."""
return {
"template_id": self.template_id,
"space_id": self.space_id,
"search_order": self.search_order,
"resolved_count": len(self.resolved_macros),
"unresolved_required": [m.target for m in self.unresolved_required],
"unresolved_optional": [m.target for m in self.unresolved_optional],
"generator_count": len(self.generator_macros),
"status": self.get_status().value,
"errors": self.errors,
}
@dataclass
class ResolutionResult:
"""
Result of template resolution.
Contains resolved content and metadata about the resolution process.
Attributes:
context: Resolution context with full state
success: Whether all required macros were resolved
resolved_content: Dictionary of macro -> resolved content
dependency_artifacts: List of artifact IDs used
needs_generation: Whether any generate macros were found
"""
context: ResolutionContext
success: bool
resolved_content: Dict[str, str] = field(default_factory=dict)
dependency_artifacts: List[str] = field(default_factory=list)
needs_generation: bool = False
@property
def status(self) -> ResolutionStatus:
"""Get resolution status."""
return self.context.get_status()
def get_resolved_macro(self, target: str) -> Optional[ResolvedMacro]:
"""
Get resolved macro by target name.
Args:
target: Macro target name
Returns:
ResolvedMacro if found, None otherwise
"""
for resolved in self.context.resolved_macros:
if resolved.macro.target == target:
return resolved
return None
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary."""
return {
"success": self.success,
"status": self.status.value,
"context": self.context.to_dict(),
"dependency_count": len(self.dependency_artifacts),
"needs_generation": self.needs_generation,
}

223
markitect/prompts/resolver/resolver.py Normal file
View File

@@ -0,0 +1,223 @@
"""
PromptResolver for resolving template macros.
Implements FR-3: PromptResolver Behavior
"""
from typing import List, Optional
from markitect.prompts.templates.models import PromptTemplate, ContentMacro, MacroKind
from markitect.prompts.resolver.models import (
ResolutionContext,
ResolutionResult,
ResolutionError,
ResolvedMacro,
)
from markitect.prompts.resolver.strategy import ResolutionStrategy, ResolutionConfig
from markitect.prompts.services.artifact_service import ArtifactService
class PromptResolver:
"""
Resolver for prompt template macros.
Implements FR-3: PromptResolver Behavior
- Deterministic resolution order (FR-3.1)
- Required macro validation (FR-3.2)
- Optional macro fallback (FR-3.3)
- Generate macro detection (FR-3.4)
"""
def __init__(
self,
artifact_service: ArtifactService,
strategy: ResolutionStrategy,
):
"""
Initialize resolver.
Args:
artifact_service: Service for artifact lookup
strategy: Resolution strategy for search order
"""
self.artifact_service = artifact_service
self.strategy = strategy
def resolve_template(
self,
template: PromptTemplate,
config: ResolutionConfig,
) -> ResolutionResult:
"""
Resolve all macros in a template.
Implements FR-3: PromptResolver Behavior
Args:
template: Template to resolve
config: Resolution configuration
Returns:
ResolutionResult with resolved macros and status
Raises:
ValueError: If template hasn't been analyzed
"""
if not template.analyzed:
raise ValueError(
f"Template '{template.name}' must be analyzed before resolution. "
"Call TemplateAnalyzer.analyze() first."
)
# Get search order
search_order = self.strategy.get_search_order(config)
# Create resolution context
context = ResolutionContext(
template_id=template.id,
space_id=config.space_id,
search_order=search_order,
)
# Track resolved content and dependencies
resolved_content = {}
dependency_artifacts = []
# Resolve each macro
for macro in template.macros:
resolved = self._resolve_macro(macro, search_order, context)
if resolved.resolved and resolved.artifact:
resolved_content[macro.target] = resolved.content
dependency_artifacts.append(resolved.artifact.id)
# Create result
result = ResolutionResult(
context=context,
success=len(context.unresolved_required) == 0,
resolved_content=resolved_content,
dependency_artifacts=dependency_artifacts,
needs_generation=len(context.generator_macros) > 0,
)
return result
def _resolve_macro(
self,
macro: ContentMacro,
search_order: List[str],
context: ResolutionContext,
) -> ResolvedMacro:
"""
Resolve a single macro.
Implements:
- FR-3.2: Required macro failure
- FR-3.3: Optional macro empty fallback
- FR-3.4: Generate macro detection
Args:
macro: Macro to resolve
search_order: Ordered space IDs to search
context: Resolution context
Returns:
ResolvedMacro with resolution result
"""
# Handle generate macros separately (FR-3.4)
if macro.kind == MacroKind.GENERATE:
context.add_generator(macro)
return ResolvedMacro(
macro=macro,
artifact=None,
resolved=False, # Will be resolved during generation phase
space_id=None,
content="",
)
# Try to resolve in each space
for space_id in search_order:
artifact = self.artifact_service.repository.get_by_name(
space_id,
macro.target,
)
if artifact:
# Found! Get content (would need to load from storage in real impl)
# For now, we'll use a placeholder
content = f"[Content of {artifact.name} from {space_id}]"
resolved = ResolvedMacro(
macro=macro,
artifact=artifact,
resolved=True,
space_id=space_id,
content=content,
)
context.add_resolved(resolved)
return resolved
# Not found in any space
if macro.kind == MacroKind.REQUIRED:
# FR-3.2: Required macros fail if not found
context.add_unresolved_required(macro)
return ResolvedMacro(
macro=macro,
artifact=None,
resolved=False,
space_id=None,
content="",
)
else:
# FR-3.3: Optional macros resolve to empty
context.add_unresolved_optional(macro)
return ResolvedMacro(
macro=macro,
artifact=None,
resolved=False,
space_id=None,
content="", # Empty content for optional
)
def validate_resolution(self, result: ResolutionResult) -> None:
"""
Validate resolution result.
Args:
result: Resolution result to validate
Raises:
ResolutionError: If required macros are missing
"""
if not result.success:
unresolved = result.context.unresolved_required
targets = [m.target for m in unresolved]
raise ResolutionError(
f"Failed to resolve required macros: {', '.join(targets)}"
)
def get_resolution_summary(self, result: ResolutionResult) -> dict:
"""
Get human-readable summary of resolution.
Args:
result: Resolution result
Returns:
Dictionary with summary information
"""
return {
"status": result.status.value,
"success": result.success,
"resolved_count": len(result.context.resolved_macros),
"unresolved_required": [
m.target for m in result.context.unresolved_required
],
"unresolved_optional": [
m.target for m in result.context.unresolved_optional
],
"needs_generation": result.needs_generation,
"generator_count": len(result.context.generator_macros),
"dependency_count": len(result.dependency_artifacts),
"search_order": result.context.search_order,
}

152
markitect/prompts/resolver/strategy.py Normal file
View File

@@ -0,0 +1,152 @@
"""
Resolution strategies for multi-space artifact lookup.
Implements FR-3.1: Deterministic resolution order
"""
from abc import ABC, abstractmethod
from typing import List, Optional
from dataclasses import dataclass, field
@dataclass
class ResolutionConfig:
"""
Configuration for resolution strategy.
Implements FR-3.1: Resolution order configuration
Resolution order:
1. Local InformationSpace
2. Explicitly included InformationSpaces
3. Default InformationSpace
4. Team/Shared InformationSpace (if configured)
Attributes:
space_id: Primary space ID
included_spaces: Explicitly included space IDs (ordered)
default_space_id: Default space for common artifacts
shared_space_id: Team/shared space (optional)
max_generation_depth: Maximum nesting depth for generators
"""
space_id: str
included_spaces: List[str] = field(default_factory=list)
default_space_id: Optional[str] = None
shared_space_id: Optional[str] = None
max_generation_depth: int = 3
def to_dict(self) -> dict:
"""Convert to dictionary for serialization."""
return {
"space_id": self.space_id,
"included_spaces": self.included_spaces,
"default_space_id": self.default_space_id,
"shared_space_id": self.shared_space_id,
"max_generation_depth": self.max_generation_depth,
}
@classmethod
def from_dict(cls, data: dict) -> "ResolutionConfig":
"""Create from dictionary."""
return cls(
space_id=data["space_id"],
included_spaces=data.get("included_spaces", []),
default_space_id=data.get("default_space_id"),
shared_space_id=data.get("shared_space_id"),
max_generation_depth=data.get("max_generation_depth", 3),
)
class ResolutionStrategy(ABC):
"""
Abstract base class for resolution strategies.
Defines how to search for artifacts across multiple spaces.
"""
@abstractmethod
def get_search_order(self, config: ResolutionConfig) -> List[str]:
"""
Get ordered list of space IDs to search.
Args:
config: Resolution configuration
Returns:
Ordered list of space IDs (no duplicates)
"""
pass
class MultiSpaceResolutionStrategy(ResolutionStrategy):
"""
Multi-space resolution strategy with deterministic search order.
Implements FR-3.1: Resolution order
1. Local space
2. Included spaces (in order)
3. Default space
4. Shared space
"""
def get_search_order(self, config: ResolutionConfig) -> List[str]:
"""
Get deterministic search order.
Implements FR-3.1 resolution order:
1. Local InformationSpace (config.space_id)
2. Explicitly included InformationSpaces (config.included_spaces)
3. Default InformationSpace (config.default_space_id)
4. Team/Shared InformationSpace (config.shared_space_id)
Removes duplicates while preserving order.
Args:
config: Resolution configuration
Returns:
Ordered list of unique space IDs
"""
search_order = []
seen = set()
def add_if_not_seen(space_id: Optional[str]) -> None:
"""Add space ID if not None and not already seen."""
if space_id and space_id not in seen:
search_order.append(space_id)
seen.add(space_id)
# 1. Local space (highest priority)
add_if_not_seen(config.space_id)
# 2. Included spaces (in order)
for included_id in config.included_spaces:
add_if_not_seen(included_id)
# 3. Default space
add_if_not_seen(config.default_space_id)
# 4. Shared space (lowest priority)
add_if_not_seen(config.shared_space_id)
return search_order
class SingleSpaceResolutionStrategy(ResolutionStrategy):
"""
Simple strategy that only searches the local space.
Useful for isolated execution or testing.
"""
def get_search_order(self, config: ResolutionConfig) -> List[str]:
"""
Return only the local space.
Args:
config: Resolution configuration
Returns:
List containing only the local space ID
"""
return [config.space_id]