Class ArtifactStore

uncommon_route/artifacts.py:37–164

Filesystem-backed artifact store.

class ArtifactStore:
    """Filesystem-backed artifact store."""

    def __init__(
        self,
        root: Path | None = None,
        *,
        now_fn: Any = None,
    ) -> None:
        self._root = root or DEFAULT_ARTIFACTS_DIR
        self._now = now_fn or time.time
        self._root.mkdir(parents=True, exist_ok=True, mode=0o700)

    @property
    def root(self) -> Path:
        return self._root

    def store_text(
        self,
        content: str,
        *,
        kind: str = "raw",
        role: str,
        session_id: str = "",
        tool_name: str = "",
        tool_call_id: str = "",
        content_type: str = "text/plain",
        summary: str = "",
    ) -> ArtifactRecord:
        created_at = self._now()
        sha256 = hashlib.sha256(content.encode("utf-8")).hexdigest()
        existing = self._find_existing(
            sha256=sha256,
            kind=kind,
            role=role,
            session_id=session_id,
            tool_name=tool_name,
            tool_call_id=tool_call_id,
        )
        if existing is not None:
            if summary and not existing.summary:
                updated = ArtifactRecord(**{**asdict(existing), "summary": summary})
                self._meta_path(existing.id).write_text(json.dumps(asdict(updated), indent=2))
                return updated
            return existing
        artifact_id = uuid.uuid4().hex[:12]
        preview = content[:240].replace("\r\n", "\n").replace("\r", "\n")
        record = ArtifactRecord(
            id=artifact_id,
            created_at=created_at,
            kind=kind,
            role=role,
            session_id=session_id,
            tool_name=tool_name,
            tool_call_id=tool_call_id,
            content_type=content_type,
            char_count=len(content),
            token_estimate=estimate_tokens(content),