file-system-primitives

File System Primitives for Honcho

Status: Draft Author: vineeth (design decisions), spec authored with Claude Date: 2026-03-25 Scope: Peer-scoped file storage with versioning, auto-embedding for semantic search, and agent tool integration.

---

Table of Contents

1. Problem Statement
2. Goals and Non-Goals
3. Design
   • Data Model and Schema DDL
   • API Surface
   • Embedding Strategy
   • Versioning Logic
   • Agent Tools
   • Storage Backend
   • Configuration
4. Migration Plan
5. Implementation Phases
6. Files to Modify
7. Risk Assessment
8. Verification Plan
9. Open Questions

---

Problem Statement

Honcho currently stores structured knowledge about peers through observations (the documents table) and peer cards (stored as JSONB in internal_metadata on the collections table). These are optimized for the deriver/dreamer pipeline but offer no general-purpose, user-addressable file storage.

Several emerging requirements demand first-class file primitives:

1. Peer card migration: The Composable Peers RFC proposes migrating peer cards from internal_metadata to a /card.md file, making them addressable and versionable.
2. Dream output materialization: Dream outputs (consolidation summaries, pattern reports) should be written as managed files rather than ephemeral logs, so they can be inspected and tracked over time.
3. Agent workspace: Agents need a persistent scratchpad — files they can create, read, and search across — to support multi-step reasoning and long-running tasks.
4. Developer integration: Application developers need to attach arbitrary text/markdown content to peers (custom instructions, context documents, configuration) that is automatically indexed for semantic retrieval.
5. Versioning for longitudinal analysis: Tracking how a peer card or agent-written document evolves over time is critical for debugging, auditing, and understanding representation drift.

Without file primitives, these use cases are forced into metadata fields, external systems, or ad-hoc message-based workarounds that lose versioning and searchability.

---

Goals and Non-Goals

Goals

• G1: Peer-scoped file CRUD with unique paths per peer (virtual filesystem semantics).
• G2: Automatic versioning on every write, with full version history retrieval.
• G3: Auto-embedding of text/markdown file content on creation and update for semantic search.
• G4: New file namespace type in the vector store ({prefix}.file.{hash}).
• G5: Agent tools (search_files, read_file, grep_files) added to Dialectic and Dreamer agents.
• G6: Managed file support: system and dream processes can own files (tracked via managed_by).
• G7: Efficient full-text search (grep) across a peer’s files via PostgreSQL GIN indexes.
• G8: Text/markdown stored inline in PostgreSQL; large binary references stored externally (S3/configurable backend) for future extensibility.
• G9: Clean Alembic migration with no downtime.

Non-Goals

• NG1: Binary file upload/download in v1. The schema accommodates it (via storage_uri on file_versions), but the API and embedding pipeline only handle text/markdown initially.
• NG2: File sharing across peers. Files are scoped to a single peer; cross-peer access requires explicit API calls from the other peer’s context.
• NG3: Real-time collaboration or file locking. Writes are atomic; last-write-wins.
• NG4: Directory listing with hierarchy. Paths are flat strings (e.g., /notes/2024-03-25.md); no directory objects or recursive listing.
• NG5: Permissions or ACLs on individual files. Access is controlled by workspace/peer-level JWT scoping, same as all other Honcho resources.
• NG6: Sub-peer file aggregation in search (mentioned in Composable Peers RFC). Deferred to the sub-peer implementation phase.

---

Design

Data Model and Schema DDL

Two new tables: peer_files (current state) and file_versions (history). The peer_files table holds the latest content for fast reads; every mutation also appends a row to file_versions.

peer_files Table

CREATE TABLE peer_files (
    -- Identity
    id                TEXT        NOT NULL DEFAULT nanoid() PRIMARY KEY,
    workspace_name    TEXT        NOT NULL REFERENCES workspaces(name),
    peer_name         TEXT        NOT NULL,
    path              TEXT        NOT NULL,
 
    -- Content
    content           TEXT,               -- NULL for binary-only files
    mime_type         TEXT        NOT NULL DEFAULT 'text/markdown',
    size_bytes        INTEGER     NOT NULL DEFAULT 0,
 
    -- Ownership
    managed_by        TEXT,               -- NULL = user-managed
                                          -- 'system' = Honcho system
                                          -- 'dream:{name}' = specific dream process
                                          -- 'deriver' = deriver process
 
    -- Metadata
    metadata          JSONB       NOT NULL DEFAULT '{}'::jsonb,
 
    -- Timestamps
    created_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
 
    -- Current version tracking
    current_version   INTEGER     NOT NULL DEFAULT 1,
 
    -- Constraints
    CONSTRAINT ck_peer_files_id_length CHECK (length(id) = 21),
    CONSTRAINT ck_peer_files_id_format CHECK (id ~ '^[A-Za-z0-9_-]+$'),
    CONSTRAINT ck_peer_files_path_length CHECK (length(path) <= 1024),
    CONSTRAINT ck_peer_files_path_format CHECK (path ~ '^/[^\x00]+$'),
    CONSTRAINT ck_peer_files_content_length CHECK (length(content) <= 1048576),  -- 1MB text limit
    CONSTRAINT ck_peer_files_mime_type_length CHECK (length(mime_type) <= 255),
 
    -- Composite unique: one path per peer per workspace
    CONSTRAINT uq_peer_files_workspace_peer_path
        UNIQUE (workspace_name, peer_name, path),
 
    -- Foreign keys
    CONSTRAINT fk_peer_files_workspace
        FOREIGN KEY (workspace_name) REFERENCES workspaces(name),
    CONSTRAINT fk_peer_files_peer
        FOREIGN KEY (peer_name, workspace_name)
        REFERENCES peers(name, workspace_name)
);
 
-- Indexes
CREATE INDEX ix_peer_files_workspace_peer
    ON peer_files (workspace_name, peer_name);
 
CREATE INDEX ix_peer_files_managed_by
    ON peer_files (managed_by)
    WHERE managed_by IS NOT NULL;
 
CREATE INDEX ix_peer_files_updated_at
    ON peer_files (updated_at);
 
-- Full-text search index for grep operations
CREATE INDEX ix_peer_files_content_gin
    ON peer_files USING gin (to_tsvector('english', content))
    WHERE content IS NOT NULL;

file_versions Table

CREATE TABLE file_versions (
    -- Identity
    id                TEXT        NOT NULL DEFAULT nanoid() PRIMARY KEY,
    file_id           TEXT        NOT NULL REFERENCES peer_files(id) ON DELETE CASCADE,
    version_number    INTEGER     NOT NULL,
 
    -- Content snapshot
    content           TEXT,               -- NULL for binary-only files
    size_bytes        INTEGER     NOT NULL DEFAULT 0,
 
    -- Who made this version
    created_by        TEXT        NOT NULL DEFAULT 'user',
                                          -- 'user', 'system', 'dream:{name}', 'deriver'
 
    -- Metadata snapshot at time of version
    metadata          JSONB       NOT NULL DEFAULT '{}'::jsonb,
 
    -- External storage (future: S3, GCS, etc.)
    storage_uri       TEXT,               -- NULL = content is inline in `content` column
                                          -- 's3://bucket/key' for binary references
 
    -- Timestamps
    created_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
 
    -- Constraints
    CONSTRAINT ck_file_versions_id_length CHECK (length(id) = 21),
    CONSTRAINT ck_file_versions_id_format CHECK (id ~ '^[A-Za-z0-9_-]+$'),
    CONSTRAINT ck_file_versions_content_length CHECK (length(content) <= 1048576),
 
    -- One version number per file
    CONSTRAINT uq_file_versions_file_version
        UNIQUE (file_id, version_number)
);
 
-- Indexes
CREATE INDEX ix_file_versions_file_id
    ON file_versions (file_id);
 
CREATE INDEX ix_file_versions_file_id_version
    ON file_versions (file_id, version_number DESC);
 
CREATE INDEX ix_file_versions_created_at
    ON file_versions (created_at);

file_embeddings Table

File content is chunked and embedded, following the same pattern as message_embeddings.

CREATE TABLE file_embeddings (
    -- Identity
    id                BIGINT      GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    content           TEXT        NOT NULL,
    embedding         vector(1536),
 
    -- Foreign keys / context
    file_id           TEXT        NOT NULL REFERENCES peer_files(id) ON DELETE CASCADE,
    file_version      INTEGER     NOT NULL,    -- version at time of embedding
    chunk_index       INTEGER     NOT NULL,    -- position within the file
    workspace_name    TEXT        NOT NULL REFERENCES workspaces(name),
    peer_name         TEXT        NOT NULL,
 
    -- Timestamps
    created_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
 
    -- Vector sync state (for external vector store reconciliation)
    sync_state        TEXT        NOT NULL DEFAULT 'pending',
    last_sync_at      TIMESTAMPTZ,
    sync_attempts     INTEGER     NOT NULL DEFAULT 0,
 
    -- Constraints
    CONSTRAINT fk_file_embeddings_peer
        FOREIGN KEY (peer_name, workspace_name)
        REFERENCES peers(name, workspace_name),
 
    -- One embedding per chunk per file version
    CONSTRAINT uq_file_embeddings_file_version_chunk
        UNIQUE (file_id, file_version, chunk_index)
);
 
-- HNSW index for similarity search
CREATE INDEX ix_file_embeddings_embedding_hnsw
    ON file_embeddings USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64);
 
-- Reconciliation index
CREATE INDEX ix_file_embeddings_sync_state_last_sync_at
    ON file_embeddings (sync_state, last_sync_at);
 
-- Lookup by file
CREATE INDEX ix_file_embeddings_file_id
    ON file_embeddings (file_id);

SQLAlchemy ORM Models

New models in src/models.py:

@final
class PeerFile(Base):
    __tablename__: str = "peer_files"
 
    id: Mapped[str] = mapped_column(TEXT, default=generate_nanoid, primary_key=True)
    workspace_name: Mapped[str] = mapped_column(
        ForeignKey("workspaces.name"), nullable=False, index=True
    )
    peer_name: Mapped[str] = mapped_column(TEXT, nullable=False)
    path: Mapped[str] = mapped_column(TEXT, nullable=False)
    content: Mapped[str | None] = mapped_column(TEXT, nullable=True)
    mime_type: Mapped[str] = mapped_column(
        TEXT, nullable=False, server_default=text("'text/markdown'")
    )
    size_bytes: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=text("0")
    )
    managed_by: Mapped[str | None] = mapped_column(TEXT, nullable=True)
    metadata: Mapped[dict[str, Any]] = mapped_column(
        "metadata", JSONB, default=dict, server_default=text("'{}'::jsonb")
    )
    created_at: Mapped[datetime.datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), index=True
    )
    updated_at: Mapped[datetime.datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
    )
    current_version: Mapped[int] = mapped_column(
        Integer, nullable=False, default=1, server_default=text("1")
    )
 
    versions = relationship(
        "FileVersion", back_populates="file", cascade="all, delete, delete-orphan"
    )
 
    __table_args__ = (
        UniqueConstraint("workspace_name", "peer_name", "path"),
        CheckConstraint("length(id) = 21", name="id_length"),
        CheckConstraint("id ~ '^[A-Za-z0-9_-]+$'", name="id_format"),
        CheckConstraint("length(path) <= 1024", name="path_length"),
        CheckConstraint(r"path ~ '^/[^\x00]+$'", name="path_format"),
        CheckConstraint("length(content) <= 1048576", name="content_length"),
        ForeignKeyConstraint(
            ["peer_name", "workspace_name"],
            ["peers.name", "peers.workspace_name"],
        ),
        Index(
            "ix_peer_files_content_gin",
            text("to_tsvector('english', content)"),
            postgresql_using="gin",
            postgresql_where=text("content IS NOT NULL"),
        ),
    )
 
 
@final
class FileVersion(Base):
    __tablename__: str = "file_versions"
 
    id: Mapped[str] = mapped_column(TEXT, default=generate_nanoid, primary_key=True)
    file_id: Mapped[str] = mapped_column(
        ForeignKey("peer_files.id", ondelete="CASCADE"), nullable=False, index=True
    )
    version_number: Mapped[int] = mapped_column(Integer, nullable=False)
    content: Mapped[str | None] = mapped_column(TEXT, nullable=True)
    size_bytes: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=text("0")
    )
    created_by: Mapped[str] = mapped_column(
        TEXT, nullable=False, server_default=text("'user'")
    )
    metadata: Mapped[dict[str, Any]] = mapped_column(
        "metadata", JSONB, default=dict, server_default=text("'{}'::jsonb")
    )
    storage_uri: Mapped[str | None] = mapped_column(TEXT, nullable=True)
    created_at: Mapped[datetime.datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), index=True
    )
 
    file = relationship("PeerFile", back_populates="versions")
 
    __table_args__ = (
        UniqueConstraint("file_id", "version_number"),
        CheckConstraint("length(id) = 21", name="id_length"),
        CheckConstraint("id ~ '^[A-Za-z0-9_-]+$'", name="id_format"),
        CheckConstraint("length(content) <= 1048576", name="content_length"),
        Index("ix_file_versions_file_id_version", "file_id", "version_number"),
    )
 
 
@final
class FileEmbedding(Base):
    __tablename__: str = "file_embeddings"
 
    id: Mapped[int] = mapped_column(
        BigInteger, Identity(), primary_key=True, autoincrement=True
    )
    content: Mapped[str] = mapped_column(TEXT, nullable=False)
    embedding: MappedColumn[Any] = mapped_column(Vector(1536), nullable=True)
    file_id: Mapped[str] = mapped_column(
        ForeignKey("peer_files.id", ondelete="CASCADE"), nullable=False, index=True
    )
    file_version: Mapped[int] = mapped_column(Integer, nullable=False)
    chunk_index: Mapped[int] = mapped_column(Integer, nullable=False)
    workspace_name: Mapped[str] = mapped_column(
        ForeignKey("workspaces.name"), nullable=False, index=True
    )
    peer_name: Mapped[str] = mapped_column(TEXT, nullable=False, index=True)
    created_at: Mapped[datetime.datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), index=True
    )
    sync_state: Mapped[VectorSyncState] = mapped_column(
        TEXT, nullable=False, server_default="pending", index=True
    )
    last_sync_at: Mapped[datetime.datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True
    )
    sync_attempts: Mapped[int] = mapped_column(
        Integer, nullable=False, default=0, server_default=text("0")
    )
 
    __table_args__ = (
        UniqueConstraint("file_id", "file_version", "chunk_index"),
        ForeignKeyConstraint(
            ["peer_name", "workspace_name"],
            ["peers.name", "peers.workspace_name"],
        ),
        Index(
            "ix_file_embeddings_embedding_hnsw",
            "embedding",
            postgresql_using="hnsw",
            postgresql_with={"m": 16, "ef_construction": 64},
            postgresql_ops={"embedding": "vector_cosine_ops"},
        ),
        Index(
            "ix_file_embeddings_sync_state_last_sync_at",
            "sync_state",
            "last_sync_at",
        ),
    )

API Surface

All file endpoints live under the peer namespace, consistent with Honcho’s existing routing pattern (/v3/workspaces/{workspace_id}/peers/{peer_id}/...).

Endpoint Definitions

Method	Path	Description
POST	/workspaces/{workspace_id}/peers/{peer_id}/files	Create a new file
POST	/workspaces/{workspace_id}/peers/{peer_id}/files/list	List files (with optional filters)
GET	/workspaces/{workspace_id}/peers/{peer_id}/files/{path:path}	Read current file content
PUT	/workspaces/{workspace_id}/peers/{peer_id}/files/{path:path}	Update file (creates new version)
DELETE	/workspaces/{workspace_id}/peers/{peer_id}/files/{path:path}	Delete file and all versions
GET	/workspaces/{workspace_id}/peers/{peer_id}/files/{path:path}/versions	List version history
GET	/workspaces/{workspace_id}/peers/{peer_id}/files/{path:path}/versions/{version}	Read specific version
POST	/workspaces/{workspace_id}/peers/{peer_id}/files/search	Semantic search across files
POST	/workspaces/{workspace_id}/peers/{peer_id}/files/grep	Full-text search across files

Pydantic Schemas

New schemas in src/schemas/api.py:

# ---------------------------------------------------------------------------
# File schemas
# ---------------------------------------------------------------------------
 
class FileCreate(BaseModel):
    """Create a new peer file."""
    path: Annotated[str, Field(
        min_length=2,
        max_length=1024,
        pattern=r"^/[^\x00]+$",
        description="File path, must start with /. Example: /card.md",
    )]
    content: Annotated[str, Field(
        min_length=0,
        max_length=1_048_576,
        description="Text content of the file",
    )]
    mime_type: str = Field(default="text/markdown", max_length=255)
    managed_by: str | None = Field(
        default=None,
        description="Ownership marker: null=user, 'system', 'dream:{name}'",
    )
    metadata: _SanitizedMetadata = {}
 
    @field_validator("content", mode="after")
    @classmethod
    def sanitize_content(cls, v: str) -> str:
        return v.replace("\x00", "")
 
 
class FileUpdate(BaseModel):
    """Update an existing peer file (creates a new version)."""
    content: Annotated[str, Field(
        min_length=0,
        max_length=1_048_576,
        description="New text content for the file",
    )]
    metadata: _SanitizedMetadata | None = None
    created_by: str = Field(
        default="user",
        description="Who is making this edit: 'user', 'system', 'dream:{name}'",
    )
 
    @field_validator("content", mode="after")
    @classmethod
    def sanitize_content(cls, v: str) -> str:
        return v.replace("\x00", "")
 
 
class FileGet(BaseModel):
    """Filtering options for listing files."""
    filters: dict[str, Any] | None = None
 
 
class File(BaseModel):
    """File response model."""
    id: str
    path: str
    content: str | None = None
    mime_type: str
    size_bytes: int
    managed_by: str | None = None
    metadata: dict[str, Any] = Field(default_factory=dict)
    current_version: int
    created_at: datetime.datetime
    updated_at: datetime.datetime
    workspace_name: str = Field(serialization_alias="workspace_id")
    peer_name: str = Field(serialization_alias="peer_id")
 
    model_config = ConfigDict(from_attributes=True, populate_by_name=True)
 
 
class FileVersion(BaseModel):
    """File version response model."""
    id: str
    file_id: str
    version_number: int
    content: str | None = None
    size_bytes: int
    created_by: str
    metadata: dict[str, Any] = Field(default_factory=dict)
    storage_uri: str | None = None
    created_at: datetime.datetime
 
    model_config = ConfigDict(from_attributes=True, populate_by_name=True)
 
 
class FileSearchQuery(BaseModel):
    """Semantic search across peer files."""
    query: Annotated[str, Field(
        min_length=1,
        max_length=10000,
        description="Semantic search query",
    )]
    top_k: int = Field(default=10, ge=1, le=50)
    distance: float | None = Field(
        default=None,
        ge=0.0,
        le=1.0,
        description="Maximum cosine distance threshold",
    )
 
    @field_validator("query", mode="after")
    @classmethod
    def sanitize_query(cls, v: str) -> str:
        return v.replace("\x00", "")
 
 
class FileSearchResult(BaseModel):
    """A single result from file semantic search."""
    file_id: str
    path: str
    score: float
    chunk_content: str
    chunk_index: int
    file_version: int
 
    model_config = ConfigDict(from_attributes=True, populate_by_name=True)
 
 
class FileGrepQuery(BaseModel):
    """Full-text/pattern search across peer files."""
    pattern: Annotated[str, Field(
        min_length=1,
        max_length=1000,
        description="Text pattern to search for (case-insensitive)",
    )]
    limit: int = Field(default=20, ge=1, le=100)
 
    @field_validator("pattern", mode="after")
    @classmethod
    def sanitize_pattern(cls, v: str) -> str:
        return v.replace("\x00", "")
 
 
class FileGrepResult(BaseModel):
    """A single result from file grep."""
    file_id: str
    path: str
    snippet: str  # Content around the match
    match_count: int  # Number of matches in this file
 
    model_config = ConfigDict(from_attributes=True, populate_by_name=True)

Router Implementation

New router file: src/routers/files.py

Key behaviors:

• Create (POST /files): Insert into peer_files, insert version 1 into file_versions, enqueue embedding job. Returns 201 with File response.
• Read (GET /files/{path}): Look up by (workspace_name, peer_name, path). Return content from peer_files.content (latest version). 404 if not found.
• Update (PUT /files/{path}): Increment current_version, update content/updated_at on peer_files, insert new file_versions row, enqueue re-embedding job. Returns 200 with File response.
• Delete (DELETE /files/{path}): Delete peer_files row (CASCADE deletes versions and embeddings). Also enqueue vector store cleanup for external backends. Returns 204.
• List (POST /files/list): Paginated via fastapi_pagination, filtered by managed_by, metadata, mime_type. Follows the POST /list pattern used by peers, sessions, and messages.
• Versions (GET /files/{path}/versions): Return paginated FileVersion list ordered by version_number DESC.
• Specific version (GET /files/{path}/versions/{version}): Return specific FileVersion content. 404 if version doesn’t exist.
• Search (POST /files/search): Embed query, search file_embeddings (pgvector) or external vector store under file namespace.
• Grep (POST /files/grep): Use to_tsvector/plainto_tsquery on peer_files.content for full-text search. Also support ILIKE fallback for non-English patterns.

Authentication

All file endpoints use the existing require_auth(workspace_name="workspace_id", peer_name="peer_id") dependency, consistent with other peer-scoped resources.

Embedding Strategy

Namespace Design

The vector store namespace for files follows the existing {prefix}.{type}.{hash} pattern:

{VECTOR_STORE.NAMESPACE}.file.{hash(workspace_name, peer_name)}

This is peer-scoped (not observer/observed-scoped like documents), because files belong to a single peer and are not directional relationships.

Implementation in src/vector_store/__init__.py:

def get_vector_namespace(
    self,
    namespace_type: Literal["document", "message", "file"],
    workspace_name: str,
    observer: str | None = None,
    observed: str | None = None,
    peer_name: str | None = None,  # New: for file namespaces
) -> str:
    if namespace_type == "file":
        if peer_name is None:
            raise ValueError("peer_name is required for file namespaces")
        hash_suffix = _hash_namespace_components(workspace_name, peer_name)
        return f"{self.namespace_prefix}.file.{hash_suffix}"
    # ... existing document/message logic

Chunking Strategy

File content is chunked using the same infrastructure as embedding_client.py:

1. Tokenize the file content using tiktoken (o200k_base encoding).
2. If tokens ⇐ MAX_EMBEDDING_TOKENS (currently 8192 for OpenAI, 2048 for Gemini): embed as a single chunk.
3. If tokens > limit: split using _chunk_text_with_tokens() with 20% overlap, identical to the existing document/message chunking.
4. Batch embed all chunks using embedding_client.batch_embed().

Each chunk produces one file_embeddings row with:

• file_id: links to the parent file
• file_version: version number at time of embedding
• chunk_index: ordinal position (0-indexed)
• content: the chunk text (for display in search results)
• embedding: the vector

Re-embedding on Update

When a file is updated:

1. Mark all existing file_embeddings rows for this file_id with sync_state = 'pending' (they will be cleaned up by the reconciler when using external stores) OR delete them directly for pgvector mode.
2. Create new file_embeddings rows for the new content.
3. For external vector stores (Turbopuffer/LanceDB), the reconciler handles upserts on its next cycle, identical to how sync_vectors.py handles documents and message embeddings.

Embedding Flow

The embedding happens synchronously during the API request for small files (< 50 chunks) and is enqueued as a background task for large files (>= 50 chunks). This matches the pattern used for conclusion creation (crud.create_observations embeds inline).

API Request (create/update file)
  |
  +--> Write peer_files + file_versions rows
  |
  +--> If content is text/markdown:
  |      |
  |      +--> Tokenize and chunk content
  |      +--> Embed all chunks (batch API call)
  |      +--> Write file_embeddings rows
  |      +--> If external vector store:
  |             +--> Reconciler picks up pending rows
  |
  +--> Return response

Versioning Logic

Version Numbering

Versions are monotonically increasing integers per file, starting at 1. The current_version field on peer_files is the authoritative latest version number.

Write Path

Every file mutation follows this sequence (within a single database transaction):

async def update_file(
    db: AsyncSession,
    workspace_name: str,
    peer_name: str,
    path: str,
    update: FileUpdate,
) -> PeerFile:
    # 1. Lock the file row (SELECT ... FOR UPDATE)
    file = await _get_file_for_update(db, workspace_name, peer_name, path)
 
    # 2. Increment version
    new_version = file.current_version + 1
 
    # 3. Create version snapshot
    version = FileVersion(
        file_id=file.id,
        version_number=new_version,
        content=update.content,
        size_bytes=len(update.content.encode("utf-8")),
        created_by=update.created_by,
        metadata=update.metadata or file.metadata,
    )
    db.add(version)
 
    # 4. Update current state
    file.content = update.content
    file.size_bytes = len(update.content.encode("utf-8"))
    file.current_version = new_version
    file.updated_at = func.now()
    if update.metadata is not None:
        file.metadata = update.metadata
 
    # 5. Re-embed (delete old embeddings, create new ones)
    await _reembed_file(db, file, new_version)
 
    await db.flush()
    return file

Version Retention

All versions are retained indefinitely. Future work may add configurable retention policies (e.g., keep last N versions, or versions newer than X days). The file_versions table is append-only; no versions are ever mutated.

Concurrency

The file row is locked with SELECT ... FOR UPDATE during writes to prevent lost updates. This is a short-lived lock (held only for the duration of the transaction). Concurrent reads are not blocked.

Agent Tools

Three new tools are added to src/utils/agent_tools.py in the TOOLS dict:

search_files

"search_files": {
    "name": "search_files",
    "description": (
        "Search across the peer's files using semantic similarity. "
        "Returns matching file chunks ranked by relevance. "
        "Use this to find relevant documents, notes, or configuration "
        "that the peer has stored."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "Semantic search query",
            },
            "top_k": {
                "type": "integer",
                "description": "Number of results to return (default: 5, max: 20)",
                "default": 5,
            },
        },
        "required": ["query"],
    },
}

read_file

"read_file": {
    "name": "read_file",
    "description": (
        "Read the current content of a specific file by path. "
        "Use this when you know the file path (e.g., from search results) "
        "and need to read its full content."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "path": {
                "type": "string",
                "description": "File path (e.g., '/card.md', '/notes/summary.md')",
            },
        },
        "required": ["path"],
    },
}

grep_files

"grep_files": {
    "name": "grep_files",
    "description": (
        "Search for files containing specific text (case-insensitive). "
        "Unlike semantic search, this finds EXACT text matches. "
        "Use for finding specific names, dates, phrases, or keywords in files."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "pattern": {
                "type": "string",
                "description": "Text to search for (case-insensitive)",
            },
            "limit": {
                "type": "integer",
                "description": "Maximum files to return (default: 10, max: 20)",
                "default": 10,
            },
        },
        "required": ["pattern"],
    },
}

Tool Assignment

Agent	New Tools Added
Dialectic (all levels)	search_files, read_file
Dialectic (minimal level)	None (keep minimal)
Dreamer	search_files, read_file, grep_files
Deriver	None (deriver processes messages, not files)

The tool lists are updated in agent_tools.py:

DIALECTIC_TOOLS: list[dict[str, Any]] = [
    # ... existing tools ...
    TOOLS["search_files"],
    TOOLS["read_file"],
]
 
DREAMER_TOOLS: list[dict[str, Any]] = [
    # ... existing tools ...
    TOOLS["search_files"],
    TOOLS["read_file"],
    TOOLS["grep_files"],
]

Tool Handler Implementations

New handlers in agent_tools.py, following the existing _handle_* pattern:

async def _handle_search_files(ctx: ToolContext, tool_input: dict[str, Any]) -> str:
    """Handle search_files tool -- semantic search across peer's files."""
    query = tool_input.get("query", "")
    top_k = _safe_int(tool_input.get("top_k", 5), 5)
    top_k = min(top_k, 20)
 
    if not query:
        return "ERROR: query is required"
 
    # Embed query
    query_embedding = await embedding_client.embed(query)
 
    # Search file embeddings
    results = await crud.search_file_embeddings(
        workspace_name=ctx.workspace_name,
        peer_name=ctx.observed,  # Files belong to the observed peer
        query_embedding=query_embedding,
        top_k=top_k,
    )
 
    if not results:
        return "No matching files found."
 
    output_parts = []
    for r in results:
        output_parts.append(
            f"[file:{r.path}] (score: {r.score:.3f}, chunk {r.chunk_index})\n"
            f"{_truncate_tool_output(r.chunk_content, max_chars=500)}"
        )
    return "\n\n---\n\n".join(output_parts)
 
 
async def _handle_read_file(ctx: ToolContext, tool_input: dict[str, Any]) -> str:
    """Handle read_file tool -- read current content of a file by path."""
    path = tool_input.get("path", "")
    if not path:
        return "ERROR: path is required"
 
    # Ensure path starts with /
    if not path.startswith("/"):
        path = "/" + path
 
    async with tracked_db("agent.read_file") as file_db:
        file = await crud.get_file_by_path(
            file_db,
            workspace_name=ctx.workspace_name,
            peer_name=ctx.observed,
            path=path,
        )
 
    if file is None:
        return f"File not found: {path}"
 
    content = file.content or "(empty file)"
    return _truncate_tool_output(
        f"[file:{path}] (version {file.current_version}, "
        f"updated {file.updated_at.isoformat()})\n\n{content}"
    )
 
 
async def _handle_grep_files(ctx: ToolContext, tool_input: dict[str, Any]) -> str:
    """Handle grep_files tool -- full-text search across peer's files."""
    pattern = tool_input.get("pattern", "")
    limit = _safe_int(tool_input.get("limit", 10), 10)
    limit = min(limit, 20)
 
    if not pattern:
        return "ERROR: pattern is required"
 
    async with tracked_db("agent.grep_files") as grep_db:
        results = await crud.grep_files(
            grep_db,
            workspace_name=ctx.workspace_name,
            peer_name=ctx.observed,
            pattern=pattern,
            limit=limit,
        )
 
    if not results:
        return f"No files contain '{pattern}'."
 
    output_parts = []
    for r in results:
        snippet = _extract_pattern_snippet(r.content or "", pattern, max_chars=500)
        output_parts.append(
            f"[file:{r.path}] ({r.match_count} matches)\n{snippet}"
        )
    return "\n\n---\n\n".join(output_parts)

The tool dispatch map in create_tool_executor is extended:

tool_handlers: dict[str, Callable] = {
    # ... existing handlers ...
    "search_files": _handle_search_files,
    "read_file": _handle_read_file,
    "grep_files": _handle_grep_files,
}

Storage Backend

Phase 1: PostgreSQL-only

All text/markdown content is stored inline in the content TEXT columns of peer_files and file_versions. The 1MB limit per file (enforced by CHECK constraint) is generous for text documents while preventing abuse.

Phase 2 (Future): External Storage

The storage_uri column on file_versions enables future support for external storage:

• NULL = content is inline in the content column (default).
• s3://bucket/path = content is stored in S3; the content column is NULL.
• gs://bucket/path = Google Cloud Storage.
• file:///path = Local filesystem (dev/self-hosting).

A StorageBackend protocol will be added when this is needed:

class StorageBackend(Protocol):
    async def put(self, key: str, content: bytes) -> str: ...
    async def get(self, uri: str) -> bytes: ...
    async def delete(self, uri: str) -> None: ...

This is a non-goal for the initial implementation but the schema is designed to accommodate it without migration.

Configuration

New settings section in src/config.py:

class FileSettings(HonchoSettings):
    model_config = SettingsConfigDict(env_prefix="FILE_", extra="ignore")
 
    # Master toggle
    ENABLED: bool = True
 
    # Maximum file size in bytes (1MB default)
    MAX_FILE_SIZE: Annotated[int, Field(default=1_048_576, gt=0, le=10_485_760)] = 1_048_576
 
    # Maximum files per peer
    MAX_FILES_PER_PEER: Annotated[int, Field(default=1000, gt=0, le=10_000)] = 1000
 
    # Maximum versions retained per file (0 = unlimited)
    MAX_VERSIONS_PER_FILE: Annotated[int, Field(default=0, ge=0, le=10_000)] = 0
 
    # Whether to auto-embed file content
    AUTO_EMBED: bool = True
 
    # Chunk threshold for background embedding (above this, enqueue instead of inline)
    BACKGROUND_EMBED_CHUNK_THRESHOLD: Annotated[int, Field(default=50, gt=0)] = 50

Added to AppSettings:

class AppSettings(HonchoSettings):
    # ... existing fields ...
    FILE: FileSettings = Field(default_factory=FileSettings)

TOML section mapping addition:

SECTION_MAP: ClassVar[dict[str, str]] = {
    # ... existing ...
    "FILE": "file",
}

Example config.toml section:

[file]
enabled = true
max_file_size = 1048576
max_files_per_peer = 1000
auto_embed = true

---

Migration Plan

Alembic Migration

A single Alembic migration creates all three tables. Because these are net-new tables with no data migration, the migration is safe and fast.

Migration file: migrations/versions/xxxx_add_peer_files_tables.py

"""Add peer_files, file_versions, and file_embeddings tables.
 
Revision ID: <auto-generated>
Revises: <current head>
Create Date: <auto-generated>
"""
 
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import JSONB, TEXT
from pgvector.sqlalchemy import Vector
 
 
def upgrade() -> None:
    # peer_files
    op.create_table(
        "peer_files",
        sa.Column("id", TEXT, primary_key=True),
        sa.Column("workspace_name", TEXT, sa.ForeignKey("workspaces.name"), nullable=False),
        sa.Column("peer_name", TEXT, nullable=False),
        sa.Column("path", TEXT, nullable=False),
        sa.Column("content", TEXT, nullable=True),
        sa.Column("mime_type", TEXT, nullable=False, server_default="text/markdown"),
        sa.Column("size_bytes", sa.Integer, nullable=False, server_default="0"),
        sa.Column("managed_by", TEXT, nullable=True),
        sa.Column("metadata", JSONB, nullable=False, server_default="{}"),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
        sa.Column("current_version", sa.Integer, nullable=False, server_default="1"),
        sa.UniqueConstraint("workspace_name", "peer_name", "path"),
        sa.ForeignKeyConstraint(["peer_name", "workspace_name"], ["peers.name", "peers.workspace_name"]),
        sa.CheckConstraint("length(id) = 21"),
        sa.CheckConstraint("id ~ '^[A-Za-z0-9_-]+$'"),
        sa.CheckConstraint("length(path) <= 1024"),
        sa.CheckConstraint(r"path ~ '^/[^\x00]+$'"),
        sa.CheckConstraint("length(content) <= 1048576"),
    )
    op.create_index("ix_peer_files_workspace_peer", "peer_files", ["workspace_name", "peer_name"])
    op.create_index("ix_peer_files_managed_by", "peer_files", ["managed_by"], postgresql_where=sa.text("managed_by IS NOT NULL"))
    op.create_index("ix_peer_files_updated_at", "peer_files", ["updated_at"])
    op.create_index(
        "ix_peer_files_content_gin",
        "peer_files",
        [sa.text("to_tsvector('english', content)")],
        postgresql_using="gin",
        postgresql_where=sa.text("content IS NOT NULL"),
    )
 
    # file_versions
    op.create_table(
        "file_versions",
        sa.Column("id", TEXT, primary_key=True),
        sa.Column("file_id", TEXT, sa.ForeignKey("peer_files.id", ondelete="CASCADE"), nullable=False),
        sa.Column("version_number", sa.Integer, nullable=False),
        sa.Column("content", TEXT, nullable=True),
        sa.Column("size_bytes", sa.Integer, nullable=False, server_default="0"),
        sa.Column("created_by", TEXT, nullable=False, server_default="user"),
        sa.Column("metadata", JSONB, nullable=False, server_default="{}"),
        sa.Column("storage_uri", TEXT, nullable=True),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
        sa.UniqueConstraint("file_id", "version_number"),
        sa.CheckConstraint("length(id) = 21"),
        sa.CheckConstraint("id ~ '^[A-Za-z0-9_-]+$'"),
        sa.CheckConstraint("length(content) <= 1048576"),
    )
    op.create_index("ix_file_versions_file_id", "file_versions", ["file_id"])
    op.create_index("ix_file_versions_file_id_version", "file_versions", ["file_id", "version_number"])
    op.create_index("ix_file_versions_created_at", "file_versions", ["created_at"])
 
    # file_embeddings
    op.create_table(
        "file_embeddings",
        sa.Column("id", sa.BigInteger, sa.Identity(), primary_key=True),
        sa.Column("content", TEXT, nullable=False),
        sa.Column("embedding", Vector(1536), nullable=True),
        sa.Column("file_id", TEXT, sa.ForeignKey("peer_files.id", ondelete="CASCADE"), nullable=False),
        sa.Column("file_version", sa.Integer, nullable=False),
        sa.Column("chunk_index", sa.Integer, nullable=False),
        sa.Column("workspace_name", TEXT, sa.ForeignKey("workspaces.name"), nullable=False),
        sa.Column("peer_name", TEXT, nullable=False),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
        sa.Column("sync_state", TEXT, nullable=False, server_default="pending"),
        sa.Column("last_sync_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("sync_attempts", sa.Integer, nullable=False, server_default="0"),
        sa.UniqueConstraint("file_id", "file_version", "chunk_index"),
        sa.ForeignKeyConstraint(["peer_name", "workspace_name"], ["peers.name", "peers.workspace_name"]),
    )
    op.create_index(
        "ix_file_embeddings_embedding_hnsw",
        "file_embeddings",
        ["embedding"],
        postgresql_using="hnsw",
        postgresql_with={"m": 16, "ef_construction": 64},
        postgresql_ops={"embedding": "vector_cosine_ops"},
    )
    op.create_index("ix_file_embeddings_sync_state_last_sync_at", "file_embeddings", ["sync_state", "last_sync_at"])
    op.create_index("ix_file_embeddings_file_id", "file_embeddings", ["file_id"])
 
 
def downgrade() -> None:
    op.drop_table("file_embeddings")
    op.drop_table("file_versions")
    op.drop_table("peer_files")

Zero-Downtime Deployment

1. Deploy migration first (creates empty tables, no lock contention).
2. Deploy application code with file endpoints.
3. No existing data is modified. Feature is additive.

---

Implementation Phases

Phase 1: Core File CRUD and Versioning (Week 1-2)

Deliverables:

• Alembic migration for peer_files, file_versions, file_embeddings
• SQLAlchemy ORM models (PeerFile, FileVersion, FileEmbedding)
• Pydantic schemas (create, update, response, search, grep)
• CRUD module (src/crud/file.py): create, read, update, delete, list, get versions, get specific version
• Router (src/routers/files.py): all 9 endpoints
• Configuration (FileSettings in config.py)
• Register router in src/main.py

Files created:

• src/crud/file.py
• src/routers/files.py
• migrations/versions/xxxx_add_peer_files_tables.py

Files modified:

• src/models.py (add PeerFile, FileVersion, FileEmbedding)
• src/schemas/api.py (add file schemas)
• src/schemas/__init__.py (re-export file schemas)
• src/config.py (add FileSettings)
• src/main.py (register files router)

Phase 2: Embedding Pipeline (Week 2-3)

Deliverables:

• Auto-embed on file create/update
• File namespace type in vector store ({prefix}.file.{hash})
• Reconciler support for file_embeddings table
• Semantic search endpoint implementation
• Grep (full-text search) endpoint implementation

Files modified:

• src/vector_store/__init__.py (add "file" namespace type to get_vector_namespace)
• src/reconciler/sync_vectors.py (add _get_file_embeddings_needing_sync, _sync_file_embeddings_batch)
• src/crud/file.py (add embedding creation and search functions)

Phase 3: Agent Tool Integration (Week 3-4)

Deliverables:

• Three new tool definitions (search_files, read_file, grep_files)
• Tool handler implementations
• Add tools to DIALECTIC_TOOLS and DREAMER_TOOLS lists
• Register in tool dispatch map

Files modified:

• src/utils/agent_tools.py (tool definitions, handlers, assignment to agent tool lists, dispatch map)

Phase 4: Testing and SDK Support (Week 4-5)

Deliverables:

• Unit tests for CRUD operations
• Integration tests for API endpoints
• Integration tests for embedding pipeline
• Integration tests for agent tools
• Python SDK update (new methods on Peer resource)
• TypeScript SDK update (new methods on Peer resource)

Files created:

• tests/test_files.py
• tests/test_file_embeddings.py

Files modified:

• sdks/python/ (add file methods to peer resource)
• sdks/typescript/ (add file methods to peer resource)

Phase 5 (Future): Composable Peers Integration

Deliverables (deferred):

• Migrate peer cards from internal_metadata to /card.md managed files
• Dream output materialization as managed files (managed_by='dream:{name}')
• Sub-peer file aggregation in search
• External storage backend (S3)

---

Files to Modify

New Files

File	Purpose
src/crud/file.py	All file CRUD operations and search functions
src/routers/files.py	FastAPI router with all file endpoints
migrations/versions/xxxx_add_peer_files_tables.py	Alembic migration
tests/test_files.py	Unit and integration tests
tests/test_file_embeddings.py	Embedding pipeline tests

Modified Files

File	Changes
src/models.py	Add PeerFile, FileVersion, FileEmbedding ORM models
src/schemas/api.py	Add FileCreate, FileUpdate, FileGet, File, FileVersion, FileSearchQuery, FileSearchResult, FileGrepQuery, FileGrepResult
src/schemas/__init__.py	Re-export new file schemas
src/config.py	Add FileSettings class, add to AppSettings, add to SECTION_MAP
src/main.py	Register files.router with /v3 prefix
src/vector_store/__init__.py	Add "file" to get_vector_namespace() namespace_type literal, implement file namespace logic
src/reconciler/sync_vectors.py	Add file embedding sync (query file_embeddings with sync_state='pending', upsert to external store)
src/utils/agent_tools.py	Add 3 tool definitions, 3 handler functions, update DIALECTIC_TOOLS, DREAMER_TOOLS, update tool dispatch map in create_tool_executor
src/crud/__init__.py	Re-export file CRUD functions
src/utils/types.py	No changes needed (existing VectorSyncState is reused)

---

Risk Assessment

High Risk

Risk	Impact	Mitigation
Embedding cost explosion: Large files with frequent updates generate many embedding API calls	High API cost, potential rate limiting	Enforce MAX_FILE_SIZE (1MB default), debounce re-embedding (only embed on update, not intermediate saves), use BACKGROUND_EMBED_CHUNK_THRESHOLD to async large files
Database bloat from version history: Unlimited version retention fills disk	Storage pressure over time	MAX_VERSIONS_PER_FILE config (default 0 = unlimited), future retention policy can prune old versions

Medium Risk

Risk	Impact	Mitigation
GIN index performance on large content: Full-text search on multi-MB files with many peers	Slow grep queries	Content size limit (1MB), partial index (WHERE content IS NOT NULL), consider pg_trgm index as supplement for non-English text
Vector store namespace proliferation: One namespace per peer for files	Namespace management complexity	Follows existing pattern (documents already have per-observer/observed namespaces); reconciler handles cleanup
HNSW index build time: Adding a new HNSW index on a production table	Migration takes minutes	Table is empty at migration time (net-new table), so index build is instant

Low Risk

Risk	Impact	Mitigation
Path collision: Two concurrent requests creating the same file path	Duplicate error	UNIQUE(workspace_name, peer_name, path) constraint, upsert semantics if needed
Agent tool output size: read_file returns large content	Token explosion in agent context	_truncate_tool_output() with existing MAX_TOOL_OUTPUT_CHARS (10,000 chars default)

---

Verification Plan

Unit Tests

1. CRUD operations:

   • Create file, verify peer_files and file_versions rows created
   • Update file, verify new version created, current_version incremented
   • Delete file, verify CASCADE deletes versions and embeddings
   • List files with filters (managed_by, metadata, mime_type)
   • Get specific version by number
   • Path uniqueness constraint enforcement
   • MAX_FILES_PER_PEER enforcement

2. Embedding pipeline:

   • Create file with short content, verify single file_embeddings row
   • Create file with long content, verify multiple chunks with correct indices
   • Update file, verify old embeddings replaced with new ones
   • Verify namespace format: {prefix}.file.{hash(workspace, peer)}

3. Search operations:

   • Semantic search returns relevant chunks ranked by score
   • Grep search returns files containing pattern with snippets
   • Empty results return appropriate empty responses

Integration Tests

1. API endpoints (all 9 endpoints):

   • Round-trip: create → read → update → read (verify new content) → list versions → read version 1 → delete → 404
   • Authentication: verify JWT scoping (workspace + peer level)
   • Validation: path format, content size limits, mime type

2. Agent tools (within agent test harness):

   • search_files returns relevant results from embedded files
   • read_file returns file content, handles missing files gracefully
   • grep_files finds exact text matches, handles no results

3. Reconciler (for external vector store configurations):

   • File embeddings with sync_state='pending' are picked up and synced
   • After sync, sync_state transitions to 'synced'

Manual Verification

1. Create a peer, write 5 markdown files via API
2. Query semantic search — verify results are relevant
3. Grep for a specific phrase — verify exact matches returned
4. Update a file 3 times — verify version history is complete
5. Use dialectic /chat endpoint — verify search_files tool is invoked when relevant

---

Open Questions

1. Should file creation be idempotent (upsert) like peer creation? The current design requires the path to not exist on POST /files (returns 409 on conflict). An alternative is upsert semantics: if the path exists, create a new version instead. This would simplify client logic but blur the distinction between create and update. Recommendation: Keep create/update separate for clarity. Consider adding a PUT /files/{path} with create-if-not-exists semantics (like get_or_create_peer).

2. Should the Deriver also write files? The current design does not give the Deriver file tools, since it processes incoming messages. However, the Deriver could write a per-session observation summary as a managed file. Recommendation: Defer to Phase 5; the dream system is the more natural file producer.

3. Embedding dimension compatibility. The file_embeddings table hardcodes Vector(1536), matching the current text-embedding-3-small model. If the embedding model changes (e.g., to a 3072-dim model), a migration would be needed. Recommendation: Use VECTOR_STORE.DIMENSIONS setting consistently; add a migration when dimensions change (same pattern as existing documents and message_embeddings tables).

4. Should grep support regex or just plain text? The current design uses PostgreSQL full-text search (to_tsvector/plainto_tsquery) with ILIKE fallback. Regex support (~* operator) would be more powerful but risks ReDoS on untrusted input. Recommendation: Start with ILIKE for the agent tool (safe, simple) and plainto_tsquery for the API endpoint. Add regex as a future option with a query timeout guard.

5. Rate limiting for file operations. Frequent file updates trigger re-embedding, which incurs API costs. Should there be rate limiting on file writes? Recommendation: Not in v1. The MAX_FILES_PER_PEER limit and MAX_FILE_SIZE provide coarse controls. If abuse is observed, add per-peer write rate limiting.

6. Peer card migration timing. The Composable Peers RFC proposes migrating peer cards to /card.md. Should this happen as part of the file system implementation, or as a separate follow-up? Recommendation: Separate follow-up (Phase 5). The file system should be stable before migrating critical peer card data onto it.