multi-modal

Status: Draft Owner: vineeth Last Updated: 2026-03-25 Target Release: v3.1 Codename: n/a (core Honcho feature)

1. Problem Statement

Honcho messages are currently text-only. The Message model stores a content column of type TEXT (max 65,535 characters), with token_count computed exclusively via tiktoken text encoding. There is no schema support for binary content, image references, or multi-modal payloads.

Modern AI applications routinely work with images, screenshots, PDFs, and other media. Users who want Honcho to form memories from visual content today have no path forward — they must pre-process images into text descriptions externally before ingesting into Honcho, losing fidelity and creating friction.

Current Limitations

No attachment model: The messages table has no column for referencing external files. The existing /upload endpoint extracts text from PDFs via pdfplumber and stores it inline in content, discarding the original binary entirely.
No storage backend: Honcho has no facility for storing or referencing binary objects. There is no S3, GCS, or filesystem integration.
Deriver is text-only: The deriver prompt (src/deriver/prompts.py) passes format_new_turn_with_timestamp(msg.content, ...) — a plain text string. There is no vision model call and no mechanism to describe images for downstream processing.
Token counting ignores media: token_count is computed in MessageCreate.validate_and_set_token_count() using tiktoken.get_encoding("o200k_base").encode(self.content). Images have no token budget impact.
SDKs have no upload-to-reference path: The Python SDK’s session.upload_file() and TypeScript’s session.uploadFile() send raw file bytes to the /upload endpoint, which extracts text. There is no concept of storing the file and attaching a reference to a message.

Why This Matters

Memory from visual context: Users sharing screenshots, photos, diagrams, or documents should have those reflected in Honcho’s representation of them.
Multi-modal LLMs are standard: Claude, GPT-4, Gemini all accept images natively. Honcho should leverage this for observation extraction.
PDF fidelity: The current text-only PDF extraction loses layout, tables, and images within PDFs. Storing the original enables richer future processing.

2. Goals / Non-Goals

Goals

Attachments on messages: Extend the message schema with a JSONB attachments array that references externally stored files, including type, URL, MIME type, byte size, and an LLM-generated text description.
Pluggable storage backend: Implement a FileStorageBackend protocol with S3, GCS, and local filesystem implementations. Backend selection is configuration-driven and vendor-agnostic.
Vision-augmented deriver: When a message has image attachments, invoke a vision-capable model to produce a text description of the image, then feed content + descriptions to the existing deriver pipeline for observation extraction.
PDF text extraction into attachments: Refactor the existing PDF upload flow so that extracted text is stored as the attachment’s description field, the original PDF is stored in the storage backend, and the reference is attached to the message.
SDK upload helpers: Provide honcho.upload(file) methods in both Python and TypeScript SDKs that handle file upload to the storage backend and return an Attachment object ready for inclusion in message creation.
Security hardening: Enforce MIME type allowlists, file size limits, magic byte verification, and block executable file types.
CloudEvents for billing: Emit telemetry events for storage operations (upload, delete) to enable billing metering in Groudon/Xatu.

Non-Goals

Inline binary storage in PostgreSQL: We will not store file bytes in the database. Files go to external storage; Honcho stores references.
Audio/video support: Out of scope for v1. The architecture should not preclude future addition, but we are not building it now.
OCR for images: Vision model description is sufficient for v1. Dedicated OCR (Tesseract, etc.) is a future enhancement.
Full-text search on attachment descriptions: GIN index on attachments->description is deferred to a future iteration.
Attachment-only messages: Messages must still have a content field (can be empty string). Attachments are supplementary.
Multi-file messages via SDK in a single call: v1 SDK upload() returns one attachment at a time. Batch upload is a future enhancement.

3. Design

3.1 Message Schema Extension

3.1.1 New Column: `attachments`

Add a JSONB column to the messages table:

ALTER TABLE messages
ADD COLUMN attachments JSONB
DEFAULT NULL;

When present, attachments is a JSON array of attachment objects. When a message has no attachments, the column is NULL (not an empty array) to avoid unnecessary storage on the vast majority of text-only messages.

3.1.2 Attachment Object Schema

Each element in the attachments array conforms to:

{
  "type": "image" | "pdf" | "text",
  "storage_url": "s3://bucket/honcho/ws/abc123/att/def456.png",
  "mime_type": "image/png",
  "filename": "screenshot.png",
  "size_bytes": 102400,
  "description": "A screenshot showing a login form with email and password fields...",
  "checksum": "sha256:a1b2c3d4...",
  "created_at": "2026-03-25T12:00:00Z"
}

Field	Type	Required	Description
`type`	string	yes	One of `"image"`, `"pdf"`, `"text"`. Extensible for future types.
`storage_url`	string	yes	Full storage URL (e.g., `s3://bucket/key`, `gs://bucket/key`, `file:///path`).
`mime_type`	string	yes	IANA MIME type (e.g., `image/png`, `application/pdf`).
`filename`	string	yes	Original filename as provided by the client.
`size_bytes`	integer	yes	File size in bytes.
`description`	string	no	LLM-generated or extracted text description. Populated asynchronously by the deriver for images; populated synchronously for PDFs (via `pdfplumber`).
`checksum`	string	no	`sha256:<hex>` hash of the file bytes. Used for deduplication and integrity verification.
`created_at`	string	yes	ISO 8601 timestamp of when the attachment was stored.

3.1.3 SQLAlchemy Model Change

In src/models.py, add to the Message class:

from sqlalchemy.dialects.postgresql import JSONB
 
class Message(Base):
    __tablename__: str = "messages"
    # ... existing columns ...
 
    attachments: Mapped[list[dict[str, Any]] | None] = mapped_column(
        JSONB, nullable=True, default=None, server_default=text("NULL")
    )

No new check constraints are needed on the column itself. Validation happens at the API schema layer.

3.1.4 Token Counting Update

Token counting must account for image tokens. The formula:

total_tokens = text_tokens + sum(image_token_estimate for each image attachment)

Image token estimates follow Claude’s pricing model as a reasonable default:

Resolution	Estimated Tokens
Any image (default)	1,600

This is a conservative high estimate. The exact value depends on image dimensions and the model, but since Honcho’s token count is used for billing estimation and context budgeting (not exact prompt construction), a fixed per-image estimate is acceptable.

For PDFs, the description field (extracted text) is tokenized normally and added to the count.

The MessageCreate schema validator becomes:

IMAGE_TOKEN_ESTIMATE = 1600  # Conservative estimate per image
 
@model_validator(mode="after")
def validate_and_set_token_count(self) -> Self:
    encoding = tiktoken.get_encoding("o200k_base")
    encoded_message = encoding.encode(self.content)
    self._encoded_message = encoded_message
 
    # Add estimated image tokens
    image_tokens = 0
    if self.attachments:
        for att in self.attachments:
            if att.type == "image":
                image_tokens += IMAGE_TOKEN_ESTIMATE
            elif att.type in ("pdf", "text") and att.description:
                image_tokens += len(encoding.encode(att.description))
 
    self._total_token_count = len(encoded_message) + image_tokens
    return self

3.1.5 Alembic Migration

"""Add attachments column to messages
 
Revision ID: <generated>
Revises: <latest>
Create Date: <generated>
"""
 
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import postgresql
 
revision: str = "<generated>"
down_revision: str = "<latest>"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
 
 
def upgrade() -> None:
    op.add_column(
        "messages",
        sa.Column(
            "attachments",
            postgresql.JSONB(astext_type=sa.Text()),
            nullable=True,
            server_default=None,
        ),
    )
 
 
def downgrade() -> None:
    op.drop_column("messages", "attachments")

This migration is safe to run online. The column is nullable with no default, so PostgreSQL adds it without rewriting the table (instant ADD COLUMN for nullable columns with no non-NULL default).

3.2 Storage Backend Abstraction

3.2.1 Protocol Definition

New file: src/storage/__init__.py

"""
Pluggable file storage backend for Honcho.
 
Supports S3, GCS, and local filesystem storage for message attachments.
"""

New file: src/storage/backend.py

from typing import Protocol, runtime_checkable
 
@runtime_checkable
class FileStorageBackend(Protocol):
    """Protocol for file storage backends.
 
    All methods are async to support both local and cloud storage uniformly.
    Implementations must be safe for concurrent use.
    """
 
    async def upload(
        self,
        key: str,
        data: bytes,
        mime_type: str,
        *,
        checksum: str | None = None,
    ) -> str:
        """Upload file bytes and return the storage URL.
 
        Args:
            key: Storage key/path (e.g., "ws/abc/att/def.png").
            data: Raw file bytes.
            mime_type: IANA MIME type.
            checksum: Optional SHA-256 hex digest for integrity check.
 
        Returns:
            Full storage URL (e.g., "s3://bucket/ws/abc/att/def.png").
 
        Raises:
            StorageError: If upload fails.
        """
        ...
 
    async def download(self, url: str) -> bytes:
        """Download file bytes from a storage URL.
 
        Args:
            url: Full storage URL as returned by upload().
 
        Returns:
            Raw file bytes.
 
        Raises:
            StorageError: If file not found or download fails.
        """
        ...
 
    async def delete(self, url: str) -> None:
        """Delete a file from storage.
 
        Args:
            url: Full storage URL as returned by upload().
 
        Raises:
            StorageError: If deletion fails (missing files are not errors).
        """
        ...
 
    async def generate_presigned_url(
        self, url: str, expires_in: int = 3600
    ) -> str:
        """Generate a time-limited presigned URL for direct download.
 
        Args:
            url: Full storage URL as returned by upload().
            expires_in: Expiry time in seconds (default: 1 hour).
 
        Returns:
            Presigned HTTPS URL.
 
        Raises:
            StorageError: If URL generation fails.
            NotImplementedError: If backend doesn't support presigned URLs.
        """
        ...

3.2.2 Key Generation

Storage keys follow a deterministic path structure:

{prefix}/{workspace_name}/attachments/{nanoid}.{ext}

Example: honcho/my_workspace/attachments/V1StGXR8_Z5jdHi6B-myT.png

prefix: From STORAGE_PATH_PREFIX config (default: "honcho").
workspace_name: Scopes files to their workspace for access control and cleanup.
nanoid: 21-character nanoid (matches existing ID format).
ext: File extension derived from MIME type.

This structure enables:

Workspace-scoped IAM policies or bucket prefixes.
Bulk deletion on workspace teardown (delete by prefix).
No collision risk (nanoid has ~70 bits of entropy).

3.2.3 S3 Backend

New file: src/storage/s3.py

import logging
from urllib.parse import urlparse
 
import boto3  # type: ignore[import-untyped]
from botocore.exceptions import ClientError
 
from src.storage.exceptions import StorageError
 
logger = logging.getLogger(__name__)
 
 
class S3Backend:
    """AWS S3 storage backend.
 
    Uses boto3 with standard AWS credential resolution
    (env vars, IAM role, credentials file).
    """
 
    def __init__(
        self,
        bucket: str,
        region: str | None = None,
        endpoint_url: str | None = None,
    ):
        self.bucket = bucket
        self._client = boto3.client(
            "s3",
            region_name=region,
            endpoint_url=endpoint_url,  # For MinIO/LocalStack
        )
 
    async def upload(
        self,
        key: str,
        data: bytes,
        mime_type: str,
        *,
        checksum: str | None = None,
    ) -> str:
        try:
            put_kwargs: dict = {
                "Bucket": self.bucket,
                "Key": key,
                "Body": data,
                "ContentType": mime_type,
            }
            if checksum:
                put_kwargs["Metadata"] = {"sha256": checksum}
 
            self._client.put_object(**put_kwargs)
            return f"s3://{self.bucket}/{key}"
        except ClientError as e:
            raise StorageError(f"S3 upload failed: {e}") from e
 
    async def download(self, url: str) -> bytes:
        bucket, key = self._parse_url(url)
        try:
            response = self._client.get_object(Bucket=bucket, Key=key)
            return response["Body"].read()
        except ClientError as e:
            raise StorageError(f"S3 download failed: {e}") from e
 
    async def delete(self, url: str) -> None:
        bucket, key = self._parse_url(url)
        try:
            self._client.delete_object(Bucket=bucket, Key=key)
        except ClientError as e:
            raise StorageError(f"S3 delete failed: {e}") from e
 
    async def generate_presigned_url(
        self, url: str, expires_in: int = 3600
    ) -> str:
        bucket, key = self._parse_url(url)
        try:
            return self._client.generate_presigned_url(
                "get_object",
                Params={"Bucket": bucket, "Key": key},
                ExpiresIn=expires_in,
            )
        except ClientError as e:
            raise StorageError(f"S3 presigned URL generation failed: {e}") from e
 
    @staticmethod
    def _parse_url(url: str) -> tuple[str, str]:
        parsed = urlparse(url)
        if parsed.scheme != "s3":
            raise StorageError(f"Not an S3 URL: {url}")
        return parsed.netloc, parsed.path.lstrip("/")

Note on async: boto3 is synchronous. The S3Backend wraps synchronous calls. For production workloads, consider aioboto3 or running boto3 calls in a thread executor. The async def signature maintains protocol compatibility. The initial implementation can use asyncio.to_thread() to avoid blocking the event loop:

import asyncio
 
async def upload(self, key, data, mime_type, *, checksum=None) -> str:
    return await asyncio.to_thread(self._sync_upload, key, data, mime_type, checksum)

3.2.4 GCS Backend

New file: src/storage/gcs.py

Follows the same pattern as S3Backend using google-cloud-storage. Deferred to Phase 2 — S3 is the primary target. The protocol ensures GCS can be added without any changes to calling code.

3.2.5 Local Filesystem Backend

New file: src/storage/local.py

import hashlib
import logging
import os
from pathlib import Path
from urllib.parse import urlparse
 
from src.storage.exceptions import StorageError
 
logger = logging.getLogger(__name__)
 
 
class LocalFilesystemBackend:
    """Local filesystem storage backend.
 
    Intended for development and testing. Not recommended for production.
    """
 
    def __init__(self, base_path: str = "/tmp/honcho-storage"):
        self.base_path = Path(base_path)
        self.base_path.mkdir(parents=True, exist_ok=True)
 
    async def upload(
        self,
        key: str,
        data: bytes,
        mime_type: str,
        *,
        checksum: str | None = None,
    ) -> str:
        file_path = self.base_path / key
        file_path.parent.mkdir(parents=True, exist_ok=True)
        file_path.write_bytes(data)
        return f"file://{file_path}"
 
    async def download(self, url: str) -> bytes:
        path = self._parse_url(url)
        try:
            return Path(path).read_bytes()
        except FileNotFoundError as e:
            raise StorageError(f"File not found: {path}") from e
 
    async def delete(self, url: str) -> None:
        path = self._parse_url(url)
        try:
            Path(path).unlink(missing_ok=True)
        except OSError as e:
            raise StorageError(f"Delete failed: {path}") from e
 
    async def generate_presigned_url(
        self, url: str, expires_in: int = 3600
    ) -> str:
        raise NotImplementedError(
            "Local filesystem backend does not support presigned URLs"
        )
 
    @staticmethod
    def _parse_url(url: str) -> str:
        parsed = urlparse(url)
        if parsed.scheme != "file":
            raise StorageError(f"Not a file URL: {url}")
        return parsed.path

3.2.6 Storage Configuration

Add to src/config.py:

class StorageSettings(HonchoSettings):
    model_config = SettingsConfigDict(env_prefix="STORAGE_", extra="ignore")
 
    # Backend type: "s3", "gcs", "local"
    BACKEND: Literal["s3", "gcs", "local"] = "local"
 
    # Bucket/container name (S3/GCS)
    BUCKET: str | None = None
 
    # Region (S3/GCS)
    REGION: str | None = None
 
    # Custom endpoint URL (for MinIO, LocalStack, etc.)
    ENDPOINT_URL: str | None = None
 
    # Path prefix within the bucket/filesystem
    PATH_PREFIX: str = "honcho"
 
    # Base path for local filesystem backend
    LOCAL_BASE_PATH: str = "/tmp/honcho-storage"
 
    # File size limits
    MAX_FILE_SIZE: Annotated[int, Field(default=10_485_760, gt=0)] = (
        10_485_760  # 10 MB
    )
 
    # Maximum number of attachments per message
    MAX_ATTACHMENTS_PER_MESSAGE: Annotated[int, Field(default=5, gt=0, le=20)] = 5
 
    @model_validator(mode="after")
    def _require_bucket_for_cloud(self) -> "StorageSettings":
        if self.BACKEND in ("s3", "gcs") and not self.BUCKET:
            raise ValueError(
                f"STORAGE_BUCKET must be set when BACKEND is '{self.BACKEND}'"
            )
        return self

Add to AppSettings:

class AppSettings(HonchoSettings):
    # ... existing fields ...
    STORAGE: StorageSettings = Field(default_factory=StorageSettings)

Add to TomlConfigSettingsSource.SECTION_MAP:

"STORAGE": "storage",

TOML example:

[storage]
backend = "s3"
bucket = "honcho-attachments"
region = "us-east-1"
path_prefix = "honcho"
max_file_size = 10485760

Environment variable example:

STORAGE_BACKEND=s3
STORAGE_BUCKET=honcho-attachments
STORAGE_REGION=us-east-1

3.2.7 Storage Backend Factory

New file: src/storage/factory.py

from functools import lru_cache
 
from src.config import settings
from src.storage.backend import FileStorageBackend
 
 
@lru_cache(maxsize=1)
def get_storage_backend() -> FileStorageBackend:
    """Get the configured storage backend singleton."""
    backend_type = settings.STORAGE.BACKEND
 
    if backend_type == "s3":
        from src.storage.s3 import S3Backend
        return S3Backend(
            bucket=settings.STORAGE.BUCKET,  # type: ignore (validated)
            region=settings.STORAGE.REGION,
            endpoint_url=settings.STORAGE.ENDPOINT_URL,
        )
    elif backend_type == "gcs":
        from src.storage.gcs import GCSBackend
        return GCSBackend(
            bucket=settings.STORAGE.BUCKET,  # type: ignore (validated)
        )
    elif backend_type == "local":
        from src.storage.local import LocalFilesystemBackend
        return LocalFilesystemBackend(
            base_path=settings.STORAGE.LOCAL_BASE_PATH,
        )
    else:
        raise ValueError(f"Unknown storage backend: {backend_type}")

3.3 API Schema Changes

3.3.1 Attachment Input Schema

New addition to src/schemas/api.py:

ALLOWED_MIME_TYPES: dict[str, str] = {
    # Images
    "image/png": "image",
    "image/jpeg": "image",
    "image/gif": "image",
    "image/webp": "image",
    # PDFs
    "application/pdf": "pdf",
    # Text
    "text/plain": "text",
    "text/csv": "text",
    "text/markdown": "text",
    "application/json": "text",
}
 
 
class AttachmentInput(BaseModel):
    """Attachment reference for message creation.
 
    Clients provide this after uploading a file via the storage API.
    """
    storage_url: str = Field(
        ..., description="Storage URL returned by the upload endpoint"
    )
    mime_type: str = Field(
        ..., description="IANA MIME type of the file"
    )
    filename: str = Field(
        ..., max_length=255, description="Original filename"
    )
    size_bytes: int = Field(
        ..., gt=0, description="File size in bytes"
    )
    checksum: str | None = Field(
        None, description="SHA-256 checksum in format 'sha256:<hex>'"
    )
 
    @field_validator("mime_type")
    @classmethod
    def validate_mime_type(cls, v: str) -> str:
        if v not in ALLOWED_MIME_TYPES:
            raise ValueError(
                f"Unsupported MIME type: {v}. Allowed: {list(ALLOWED_MIME_TYPES.keys())}"
            )
        return v
 
    @field_validator("checksum")
    @classmethod
    def validate_checksum_format(cls, v: str | None) -> str | None:
        if v is not None and not v.startswith("sha256:"):
            raise ValueError("Checksum must be in format 'sha256:<hex>'")
        return v
 
    def to_attachment_dict(self) -> dict[str, Any]:
        """Convert to the JSONB attachment format stored in the database."""
        from src.utils.formatting import utc_now_iso
        return {
            "type": ALLOWED_MIME_TYPES[self.mime_type],
            "storage_url": self.storage_url,
            "mime_type": self.mime_type,
            "filename": self.filename,
            "size_bytes": self.size_bytes,
            "description": None,  # Populated later by deriver (images) or synchronously (PDFs)
            "checksum": self.checksum,
            "created_at": utc_now_iso(),
        }

3.3.2 Updated MessageCreate Schema

class MessageCreate(MessageBase):
    content: Annotated[str, Field(min_length=0, max_length=settings.MAX_MESSAGE_SIZE)]
    peer_name: str = Field(alias="peer_id")
    metadata: _SanitizedMetadata | None = None
    configuration: MessageConfiguration | None = None
    created_at: datetime.datetime | None = None
    attachments: list[AttachmentInput] | None = Field(
        None,
        max_length=settings.STORAGE.MAX_ATTACHMENTS_PER_MESSAGE,
        description="File attachments for this message",
    )
 
    _encoded_message: list[int] = PrivateAttr(default=[])
    _attachment_dicts: list[dict[str, Any]] = PrivateAttr(default=[])
 
    @field_validator("content", mode="after")
    @classmethod
    def sanitize_content(cls, v: str) -> str:
        return v.replace("\x00", "")
 
    @property
    def encoded_message(self) -> list[int]:
        return self._encoded_message
 
    @property
    def attachment_dicts(self) -> list[dict[str, Any]]:
        return self._attachment_dicts
 
    @model_validator(mode="after")
    def validate_and_set_token_count(self) -> Self:
        encoding = tiktoken.get_encoding("o200k_base")
        encoded_message = encoding.encode(self.content)
        self._encoded_message = encoded_message
 
        # Convert attachments to storage format
        if self.attachments:
            self._attachment_dicts = [
                att.to_attachment_dict() for att in self.attachments
            ]
 
        return self

3.3.3 Updated Message Response Schema

class Message(MessageBase):
    public_id: str = Field(serialization_alias="id")
    content: str
    peer_name: str = Field(serialization_alias="peer_id")
    session_name: str = Field(serialization_alias="session_id")
    h_metadata: dict[str, Any] = Field(
        default_factory=dict, serialization_alias="metadata"
    )
    created_at: datetime.datetime
    workspace_name: str = Field(serialization_alias="workspace_id")
    token_count: int
    attachments: list[dict[str, Any]] | None = Field(
        default=None, description="File attachments"
    )
 
    model_config = ConfigDict(
        from_attributes=True, populate_by_name=True
    )

3.3.4 Upload Endpoint

New router: src/routers/storage.py

@router.post("/upload", response_model=AttachmentUploadResponse, status_code=201)
async def upload_attachment(
    workspace_id: str = Path(...),
    file: UploadFile = File(...),
):
    """Upload a file and receive a storage reference for use in message attachments.
 
    The file is stored in the configured backend (S3, GCS, or local filesystem).
    The returned reference should be included in the `attachments` array when
    creating messages.
 
    Supported file types: PNG, JPEG, GIF, WebP, PDF, plain text, CSV, Markdown, JSON.
    Maximum file size: configurable (default 10MB).
    """

Response:

class AttachmentUploadResponse(BaseModel):
    storage_url: str
    mime_type: str
    filename: str
    size_bytes: int
    checksum: str
    type: str  # "image", "pdf", "text"

3.4 Deriver Integration

3.4.1 Vision Description Pipeline

When a message has image attachments without a description, the deriver must generate one before running observation extraction. This happens as a pre-processing step within process_representation_tasks_batch().

Flow:

Message arrives with image attachment (description=None)
    |
    v
Deriver picks up message from queue
    |
    v
Pre-process: For each image attachment without description:
    1. Download image bytes from storage backend
    2. Call vision model with image + prompt: "Describe this image in detail..."
    3. Store description back in the attachment JSONB
    4. Commit to DB
    |
    v
Format messages: content + attachment descriptions
    |
    v
Standard deriver prompt + LLM call for observation extraction

3.4.2 Vision Model Configuration

Add to DeriverSettings:

class DeriverSettings(BackupLLMSettingsMixin, HonchoSettings):
    # ... existing fields ...
 
    # Vision model for describing image attachments
    VISION_MODEL: str = "claude-sonnet-4-20250514"
    VISION_PROVIDER: SupportedProviders = "anthropic"
    VISION_MAX_TOKENS: Annotated[int, Field(default=1024, gt=0, le=4096)] = 1024

Config via environment:

DERIVER_VISION_MODEL=claude-sonnet-4-20250514
DERIVER_VISION_PROVIDER=anthropic

Config via TOML:

[deriver]
vision_model = "claude-sonnet-4-20250514"
vision_provider = "anthropic"

3.4.3 Vision Description Function

New file: src/deriver/vision.py

"""
Vision processing for image attachments.
 
Generates text descriptions of images using a vision-capable LLM.
These descriptions are stored in the attachment's 'description' field
and used by the deriver for observation extraction.
"""
 
import logging
from typing import Any
 
from src.config import settings
from src.storage.factory import get_storage_backend
from src.utils.clients import honcho_llm_call
 
logger = logging.getLogger(__name__)
 
VISION_PROMPT = """Describe this image in detail. Focus on:
- What is depicted (objects, people, scenes, text)
- Any text visible in the image (transcribe it)
- Context and setting
- Notable details that would help understand who shared this image and why
 
Be factual and concise. Do not speculate beyond what is visible."""
 
 
async def describe_image(storage_url: str) -> str:
    """Download an image and generate a text description using a vision model.
 
    Args:
        storage_url: Storage URL of the image.
 
    Returns:
        Text description of the image.
 
    Raises:
        StorageError: If image download fails.
        LLMError: If vision model call fails.
    """
    backend = get_storage_backend()
    image_bytes = await backend.download(storage_url)
 
    # Determine media type from URL extension or stored mime_type
    # The caller should pass mime_type, but we can infer from URL as fallback
    media_type = _infer_media_type(storage_url)
 
    response = await honcho_llm_call(
        llm_settings=_build_vision_settings(),
        prompt=VISION_PROMPT,
        max_tokens=settings.DERIVER.VISION_MAX_TOKENS,
        track_name="Vision Description",
        images=[(image_bytes, media_type)],
        trace_name="vision_description",
    )
 
    return response.content
 
 
async def process_image_attachments(
    attachments: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    """Process image attachments that lack descriptions.
 
    Args:
        attachments: List of attachment dicts from the message.
 
    Returns:
        Updated attachment list with descriptions populated.
    """
    updated = []
    for att in attachments:
        if att["type"] == "image" and not att.get("description"):
            try:
                description = await describe_image(att["storage_url"])
                att["description"] = description
                logger.info(
                    "Generated description for image %s (%d chars)",
                    att["filename"],
                    len(description),
                )
            except Exception:
                logger.exception(
                    "Failed to describe image %s", att["filename"]
                )
                att["description"] = "[Image description unavailable]"
        updated.append(att)
    return updated

3.4.4 Deriver Prompt Modification

The existing format_new_turn_with_timestamp in src/utils/formatting.py must be extended to include attachment descriptions:

def format_new_turn_with_timestamp(
    new_turn: str,
    current_time: datetime,
    speaker: str,
    attachments: list[dict[str, Any]] | None = None,
) -> str:
    """Format new turn message with optional timestamp and attachment descriptions.
 
    Args:
        new_turn: The message content.
        current_time: Message timestamp.
        speaker: The speaker's name.
        attachments: Optional list of attachment dicts.
 
    Returns:
        Formatted string including attachment descriptions.
    """
    current_time_str = current_time.strftime("%Y-%m-%d %H:%M:%S")
    parts = [f"{current_time_str} {speaker}: {new_turn}"]
 
    if attachments:
        for att in attachments:
            desc = att.get("description")
            if desc:
                parts.append(f"  [Attached {att['type']}: {att.get('filename', 'unnamed')}] {desc}")
 
    return "\n".join(parts)

The call site in src/deriver/deriver.py changes from:

formatted_messages = "\n".join(
    format_new_turn_with_timestamp(msg.content, msg.created_at, msg.peer_name)
    for msg in messages
)

To:

formatted_messages = "\n".join(
    format_new_turn_with_timestamp(
        msg.content, msg.created_at, msg.peer_name, msg.attachments
    )
    for msg in messages
)

3.4.5 Deriver Pre-Processing Step

In src/deriver/deriver.py, add pre-processing before the main LLM call:

async def process_representation_tasks_batch(
    messages: list[Message],
    ...
) -> None:
    # ... existing setup ...
 
    # Pre-process: generate descriptions for image attachments
    from src.deriver.vision import process_image_attachments
    from src.dependencies import tracked_db
 
    for msg in messages:
        if msg.attachments:
            needs_description = any(
                att["type"] == "image" and not att.get("description")
                for att in msg.attachments
            )
            if needs_description:
                msg.attachments = await process_image_attachments(msg.attachments)
                # Persist the updated descriptions
                async with tracked_db("vision_description") as db:
                    from sqlalchemy import update
                    await db.execute(
                        update(models.Message)
                        .where(models.Message.id == msg.id)
                        .values(attachments=msg.attachments)
                    )
                    await db.commit()
 
    # ... rest of existing logic (format messages, LLM call, etc.) ...

3.4.6 PDF Processing at Upload Time

PDFs are processed synchronously during the upload/message-creation step, not in the deriver. This is because pdfplumber is fast and we already have the dependency.

In the upload endpoint or the message creation CRUD, when an attachment with mime_type: "application/pdf" is provided:

async def extract_pdf_description(storage_url: str) -> str:
    """Extract text from a PDF stored in the storage backend."""
    backend = get_storage_backend()
    pdf_bytes = await backend.download(storage_url)
 
    import pdfplumber
    from io import BytesIO
 
    with pdfplumber.open(BytesIO(pdf_bytes)) as pdf:
        text_parts = []
        for page_num, page in enumerate(pdf.pages):
            text = page.extract_text()
            if text and text.strip():
                text_parts.append(f"[Page {page_num + 1}]\n{text}")
        return "\n\n".join(text_parts)

This reuses the exact logic from the existing PDFProcessor in src/utils/files.py. The description is populated into the attachment dict before the message is committed to the database, so the deriver receives it pre-processed.

3.5 SDK Changes

3.5.1 Python SDK

New Type: `Attachment`

New file: sdks/python/src/honcho/attachment.py

"""Attachment class for Honcho SDK."""
 
from __future__ import annotations
 
from typing import Any
 
 
class Attachment:
    """Represents a file attachment for a Honcho message.
 
    Typically created via `honcho.upload()` or `session.upload()`.
 
    Attributes:
        storage_url: Storage URL for the file.
        mime_type: IANA MIME type.
        filename: Original filename.
        size_bytes: File size in bytes.
        checksum: SHA-256 checksum.
        type: Attachment type ("image", "pdf", "text").
        description: Text description (if available).
    """
 
    def __init__(
        self,
        storage_url: str,
        mime_type: str,
        filename: str,
        size_bytes: int,
        checksum: str,
        type: str,
        description: str | None = None,
    ) -> None:
        self.storage_url = storage_url
        self.mime_type = mime_type
        self.filename = filename
        self.size_bytes = size_bytes
        self.checksum = checksum
        self.type = type
        self.description = description
 
    def to_api_dict(self) -> dict[str, Any]:
        """Convert to the API request format."""
        return {
            "storage_url": self.storage_url,
            "mime_type": self.mime_type,
            "filename": self.filename,
            "size_bytes": self.size_bytes,
            "checksum": self.checksum,
        }
 
    @classmethod
    def from_api_response(cls, data: dict[str, Any]) -> "Attachment":
        """Create from upload API response."""
        return cls(
            storage_url=data["storage_url"],
            mime_type=data["mime_type"],
            filename=data["filename"],
            size_bytes=data["size_bytes"],
            checksum=data["checksum"],
            type=data["type"],
        )
 
    def __repr__(self) -> str:
        return f"Attachment(type='{self.type}', filename='{self.filename}')"

Upload Method on Honcho Client

# In sdks/python/src/honcho/client.py
 
class Honcho:
    # ... existing ...
 
    def upload(
        self,
        file: str | IOBase | tuple[str, bytes, str],
        *,
        workspace_id: str | None = None,
    ) -> Attachment:
        """Upload a file and return an Attachment reference.
 
        The returned Attachment can be passed to session.add_messages()
        in the `attachments` parameter.
 
        Args:
            file: File to upload. Can be:
                - A file path (str)
                - A file object (must have .name and .read())
                - A tuple (filename, bytes, content_type)
            workspace_id: Override workspace (defaults to client workspace).
 
        Returns:
            Attachment object ready for inclusion in messages.
 
        Example:
            attachment = honcho.upload("photo.png")
            session.add_messages([
                user.message("Check out this photo", attachments=[attachment])
            ])
        """

Updated Message Creation

# In sdks/python/src/honcho/session.py
 
class Session:
    def add_messages(
        self,
        messages: list[MessageInput] | MessageInput,
        *,
        attachments: list[Attachment] | None = None,  # NEW
    ) -> list[Message]:
        """Add messages to this session.
 
        Args:
            messages: One or more messages to add.
            attachments: Optional attachments for the first message.
                For per-message attachments, use MessageInput.attachments.
        """

Usage Example

from honcho import Honcho
 
honcho = Honcho(workspace="my_workspace")
session = honcho.session("chat-1")
user = honcho.peer("alice")
 
# Upload a file
attachment = honcho.upload("screenshot.png")
 
# Create a message with the attachment
session.add_messages([
    user.message(
        "Here's what I'm seeing on my screen",
        attachments=[attachment],
    )
])
 
# Or upload and attach in one step
session.upload_file("report.pdf", peer=user)

3.5.2 TypeScript SDK

New Type: `Attachment`

New file: sdks/typescript/src/attachment.ts

/**
 * Represents a file attachment for a Honcho message.
 */
export interface AttachmentData {
  storageUrl: string
  mimeType: string
  filename: string
  sizeBytes: number
  checksum: string
  type: 'image' | 'pdf' | 'text'
  description?: string
}
 
export class Attachment {
  readonly storageUrl: string
  readonly mimeType: string
  readonly filename: string
  readonly sizeBytes: number
  readonly checksum: string
  readonly type: 'image' | 'pdf' | 'text'
  readonly description?: string
 
  constructor(data: AttachmentData) {
    this.storageUrl = data.storageUrl
    this.mimeType = data.mimeType
    this.filename = data.filename
    this.sizeBytes = data.sizeBytes
    this.checksum = data.checksum
    this.type = data.type
    this.description = data.description
  }
 
  /** Convert to API request format */
  toApiDict(): Record<string, unknown> {
    return {
      storage_url: this.storageUrl,
      mime_type: this.mimeType,
      filename: this.filename,
      size_bytes: this.sizeBytes,
      checksum: this.checksum,
    }
  }
 
  static fromApiResponse(data: Record<string, unknown>): Attachment {
    return new Attachment({
      storageUrl: data.storage_url as string,
      mimeType: data.mime_type as string,
      filename: data.filename as string,
      sizeBytes: data.size_bytes as number,
      checksum: data.checksum as string,
      type: data.type as 'image' | 'pdf' | 'text',
    })
  }
}

Upload Method

class Honcho {
  /**
   * Upload a file and return an Attachment reference.
   *
   * @param file - File to upload (Buffer, ReadableStream, or File object)
   * @param options - Upload options including filename and content type
   * @returns Attachment ready for inclusion in messages
   *
   * @example
   * ```typescript
   * const attachment = await honcho.upload(fs.readFileSync('photo.png'), {
   *   filename: 'photo.png',
   *   contentType: 'image/png',
   * })
   * await session.addMessages([
   *   user.message('Check this out', { attachments: [attachment] })
   * ])
   * ```
   */
  async upload(
    file: Buffer | ReadableStream | File,
    options: { filename: string; contentType: string }
  ): Promise<Attachment>
}

Updated MessageInput

export interface MessageInput {
  peerId: string
  content: string
  metadata?: Record<string, unknown>
  configuration?: MessageConfiguration
  createdAt?: string
  attachments?: Attachment[]  // NEW
}

3.6 Security

3.6.1 MIME Type Allowlist

Only the following MIME types are accepted:

MIME Type	Category	Extension
`image/png`	image	`.png`
`image/jpeg`	image	`.jpg`, `.jpeg`
`image/gif`	image	`.gif`
`image/webp`	image	`.webp`
`application/pdf`	pdf	`.pdf`
`text/plain`	text	`.txt`
`text/csv`	text	`.csv`
`text/markdown`	text	`.md`
`application/json`	text	`.json`

All other MIME types are rejected with HTTP 415 Unsupported Media Type.

3.6.2 Magic Byte Verification

MIME type declared by the client must match magic bytes in the file content. This prevents disguised executables.

New file: src/storage/validation.py

"""File validation utilities for storage uploads."""
 
MAGIC_BYTES: dict[str, list[bytes]] = {
    "image/png": [b"\x89PNG\r\n\x1a\n"],
    "image/jpeg": [b"\xff\xd8\xff"],
    "image/gif": [b"GIF87a", b"GIF89a"],
    "image/webp": [b"RIFF"],  # Check for RIFF header + WEBP at offset 8
    "application/pdf": [b"%PDF"],
}
 
 
def verify_magic_bytes(data: bytes, declared_mime_type: str) -> bool:
    """Verify that file content matches the declared MIME type.
 
    Returns True if:
    - The MIME type has no registered magic bytes (text types)
    - The file starts with one of the expected magic byte sequences
 
    Returns False if:
    - The file content doesn't match any expected magic bytes
    """
    expected = MAGIC_BYTES.get(declared_mime_type)
    if expected is None:
        return True  # No magic bytes check for text types
 
    for magic in expected:
        if data[:len(magic)] == magic:
            if declared_mime_type == "image/webp":
                # Additional check: WEBP marker at offset 8
                return len(data) >= 12 and data[8:12] == b"WEBP"
            return True
    return False

3.6.3 File Size Limits

Default maximum: 10 MB per file (configurable via STORAGE_MAX_FILE_SIZE).
Enforced at two points: (1) in the upload endpoint before reading the full body, (2) after reading, before storage.
Content-Length header is validated if present but not trusted (full body is measured).

3.6.4 Filename Sanitization

import re
import os
 
def sanitize_filename(filename: str) -> str:
    """Sanitize a filename to prevent path traversal and other attacks."""
    # Remove path components
    filename = os.path.basename(filename)
    # Remove null bytes
    filename = filename.replace("\x00", "")
    # Allow only safe characters
    filename = re.sub(r"[^a-zA-Z0-9._-]", "_", filename)
    # Prevent hidden files
    filename = filename.lstrip(".")
    # Limit length
    filename = filename[:255]
    return filename or "unnamed"

3.6.5 No Executable File Types

The allowlist inherently blocks executable types (.exe, .sh, .bat, .py, etc.). Additionally, even if a client sends a valid image MIME type, the magic byte check ensures the content is actually an image.

3.6.6 Storage URL Validation

When messages reference storage URLs in the attachments field, the URL scheme must match the configured backend:

s3:// for S3 backend
gs:// for GCS backend
file:// for local backend

Cross-backend URL references are rejected.

3.6.7 Security Checklist

MIME type allowlist (no executables)
Magic byte verification (content matches declared type)
File size limits (configurable, default 10MB)
Filename sanitization (no path traversal)
Storage URL scheme validation
No inline binary in database (references only)
Workspace-scoped storage paths (isolation)
SHA-256 checksums for integrity

3.7 CloudEvents for Billing

New telemetry events for storage operations, enabling Groudon/Xatu to meter storage usage.

3.7.1 StorageUploadEvent

New file: src/telemetry/events/storage.py

from pydantic import Field
from src.telemetry.events.base import BaseEvent
 
 
class StorageUploadEvent(BaseEvent):
    """Emitted when a file is uploaded to storage."""
 
    _event_type = "honcho.storage.upload"
    _schema_version = 1
    _category = "resource"
 
    workspace_name: str
    storage_url: str
    mime_type: str
    size_bytes: int
    attachment_type: str  # "image", "pdf", "text"
 
    def get_resource_id(self) -> str:
        return f"{self.workspace_name}:{self.storage_url}"
 
 
class StorageDeleteEvent(BaseEvent):
    """Emitted when a file is deleted from storage."""
 
    _event_type = "honcho.storage.delete"
    _schema_version = 1
    _category = "resource"
 
    workspace_name: str
    storage_url: str
 
    def get_resource_id(self) -> str:
        return f"{self.workspace_name}:{self.storage_url}"
 
 
class VisionDescriptionEvent(BaseEvent):
    """Emitted when a vision model generates an image description."""
 
    _event_type = "honcho.work.vision_description.completed"
    _schema_version = 1
    _category = "work"
 
    workspace_name: str
    session_name: str
    message_id: int
    attachment_filename: str
    description_length: int
    input_tokens: int
    output_tokens: int
    duration_ms: float
 
    def get_resource_id(self) -> str:
        return f"{self.workspace_name}:{self.message_id}:{self.attachment_filename}"

from src.telemetry.events.storage import (
    StorageUploadEvent,
    StorageDeleteEvent,
    VisionDescriptionEvent,
)

3.8 Existing Upload Endpoint Refactor

The current /upload endpoint in src/routers/messages.py extracts text from files and stores them as plain-text messages. This must be updated to use the new storage + attachment system.

3.8.1 New Behavior

The existing create_messages_with_file endpoint changes:

Store the original file in the storage backend.
Create an attachment dict with the storage URL.
For PDFs: Extract text via pdfplumber, populate description in the attachment.
For text files: Read content, populate description in the attachment.
For images: Leave description as None (deriver will populate).
Create a single message with the file content as content (for PDFs/text) or empty string (for images), plus the attachment in attachments.

This is a breaking change to the upload endpoint behavior. The old behavior (text extraction into content, file metadata in internal_metadata) is replaced. The old endpoint path is preserved for backward compatibility but the response shape changes (messages now include attachments).

3.8.2 Backward Compatibility

The content field for PDF uploads continues to contain extracted text (same as before). The addition of attachments is purely additive. Clients that ignore attachments see no change.

For image uploads (new capability), content will be an empty string or a user-provided description.

4. Migration Plan

4.1 Database Migration

The only schema change is adding the nullable attachments column. This is a non-blocking online migration:

ALTER TABLE messages ADD COLUMN attachments JSONB DEFAULT NULL;

PostgreSQL adds nullable columns with no default instantly (no table rewrite). Existing messages have attachments = NULL.

4.2 Configuration Migration

New environment variables / TOML sections are all optional with sensible defaults:

STORAGE_BACKEND=local (default for development)
DERIVER_VISION_MODEL=claude-sonnet-4-20250514
DERIVER_VISION_PROVIDER=anthropic

No existing configuration needs to change.

4.3 SDK Migration

The SDK changes are purely additive:

New Attachment class and upload() method.
MessageInput gains optional attachments field.
Message response gains optional attachments field.
No breaking changes to existing method signatures.

SDK version bump: minor version (v2.1.0).

5. Implementation Phases

Phase 1: Storage Backend + Schema (1 week)

Files to create:

src/storage/__init__.py
src/storage/backend.py (protocol)
src/storage/s3.py
src/storage/local.py
src/storage/factory.py
src/storage/exceptions.py
src/storage/validation.py

Files to modify:

src/config.py — add StorageSettings, add to AppSettings
src/models.py — add attachments column to Message
src/schemas/api.py — add AttachmentInput, update MessageCreate, update Message
pyproject.toml — move boto3 from dev to main dependencies
New Alembic migration

Tests:

Unit tests for storage backends (S3 with moto, local with tmpdir)
Unit tests for MIME validation, magic byte checks, filename sanitization
Unit tests for attachment schema validation

Definition of done: uv run pytest tests/ passes. uv run basedpyright clean. Files can be uploaded to local backend and referenced in messages.

Phase 2: Upload API Endpoint (3 days)

Files to create:

src/routers/storage.py — upload endpoint

Files to modify:

src/main.py — register storage router
src/routers/messages.py — update create_messages_with_file to use new system
src/crud/message.py — pass attachments through create_messages()

Tests:

Integration tests for upload endpoint
Integration tests for message creation with attachments
Test file size rejection, MIME type rejection

Definition of done: Upload endpoint works end-to-end. Messages with attachments stored and returned correctly in API responses.

Phase 3: Deriver Vision Integration (1 week)

Files to create:

src/deriver/vision.py — image description generation

Files to modify:

src/deriver/deriver.py — pre-processing step for image attachments
src/utils/formatting.py — extend format_new_turn_with_timestamp for attachments
src/config.py — add vision model settings to DeriverSettings

Tests:

Unit tests for vision description (mocked LLM)
Integration test: message with image attachment → deriver processes → description populated
Test that deriver prompt includes attachment descriptions

Definition of done: Image messages processed end-to-end. Observations extracted from image descriptions. Descriptions persisted.

Phase 4: CloudEvents + Telemetry (2 days)

Files to create:

src/telemetry/events/storage.py

Files to modify:

src/telemetry/events/__init__.py — register new events
src/routers/storage.py — emit StorageUploadEvent
src/deriver/vision.py — emit VisionDescriptionEvent

Tests:

Unit tests for event creation and serialization
Verify events are emitted in integration tests

Definition of done: Storage and vision events emitted and visible in telemetry output.

Phase 5: SDK Updates (1 week)

Python SDK files to create:

sdks/python/src/honcho/attachment.py

Python SDK files to modify:

sdks/python/src/honcho/client.py — add upload() method
sdks/python/src/honcho/session.py — update add_messages() for attachments
sdks/python/src/honcho/message.py — add attachments field
sdks/python/src/honcho/api_types.py — add attachment types
sdks/python/src/honcho/__init__.py — export Attachment

TypeScript SDK files to create:

sdks/typescript/src/attachment.ts

TypeScript SDK files to modify:

sdks/typescript/src/client.ts — add upload() method
sdks/typescript/src/session.ts — update addMessages() for attachments
sdks/typescript/src/message.ts — add attachments field
sdks/typescript/src/types/api.ts — add attachment types
sdks/typescript/src/index.ts — export Attachment

Tests:

Python SDK: test upload + message creation + attachment in response
TypeScript SDK: test upload + message creation (via pytest orchestration)

Definition of done: Both SDKs can upload files and create messages with attachments. Examples updated.

Phase 6: GCS Backend (deferred)

Implement src/storage/gcs.py
Add google-cloud-storage to pyproject.toml
Tests with mock GCS

6. Files to Modify (Complete List)

Server (repos/honcho/)

File	Change
`src/models.py`	Add `attachments` column to `Message`
`src/schemas/api.py`	Add `AttachmentInput`, `AttachmentUploadResponse`, update `MessageCreate`, `Message`
`src/config.py`	Add `StorageSettings`, add to `AppSettings`, add vision settings to `DeriverSettings`
`src/crud/message.py`	Pass `attachments` in `create_messages()`
`src/routers/messages.py`	Update `create_messages_with_file`
`src/main.py`	Register storage router
`src/deriver/deriver.py`	Add vision pre-processing step
`src/utils/formatting.py`	Extend `format_new_turn_with_timestamp`
`src/telemetry/events/__init__.py`	Register storage events
`pyproject.toml`	Move `boto3` to main deps
`migrations/versions/<new>.py`	Add `attachments` column
New `src/storage/__init__.py`	Storage package
New `src/storage/backend.py`	`FileStorageBackend` protocol
New `src/storage/s3.py`	S3 implementation
New `src/storage/local.py`	Local filesystem implementation
New `src/storage/factory.py`	Backend factory
New `src/storage/exceptions.py`	Storage exceptions
New `src/storage/validation.py`	Magic bytes, filename sanitization
New `src/routers/storage.py`	Upload endpoint
New `src/deriver/vision.py`	Vision description generation
New `src/telemetry/events/storage.py`	Storage telemetry events

Python SDK (repos/honcho/sdks/python/)

File	Change
`src/honcho/client.py`	Add `upload()` method
`src/honcho/session.py`	Update `add_messages()`
`src/honcho/message.py`	Add `attachments` field
`src/honcho/api_types.py`	Add attachment response type
`src/honcho/__init__.py`	Export `Attachment`
New `src/honcho/attachment.py`	`Attachment` class

TypeScript SDK (repos/honcho/sdks/typescript/)

File	Change
`src/client.ts`	Add `upload()` method
`src/session.ts`	Update `addMessages()`
`src/message.ts`	Add `attachments` field
`src/types/api.ts`	Add attachment types
`src/index.ts`	Export `Attachment`
New `src/attachment.ts`	`Attachment` class

7. Risk Assessment

7.1 High Risk

Risk	Impact	Mitigation
Vision model latency adds to deriver processing time	Deriver throughput decreases for image-heavy workloads	Vision processing is async in the deriver; images without descriptions don’t block text-only messages in the same batch. Consider a separate vision queue if latency becomes problematic.
Storage backend outage blocks message creation	Users can’t create messages with attachments	Storage upload is a separate API call. If upload fails, the user gets an error before message creation. Messages without attachments are unaffected.
Large attachments increase storage costs	Unexpected billing for Groudon tenants	CloudEvents enable metering. File size limits are enforced (default 10MB). Groudon can set per-tenant limits.

7.2 Medium Risk

Risk	Impact	Mitigation
JSONB `attachments` column grows large for attachment-heavy messages	Query performance degrades	Max 5 attachments per message. Each attachment is ~500 bytes of JSON. 5 attachments = ~2.5KB, well within JSONB TOAST threshold.
Vision model hallucination in image descriptions	Bad observations extracted by deriver	The vision prompt is fact-focused. Descriptions are stored and auditable. Future: confidence scoring.
S3 credentials misconfigured in self-hosted deployments	Upload endpoint returns 500s	Clear error messages on startup validation. Local backend as default.
`boto3` synchronous calls block event loop	API latency spikes during upload	Wrap boto3 calls in `asyncio.to_thread()`. Document in CLAUDE.md. Consider `aioboto3` in a future iteration.

7.3 Low Risk

Risk	Impact	Mitigation
Existing `/upload` endpoint behavior change	SDK users who rely on current upload behavior see different response shape	`content` field still contains extracted text for PDFs. `attachments` is additive. SDK version bump signals the change.
Token counting estimate diverges from actual model token usage	Billing inaccuracy	Conservative estimate (1,600 tokens/image). Documented as estimate.

8. Verification Plan

8.1 Unit Tests

Storage backends: Upload, download, delete for S3 (moto mock), local filesystem.
Validation: MIME type allowlist, magic byte verification for all supported types, filename sanitization edge cases (path traversal, null bytes, Unicode, empty).
Schema: AttachmentInput validation (valid/invalid MIME types, checksum format, size limits).
Token counting: Verify image token estimates are added to total.
Formatting: format_new_turn_with_timestamp with and without attachments.

8.2 Integration Tests

Upload flow: Upload file via /upload endpoint → verify file in storage → verify response contains correct metadata.
Message with attachment: Upload file, create message with attachment reference → verify message in DB has attachments JSONB populated.
Deriver with image: Create message with image attachment → deriver processes → verify description field populated → verify observations extracted.
Deriver with PDF: Create message with PDF attachment → verify description extracted synchronously → deriver processes with description.
Deriver with text-only message: Ensure existing text-only flow is unchanged (no regression).
Security rejection: Upload file with wrong magic bytes → verify 415 response. Upload oversized file → verify 413 response.
SDK upload: Python SDK honcho.upload() → session.add_messages() with attachment → verify end-to-end.

8.3 Performance Tests

Upload latency: Benchmark upload endpoint with 1MB, 5MB, 10MB files.
Deriver throughput: Compare deriver processing time for batches with and without image attachments.
Vision model latency: Measure average time for vision model description.

8.4 Security Tests

Path traversal filenames: ../../etc/passwd, ../../../root/.ssh/id_rsa.
MIME type spoofing: .exe renamed to .png with wrong magic bytes.
Oversized files: Files exactly at and above the size limit.
Null bytes in filenames: photo\x00.exe.png.
Cross-backend URL injection: S3 URL when backend is local.

9. Open Questions

9.1 Resolved

Storage backend choice: External (S3/GCS/local), not PostgreSQL BYTEA. Decided by vineeth.
PDF processing location: Synchronous at upload time (not in deriver). Existing pdfplumber dependency confirms this.
Image description approach: Vision model in deriver, description stored in attachment. Decided by vineeth.
Attachment model: JSONB array on message (not a separate table). Simpler, fewer joins, attachments are always accessed with their message.

9.2 Open

Should the deriver vision step be a separate queue task type? Currently proposed as a pre-processing step within the representation task. A separate vision task type would enable independent scaling and retry, but adds queue complexity. Recommendation: Start with pre-processing step; extract to separate task type if vision latency becomes a bottleneck.
Should attachment descriptions be included in message embeddings? Currently, MessageEmbedding content is set to message_obj.content. Should it include attachment descriptions for semantic search? Recommendation: Yes, concatenate content + descriptions for embedding. This enables semantic search to find messages by their visual content.
How should workspace deletion handle stored files? When a workspace is deleted, all messages are cascade-deleted. Should we also delete stored files? Recommendation: Yes, emit a StorageBulkDeleteEvent and delete by prefix ({prefix}/{workspace_name}/). This should be handled by the existing deletion queue task type.
Should presigned URLs be exposed via a public API endpoint? This would enable clients to download attachments directly without going through Honcho. Recommendation: Yes, add GET /workspaces/{workspace_id}/attachments/{attachment_id}/url that returns a time-limited presigned URL. Defer to Phase 2.
What is the token cost accounting for vision model calls in Groudon billing? Vision model input tokens (image tokens) are significantly more expensive than text tokens. Recommendation: The VisionDescriptionEvent includes input_tokens and output_tokens. Xatu should meter vision tokens separately from deriver text tokens, and Groudon’s Stripe metering should reflect the higher per-token cost.
Should the honcho_llm_call utility support image inputs natively? The current honcho_llm_call in src/utils/clients.py only handles text prompts. It needs an images parameter. Recommendation: Yes, extend honcho_llm_call with an optional images: list[tuple[bytes, str]] | None parameter. The underlying provider-specific dispatch already supports multimodal for Anthropic (content blocks with image type) and Google (inline_data parts).
Rate limiting on upload endpoint? Large files consume bandwidth and storage. Should there be per-workspace upload rate limits? Recommendation: Defer to Groudon gateway-level rate limiting. The upload endpoint should respect the existing auth and workspace scoping.

Plastic Labs Internal Wiki

Explorer

Recently Updated

Company

Operating Model — Decisions Log

Operating Model

Security & Compliance

ideal-reasoned-representations

multi-modal

Multi-Modal Message Support: Images and PDFs

1. Problem Statement

Current Limitations

Why This Matters

2. Goals / Non-Goals

Goals

Non-Goals

3. Design

3.1 Message Schema Extension

3.1.1 New Column: attachments

3.1.2 Attachment Object Schema

3.1.3 SQLAlchemy Model Change

3.1.4 Token Counting Update

3.1.5 Alembic Migration

3.2 Storage Backend Abstraction

3.2.1 Protocol Definition

3.2.2 Key Generation

3.2.3 S3 Backend

3.2.4 GCS Backend

3.2.5 Local Filesystem Backend

3.2.6 Storage Configuration

3.2.7 Storage Backend Factory

3.3 API Schema Changes

3.3.1 Attachment Input Schema

3.3.2 Updated MessageCreate Schema

3.3.3 Updated Message Response Schema

3.3.4 Upload Endpoint

3.4 Deriver Integration

3.4.1 Vision Description Pipeline

3.4.2 Vision Model Configuration

3.4.3 Vision Description Function

3.4.4 Deriver Prompt Modification

3.4.5 Deriver Pre-Processing Step

3.4.6 PDF Processing at Upload Time

3.5 SDK Changes

3.5.1 Python SDK

New Type: Attachment

Upload Method on Honcho Client

Updated Message Creation

Usage Example

3.5.2 TypeScript SDK

New Type: Attachment

Upload Method

Updated MessageInput

3.6 Security

3.6.1 MIME Type Allowlist

3.6.2 Magic Byte Verification

3.6.3 File Size Limits

3.6.4 Filename Sanitization

3.6.5 No Executable File Types

3.6.6 Storage URL Validation

3.6.7 Security Checklist

3.7 CloudEvents for Billing

3.7.1 StorageUploadEvent

3.8 Existing Upload Endpoint Refactor

3.8.1 New Behavior

3.8.2 Backward Compatibility

4. Migration Plan

4.1 Database Migration

4.2 Configuration Migration

4.3 SDK Migration

5. Implementation Phases

Phase 1: Storage Backend + Schema (1 week)

Phase 2: Upload API Endpoint (3 days)

Phase 3: Deriver Vision Integration (1 week)

Phase 4: CloudEvents + Telemetry (2 days)

Phase 5: SDK Updates (1 week)

Phase 6: GCS Backend (deferred)

6. Files to Modify (Complete List)

Server (repos/honcho/)

Python SDK (repos/honcho/sdks/python/)

3.1.1 New Column: `attachments`

New Type: `Attachment`

New Type: `Attachment`