feat(sessions): sessions integration — storage rework, reliable approvals, and approval UX

api/entrypoints/routers.py

@@ -137,10 +137,6 @@
 from oss.src.core.ai_services.service import AIServicesService
 from oss.src.apis.fastapi.ai_services.router import AIServicesRouter
 
-from oss.src.dbs.postgres.sessions.states.dbes import SessionStateDBE  # noqa: F401
-from oss.src.dbs.postgres.sessions.states.dao import SessionStatesDAO
-from oss.src.core.sessions.states.service import SessionStatesService
-
 from oss.src.core.accounts.service import PlatformAdminAccountsService
 from oss.src.apis.fastapi.accounts.router import PlatformAdminAccountsRouter
 from oss.src.dbs.postgres.gateway.connections.dao import ConnectionsDAO
@@ -178,10 +174,15 @@
 from oss.src.tasks.asyncio.sessions.orphan_sweep import orphan_sweep_loop
 from oss.src.dbs.redis.shared.engine import get_lock_engine
 
+from oss.src.dbs.postgres.sessions.turns.dbes import SessionTurnDBE  # noqa: F401
+from oss.src.dbs.postgres.sessions.turns.dao import SessionTurnsDAO
+from oss.src.core.sessions.turns.service import SessionTurnsService
+
 # Interactions
 from oss.src.dbs.postgres.sessions.interactions.dbes import SessionInteractionDBE  # noqa: F401
 from oss.src.dbs.postgres.sessions.interactions.dao import SessionInteractionsDAO
 from oss.src.core.sessions.interactions.service import SessionInteractionsService
+from oss.src.core.sessions.service import SessionsService
 from oss.src.tasks.asyncio.sessions.interactions_dispatcher import (
     InteractionsDispatcher,
 )
@@ -549,6 +550,7 @@ async def lifespan(*args, **kwargs):
 evaluations_dao = EvaluationsDAO(engine=_transactions_engine)
 folders_dao = FoldersDAO(engine=_transactions_engine)
 session_streams_dao = SessionStreamsDAO(engine=_transactions_engine)
+session_turns_dao = SessionTurnsDAO(engine=_transactions_engine)
 
 connections_dao = ConnectionsDAO(engine=_transactions_engine)
 mounts_dao = MountsDAO(engine=_transactions_engine)
@@ -616,6 +618,10 @@ async def lifespan(*args, **kwargs):
     lock_engine=_lock_engine,
 )
 
+session_turns_service = SessionTurnsService(
+    turns_dao=session_turns_dao,
+)
+
 workflows_service = WorkflowsService(
     workflows_dao=workflows_dao,
     static_catalog=StaticWorkflowCatalog(),
@@ -1026,22 +1032,26 @@ async def _dispatch_detached_run(*, project_id, user_id, request) -> str:
     ai_services_service=ai_services_service,
 )
 
-# SESSION STATES ---------------------------------------------------------------
+# SESSIONS ---------------------------------------------------------------------
+# Session header rename (name/description) lives on the streams router
+# (PUT /sessions/streams/header) via streams_service.set_header.
 
-session_states_dao = SessionStatesDAO(engine=_transactions_engine)
-
-session_states_service = SessionStatesService(
-    session_states_dao=session_states_dao,
+sessions_service = SessionsService(
+    streams_service=session_streams_service,
+    turns_service=session_turns_service,
+    interactions_service=interactions_service,
+    mounts_service=mounts_service,
 )
 
 sessions = SessionsRouter(
     streams_service=session_streams_service,
-    states_service=session_states_service,
     records_service=records_service,
     interactions_service=interactions_service,
     workflows_service=workflows_service,
     session_mounts_service=session_mounts_service,
     mounts_service=mounts_service,
+    turns_service=session_turns_service,
+    sessions_service=sessions_service,
     respond_task=_interactions_worker.respond_interaction,
 )
 
@@ -1479,15 +1489,20 @@ async def _dispatch_detached_run(*, project_id, user_id, request) -> str:
     tags=["Sessions"],
 )
 
+app.include_router(
+    router=sessions.turns.router,
+    prefix="/sessions/turns",
+    tags=["Sessions"],
+)
+
 app.include_router(
     router=platform_admin_accounts.router,
     prefix="/admin",
     tags=["Admin"],
 )
 
 app.include_router(
-    router=sessions.states.router,
-    prefix="/sessions",
+    router=sessions.root.router,
     tags=["Sessions"],
 )
 

api/oss/databases/postgres/migrations/core/env.py

@@ -22,7 +22,6 @@
 import oss.src.dbs.postgres.users.dbes  # noqa: F401
 import oss.src.dbs.postgres.workflows.dbes  # noqa: F401
 import oss.src.dbs.postgres.webhooks.dbes  # noqa: F401
-import oss.src.dbs.postgres.sessions.states.dbes  # noqa: F401
 
 
 # this is the Alembic Config object, which provides

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

@@ -0,0 +1,98 @@
+"""add session_turns table
+
+Revision ID: oss000000014
+Revises: oss000000013
+Create Date: 2026-07-17 00:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+revision: str = "oss000000014"
+down_revision: Union[str, None] = "oss000000013"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "session_turns",
+        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("stream_id", sa.UUID(as_uuid=True), nullable=False),
+        sa.Column("turn_index", sa.Integer(), nullable=False),
+        sa.Column("harness_kind", sa.String(), nullable=False),
+        sa.Column("agent_session_id", sa.String(), nullable=True),
+        sa.Column("sandbox_id", sa.String(), nullable=True),
+        sa.Column(
+            "references",
+            postgresql.JSONB(none_as_null=True),
+            nullable=True,
+        ),
+        sa.Column("trace_id", sa.UUID(as_uuid=True), nullable=True),
+        # OTel span id: 64-bit / 16 hex chars, NOT a UUID (128-bit / 32 hex). Text, not UUID.
+        sa.Column("span_id", sa.String(), nullable=True),
+        sa.Column("start_time", sa.TIMESTAMP(timezone=True), nullable=True),
+        sa.Column("end_time", 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.ForeignKeyConstraint(
+            ["project_id"],
+            ["projects.id"],
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["project_id", "stream_id"],
+            ["session_streams.project_id", "session_streams.id"],
+            ondelete="NO ACTION",
+        ),
+        sa.PrimaryKeyConstraint("project_id", "id"),
+    )
+    op.create_index(
+        "ix_session_turns_project_id_session_id",
+        "session_turns",
+        ["project_id", "session_id"],
+    )
+    op.create_index(
+        "ix_session_turns_project_id_session_id_turn_index",
+        "session_turns",
+        ["project_id", "session_id", "turn_index"],
+        unique=True,
+    )
+    op.create_index(
+        "ix_session_turns_references",
+        "session_turns",
+        ["references"],
+        postgresql_using="gin",
+        postgresql_ops={"references": "jsonb_path_ops"},
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        "ix_session_turns_references",
+        table_name="session_turns",
+    )
+    op.drop_index(
+        "ix_session_turns_project_id_session_id_turn_index",
+        table_name="session_turns",
+    )
+    op.drop_index(
+        "ix_session_turns_project_id_session_id",
+        table_name="session_turns",
+    )
+    op.drop_table("session_turns")

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

@@ -0,0 +1,33 @@
+"""add name/description header to session_streams
+
+Revision ID: oss000000015
+Revises: oss000000014
+Create Date: 2026-07-17 00:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = "oss000000015"
+down_revision: Union[str, None] = "oss000000014"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "session_streams",
+        sa.Column("name", sa.String(), nullable=True),
+    )
+    op.add_column(
+        "session_streams",
+        sa.Column("description", sa.String(), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("session_streams", "description")
+    op.drop_column("session_streams", "name")

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

@@ -0,0 +1,59 @@
+"""add mounts agent_id column, index, and backfill
+
+Mirrors session_id: a bare, nullable varchar populated only for agent mounts
+(session mounts stay agent_id-null), with a partial index for querying "mounts
+for agent X" the same way session_id already supports "mounts for session X".
+
+Backfill: mounts are core DB, so a data migration is allowed here (unlike the
+tracing DB). Agent-mount slugs are minted as
+`__ag__agent__<canonical_artifact_id>__<name>` (`mint_agent_slug`,
+`core/mounts/service.py`) — the artifact id is the raw canonical (lowercase)
+UUID, not a hash, so it is deterministically recoverable straight out of the
+slug. Session-mount slugs (`__ag__session__<uuid5(session_id)>__<name>`) hash
+the session id, so no session_id backfill is possible or attempted here.
+
+Revision ID: oss000000016
+Revises: oss000000015
+Create Date: 2026-07-17 00:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = "oss000000016"
+down_revision: Union[str, None] = "oss000000015"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+_AGENT_SLUG_PREFIX = "__ag__agent__"
+
+
+def upgrade() -> None:
+    op.add_column("mounts", sa.Column("agent_id", sa.String(), nullable=True))
+
+    op.create_index(
+        "ix_mounts_project_id_agent_id",
+        "mounts",
+        ["project_id", "agent_id"],
+        unique=False,
+        postgresql_where=sa.text("agent_id IS NOT NULL"),
+    )
+
+    conn = op.get_bind()
+    conn.execute(
+        sa.text(
+            f"""
+            UPDATE mounts
+            SET agent_id = split_part(substr(slug, {len(_AGENT_SLUG_PREFIX) + 1}), '__', 1)
+            WHERE left(slug, {len(_AGENT_SLUG_PREFIX)}) = '{_AGENT_SLUG_PREFIX}'
+            """
+        )
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_mounts_project_id_agent_id", table_name="mounts")
+    op.drop_column("mounts", "agent_id")

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

@@ -0,0 +1,61 @@
+"""drop session_states table
+
+Superseded by the session_streams header (name/description) — the /sessions/states/
+router now reads/writes the merged stream row via streams_service. The standalone
+table, its DAO/service, and DBE are orphaned; drop the physical table.
+
+Revision ID: oss000000017
+Revises: oss000000016
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+revision: str = "oss000000017"
+down_revision: Union[str, None] = "oss000000016"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.drop_table("session_states")
+
+
+def downgrade() -> None:
+    op.create_table(
+        "session_states",
+        sa.Column("project_id", sa.UUID(), nullable=False),
+        sa.Column("id", sa.UUID(), nullable=False),
+        sa.Column("session_id", sa.String(), nullable=False),
+        sa.Column("sandbox_id", sa.String(), nullable=True),
+        sa.Column(
+            "data",
+            postgresql.JSON(astext_type=sa.Text()),
+            nullable=True,
+        ),
+        sa.Column("flags", postgresql.JSONB(none_as_null=True), nullable=True),
+        sa.Column("tags", postgresql.JSONB(none_as_null=True), nullable=True),
+        sa.Column("meta", postgresql.JSON(none_as_null=True), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.TIMESTAMP(timezone=True),
+            server_default=sa.text("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(), nullable=True),
+        sa.Column("updated_by_id", sa.UUID(), nullable=True),
+        sa.Column("deleted_by_id", sa.UUID(), nullable=True),
+        sa.ForeignKeyConstraint(["project_id"], ["projects.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("project_id", "id"),
+        sa.UniqueConstraint(
+            "project_id",
+            "session_id",
+            name="uq_session_states_project_session_id",
+        ),
+    )