fix(prompts): fix three infrastructure bugs in prompt dependency resolution
- ContentMacro: add __post_init__ to auto-derive raw_text when built
programmatically, preventing str.replace("", X) corruption
- MacroParser: add @{target} shorthand syntax support mapped to REQUIRED kind,
updating parse, has_macros, count_macros, and find_macro_positions
- Artifact: store content in model and SQLite DB, replace resolver placeholder
with actual artifact content, add migration for existing databases
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -89,6 +89,7 @@ class Artifact:
|
||||
artifact_type: ArtifactType
|
||||
content_digest: str
|
||||
content_size: int = 0
|
||||
content: str = ""
|
||||
metadata: ArtifactMetadata = field(default_factory=ArtifactMetadata)
|
||||
created_at: datetime = field(default_factory=datetime.utcnow)
|
||||
updated_at: datetime = field(default_factory=datetime.utcnow)
|
||||
@@ -126,6 +127,7 @@ class Artifact:
|
||||
artifact_type=artifact_type,
|
||||
content_digest=content_digest,
|
||||
content_size=content_size,
|
||||
content=content,
|
||||
metadata=metadata or ArtifactMetadata(),
|
||||
)
|
||||
|
||||
@@ -136,6 +138,7 @@ class Artifact:
|
||||
Args:
|
||||
new_content: New content string
|
||||
"""
|
||||
self.content = new_content
|
||||
self.content_digest = calculate_content_digest(new_content)
|
||||
self.content_size = len(new_content.encode('utf-8'))
|
||||
self.updated_at = datetime.utcnow()
|
||||
@@ -161,6 +164,7 @@ class Artifact:
|
||||
"artifact_type": self.artifact_type.value,
|
||||
"content_digest": self.content_digest,
|
||||
"content_size": self.content_size,
|
||||
"content": self.content,
|
||||
"metadata": self.metadata.to_dict(),
|
||||
"created_at": self.created_at.isoformat(),
|
||||
"updated_at": self.updated_at.isoformat(),
|
||||
@@ -176,6 +180,7 @@ class Artifact:
|
||||
artifact_type=ArtifactType(data["artifact_type"]),
|
||||
content_digest=data["content_digest"],
|
||||
content_size=data.get("content_size", 0),
|
||||
content=data.get("content", ""),
|
||||
metadata=ArtifactMetadata.from_dict(data.get("metadata", {})),
|
||||
created_at=datetime.fromisoformat(data["created_at"]),
|
||||
updated_at=datetime.fromisoformat(data["updated_at"]),
|
||||
|
||||
@@ -30,6 +30,7 @@ CREATE TABLE IF NOT EXISTS prompt_artifacts (
|
||||
artifact_type TEXT NOT NULL,
|
||||
content_digest TEXT NOT NULL,
|
||||
content_size INTEGER DEFAULT 0,
|
||||
content TEXT DEFAULT '',
|
||||
metadata JSON,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
@@ -59,6 +60,13 @@ def initialize_artifact_tables(db_path: str) -> None:
|
||||
conn = sqlite3.connect(db_path)
|
||||
try:
|
||||
conn.executescript(ARTIFACT_TABLES_SQL)
|
||||
# Migration: add content column to existing databases
|
||||
try:
|
||||
conn.execute(
|
||||
"ALTER TABLE prompt_artifacts ADD COLUMN content TEXT DEFAULT ''"
|
||||
)
|
||||
except sqlite3.OperationalError:
|
||||
pass # Column already exists
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
@@ -107,8 +115,8 @@ class SQLiteArtifactRepository(IArtifactRepository):
|
||||
"""
|
||||
INSERT INTO prompt_artifacts (
|
||||
id, space_id, name, artifact_type, content_digest,
|
||||
content_size, metadata, created_at, updated_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
content_size, content, metadata, created_at, updated_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
(
|
||||
artifact.id,
|
||||
@@ -117,6 +125,7 @@ class SQLiteArtifactRepository(IArtifactRepository):
|
||||
artifact.artifact_type.value,
|
||||
artifact.content_digest,
|
||||
artifact.content_size,
|
||||
artifact.content,
|
||||
json.dumps(artifact.metadata.to_dict()),
|
||||
artifact.created_at.isoformat(),
|
||||
artifact.updated_at.isoformat(),
|
||||
@@ -251,12 +260,14 @@ class SQLiteArtifactRepository(IArtifactRepository):
|
||||
cursor = conn.execute(
|
||||
"""
|
||||
UPDATE prompt_artifacts
|
||||
SET content_digest = ?, content_size = ?, metadata = ?, updated_at = ?
|
||||
SET content_digest = ?, content_size = ?, content = ?,
|
||||
metadata = ?, updated_at = ?
|
||||
WHERE id = ?
|
||||
""",
|
||||
(
|
||||
artifact.content_digest,
|
||||
artifact.content_size,
|
||||
artifact.content,
|
||||
json.dumps(artifact.metadata.to_dict()),
|
||||
artifact.updated_at.isoformat(),
|
||||
artifact.id,
|
||||
@@ -326,6 +337,7 @@ class SQLiteArtifactRepository(IArtifactRepository):
|
||||
artifact_type=ArtifactType(row["artifact_type"]),
|
||||
content_digest=row["content_digest"],
|
||||
content_size=row["content_size"],
|
||||
content=row["content"] or "",
|
||||
metadata=ArtifactMetadata.from_dict(metadata_dict),
|
||||
created_at=datetime.fromisoformat(row["created_at"]),
|
||||
updated_at=datetime.fromisoformat(row["updated_at"]),
|
||||
|
||||
@@ -143,9 +143,7 @@ class PromptResolver:
|
||||
)
|
||||
|
||||
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}]"
|
||||
content = artifact.content
|
||||
|
||||
resolved = ResolvedMacro(
|
||||
macro=macro,
|
||||
|
||||
@@ -46,6 +46,11 @@ class ContentMacro:
|
||||
raw_text: str = ""
|
||||
line_number: int = 0
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
"""Auto-derive raw_text when built programmatically."""
|
||||
if not self.raw_text:
|
||||
self.raw_text = f"@{{{self.target}}}"
|
||||
|
||||
def __str__(self) -> str:
|
||||
"""String representation of macro."""
|
||||
params = ''.join(f"|{k}={v}" for k, v in self.parameters.items())
|
||||
|
||||
@@ -38,6 +38,9 @@ class MacroParser:
|
||||
re.IGNORECASE
|
||||
)
|
||||
|
||||
# Shorthand pattern: @{target} — maps to MacroKind.REQUIRED
|
||||
SHORTHAND_PATTERN = re.compile(r'@\{([^}]+)\}')
|
||||
|
||||
# Parameter pattern: |key=value
|
||||
PARAM_PATTERN = re.compile(r'\|([^=]+)=([^|]+)')
|
||||
|
||||
@@ -95,6 +98,19 @@ class MacroParser:
|
||||
f"Line {line_number}: {e}"
|
||||
) from e
|
||||
|
||||
# Scan for @{target} shorthand syntax
|
||||
for match in self.SHORTHAND_PATTERN.finditer(line):
|
||||
target = match.group(1).strip()
|
||||
raw_text = match.group(0)
|
||||
if target:
|
||||
macros.append(ContentMacro(
|
||||
kind=MacroKind.REQUIRED,
|
||||
target=target,
|
||||
parameters={},
|
||||
raw_text=raw_text,
|
||||
line_number=line_number,
|
||||
))
|
||||
|
||||
return macros
|
||||
|
||||
def _parse_match(self, match: re.Match, line_number: int) -> ContentMacro:
|
||||
@@ -172,7 +188,7 @@ class MacroParser:
|
||||
content: Template content
|
||||
|
||||
Returns:
|
||||
List of (start_pos, end_pos, macro_text) tuples
|
||||
List of (start_pos, end_pos, macro_text) tuples sorted by position
|
||||
"""
|
||||
positions = []
|
||||
for match in self.MACRO_PATTERN.finditer(content):
|
||||
@@ -181,6 +197,13 @@ class MacroParser:
|
||||
match.end(),
|
||||
match.group(0)
|
||||
))
|
||||
for match in self.SHORTHAND_PATTERN.finditer(content):
|
||||
positions.append((
|
||||
match.start(),
|
||||
match.end(),
|
||||
match.group(0)
|
||||
))
|
||||
positions.sort(key=lambda p: p[0])
|
||||
return positions
|
||||
|
||||
def count_macros(self, content: str) -> dict:
|
||||
@@ -211,4 +234,4 @@ class MacroParser:
|
||||
Returns:
|
||||
True if any macros found
|
||||
"""
|
||||
return bool(self.MACRO_PATTERN.search(content))
|
||||
return bool(self.MACRO_PATTERN.search(content) or self.SHORTHAND_PATTERN.search(content))
|
||||
|
||||
Reference in New Issue
Block a user