ArtifactStore

File: brain/agent/artifact_store.py

---

Purpose

ArtifactStore handles oversized tool outputs — outputs whose serialized JSON exceeds TOOL_MESSAGE_MAX_CHARS = 12_000 characters. Instead of truncating the output or injecting a 12KB blob into the LLM context, the store:

1. Writes the full content to a managed file on disk.
2. Returns an ArtifactRef that gets embedded in a ToolArtifactReference outcome.
3. Injects a compact reference message into the LLM context so the model can request the full content via read_file.

---

Key Properties

Property	Value
Base directory	/tmp/sb_artifacts/<session_id>/
Default TTL	3600 seconds (1 hour)
Thread safety	Internal threading.Lock on the index
Session scope	Each artifact tagged with session_id
Global singleton	get_global_artifact_store() returns a process-wide instance

---

ArtifactRef

@dataclass
class ArtifactRef:
    artifact_id: str      # e.g. "art_3f9b2a1c8e4d7f6a"
    session_id: str
    file_path: str        # Absolute path on disk
    created_at: float     # time.time()
    ttl_s: int
    size_bytes: int
    tool_name: str
    summary: str          # ≤200 char preview

    @property
    def is_expired(self) -> bool: ...

    def to_model_hint(self) -> dict: ...

---

API

Write

ref: ArtifactRef = store.store(
    session_id="sess_abc",
    content=large_json_string,
    tool_name="local_search",
    ttl_s=3600,        # optional; defaults to 1 hour
)

• Writes content to <base_dir>/<session_id>/<artifact_id>.txt.
• Registers the ref in the in-memory index.
• Returns the ref without raising on write failure (callers should have a fallback truncation path if storage fails).
• summary is the first 200 characters of content, newlines replaced with spaces.

Read

content: str | None = store.retrieve(artifact_id)
# Returns None if: not found, expired, or file missing

ref: ArtifactRef | None = store.get_ref(artifact_id)
# Returns the ArtifactRef without reading file content

Cleanup

removed: int = store.cleanup_session(session_id)
# Removes all artifacts for the session from index + disk
# Also sweeps the session directory for orphaned files

removed: int = store.cleanup_expired()
# Removes all expired artifacts from index + disk
# Call periodically (e.g. on harness init, or in a background task)

Stats

stats: dict = store.stats()
# → {"artifact_count": 3, "total_bytes": 45032}

---

BoundedToolExecutor Integration

# In BoundedToolExecutor._run_tool_call():
serialized = _serialize_output(result)

if len(serialized) > _MAX_INLINE_CHARS and state.is_accepting_writes():
    ref = self._artifact_store.store(
        state.session_id, serialized,
        tool_name=call.name,
    )
    return ToolArtifactReference(
        call_id=call.id,
        tool_name=call.name,
        artifact_id=ref.artifact_id,
        summary=ref.summary,
        size_bytes=ref.size_bytes,
    )

return ToolExecutionResult(call_id=call.id, ..., output=result, ...)

Note the state.is_accepting_writes() check: if the turn is already past its deadline, oversized outputs are not stored (the turn is abandoned anyway).

---

Global Singleton

store = get_global_artifact_store()

The singleton is initialised lazily with a double-checked lock. The harness always uses the global store so that model-requested reads within the same session still work even after the producing turn's TurnRuntimeState is discarded.

# In AgentHarness.__init__():
self._artifact_store: ArtifactStore = get_global_artifact_store()

---

Model Interaction Flow

Tool returns 50KB JSON
        ↓
BoundedToolExecutor detects len > 12_000
        ↓
ArtifactStore.store() → ArtifactRef(artifact_id="art_abc123")
        ↓
ToolArtifactReference outcome → outcome_to_model_content() →
{
  "artifact_reference": "art_abc123",
  "summary": "First 200 chars...",
  "hint": "Use the read_file tool to retrieve the full content.",
  "size_bytes": 51200
}
        ↓
LLM sees compact hint, calls read_file(path="/tmp/sb_artifacts/...")
        ↓
Full content returned inline