feat(api): session attachment resource with protected mount, quotas, and sweep

api/entrypoints/routers.py

@@ -165,6 +165,10 @@
 from oss.src.core.mounts.service import MountsService
 from oss.src.core.store.storage import ObjectStore
 from oss.src.core.sessions.mounts.service import SessionMountsService
+from oss.src.core.sessions.attachments.dtos import AttachmentLimits
+from oss.src.core.sessions.attachments.service import SessionAttachmentsService
+from oss.src.dbs.postgres.sessions.attachments.dao import SessionAttachmentsDAO
+from oss.src.tasks.asyncio.sessions.attachment_sweep import attachment_sweep_loop
 from oss.src.apis.fastapi.mounts.router import MountsRouter
 
 # Session streams
@@ -271,6 +275,16 @@ async def lifespan(*args, **kwargs):
         orphan_sweep_loop(_transactions_engine, _lock_engine)
     )
 
+    _attachment_sweep_task = asyncio.create_task(
+        attachment_sweep_loop(
+            attachments_dao=session_attachments_dao,
+            original_store=mounts_service,
+            lock_engine=_lock_engine,
+            pending_ttl_seconds=env.agenta.sessions.attachments.pending_ttl_seconds,
+            unreferenced_ttl_seconds=env.agenta.sessions.attachments.unreferenced_ttl_seconds,
+            sweep_interval_seconds=env.agenta.sessions.attachments.sweep_interval_seconds,
+        )
+    )
     # Best-effort: ingestion re-resolves on demand if this fails.
     if env.composio.enabled:
         try:
@@ -282,7 +296,16 @@ async def lifespan(*args, **kwargs):
 
     yield
 
-    _orphan_sweep_task.cancel()
+    for task in (
+        _orphan_sweep_task,
+        _attachment_sweep_task,
+    ):
+        task.cancel()
+    await asyncio.gather(
+        _orphan_sweep_task,
+        _attachment_sweep_task,
+        return_exceptions=True,
+    )
 
     await _triggers_broker.shutdown()
 
@@ -559,6 +582,7 @@ async def lifespan(*args, **kwargs):
 
 connections_dao = ConnectionsDAO(engine=_transactions_engine)
 mounts_dao = MountsDAO(engine=_transactions_engine)
+session_attachments_dao = SessionAttachmentsDAO(engine=_transactions_engine)
 
 # SERVICES ---------------------------------------------------------------------
 
@@ -883,6 +907,21 @@ async def _dispatch_detached_run(*, project_id, user_id, request) -> str:
     mounts_service=mounts_service,
 )
 
+session_attachments_service = SessionAttachmentsService(
+    attachments_dao=session_attachments_dao,
+    original_store=mounts_service,
+    limits=AttachmentLimits(
+        max_image_bytes=env.agenta.sessions.attachments.max_image_bytes,
+        max_audio_bytes=env.agenta.sessions.attachments.max_audio_bytes,
+        max_document_bytes=env.agenta.sessions.attachments.max_document_bytes,
+        max_other_bytes=env.agenta.sessions.attachments.max_other_bytes,
+        max_per_session_count=env.agenta.sessions.attachments.max_per_session_count,
+        max_per_session_bytes=env.agenta.sessions.attachments.max_per_session_bytes,
+        max_pending_per_session=env.agenta.sessions.attachments.max_pending_per_session,
+        pending_ttl_seconds=env.agenta.sessions.attachments.pending_ttl_seconds,
+    ),
+)
+
 _t_services_done = time.perf_counter() - _t_services
 print(f"[STARTUP] Service initialization completed (+{_t_services_done:.3f}s)")
 _t_routers = time.perf_counter()
@@ -1053,6 +1092,7 @@ async def _dispatch_detached_run(*, project_id, user_id, request) -> str:
     records_service=records_service,
     interactions_service=interactions_service,
     workflows_service=workflows_service,
+    attachments_service=session_attachments_service,
     session_mounts_service=session_mounts_service,
     mounts_service=mounts_service,
     turns_service=session_turns_service,
@@ -1482,6 +1522,12 @@ async def _dispatch_detached_run(*, project_id, user_id, request) -> str:
     tags=["Mounts"],
 )
 
+app.include_router(
+    router=sessions.attachments.router,
+    prefix="/sessions",
+    tags=["Sessions"],
+)
+
 app.include_router(
     router=sessions.mounts.router,
     prefix="/sessions",

api/oss/databases/postgres/migrations/core_oss/versions/oss000000020_add_session_attachments.py

@@ -0,0 +1,120 @@
+"""add session attachments
+
+Revision ID: oss000000020
+Revises: oss000000019
+Create Date: 2026-07-31 00:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "oss000000020"
+down_revision: Union[str, None] = "oss000000019"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "mounts",
+        sa.Column("purpose", sa.String(), nullable=True),
+    )
+    op.create_table(
+        "session_attachments",
+        sa.Column("id", sa.UUID(as_uuid=True), nullable=False),
+        sa.Column("project_id", sa.UUID(as_uuid=True), nullable=False),
+        sa.Column("session_id", sa.String(), nullable=False),
+        sa.Column("mount_id", sa.UUID(as_uuid=True), nullable=False),
+        sa.Column("path", sa.String(), nullable=False),
+        sa.Column("filename", sa.String(), nullable=False),
+        sa.Column("media_type", sa.String(), nullable=False),
+        sa.Column("size", sa.BigInteger(), nullable=False),
+        sa.Column("kind", sa.String(), nullable=False),
+        sa.Column("state", sa.String(), nullable=False),
+        sa.Column("idempotency_key", sa.String(), nullable=False),
+        sa.Column("content_digest", sa.String(), nullable=False),
+        sa.Column("referenced_at", sa.TIMESTAMP(timezone=True), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.TIMESTAMP(timezone=True),
+            server_default=sa.func.current_timestamp(),
+            nullable=True,
+        ),
+        sa.Column("updated_at", sa.TIMESTAMP(timezone=True), nullable=True),
+        sa.Column("deleted_at", sa.TIMESTAMP(timezone=True), nullable=True),
+        sa.Column("created_by_id", sa.UUID(as_uuid=True), nullable=True),
+        sa.Column("updated_by_id", sa.UUID(as_uuid=True), nullable=True),
+        sa.Column("deleted_by_id", sa.UUID(as_uuid=True), nullable=True),
+        sa.CheckConstraint(
+            "state IN ('pending', 'ready', 'deleting')",
+            name="ck_session_attachments_state",
+        ),
+        sa.CheckConstraint(
+            "kind IN ('image', 'audio', 'document', 'other')",
+            name="ck_session_attachments_kind",
+        ),
+        sa.ForeignKeyConstraint(
+            ["project_id"],
+            ["projects.id"],
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["project_id", "mount_id"],
+            ["mounts.project_id", "mounts.id"],
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("project_id", "id"),
+        sa.UniqueConstraint(
+            "project_id",
+            "session_id",
+            "idempotency_key",
+            name="uq_session_attachments_idempotency",
+        ),
+    )
+    op.create_index(
+        "ix_session_attachments_session_state",
+        "session_attachments",
+        ["project_id", "session_id", "state"],
+    )
+    op.create_index(
+        "ix_session_attachments_project_mount",
+        "session_attachments",
+        ["project_id", "mount_id"],
+    )
+    op.create_index(
+        "ix_session_attachments_pending_created",
+        "session_attachments",
+        ["created_at"],
+        postgresql_where=sa.text("state IN ('pending', 'deleting')"),
+    )
+    op.create_index(
+        "ix_session_attachments_ready_unreferenced",
+        "session_attachments",
+        ["created_at"],
+        postgresql_where=sa.text("state = 'ready' AND referenced_at IS NULL"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        "ix_session_attachments_ready_unreferenced",
+        table_name="session_attachments",
+    )
+    op.drop_index(
+        "ix_session_attachments_pending_created",
+        table_name="session_attachments",
+    )
+    op.drop_index(
+        "ix_session_attachments_session_state",
+        table_name="session_attachments",
+    )
+    op.drop_index(
+        "ix_session_attachments_project_mount",
+        table_name="session_attachments",
+    )
+    op.drop_table("session_attachments")
+    op.drop_column("mounts", "purpose")

api/oss/src/apis/fastapi/mounts/models.py

@@ -1,26 +1,36 @@
-from typing import List, Optional
+from typing import Any, Dict, List, Optional
 from uuid import UUID
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 
 from oss.src.core.mounts.dtos import (
     Mount,
-    MountCreate,
     MountCredentials,
     MountEdit,
     MountFile,
+    MountFlags,
     MountQuery,
 )
-from oss.src.core.shared.dtos import Windowing
+from oss.src.core.shared.dtos import Header, Slug, Windowing
 
 
 # ---------------------------------------------------------------------------
 # Request models
 # ---------------------------------------------------------------------------
 
 
+class PublicMountCreate(Slug, Header):
+    model_config = ConfigDict(extra="forbid")
+
+    session_id: Optional[str] = None
+    agent_id: Optional[str] = None
+    flags: MountFlags = Field(default_factory=MountFlags)
+    tags: Optional[Dict[str, Any]] = None
+    meta: Optional[Dict[str, Any]] = None
+
+
 class MountCreateRequest(BaseModel):
-    mount: MountCreate
+    mount: PublicMountCreate
 
 
 class MountEditRequest(BaseModel):

api/oss/src/apis/fastapi/mounts/router.py

@@ -19,7 +19,7 @@
 from oss.src.core.access.permissions.service import check_action_access
 from oss.src.apis.fastapi.shared.exceptions import FORBIDDEN_EXCEPTION
 
-from oss.src.core.mounts.dtos import MountArchiveSource
+from oss.src.core.mounts.dtos import MountArchiveSource, MountCreate
 from oss.src.core.mounts.service import MountsService
 from oss.src.core.mounts.types import (
     MountArtifactIdInvalid,
@@ -29,6 +29,7 @@
     MountImmutableField,
     MountNameInvalid,
     MountNotFound,
+    MountProtected,
     MountPathInvalid,
     MountSlugConflict,
     MountSlugReserved,
@@ -112,6 +113,11 @@ async def wrapper(*args, **kwargs):
                     status_code=status.HTTP_404_NOT_FOUND,
                     detail=e.message,
                 ) from e
+            except MountProtected as e:
+                raise HTTPException(
+                    status_code=status.HTTP_404_NOT_FOUND,
+                    detail=MountNotFound().message,
+                ) from e
             except MountNotFound as e:
                 raise HTTPException(
                     status_code=status.HTTP_404_NOT_FOUND,
@@ -314,7 +320,7 @@ async def create_mount(
             project_id=UUID(request.state.project_id),
             user_id=UUID(str(request.state.user_id)),
             #
-            mount_create=body.mount,
+            mount_create=MountCreate(**body.mount.model_dump()),
         )
 
         return MountResponse(count=1, mount=mount)
@@ -435,6 +441,7 @@ async def edit_mount(
         return MountResponse(count=1, mount=mount)
 
     @intercept_exceptions()
+    @handle_mount_exceptions()
     async def archive_mount(
         self,
         request: Request,
@@ -457,6 +464,7 @@ async def archive_mount(
         return MountResponse(count=1, mount=mount)
 
     @intercept_exceptions()
+    @handle_mount_exceptions()
     async def unarchive_mount(
         self,
         request: Request,

api/oss/src/apis/fastapi/sessions/models.py

@@ -177,6 +177,34 @@ class SessionMountsResponse(BaseModel):
     mounts: List[SessionMount] = Field(default_factory=list)
 
 
+# ---------------------------------------------------------------------------
+# Attachments request/response models
+# ---------------------------------------------------------------------------
+
+
+class SessionAttachment(BaseModel):
+    attachment_id: UUID
+    filename: str
+    media_type: str
+    size: int
+    created_at: datetime
+
+
+class SessionAttachmentResponse(BaseModel):
+    count: int = 0
+    attachment: SessionAttachment
+
+
+class SessionAttachmentReferenceRequest(BaseModel):
+    session_id: str
+    attachment_ids: List[UUID] = Field(max_length=100)
+
+
+class SessionAttachmentsResponse(BaseModel):
+    count: int = 0
+    attachments: List[SessionAttachment] = Field(default_factory=list)
+
+
 # ---------------------------------------------------------------------------
 # Turns request/response models
 # ---------------------------------------------------------------------------