feat(diagnostics): debug sessions — start, stop, the tag on every line, screenshots (E16 S1)

A session is one recording, capped at 60 minutes by the server. While it is
open, a request naming the caller's own session in X-Debug-Session is logged
at the session's level with every line tagged; a header naming anything else
is ignored. The browser's batches count a session as consent to its level and
keep the tag only on its own events. Screenshots are judged by their bytes,
capped in size and count, and served back to their owner only.

- debug_sessions / debug_screenshots tables, one open session per user
  (partial unique index), log_events.debug_session_id indexed
- /api/debug-sessions: start (joins a running one), stop, list, get, events,
  screenshots; /api/me reports the live session
- the hourly log job closes expired sessions and drops them with the logs
- logsink.serialize and routes._common.capped_body shared by their readers

backend/app/README.md

@@ -21,8 +21,8 @@ Three layers, by *who* runs the code — not by feature:
 
 3. **Flat `app/*.py` core modules — cross-cutting primitives.** `auth`, `sysconfig`, `quota`, `audit`,
    `notifications`, `userscope`, `realtime`, `ratelimit`, `progress`, `logs` (the one logging
-   configuration), `logsink` (its `log_events` table) and `diagnostics` (a user's consented log
-   level), … are single-file primitives many layers import. They depend on nothing above them (a core
+   configuration), `logsink` (its `log_events` table), `diagnostics` (a user's consented log
+   level) and `debugsession` (a recording that tags its user's lines, E16), … are single-file primitives many layers import. They depend on nothing above them (a core
    module must never import a `routes/…` module).
 
 **Extraction trigger: the SECOND consumer.** Keep new logic inline in its first caller. The moment a

backend/app/auth.py

@@ -4,7 +4,7 @@ from sqlalchemy import delete, func, select
 from sqlalchemy.orm import Session
 from starlette.requests import HTTPConnection
 
-from app import diagnostics
+from app import debugsession
 from app import email as email_mod
 from app import logs
 from app import mailtext
@@ -367,7 +367,11 @@ def _session_user(request: Request, db: Session, user_id: int | None, is_default
     # session middleware and this task's context. `request.state` is shared across middleware layers
     # (it lives on the ASGI scope), so the access-log line can read it back. The user's consented
     # diagnostics level rides along: their requests are logged — and stored — in that detail (S5).
-    logs.set_user(user.id, diagnostics.effective_level(user), user.diagnostics_until)
+    # So does a debug session the request names (E16): only the caller's own open one counts, and a
+    # header naming anything else is ignored rather than refused — a stale tab must keep working.
+    session = debugsession.from_header(db, user, request.headers.get(debugsession.HEADER))
+    level, until = debugsession.level_for(user, session)
+    logs.set_user(user.id, level, until, session.id if session else None)
     request.state.user_id = user.id
     return user
 

backend/app/debugsession.py

@@ -0,0 +1,261 @@
+"""Debug sessions (E16): Start → reproduce → Stop, and the evidence a bug report is built from.
+
+While a session is open its user's browser names it in `X-Debug-Session` on every request. The auth
+dependency accepts the header only for the caller's OWN open session (anything else is ignored, never
+an error — a stale tab must not break the app) and then logs that request at the session's level with
+every line tagged by its id; the browser tags its own events the same way. The issue is later built
+from exactly the rows carrying the tag, so what the review list shows is what gets sent.
+
+Starting a session IS the consent for its level, for its duration: the start dialog carries the
+consent text. It needs no change to the user's own diagnostics level, and none is made — when the
+session ends the level in force is simply the user's own again.
+
+A session is capped at `MAX_MINUTES`, set by the server at Start. An open one past its cap is over
+whether or not anything has closed it yet: it reads as over at once, a new Start closes it, and the
+hourly log job sweeps the rest. Sessions go with the logs they point at, after `log_retention_days`.
+
+Core layer: imports `models`, `diagnostics`.
+"""
+
+import secrets
+from datetime import datetime, timedelta, timezone
+from enum import StrEnum
+from typing import Literal, get_args
+
+from sqlalchemy import delete, func, select, update
+from sqlalchemy.orm import Session, defer
+
+from app import diagnostics
+from app.models import DebugScreenshot, DebugSession, LogEvent, User
+
+HEADER = "x-debug-session"
+MAX_MINUTES = 60
+SCREENSHOT_MAX_BYTES = 10 * 1024 * 1024
+SCREENSHOTS_MAX = 8
+# Rows an issue can carry. A 60-minute session at Teljes is a few thousand rows; past this the
+# bundle says how many it left out rather than growing without bound.
+BUNDLE_MAX = 5_000
+
+SessionLevel = Literal["DEBUG", "TRACE"]
+SESSION_LEVELS: tuple[str, ...] = get_args(SessionLevel)
+# Detail order of the diagnostics levels: a session raises the level in force, never lowers it.
+_DETAIL = {"INFO": 0, "DEBUG": 1, "TRACE": 2}
+_ID_LEN = 16
+
+
+class EndReason(StrEnum):
+    STOPPED = "stopped"
+    EXPIRED = "expired"
+
+
+def _is_live(session: DebugSession, now: datetime) -> bool:
+    return session.ended_at is None and session.until > now
+
+
+def open_session(db: Session, user: User) -> DebugSession | None:
+    """The user's live session, or None. Read-only: one past its cap is over and reads as None;
+    `start` and the hourly sweep are what close it."""
+    session = _unclosed(db, user)
+    if session is None or not _is_live(session, datetime.now(timezone.utc)):
+        return None
+    return session
+
+
+def _unclosed(db: Session, user: User) -> DebugSession | None:
+    return db.scalar(
+        select(DebugSession).where(DebugSession.user_id == user.id, DebugSession.ended_at.is_(None))
+    )
+
+
+def start(db: Session, user: User, level: SessionLevel) -> DebugSession:
+    """Open a session, or return the live one — a second tab pressing Start joins the recording
+    rather than ending it. One past its cap is closed first: the one-open-per-user index would
+    refuse the new row beside it. Flushes; the caller commits. Two concurrent Starts can still both
+    get here: the index refuses the second, and the caller retries once."""
+    now = datetime.now(timezone.utc)
+    current = _unclosed(db, user)
+    if current is not None:
+        if _is_live(current, now):
+            return current
+        end(current, EndReason.EXPIRED)
+        db.flush()
+    session = DebugSession(
+        id=secrets.token_hex(_ID_LEN // 2),
+        user_id=user.id,
+        level=level,
+        started_at=now,
+        until=now + timedelta(minutes=MAX_MINUTES),
+    )
+    db.add(session)
+    db.flush()
+    return session
+
+
+def end(session: DebugSession, reason: EndReason) -> None:
+    """Close a session. An expired one is dated at its cap, not at whenever it was noticed."""
+    if session.ended_at is not None:
+        return
+    session.ended_at = (
+        session.until
+        if reason is EndReason.EXPIRED
+        else min(datetime.now(timezone.utc), session.until)
+    )
+    session.ended_reason = reason
+
+
+def from_header(db: Session, user: User, value: str | None) -> DebugSession | None:
+    """The session a request names, if it is the caller's own and still live; else None."""
+    if not value or len(value) != _ID_LEN or not all(c in "0123456789abcdef" for c in value):
+        return None
+    session = db.get(DebugSession, value)
+    if (
+        session is None
+        or session.user_id != user.id
+        or not _is_live(session, datetime.now(timezone.utc))
+    ):
+        return None
+    return session
+
+
+def level_for(user: User, session: DebugSession | None) -> tuple[str, datetime | None]:
+    """The diagnostics level (and its expiry) a request is logged at: the user's own, raised to the
+    session's while one is named — whichever records more."""
+    own = diagnostics.effective_level(user)
+    until = user.diagnostics_until if own != diagnostics.BASE else None
+    if session is None or _DETAIL[session.level] <= _DETAIL[own]:
+        return own, until
+    return session.level, session.until
+
+
+def serialize(session: DebugSession | None, screenshots: int = 0) -> dict | None:
+    if session is None:
+        return None
+    return {
+        "id": session.id,
+        "level": session.level,
+        "started_at": session.started_at.isoformat(),
+        "until": session.until.isoformat(),
+        "ended_at": session.ended_at.isoformat() if session.ended_at else None,
+        "ended_reason": session.ended_reason,
+        "issue_url": session.issue_url,
+        "screenshots": screenshots,
+    }
+
+
+def screenshot_count(db: Session, session: DebugSession) -> int:
+    return (
+        db.scalar(
+            select(func.count())
+            .select_from(DebugScreenshot)
+            .where(DebugScreenshot.session_id == session.id)
+        )
+        or 0
+    )
+
+
+# --- screenshots ---------------------------------------------------------------------------
+# By content, not by the declared type: what is stored is served back to a browser as an image.
+_SIGNATURES = (
+    (b"\x89PNG\r\n\x1a\n", "image/png"),
+    (b"\xff\xd8\xff", "image/jpeg"),
+)
+
+
+def sniff(data: bytes) -> str | None:
+    """The image type `data` really is — PNG, JPEG or WebP — or None."""
+    for magic, mime in _SIGNATURES:
+        if data.startswith(magic):
+            return mime
+    if len(data) >= 12 and data[:4] == b"RIFF" and data[8:12] == b"WEBP":
+        return "image/webp"
+    return None
+
+
+class ScreenshotRefused(Exception):
+    """A screenshot that cannot be kept; `code` is the refusal's machine word."""
+
+    def __init__(self, code: str) -> None:
+        super().__init__(code)
+        self.code = code
+
+
+def add_screenshot(db: Session, session: DebugSession, data: bytes) -> DebugScreenshot:
+    """Keep one screenshot. Flushes; the caller commits. Refused once the issue was sent (it can no
+    longer carry it), past the per-session count, or when it is not an image we can show."""
+    if session.issue_url is not None:
+        raise ScreenshotRefused("already_sent")
+    if not data:
+        raise ScreenshotRefused("empty")
+    mime = sniff(data)
+    if mime is None:
+        raise ScreenshotRefused("not_an_image")
+    if screenshot_count(db, session) >= SCREENSHOTS_MAX:
+        raise ScreenshotRefused("too_many")
+    shot = DebugScreenshot(
+        session_id=session.id,
+        created_at=datetime.now(timezone.utc),
+        mime=mime,
+        size=len(data),
+        data=data,
+    )
+    db.add(shot)
+    db.flush()
+    return shot
+
+
+def screenshots(db: Session, session: DebugSession) -> list[DebugScreenshot]:
+    """The session's screenshots, oldest first, WITHOUT their bytes loaded."""
+    return list(
+        db.scalars(
+            select(DebugScreenshot)
+            .where(DebugScreenshot.session_id == session.id)
+            .options(defer(DebugScreenshot.data))
+            .order_by(DebugScreenshot.id)
+        )
+    )
+
+
+# --- the evidence --------------------------------------------------------------------------
+def bundle(db: Session, session: DebugSession) -> tuple[list[LogEvent], int]:
+    """The session's log rows, browser and server interleaved in time, capped at BUNDLE_MAX — and
+    how many rows it has in all, so a cut bundle can say so."""
+    rows = list(
+        db.scalars(
+            select(LogEvent)
+            .where(LogEvent.debug_session_id == session.id)
+            .order_by(LogEvent.ts, LogEvent.id)
+            .limit(BUNDLE_MAX)
+        )
+    )
+    total = len(rows)
+    if total == BUNDLE_MAX:
+        total = (
+            db.scalar(
+                select(func.count())
+                .select_from(LogEvent)
+                .where(LogEvent.debug_session_id == session.id)
+            )
+            or total
+        )
+    return rows, total
+
+
+# --- housekeeping (the hourly log job) -----------------------------------------------------
+def sweep(db: Session) -> int:
+    """Stage closing every open session past its cap; returns how many."""
+    now = datetime.now(timezone.utc)
+    result = db.execute(
+        update(DebugSession)
+        .where(DebugSession.ended_at.is_(None), DebugSession.until <= now)
+        .values(ended_at=DebugSession.until, ended_reason=EndReason.EXPIRED.value)
+        .execution_options(synchronize_session=False)
+    )
+    return result.rowcount or 0
+
+
+def gc(db: Session, retention_days: int) -> int:
+    """Stage deleting sessions (and by cascade their screenshots) started before the log retention
+    window — the rows they would bundle are gone by then too. Returns how many."""
+    cutoff = datetime.now(timezone.utc) - timedelta(days=max(1, retention_days))
+    result = db.execute(delete(DebugSession).where(DebugSession.started_at < cutoff))
+    return result.rowcount or 0

backend/app/logs.py

@@ -100,12 +100,14 @@ class _RequestScope:
     request; every copied context shares the same object, so what the dependency writes is seen by
     the handler and by the request line itself."""
 
-    __slots__ = ("user_id", "level")
+    __slots__ = ("user_id", "level", "debug_session_id")
 
     def __init__(self) -> None:
         self.user_id: int | None = None
         # The user's consented diagnostics level as a logging number; None at the base level.
         self.level: int | None = None
+        # The user's own open debug session this request named (E16), else None.
+        self.debug_session_id: str | None = None
 
 
 _scope: contextvars.ContextVar[_RequestScope | None] = contextvars.ContextVar(
@@ -113,11 +115,19 @@ _scope: contextvars.ContextVar[_RequestScope | None] = contextvars.ContextVar(
 )
 
 
-def set_user(user_id: int | None, level: str = "INFO", until: datetime | None = None) -> None:
+def set_user(
+    user_id: int | None,
+    level: str = "INFO",
+    until: datetime | None = None,
+    debug_session: str | None = None,
+) -> None:
     """Attribute the rest of this request's log lines to `user_id`, and log them at the user's
     consented diagnostics `level` until `until` (E22 S5; "INFO" = no elevation). Called from the
     auth dependency — the one place the session is actually resolved into an account — rather than
-    from the request middleware, which runs OUTSIDE the session middleware."""
+    from the request middleware, which runs OUTSIDE the session middleware.
+
+    `debug_session` tags every line with the user's open debug session (E16). The caller has
+    already checked it is theirs and open; a session's level arrives here as `level`/`until`."""
     _user_id.set(user_id)
     scope = _scope.get()
     if scope is None:  # outside log_requests (a script, a unit test): this context only
@@ -125,6 +135,7 @@ def set_user(user_id: int | None, level: str = "INFO", until: datetime | None =
         _scope.set(scope)
     elevated = level != "INFO" and until is not None and until > datetime.now(timezone.utc)
     scope.user_id = user_id
+    scope.debug_session_id = debug_session
     scope.level = resolve_level(level) if elevated else None
     if user_id is not None:
         _track_elevation(user_id, scope.level, until)
@@ -328,6 +339,12 @@ class ContextFilter(logging.Filter):
         scope = _scope.get()
         if scope is not None and scope.user_id is not None and not hasattr(record, "user_id"):
             record.user_id = scope.user_id
+        if (
+            scope is not None
+            and scope.debug_session_id is not None
+            and not hasattr(record, "debug_session_id")
+        ):
+            record.debug_session_id = scope.debug_session_id
         return True
 
 

backend/app/logsink.py

@@ -44,6 +44,7 @@ from app import logs
 from app.config import settings
 from app.db import engine
 from app.models import LogEvent
+from app.utils import iso
 
 QUEUE_MAX = 10_000
 BATCH_MAX = 500
@@ -343,3 +344,24 @@ def gc(db: Session, retention_days: int) -> int:
     cutoff = datetime.now(timezone.utc) - timedelta(days=max(1, retention_days))
     result = db.execute(delete(LogEvent).where(LogEvent.ts < cutoff))
     return result.rowcount or 0
+
+
+def serialize(row: LogEvent) -> dict:
+    """A stored row as the API sends it — one shape for every reader: the admin console (E22.6) and
+    a debug session's evidence list (E16)."""
+    return {
+        "id": row.id,
+        "ts": iso(row.ts),
+        "level": row.level,
+        "source": row.source,
+        "logger": row.logger,
+        "event": row.event,
+        "msg": row.msg,
+        "attrs": row.attrs,
+        "exc": row.exc,
+        "request_id": row.request_id,
+        "user_id": row.user_id,
+        "job_id": row.job_id,
+        "debug_session_id": row.debug_session_id,
+        "version": row.version,
+    }

backend/app/main.py

@@ -37,6 +37,7 @@ from app.routes import (
     auth as auth_routes,
     channels,
     config as config_routes,
+    debug_sessions as debug_sessions_routes,
     diagnostics as diagnostics_routes,
     discover as discover_routes,
     downloads,
@@ -273,6 +274,7 @@ app.include_router(discover_routes.router)
 app.include_router(search_routes.router)
 app.include_router(me.router)
 app.include_router(diagnostics_routes.router)
+app.include_router(debug_sessions_routes.router)
 app.include_router(notifications.router)
 app.include_router(messages.router)
 app.include_router(channels.router)

backend/app/models.py

@@ -12,6 +12,7 @@ from sqlalchemy import (
     Index,
     Integer,
     JSON,
+    LargeBinary,
     SmallInteger,
     String,
     Text,
@@ -565,7 +566,15 @@ class LogEvent(Base):
     the retention job (`log_retention_days`) bounds whatever another process writes late."""
 
     __tablename__ = "log_events"
-    __table_args__ = (Index("ix_log_events_user_ts", "user_id", "ts"),)
+    __table_args__ = (
+        Index("ix_log_events_user_ts", "user_id", "ts"),
+        # E16: an issue's bundle is `WHERE debug_session_id = …`; nearly every row has none.
+        Index(
+            "ix_log_events_debug_session",
+            "debug_session_id",
+            postgresql_where="debug_session_id IS NOT NULL",
+        ),
+    )
 
     id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
     ts: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True)
@@ -584,6 +593,50 @@ class LogEvent(Base):
     version: Mapped[str | None] = mapped_column(String(64))
 
 
+class DebugSession(Base):
+    """One recording: Start → reproduce → Stop (E16, app/debugsession.py). While it is open, every
+    request of its user that names it (`X-Debug-Session`) is logged at `level` and tagged with `id`,
+    and the issue it ends in is built from the rows carrying that tag. `until` is the hard cap the
+    server set at Start; an open session past it is over whether or not the sweep has closed it yet.
+    One OPEN session per user (`uq_debug_sessions_open_per_user`, partial on `ended_at IS NULL`)."""
+
+    __tablename__ = "debug_sessions"
+    __table_args__ = (
+        Index(
+            "uq_debug_sessions_open_per_user",
+            "user_id",
+            unique=True,
+            postgresql_where="ended_at IS NULL",
+        ),
+    )
+
+    id: Mapped[str] = mapped_column(String(16), primary_key=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)
+    level: Mapped[str] = mapped_column(String(8))
+    started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))
+    until: Mapped[datetime] = mapped_column(DateTime(timezone=True))
+    ended_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    # app/debugsession.py `EndReason`: stopped by the user, or past `until`.
+    ended_reason: Mapped[str | None] = mapped_column(String(16))
+    issue_url: Mapped[str | None] = mapped_column(String(512))
+
+
+class DebugScreenshot(Base):
+    """A screenshot attached to a debug session (E16) — pasted, dropped, or captured from the pill.
+    Stored in the row: bounded per session (app/debugsession.py), and gone with it by cascade."""
+
+    __tablename__ = "debug_screenshots"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    session_id: Mapped[str] = mapped_column(
+        ForeignKey("debug_sessions.id", ondelete="CASCADE"), index=True
+    )
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))
+    mime: Mapped[str] = mapped_column(String(16))
+    size: Mapped[int] = mapped_column(Integer)
+    data: Mapped[bytes] = mapped_column(LargeBinary)
+
+
 class AppState(Base):
     """Single-row global app state (admin-controlled)."""
 

backend/app/routes/_common.py

@@ -8,7 +8,7 @@ one existence-leak policy.
 
 from typing import Any, TypeVar
 
-from fastapi import HTTPException
+from fastapi import HTTPException, Request
 from sqlalchemy.orm import Session
 
 from app.db import Base
@@ -63,3 +63,15 @@ def youtube_refusal(exc: Exception) -> HTTPException:
     YouTube being unavailable to us — the retryable answer."""
     reason = refusal_reason(exc) if isinstance(exc, YouTubeError) else "unavailable"
     return HTTPException(status_code=422, detail={"code": "youtube", "reason": reason})
+
+
+async def capped_body(request: Request, limit: int) -> bytes:
+    """The request body, refused (413 `too_large`) the moment it passes `limit` bytes while still
+    arriving. Counted, not taken from Content-Length: a declared length can be absent (chunked) or
+    wrong, and counting stops reading at the same point a declared-length check would have."""
+    body = bytearray()
+    async for chunk in request.stream():
+        body += chunk
+        if len(body) > limit:
+            raise HTTPException(status_code=413, detail={"code": "too_large"})
+    return bytes(body)

backend/app/routes/admin_logs.py

@@ -26,10 +26,11 @@ from fastapi import APIRouter, Depends, HTTPException, Query
 from sqlalchemy import Text, and_, cast, func, or_, select
 from sqlalchemy.orm import Session
 
+from app import logsink
 from app.auth import admin_user
 from app.db import get_db
 from app.models import LogEvent, User
-from app.utils import iso, like_escape
+from app.utils import like_escape
 
 router = APIRouter(prefix="/api/admin/logs", tags=["admin"])
 
@@ -66,25 +67,6 @@ def _decode_cursor(cursor: str) -> tuple[datetime, int]:
         raise HTTPException(status_code=400, detail={"code": "bad_cursor"}) from exc
 
 
-def _serialize(row: LogEvent) -> dict:
-    return {
-        "id": row.id,
-        "ts": iso(row.ts),
-        "level": row.level,
-        "source": row.source,
-        "logger": row.logger,
-        "event": row.event,
-        "msg": row.msg,
-        "attrs": row.attrs,
-        "exc": row.exc,
-        "request_id": row.request_id,
-        "user_id": row.user_id,
-        "job_id": row.job_id,
-        "debug_session_id": row.debug_session_id,
-        "version": row.version,
-    }
-
-
 def _accounts(db: Session, user_ids: set[int]) -> dict[str, str]:
     """The accounts a page names, by id → address. An id with no account left (erased after the
     line was written) is simply absent, and the client shows the bare id."""
@@ -196,7 +178,7 @@ def list_logs(
     more = len(rows) > limit
     rows = rows[:limit]
     return {
-        "items": [_serialize(r) for r in rows],
+        "items": [logsink.serialize(r) for r in rows],
         "accounts": _accounts(db, {r.user_id for r in rows if r.user_id is not None}),
         "next_cursor": _encode_cursor(rows[-1]) if more else None,
     }

backend/app/routes/debug_sessions.py

@@ -0,0 +1,183 @@
+"""Debug sessions (E16): Start → reproduce → Stop, the screenshots, and the evidence under review.
+
+The session itself — what tags a request, its cap, its level — is `app/debugsession.py`. These routes
+are the user's handle on their OWN sessions only: another user's id 404s (`owned_or_404`), including
+for an admin, who reads everyone's lines in the diagnostics console instead. The shared demo account
+cannot record: its log would interleave every visitor under one account.
+
+Screenshots arrive as a raw image body (`image/png`, `image/jpeg`, `image/webp`), not a multipart
+form — one image per request, capped while it arrives, and kept only if its bytes really are one of
+those types, whatever the Content-Type claimed.
+"""
+
+from datetime import datetime, timezone
+
+from fastapi import APIRouter, Depends, HTTPException, Request, Response
+from pydantic import BaseModel
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app import debugsession, logs, logsink
+from app.auth import require_human
+from app.db import get_db
+from app.models import DebugScreenshot, DebugSession, User
+from app.routes._common import capped_body, not_found, owned_or_404
+from app.utils import iso
+
+router = APIRouter(prefix="/api/debug-sessions", tags=["diagnostics"])
+log = logs.get_logger(__name__)
+
+# Sessions listed in Settings → Diagnosztika, newest first: enough to reach last week's unsent draft.
+LIST_MAX = 20
+
+
+class StartIn(BaseModel):
+    level: debugsession.SessionLevel
+
+
+def _view(db: Session, session: DebugSession) -> dict:
+    return debugsession.serialize(session, debugsession.screenshot_count(db, session)) or {}
+
+
+@router.post("")
+def start_session(
+    body: StartIn,
+    user: User = Depends(require_human),
+    db: Session = Depends(get_db),
+) -> dict:
+    """Start recording, or join the recording already running (another tab started it)."""
+    try:
+        session = debugsession.start(db, user, body.level)
+        db.commit()
+    except IntegrityError:
+        # A concurrent Start won the one-open-per-user index: join that one.
+        db.rollback()
+        session = debugsession.open_session(db, user)
+        if session is None:
+            raise HTTPException(status_code=409, detail={"code": "session_conflict"})
+    # The rest of THIS request is part of the session too.
+    level, until = debugsession.level_for(user, session)
+    logs.set_user(user.id, level, until, session.id)
+    log.info(
+        "debug session started",
+        extra=logs.event("debug.session.started", level=session.level, until=iso(session.until)),
+    )
+    return {"ok": True, "session": _view(db, session)}
+
+
+@router.post("/{sid}/stop")
+def stop_session(
+    sid: str, user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> dict:
+    session = owned_or_404(db, DebugSession, sid, user, what="debug_session")
+    if session.ended_at is None:
+        # Logged BEFORE the session closes, while this request still carries its tag: the last line
+        # of the evidence is the Stop itself.
+        log.info("debug session stopped", extra=logs.event("debug.session.stopped"))
+        # Past its cap it had already ended on its own; a late Stop must not rename that.
+        live = session.until > datetime.now(timezone.utc)
+        reason = debugsession.EndReason.STOPPED if live else debugsession.EndReason.EXPIRED
+        debugsession.end(session, reason)
+        db.commit()
+    return {"ok": True, "session": _view(db, session)}
+
+
+@router.get("")
+def list_sessions(user: User = Depends(require_human), db: Session = Depends(get_db)) -> dict:
+    sessions = db.scalars(
+        select(DebugSession)
+        .where(DebugSession.user_id == user.id)
+        .order_by(DebugSession.started_at.desc())
+        .limit(LIST_MAX)
+    ).all()
+    return {"items": [_view(db, s) for s in sessions]}
+
+
+@router.get("/{sid}")
+def get_session(
+    sid: str, user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> dict:
+    return _view(db, owned_or_404(db, DebugSession, sid, user, what="debug_session"))
+
+
+@router.get("/{sid}/events")
+def session_events(
+    sid: str, user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> dict:
+    """The rows an issue from this session would carry — browser and server, in time order."""
+    session = owned_or_404(db, DebugSession, sid, user, what="debug_session")
+    rows, total = debugsession.bundle(db, session)
+    return {"items": [logsink.serialize(r) for r in rows], "total": total}
+
+
+# --- screenshots ---------------------------------------------------------------------------
+def _shot_view(shot: DebugScreenshot) -> dict:
+    return {"id": shot.id, "mime": shot.mime, "size": shot.size, "created_at": iso(shot.created_at)}
+
+
+@router.get("/{sid}/screenshots")
+def list_screenshots(
+    sid: str, user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> dict:
+    session = owned_or_404(db, DebugSession, sid, user, what="debug_session")
+    return {"items": [_shot_view(s) for s in debugsession.screenshots(db, session)]}
+
+
+def _upload_target(
+    sid: str, user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> DebugSession:
+    return owned_or_404(db, DebugSession, sid, user, what="debug_session")
+
+
+async def _image_body(request: Request) -> bytes:
+    return await capped_body(request, debugsession.SCREENSHOT_MAX_BYTES)
+
+
+@router.post("/{sid}/screenshots")
+def add_screenshot(
+    # The owner check FIRST: dependencies resolve in order, so a stranger's upload is refused
+    # before a byte of it is read.
+    session: DebugSession = Depends(_upload_target),
+    data: bytes = Depends(_image_body),
+    db: Session = Depends(get_db),
+) -> dict:
+    try:
+        shot = debugsession.add_screenshot(db, session, data)
+    except debugsession.ScreenshotRefused as exc:
+        raise HTTPException(
+            status_code=422, detail={"code": "screenshot_refused", "reason": exc.code}
+        ) from exc
+    db.commit()
+    return {"ok": True, "screenshot": _shot_view(shot)}
+
+
+def _session_screenshot(db: Session, sid: str, shot_id: int, user: User) -> DebugScreenshot:
+    session = owned_or_404(db, DebugSession, sid, user, what="debug_session")
+    shot = db.get(DebugScreenshot, shot_id)
+    if shot is None or shot.session_id != session.id:
+        raise not_found("screenshot")
+    return shot
+
+
+@router.get("/{sid}/screenshots/{shot_id}")
+def get_screenshot(
+    sid: str, shot_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> Response:
+    shot = _session_screenshot(db, sid, shot_id, user)
+    # `nosniff`: the type was checked against the bytes at upload, and the browser must not guess.
+    return Response(
+        content=shot.data,
+        media_type=shot.mime,
+        headers={"Cache-Control": "private, max-age=3600", "X-Content-Type-Options": "nosniff"},
+    )
+
+
+@router.delete("/{sid}/screenshots/{shot_id}")
+def delete_screenshot(
+    sid: str, shot_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> dict:
+    shot = _session_screenshot(db, sid, shot_id, user)
+    db.delete(shot)
+    db.commit()
+    return {"ok": True, "deleted": shot_id}

backend/app/routes/diagnostics.py

@@ -13,7 +13,9 @@ Nothing from the body is trusted as-is:
   * everything is redacted AGAIN with the server's own rules — the client redacts at capture, but a
     modified client would not;
   * an event below the consented level is dropped (a Részletes consent does not store what only
-    Teljes may carry), and so is the whole batch once the level has lapsed;
+    Teljes may carry), and so is the whole batch once the level has lapsed — where an open debug
+    session the request names counts as consent to ITS level (E16), and only its own events keep
+    its tag;
   * the timestamp is clamped to the consent window, so a row cannot dodge the retention job by
     claiming to come from the future.
 """
@@ -27,11 +29,12 @@ from pydantic import BaseModel, Field, ValidationError
 from sqlalchemy import insert
 from sqlalchemy.orm import Session
 
-from app import diagnostics, logs, logsink
+from app import debugsession, diagnostics, logs, logsink
 from app.auth import current_user
 from app.db import get_db
-from app.models import LogEvent, User
+from app.models import DebugSession, LogEvent, User
 from app.ratelimit import RateLimiter
+from app.routes._common import capped_body
 
 router = APIRouter(prefix="/api/diagnostics", tags=["diagnostics"])
 
@@ -60,6 +63,9 @@ class ClientEvent(BaseModel):
     attrs: dict[str, Any] | None = None
     request_id: str | None = Field(default=None, pattern=r"^[A-Za-z0-9._-]{8,64}$")
     tab_id: str | None = Field(default=None, pattern=r"^[A-Za-z0-9_-]{1,32}$")
+    # The debug session the event was captured in (E16); kept only when it is the one this request
+    # names in `X-Debug-Session` and the auth dependency found it the caller's own and open.
+    debug_session_id: str | None = Field(default=None, pattern=r"^[0-9a-f]{16}$")
     version: str | None = Field(default=None, max_length=64)
 
 
@@ -67,18 +73,6 @@ class ClientEventsIn(BaseModel):
     events: list[ClientEvent] = Field(max_length=EVENTS_MAX)
 
 
-async def _capped_body(request: Request) -> bytes:
-    """The request body, refused the moment it passes BODY_MAX_BYTES while still arriving. Counted,
-    not taken from Content-Length: a declared length can be absent (chunked) or wrong, and counting
-    stops reading at the same point a declared-length check would have."""
-    body = bytearray()
-    async for chunk in request.stream():
-        body += chunk
-        if len(body) > BODY_MAX_BYTES:
-            raise HTTPException(status_code=413, detail={"code": "too_large"})
-    return bytes(body)
-
-
 async def _parsed_events(request: Request) -> ClientEventsIn:
     """The body as the model — parsed only after the size cap, which is why this is a dependency and
     not a plain body parameter (FastAPI reads a declared body in full before anything else runs).
@@ -89,14 +83,20 @@ async def _parsed_events(request: Request) -> ClientEventsIn:
     media_type = request.headers.get("content-type", "").split(";")[0].strip().lower()
     if media_type != "application/json" and not media_type.endswith("+json"):
         raise HTTPException(status_code=415, detail={"code": "json_required"})
-    raw = await _capped_body(request)
+    raw = await capped_body(request, BODY_MAX_BYTES)
     try:
         return ClientEventsIn.model_validate_json(raw)
     except ValidationError as exc:
         raise HTTPException(status_code=422, detail=json.loads(exc.json())) from exc
 
 
-def _row(item: ClientEvent, user: User, now: datetime, window_start: datetime) -> dict[str, Any]:
+def _row(
+    item: ClientEvent,
+    user: User,
+    now: datetime,
+    window_start: datetime,
+    session: DebugSession | None,
+) -> dict[str, Any]:
     attrs = logs.redact(item.attrs) if item.attrs else None
     if attrs is not None:
         size = len(json.dumps(attrs, ensure_ascii=False, default=str).encode())
@@ -117,6 +117,9 @@ def _row(item: ClientEvent, user: User, now: datetime, window_start: datetime) -
             "attrs": attrs,
             "request_id": item.request_id,
             "user_id": user.id,
+            "debug_session_id": (
+                session.id if session and item.debug_session_id == session.id else None
+            ),
             "version": item.version,
         }
     )
@@ -126,11 +129,15 @@ def _row(item: ClientEvent, user: User, now: datetime, window_start: datetime) -
 def ingest_events(
     # The account first: dependencies resolve in order, so an anonymous caller is refused before a
     # byte of its body is read.
+    request: Request,
     user: User = Depends(current_user),
     body: ClientEventsIn = Depends(_parsed_events),
     db: Session = Depends(get_db),
 ) -> dict:
-    level = diagnostics.effective_level(user)
+    # A debug session the request names raises the level for its own events (E16): starting one is
+    # the consent to that level for its duration, whatever the user's own level is.
+    session = debugsession.from_header(db, user, request.headers.get(debugsession.HEADER))
+    level, _ = debugsession.level_for(user, session)
     if level == diagnostics.BASE:
         # Lapsed or withdrawn: the client stops streaming when it sees this.
         raise HTTPException(status_code=409, detail={"code": "diagnostics_off"})
@@ -140,7 +147,7 @@ def ingest_events(
     window_start = now - timedelta(hours=diagnostics.ELEVATION_HOURS)
     floor = logs.SEVERITY[level]
     rows = [
-        _row(item, user, now, window_start)
+        _row(item, user, now, window_start, session)
         for item in body.events
         if logs.SEVERITY[item.level] >= floor
     ]

backend/app/routes/me.py

@@ -17,7 +17,7 @@ from app.auth import (
     optional_current_user,
     purge_user,
 )
-from app import diagnostics
+from app import debugsession, diagnostics
 from app import email as email_mod
 from app import logs
 from app import sysconfig
@@ -158,9 +158,18 @@ def get_me(
         # The consented diagnostics level in force and when it lapses (E22 S5). The client applies
         # it to its own log and streams that log while it is raised.
         "diagnostics": diagnostics.state(db, user),
+        # The live debug session, if one is recording (E16) — a reload resumes it.
+        "debug_session": _live_session(db, user),
     }
 
 
+def _live_session(db: Session, user: User) -> dict | None:
+    session = debugsession.open_session(db, user)
+    if session is None:
+        return None
+    return debugsession.serialize(session, debugsession.screenshot_count(db, session))
+
+
 @router.delete("/account")
 def delete_my_account(
     request: Request,

backend/app/scheduler.py

@@ -9,7 +9,7 @@ from typing import Callable
 from apscheduler.schedulers.background import BackgroundScheduler
 from sqlalchemy import select
 
-from app import audit, logs, logsink, progress, quota
+from app import audit, debugsession, logs, logsink, progress, quota
 from app.audit import AuditAction
 from app.config import settings
 from app.db import SessionLocal
@@ -386,9 +386,13 @@ def _log_gc_job() -> None:
     def work(db):
         from app import sysconfig
 
-        reaped = logsink.gc(db, sysconfig.get_int(db, "log_retention_days"))
+        days = sysconfig.get_int(db, "log_retention_days")
+        reaped = logsink.gc(db, days)
+        # Debug sessions (E16) go with the rows they bundle; an open one past its cap is closed.
+        closed = debugsession.sweep(db)
+        sessions = debugsession.gc(db, days)
         db.commit()
-        return {"reaped": reaped}
+        return {"reaped": reaped, "sessions_closed": closed, "sessions_reaped": sessions}
 
     _job("log_gc", work)