feat: persist session events to DB and add replay endpoint

Stream-json events are now batch-persisted to a session_events table
(JSONB) during SSE streaming. Completed sessions can be replayed via
GET /sessions/{id}/events. The streaming pipeline is refactored into
layered generators: stream_events() -> stream_and_persist() for
clean separation of concerns.

- Add SessionEvent model with (session_id, event_id) unique constraint
- Alembic migration for session_events table
- Refactor stream_output() into stream_events() + stream_and_persist()
- Batch INSERT with ON CONFLICT DO NOTHING for idempotent reconnects
- Add GET /events replay endpoint with SessionEventsListResponse schema
- Frontend: load stored events for terminal sessions (existingSessionId prop)
- Fix FakeDockerClient log_lines to include trailing newlines
- 14 new tests (stream_events, stream_and_persist, get_session_events)

Author: mariuspruvot

CLAUDE.md

@@ -47,6 +47,7 @@ infra/
 - **Container orchestration**: `container` module manages ephemeral Docker lifecycle, credential injection, result relay
 - **Skills as agents**: each skill is a self-contained folder with workflow definitions, mounted into containers
 - **SSE passthrough**: backend relays container output to frontend (no AI response generation in backend)
+- **Session persistence**: stream-json events are batch-persisted to `session_events` table (JSONB) during SSE streaming via `stream_and_persist()`. Completed sessions can be replayed from `GET /sessions/{id}/events`. The streaming pipeline is layered: `stream_events()` (raw tuples) -> `stream_and_persist()` (SSE + DB writes) or `stream_output()` (SSE only, for tests).
 - **Stream-json protocol**: containers emit NDJSON with 5 event types: `system` (init/retry), `assistant` (one event per content block — thinking/text/tool_use), `user` (tool_result), `result` (session end + metadata), `rate_limit_event`. The `result.result` field duplicates the last assistant text — only display assistant events, use result for status only. No `--include-partial-messages` flag, so no `stream_event` deltas.
 - **API prefix**: all routes under `/api/v1`
 - **Admin panel**: SQLAdmin at `/admin`, configured in `admin/views.py`
@@ -76,6 +77,8 @@ cd apps/api && uv run alembic revision --autogenerate -m "description"  # New mi
 
 - Tests use `AsyncClient` with `ASGITransport` (no real server)
 - `conftest.py` sets env vars (DATABASE_URL, SECRET_KEY, etc.) **before** any app imports — order matters
+- **`FakeDockerClient` log_lines**: must include trailing `\n` for the line-buffering logic in `stream_events()` to split correctly
+- **`get_db_context` in tests**: tests that call `stream_and_persist()` need a `db_with_factory` fixture that calls `set_session_factory()` / `clear_session_factory()` — see `test_service.py` for the pattern
 
 ## Environment
 

apps/api/alembic/versions/f6a7b8c9d0e1_add_session_events_table.py

@@ -0,0 +1,42 @@
+"""add session_events table
+
+Revision ID: f6a7b8c9d0e1
+Revises: e5f6a7b8c9d0
+Create Date: 2026-04-17 18:00:00.000000
+
+Persists raw stream-json events (JSONB) for each container session
+so completed sessions can be replayed from the database.
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = "f6a7b8c9d0e1"
+down_revision: str | None = "e5f6a7b8c9d0"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "session_events",
+        sa.Column("session_id", sa.Uuid(), nullable=False),
+        sa.Column("event_id", sa.Integer(), nullable=False),
+        sa.Column("data", postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(["session_id"], ["container_sessions.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("session_id", "event_id", name="uq_session_events_session_event"),
+    )
+    op.create_index(op.f("ix_session_events_session_id"), "session_events", ["session_id"])
+
+
+def downgrade() -> None:
+    op.drop_index(op.f("ix_session_events_session_id"), table_name="session_events")
+    op.drop_table("session_events")

apps/api/src/helprs/modules/container/models.py

@@ -4,7 +4,8 @@
 import uuid
 from datetime import datetime
 
-from sqlalchemy import DateTime, Enum, ForeignKey, Integer, String
+from sqlalchemy import DateTime, Enum, ForeignKey, Integer, String, UniqueConstraint
+from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.orm import Mapped, mapped_column
 
 from helprs.core.database import Base
@@ -47,3 +48,22 @@ class ContainerSession(Base):
     )
     started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
     completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+
+
+class SessionEvent(Base):
+    """A single stream-json event persisted from a container session.
+
+    Stores raw NDJSON events (assistant, system, user, result) as JSONB
+    so completed sessions can be replayed from the database.
+    """
+
+    __tablename__ = "session_events"
+    __table_args__ = (UniqueConstraint("session_id", "event_id", name="uq_session_events_session_event"),)
+
+    session_id: Mapped[uuid.UUID] = mapped_column(
+        ForeignKey("container_sessions.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    event_id: Mapped[int] = mapped_column(Integer, nullable=False)
+    data: Mapped[dict] = mapped_column(JSONB, nullable=False)

apps/api/src/helprs/modules/container/router.py

@@ -14,17 +14,20 @@
     CreateSessionRequest,
     SendMessageRequest,
     SendMessageResponse,
+    SessionEventResponse,
+    SessionEventsListResponse,
     StopSessionResponse,
 )
 from helprs.modules.container.service import (
     AioDockerClient,
     ContainerStatus,
     create_session,
+    get_session_events,
     get_session_or_404,
     send_message,
     start_container,
     stop_container,
-    stream_output,
+    stream_and_persist,
 )
 from helprs.modules.installation.service import get_byok_config, mint_installation_token
 
@@ -146,7 +149,7 @@ async def stream_container_output(
 
     async def _event_stream():
         try:
-            async for event in stream_output(docker, cs.container_id, offset=offset):
+            async for event in stream_and_persist(docker, cs.container_id, session_id=session_id, offset=offset):
                 yield event
         finally:
             await docker.close()
@@ -162,6 +165,26 @@ async def _event_stream():
     )
 
 
+@router.get("/sessions/{session_id}/events", response_model=SessionEventsListResponse)
+@limiter.limit("30/minute")
+async def get_session_events_endpoint(
+    session_id: UUID,
+    request: Request,
+    db: DbSession,
+):
+    """Retrieve persisted stream-json events for a session.
+
+    Returns all events ordered by ``event_id``, suitable for replaying
+    completed sessions in the frontend without an SSE connection.
+    """
+    events = await get_session_events(db, session_id)
+    return SessionEventsListResponse(
+        session_id=session_id,
+        events=[SessionEventResponse.model_validate(e) for e in events],
+        total=len(events),
+    )
+
+
 @router.post("/sessions/{session_id}/message", response_model=SendMessageResponse)
 @limiter.limit("30/minute")
 async def send_session_message(

apps/api/src/helprs/modules/container/schemas.py

@@ -85,3 +85,21 @@ class StopSessionResponse(BaseModel):
     id: uuid.UUID
     status: str
     message: str
+
+
+class SessionEventResponse(BaseModel):
+    """A single persisted stream-json event."""
+
+    model_config = {"from_attributes": True}
+
+    event_id: int
+    data: dict
+    created_at: datetime
+
+
+class SessionEventsListResponse(BaseModel):
+    """All persisted events for a session."""
+
+    session_id: uuid.UUID
+    events: list[SessionEventResponse]
+    total: int

apps/api/src/helprs/modules/container/service.py

@@ -22,8 +22,9 @@
 import structlog
 from sqlalchemy import select
 
+from helprs.core.database import get_db_context
 from helprs.core.exceptions import ExternalServiceError, NotFoundError
-from helprs.modules.container.models import ContainerSession, ContainerStatus
+from helprs.modules.container.models import ContainerSession, ContainerStatus, SessionEvent
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator
@@ -294,25 +295,17 @@ async def start_container(
     return cs
 
 
-async def stream_output(
+async def stream_events(
     docker: DockerClient,
     container_id: str,
-    offset: int = 0,
-) -> AsyncIterator[str]:
-    """Async generator yielding container log lines as SSE events.
-
-    Docker may split long stdout lines (e.g. stream-json tool_result
-    events with full file contents) across multiple log frames.  We
-    buffer chunks and only emit complete newline-delimited lines so
-    the frontend always receives valid, parseable JSON per SSE event.
+) -> AsyncIterator[tuple[int, str]]:
+    """Yield ``(event_id, raw_line)`` tuples from container log stream.
 
-    Each event includes an incrementing ``id:`` field so that clients
-    can resume from the last received event via the ``offset`` query
-    parameter (number of events to skip).
+    Handles Docker log frame buffering (long lines split across chunks),
+    newline splitting, and keepalive signaling.
 
-    When the container is quiet (Claude is thinking), the Docker log
-    stream produces no data.  We send SSE comments (``:``) every
-    ``KEEPALIVE_INTERVAL`` seconds to prevent idle-timeout disconnects.
+    Yields ``(0, "")`` as a sentinel for keepalive intervals (no data
+    from container for 15 seconds — Claude is thinking).
     """
     keepalive_interval = 15.0
     event_id = 0
@@ -329,19 +322,121 @@ async def stream_output(
                 if not line:
                     continue
                 event_id += 1
-                if event_id <= offset:
-                    continue
-                yield f"id: {event_id}\ndata: {line}\n\n"
+                yield (event_id, line)
         except TimeoutError:
-            # No data from container — send SSE keepalive comment to
-            # prevent the HTTP connection from being closed by proxies
-            # or the browser.  SSE comments are silently ignored by
-            # EventSource clients.
-            yield ": keepalive\n\n"
+            yield (0, "")
         except StopAsyncIteration:
             break
 
 
+async def stream_output(
+    docker: DockerClient,
+    container_id: str,
+    offset: int = 0,
+) -> AsyncIterator[str]:
+    """Yield SSE-formatted events without persistence.
+
+    Thin wrapper over :func:`stream_events` that formats tuples as SSE.
+    Used by tests and code paths that don't need DB persistence.
+    """
+    async for event_id, line in stream_events(docker, container_id):
+        if event_id == 0:
+            yield ": keepalive\n\n"
+            continue
+        if event_id <= offset:
+            continue
+        yield f"id: {event_id}\ndata: {line}\n\n"
+
+
+_PERSIST_BATCH_SIZE = 10
+
+
+async def stream_and_persist(
+    docker: DockerClient,
+    container_id: str,
+    session_id: UUID,
+    offset: int = 0,
+    batch_size: int = _PERSIST_BATCH_SIZE,
+) -> AsyncIterator[str]:
+    """Stream container output as SSE while persisting events to the DB.
+
+    Wraps :func:`stream_events`, yields the same SSE format as
+    :func:`stream_output`, but also batch-inserts events into the
+    ``session_events`` table via :func:`get_db_context`.
+
+    Events at or below *offset* are skipped for SSE output but still
+    persisted (they may not have been flushed before a prior disconnect).
+    Duplicate ``(session_id, event_id)`` pairs are silently ignored via
+    ``ON CONFLICT DO NOTHING``.
+    """
+    from sqlalchemy.dialects.postgresql import insert as pg_insert
+
+    pending: list[tuple[int, dict]] = []
+
+    async def _flush() -> None:
+        if not pending:
+            return
+        batch = pending[:]
+        pending.clear()
+        try:
+            async with get_db_context() as db:
+                stmt = pg_insert(SessionEvent.__table__).values(
+                    [{"session_id": session_id, "event_id": eid, "data": data} for eid, data in batch]
+                )
+                stmt = stmt.on_conflict_do_nothing(
+                    constraint="uq_session_events_session_event",
+                )
+                await db.execute(stmt)
+        except Exception:
+            await logger.aexception(
+                "event_persist_failed",
+                session_id=str(session_id),
+                batch_size=len(batch),
+            )
+
+    async for event_id, line in stream_events(docker, container_id):
+        if event_id == 0:
+            # Keepalive — flush pending if any
+            await _flush()
+            yield ": keepalive\n\n"
+            continue
+
+        # Parse to JSONB-safe dict
+        try:
+            parsed = json.loads(line)
+        except json.JSONDecodeError:
+            parsed = {"_raw": line}
+
+        pending.append((event_id, parsed))
+
+        if len(pending) >= batch_size:
+            await _flush()
+
+        if event_id <= offset:
+            continue
+        yield f"id: {event_id}\ndata: {line}\n\n"
+
+    # Final flush for remaining events
+    await _flush()
+
+
+# ---------------------------------------------------------------------------
+# Session event retrieval
+# ---------------------------------------------------------------------------
+
+
+async def get_session_events(
+    db: AsyncSession,
+    session_id: UUID,
+) -> list[SessionEvent]:
+    """Return all persisted events for a session, ordered by event_id."""
+    await get_session_or_404(db, session_id)
+    result = await db.execute(
+        select(SessionEvent).where(SessionEvent.session_id == session_id).order_by(SessionEvent.event_id)
+    )
+    return list(result.scalars().all())
+
+
 async def send_message(
     db: AsyncSession,
     session_id: UUID,