diff --git a/CLAUDE.md b/CLAUDE.md
index cb86892c..f1fae3cd 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -58,7 +58,7 @@ how it gets built. Concretely:
 `routes/*` (HTTP only: validation, auth-dependency, serialization) → domain packages `sync/`, `plex/`,
 `downloads/`, `youtube/` (shared by routes + scheduler + worker) → flat core (`auth`, `quota`,
 `sysconfig`, `state`, `audit`, `notifications`, `userscope`, `realtime`, `ratelimit`, `progress`,
-`security`, `logs`, `logsink`, `diagnostics`) → `models/db/config`. **Dependency direction is one-way: core never imports routes.**
+`security`, `logs`, `logsink`, `diagnostics`, `debugsession`) → `models/db/config`. **Dependency direction is one-way: core never imports routes.**
 Frontend: `main.tsx` provider tree → `App.tsx` shell → `moduleRegistry` route-chunks → module
 components → the **platform layer `components/ui/`** (domain-free primitives: the popover engine and
 its menus, `Modal`/`Overlay`, `Button`/`form`, `DataTable`/`VirtualGrid`, `QueryState`, …). That
diff --git a/backend/alembic/versions/0089_debug_sessions.py b/backend/alembic/versions/0089_debug_sessions.py
new file mode 100644
index 00000000..161d117f
--- /dev/null
+++ b/backend/alembic/versions/0089_debug_sessions.py
@@ -0,0 +1,81 @@
+"""add debug_sessions + debug_screenshots, and index log_events.debug_session_id (E16)
+
+Revision ID: 0089_debug_sessions
+Revises: 0088_channels_feed_fallback_at
+Create Date: 2026-09-19
+
+A debug session is one recording: Start → reproduce → Stop (app/debugsession.py). Every log line
+of that user's requests while it runs carries its id in `log_events.debug_session_id` — the column
+E22 created and left NULL — and the issue it ends in is built from exactly those rows, so the column
+gets its index now. Partial: nearly every row has no session, and an index of NULLs would be most
+of the table for nothing.
+
+At most one OPEN session per user, held by a partial unique index rather than by the start handler
+alone: two tabs pressing Start at once would otherwise both pass a "none open" check.
+
+Screenshots are rows, not files: bounded (size and count per session, app/debugsession.py), gone
+with their session by cascade, and a self-host has no file volume to configure for them.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0089_debug_sessions"
+down_revision: str | None = "0088_channels_feed_fallback_at"
+branch_labels: str | None = None
+depends_on: str | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "debug_sessions",
+        sa.Column("id", sa.String(16), primary_key=True),
+        sa.Column(
+            "user_id",
+            sa.Integer,
+            sa.ForeignKey("users.id", ondelete="CASCADE"),
+            nullable=False,
+            index=True,
+        ),
+        sa.Column("level", sa.String(8), nullable=False),
+        sa.Column("started_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("until", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("ended_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("ended_reason", sa.String(16), nullable=True),
+        sa.Column("issue_url", sa.String(512), nullable=True),
+    )
+    op.create_index(
+        "uq_debug_sessions_open_per_user",
+        "debug_sessions",
+        ["user_id"],
+        unique=True,
+        postgresql_where=sa.text("ended_at IS NULL"),
+    )
+    op.create_table(
+        "debug_screenshots",
+        sa.Column("id", sa.Integer, primary_key=True),
+        sa.Column(
+            "session_id",
+            sa.String(16),
+            sa.ForeignKey("debug_sessions.id", ondelete="CASCADE"),
+            nullable=False,
+            index=True,
+        ),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("mime", sa.String(16), nullable=False),
+        sa.Column("size", sa.Integer, nullable=False),
+        sa.Column("data", sa.LargeBinary, nullable=False),
+    )
+    op.create_index(
+        "ix_log_events_debug_session",
+        "log_events",
+        ["debug_session_id"],
+        postgresql_where=sa.text("debug_session_id IS NOT NULL"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_log_events_debug_session", table_name="log_events")
+    op.drop_table("debug_screenshots")
+    op.drop_index("uq_debug_sessions_open_per_user", table_name="debug_sessions")
+    op.drop_table("debug_sessions")
diff --git a/backend/app/README.md b/backend/app/README.md
index ec04ecc2..af8b21b3 100644
--- a/backend/app/README.md
+++ b/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
diff --git a/backend/app/auth.py b/backend/app/auth.py
index 763954cf..cbee737c 100644
--- a/backend/app/auth.py
+++ b/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
 
diff --git a/backend/app/debugsession.py b/backend/app/debugsession.py
new file mode 100644
index 00000000..5aaa847e
--- /dev/null
+++ b/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
diff --git a/backend/app/logs.py b/backend/app/logs.py
index 565030c8..61c2dcc6 100644
--- a/backend/app/logs.py
+++ b/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
 
 
diff --git a/backend/app/logsink.py b/backend/app/logsink.py
index 924475ee..b07acd1c 100644
--- a/backend/app/logsink.py
+++ b/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,
+    }
diff --git a/backend/app/main.py b/backend/app/main.py
index bf0e4708..c3856349 100644
--- a/backend/app/main.py
+++ b/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)
diff --git a/backend/app/models.py b/backend/app/models.py
index b4273184..a827d2d2 100644
--- a/backend/app/models.py
+++ b/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)."""
 
diff --git a/backend/app/routes/_common.py b/backend/app/routes/_common.py
index 6689f6fa..e8c530a1 100644
--- a/backend/app/routes/_common.py
+++ b/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)
diff --git a/backend/app/routes/admin_logs.py b/backend/app/routes/admin_logs.py
index 43ae3c65..07b4c44f 100644
--- a/backend/app/routes/admin_logs.py
+++ b/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,
     }
diff --git a/backend/app/routes/debug_sessions.py b/backend/app/routes/debug_sessions.py
new file mode 100644
index 00000000..a3722d74
--- /dev/null
+++ b/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}
diff --git a/backend/app/routes/diagnostics.py b/backend/app/routes/diagnostics.py
index 2cf62804..bdcfc9b9 100644
--- a/backend/app/routes/diagnostics.py
+++ b/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
     ]
diff --git a/backend/app/routes/me.py b/backend/app/routes/me.py
index 6f1784cf..34bf9cc0 100644
--- a/backend/app/routes/me.py
+++ b/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,
diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py
index c3e24d12..156fbe21 100644
--- a/backend/app/scheduler.py
+++ b/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)
 
diff --git a/backend/tests/test_debug_sessions.py b/backend/tests/test_debug_sessions.py
new file mode 100644
index 00000000..50a9bc96
--- /dev/null
+++ b/backend/tests/test_debug_sessions.py
@@ -0,0 +1,438 @@
+"""E16 S1 — debug sessions: Start → reproduce → Stop, the tag on every line, the screenshots.
+
+Four parts:
+  * the session's life: one open per user, a second Start joins it, a Stop ends it, the cap ends it
+    on its own, the demo account cannot record, nobody reaches another user's session (DB lane);
+  * the tag: a request naming the caller's own open session is logged at its level with every line
+    tagged, while a header naming anything else is ignored (a real sign-in through `_session_user`);
+  * the browser's batches: a session counts as consent to its level, and only its own events keep
+    its tag (DB lane);
+  * screenshots: judged by their bytes, capped in size and count, served back only to their owner.
+"""
+
+import logging
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from fastapi import BackgroundTasks
+from sqlalchemy import select
+
+from app import debugsession, logs
+from app.models import DebugScreenshot, DebugSession, LogEvent, User
+from app.routes import diagnostics as diagnostics_routes
+
+_PW = "correct horse battery"
+_PNG = b"\x89PNG\r\n\x1a\n" + b"\x00" * 32
+_JPEG = b"\xff\xd8\xff\xe0" + b"\x00" * 32
+_WEBP = b"RIFF\x00\x00\x00\x00WEBPVP8 " + b"\x00" * 16
+
+
+def _signed_in(client, make_user, email="rec@example.com", **kw) -> User:
+    """A real sign-in, so the auth dependency itself reads `X-Debug-Session` (`as_user` bypasses
+    it)."""
+    user = make_user(email=email, password=_PW, **kw)
+    r = client.post("/auth/password-login", json={"email": email, "password": _PW})
+    assert r.status_code == 200, r.text
+    return user
+
+
+def _start(client, level="DEBUG") -> dict:
+    r = client.post("/api/debug-sessions", json={"level": level})
+    assert r.status_code == 200, r.text
+    return r.json()["session"]
+
+
+# --- the session's life ---------------------------------------------------------------------
+
+
+def test_start_opens_a_capped_session_that_me_reports(db, client, make_user):
+    _signed_in(client, make_user)
+    session = _start(client, "TRACE")
+    assert session["level"] == "TRACE" and session["ended_at"] is None
+    started = datetime.fromisoformat(session["started_at"])
+    until = datetime.fromisoformat(session["until"])
+    assert until - started == timedelta(minutes=debugsession.MAX_MINUTES)
+    assert client.get("/api/me").json()["debug_session"]["id"] == session["id"]
+
+
+def test_a_second_start_joins_the_running_session(db, make_user, as_user):
+    client = as_user(make_user())
+    first = _start(client, "DEBUG")
+    assert _start(client, "TRACE")["id"] == first["id"]
+    assert db.scalar(select(DebugSession).where(DebugSession.id == first["id"])).level == "DEBUG"
+
+
+def test_only_the_two_raised_levels_can_be_recorded(db, make_user, as_user):
+    client = as_user(make_user())
+    for bogus in ("INFO", "WARNING", "trace"):
+        assert client.post("/api/debug-sessions", json={"level": bogus}).status_code == 422
+
+
+def test_the_demo_account_cannot_record(db, make_user, as_user):
+    r = as_user(make_user(is_demo=True)).post("/api/debug-sessions", json={"level": "DEBUG"})
+    assert r.status_code == 403 and r.json()["detail"]["code"] == "demo"
+
+
+def test_stop_ends_it_and_me_stops_reporting_it(db, client, make_user):
+    _signed_in(client, make_user)
+    sid = _start(client)["id"]
+    r = client.post(f"/api/debug-sessions/{sid}/stop")
+    assert r.status_code == 200 and r.json()["session"]["ended_reason"] == "stopped"
+    assert client.get("/api/me").json()["debug_session"] is None
+    # A second Stop changes nothing.
+    again = client.post(f"/api/debug-sessions/{sid}/stop").json()["session"]
+    assert again["ended_at"] == r.json()["session"]["ended_at"]
+
+
+def _expire(db, sid: str) -> None:
+    session = db.get(DebugSession, sid)
+    session.started_at -= timedelta(hours=2)
+    session.until = datetime.now(timezone.utc) - timedelta(minutes=1)
+    db.commit()
+
+
+def test_a_session_past_its_cap_is_over_before_anything_closes_it(db, client, make_user):
+    _signed_in(client, make_user)
+    sid = _start(client)["id"]
+    _expire(db, sid)
+    assert client.get("/api/me").json()["debug_session"] is None
+    # A late Stop records that it ended on its own, at its cap.
+    stopped = client.post(f"/api/debug-sessions/{sid}/stop").json()["session"]
+    assert stopped["ended_reason"] == "expired"
+    assert stopped["ended_at"] == stopped["until"]
+
+
+def test_a_start_after_an_expired_session_closes_it_and_opens_a_new_one(db, make_user, as_user):
+    client = as_user(make_user())
+    old = _start(client)["id"]
+    _expire(db, old)
+    new = _start(client)["id"]
+    assert new != old
+    db.expire_all()
+    assert db.get(DebugSession, old).ended_reason == "expired"
+
+
+def test_nobody_reaches_another_users_session(db, make_user, as_user):
+    owner, stranger = as_user(make_user()), as_user(make_user())
+    sid = _start(owner)["id"]
+    shot = owner.post(
+        f"/api/debug-sessions/{sid}/screenshots",
+        content=_PNG,
+        headers={"content-type": "image/png"},
+    ).json()["screenshot"]["id"]
+    for method, path in (
+        ("get", f"/api/debug-sessions/{sid}"),
+        ("post", f"/api/debug-sessions/{sid}/stop"),
+        ("get", f"/api/debug-sessions/{sid}/events"),
+        ("get", f"/api/debug-sessions/{sid}/screenshots"),
+        ("get", f"/api/debug-sessions/{sid}/screenshots/{shot}"),
+        ("delete", f"/api/debug-sessions/{sid}/screenshots/{shot}"),
+    ):
+        r = getattr(stranger, method)(path)
+        assert r.status_code == 404, (method, path, r.status_code)
+        assert r.json()["detail"] == {"code": "not_found", "reason": "debug_session"}
+    r = stranger.post(
+        f"/api/debug-sessions/{sid}/screenshots",
+        content=_PNG,
+        headers={"content-type": "image/png"},
+    )
+    assert r.status_code == 404
+    assert stranger.get("/api/debug-sessions").json()["items"] == []
+    db.expire_all()
+    assert db.get(DebugSession, sid).ended_at is None
+
+
+def test_the_list_is_the_users_own_newest_first(db, make_user, as_user):
+    client = as_user(make_user())
+    first = _start(client)["id"]
+    client.post(f"/api/debug-sessions/{first}/stop")
+    second = _start(client)["id"]
+    assert [s["id"] for s in client.get("/api/debug-sessions").json()["items"]] == [second, first]
+
+
+def test_the_hourly_job_closes_expired_sessions_and_drops_old_ones(db, make_user):
+    user = make_user()
+    now = datetime.now(timezone.utc)
+    lapsed = DebugSession(
+        id="b" * 16,
+        user_id=user.id,
+        level="DEBUG",
+        started_at=now - timedelta(hours=2),
+        until=now - timedelta(hours=1),
+    )
+    db.add(lapsed)
+    ancient = DebugSession(
+        id="c" * 16,
+        user_id=user.id,
+        level="DEBUG",
+        started_at=now - timedelta(days=30),
+        until=now - timedelta(days=30) + timedelta(hours=1),
+        ended_at=now - timedelta(days=30),
+        ended_reason="stopped",
+    )
+    db.add(ancient)
+    db.flush()  # no ORM relationship orders the inserts: the session row first
+    db.add(
+        DebugScreenshot(session_id=ancient.id, created_at=now, mime="image/png", size=4, data=b"x")
+    )
+    db.commit()
+
+    assert debugsession.sweep(db) == 1
+    assert debugsession.gc(db, 7) == 1
+    db.commit()
+    db.expire_all()
+    assert db.get(DebugSession, lapsed.id).ended_reason == "expired"
+    assert db.get(DebugSession, lapsed.id).ended_at == lapsed.until
+    assert db.get(DebugSession, ancient.id) is None
+    assert db.scalars(select(DebugScreenshot)).all() == []
+
+
+def test_an_erased_account_takes_its_sessions_with_it(db, make_user, as_user):
+    from app.auth import purge_user
+
+    user = make_user()
+    client = as_user(user)
+    sid = _start(client)["id"]
+    client.post(
+        f"/api/debug-sessions/{sid}/screenshots",
+        content=_PNG,
+        headers={"content-type": "image/png"},
+    )
+    purge_user(db, user, BackgroundTasks())
+    assert db.get(DebugSession, sid) is None
+    assert db.scalars(select(DebugScreenshot)).all() == []
+
+
+# --- the tag --------------------------------------------------------------------------------
+
+
+class _Capture(logging.Handler):
+    def __init__(self):
+        super().__init__(level=logging.NOTSET)
+        self.records: list[logging.LogRecord] = []
+        self.addFilter(logs.ContextFilter())
+        self.addFilter(logs.LevelGate())
+
+    def emit(self, record):
+        self.records.append(record)
+
+
+@pytest.fixture
+def gated():
+    """The process at INFO with a gated handler on the root, as test_diagnostics' fixture."""
+    root = logging.getLogger()
+    saved = (logs._base_level, dict(logs._elevations), logs._root_applied, root.level)
+    handler = _Capture()
+    root.addHandler(handler)
+    logs._elevations.clear()
+    logs.set_level("INFO")
+    try:
+        yield handler
+    finally:
+        root.removeHandler(handler)
+        logs._base_level, logs._root_applied = saved[0], saved[2]
+        logs._elevations.clear()
+        logs._elevations.update(saved[1])
+        root.setLevel(saved[3])
+
+
+def _request_lines(handler: _Capture) -> list[logging.LogRecord]:
+    return [r for r in handler.records if getattr(r, "event", None) == "http.request"]
+
+
+def test_a_request_naming_the_own_open_session_is_tagged_and_raised(gated, db, client, make_user):
+    _signed_in(client, make_user)
+    sid = _start(client, "DEBUG")["id"]
+    gated.records.clear()
+    client.get("/api/debug-sessions", headers={"x-debug-session": sid})
+    (line,) = _request_lines(gated)
+    assert getattr(line, "debug_session_id", None) == sid
+    # Raised to the session's level: the line is kept for storage though the user's own is Alap.
+    assert getattr(line, "elevated", False) is True
+
+
+def test_a_header_naming_anything_else_is_ignored_not_refused(gated, db, client, make_user):
+    other = make_user(email="other@example.com")
+    foreign = debugsession.start(db, other, "TRACE").id
+    db.commit()
+    _signed_in(client, make_user)
+    own = _start(client)["id"]
+    client.post(f"/api/debug-sessions/{own}/stop")
+    for value in (foreign, own, "not-a-session", "f" * 16, "A" * 16):
+        gated.records.clear()
+        # Not `/api/me`: a polled route logs at DEBUG, and an un-raised request would drop the very
+        # line this checks.
+        r = client.get("/api/debug-sessions", headers={"x-debug-session": value})
+        assert r.status_code == 200, value
+        (line,) = _request_lines(gated)
+        assert getattr(line, "debug_session_id", None) is None, value
+        assert getattr(line, "elevated", False) is False, value
+
+
+def test_a_session_never_lowers_the_users_own_level():
+    now = datetime.now(timezone.utc)
+    user = User(diagnostics_level="TRACE", diagnostics_until=now + timedelta(hours=3))
+    session = DebugSession(level="DEBUG", until=now + timedelta(minutes=30))
+    assert debugsession.level_for(user, session) == ("TRACE", user.diagnostics_until)
+    assert debugsession.level_for(User(), session) == ("DEBUG", session.until)
+    assert debugsession.level_for(User(), None) == ("INFO", None)
+
+
+# --- the browser's batches -------------------------------------------------------------------
+
+
+def _event(**over) -> dict:
+    return {
+        "ts": datetime.now(timezone.utc).isoformat(),
+        "level": "DEBUG",
+        "event": "ui.click",
+        "msg": "click",
+        "tab_id": "a1b2c3",
+        **over,
+    }
+
+
+@pytest.fixture
+def batches():
+    diagnostics_routes._batches.reset()
+    yield
+    diagnostics_routes._batches.reset()
+
+
+def test_a_session_lets_its_own_events_in_at_its_level_and_tags_them(
+    batches, db, make_user, as_user
+):
+    user = make_user()
+    client = as_user(user)
+    sid = _start(client, "DEBUG")["id"]
+    other = debugsession.start(db, make_user(), "DEBUG").id
+    db.commit()
+    events = [
+        _event(debug_session_id=sid),
+        _event(debug_session_id=other),  # not this session's: kept, untagged
+        _event(),  # captured before Start: kept, untagged
+        _event(level="TRACE", debug_session_id=sid),  # under the session's level: dropped
+    ]
+    r = client.post(
+        "/api/diagnostics/events", json={"events": events}, headers={"x-debug-session": sid}
+    )
+    assert r.json() == {"ok": True, "stored": 3}
+    rows = db.scalars(select(LogEvent).order_by(LogEvent.id)).all()
+    assert [row.debug_session_id for row in rows] == [sid, None, None]
+    assert {row.user_id for row in rows} == {user.id}
+
+
+def test_without_the_header_a_base_level_user_is_still_refused(batches, db, make_user, as_user):
+    client = as_user(make_user())
+    sid = _start(client)["id"]
+    r = client.post("/api/diagnostics/events", json={"events": [_event(debug_session_id=sid)]})
+    assert r.status_code == 409 and r.json()["detail"]["code"] == "diagnostics_off"
+
+
+def test_the_evidence_is_the_sessions_rows_in_time_order(db, make_user, as_user):
+    user = make_user()
+    client = as_user(user)
+    sid = _start(client)["id"]
+    now = datetime.now(timezone.utc)
+
+    def row(ts, source, session):
+        return LogEvent(
+            ts=ts,
+            level="INFO",
+            sev=9,
+            source=source,
+            logger=source,
+            event="x.y",
+            msg=source,
+            user_id=user.id,
+            debug_session_id=session,
+        )
+
+    db.add_all(
+        [
+            row(now + timedelta(seconds=2), "api", sid),
+            row(now + timedelta(seconds=1), "web", sid),
+            row(now, "web", None),
+        ]
+    )
+    db.commit()
+    body = client.get(f"/api/debug-sessions/{sid}/events").json()
+    assert [(i["source"], i["debug_session_id"]) for i in body["items"]] == [
+        ("web", sid),
+        ("api", sid),
+    ]
+    assert body["total"] == 2
+
+
+# --- screenshots ----------------------------------------------------------------------------
+
+
+@pytest.fixture
+def recording(db, make_user, as_user):
+    client = as_user(make_user())
+    return client, _start(client)["id"]
+
+
+def _upload(client, sid, data, content_type="image/png"):
+    return client.post(
+        f"/api/debug-sessions/{sid}/screenshots",
+        content=data,
+        headers={"content-type": content_type},
+    )
+
+
+@pytest.mark.parametrize(
+    ("data", "mime"), [(_PNG, "image/png"), (_JPEG, "image/jpeg"), (_WEBP, "image/webp")]
+)
+def test_a_screenshot_is_kept_as_what_its_bytes_are(recording, data, mime):
+    client, sid = recording
+    # The declared type is not trusted: every one of these claims to be a PNG.
+    r = _upload(client, sid, data)
+    assert r.status_code == 200 and r.json()["screenshot"]["mime"] == mime
+    shot = r.json()["screenshot"]["id"]
+    got = client.get(f"/api/debug-sessions/{sid}/screenshots/{shot}")
+    assert got.content == data
+    assert got.headers["content-type"] == mime
+    assert got.headers["x-content-type-options"] == "nosniff"
+
+
+@pytest.mark.parametrize(
+    ("data", "reason"), [(b"<svg onload=alert(1)>", "not_an_image"), (b"", "empty")]
+)
+def test_what_is_not_an_image_is_refused(recording, data, reason):
+    client, sid = recording
+    r = _upload(client, sid, data)
+    assert r.status_code == 422
+    assert r.json()["detail"] == {"code": "screenshot_refused", "reason": reason}
+
+
+def test_screenshots_are_capped_in_size_and_count(recording, monkeypatch):
+    client, sid = recording
+    monkeypatch.setattr(debugsession, "SCREENSHOT_MAX_BYTES", len(_PNG) + 1)
+    assert _upload(client, sid, _PNG + b"\x00\x00").status_code == 413
+    for _ in range(debugsession.SCREENSHOTS_MAX):
+        assert _upload(client, sid, _PNG).status_code == 200
+    r = _upload(client, sid, _PNG)
+    assert r.status_code == 422 and r.json()["detail"]["reason"] == "too_many"
+    assert (
+        client.get(f"/api/debug-sessions/{sid}").json()["screenshots"]
+        == debugsession.SCREENSHOTS_MAX
+    )
+
+
+def test_a_sent_session_takes_no_more_screenshots(db, recording):
+    client, sid = recording
+    db.get(DebugSession, sid).issue_url = "https://git.example/o/r/issues/1"
+    db.commit()
+    r = _upload(client, sid, _PNG)
+    assert r.status_code == 422 and r.json()["detail"]["reason"] == "already_sent"
+
+
+def test_a_screenshot_can_be_removed_and_is_then_gone(recording):
+    client, sid = recording
+    shot = _upload(client, sid, _PNG).json()["screenshot"]["id"]
+    assert client.get(f"/api/debug-sessions/{sid}/screenshots").json()["items"][0]["id"] == shot
+    r = client.delete(f"/api/debug-sessions/{sid}/screenshots/{shot}")
+    assert r.json() == {"ok": True, "deleted": shot}
+    r = client.get(f"/api/debug-sessions/{sid}/screenshots/{shot}")
+    assert r.status_code == 404 and r.json()["detail"]["reason"] == "screenshot"
diff --git a/frontend/src/i18n/locales/en/errors.json b/frontend/src/i18n/locales/en/errors.json
index 4e7f131b..5d123be6 100644
--- a/frontend/src/i18n/locales/en/errors.json
+++ b/frontend/src/i18n/locales/en/errors.json
@@ -20,6 +20,8 @@
     "recipient": "There's no user with that email address.",
     "job": "This job doesn't exist any more.",
     "download": "This download doesn't exist any more.",
+    "debug_session": "This debug session doesn't exist any more.",
+    "screenshot": "This screenshot doesn't exist any more.",
     "link": "This share link doesn't exist any more.",
     "notification": "This notification doesn't exist any more.",
     "account": "That account no longer exists.",
diff --git a/frontend/src/i18n/locales/hu/errors.json b/frontend/src/i18n/locales/hu/errors.json
index 57d74321..3f932915 100644
--- a/frontend/src/i18n/locales/hu/errors.json
+++ b/frontend/src/i18n/locales/hu/errors.json
@@ -20,6 +20,8 @@
     "recipient": "Nincs felhasználó ezzel az e-mail címmel.",
     "job": "Ez a feladat már nem létezik.",
     "download": "Ez a letöltés már nem létezik.",
+    "debug_session": "Ez a hibakereső munkamenet már nem létezik.",
+    "screenshot": "Ez a képernyőkép már nem létezik.",
     "link": "Ez a megosztási link már nem létezik.",
     "notification": "Ez az értesítés már nem létezik.",
     "account": "Ez a fiók már nem létezik.",