"""Project, frame, message, activity-step, and cell-log persistence."""

from __future__ import annotations

import base64
import hashlib
import json
import re
import sqlite3
import uuid
from collections.abc import Mapping
from typing import Any, Callable

from openai4s.execution.dependencies import (
    REPLAY_POLICIES,
    VISIBILITIES,
    analyze_code,
    default_replay_policy,
    default_visibility,
    normalize_string_list,
)
from openai4s.storage.deletion import SessionDeletionRepository
from openai4s.storage.like import like_contains


def visible_session_clause(
    user_id: str, *, table: str = "frames", session_expr: str | None = None
) -> tuple[str, list]:
    """The one team-mode visibility rule, as SQL over a frames-shaped table.

    Returns `(clause, params)` for a WHERE conjunct. One function because
    `browse`, `search` and `frame_detail` must answer the same question:
    three spellings of a visibility rule is three chances for one of them
    to be wrong, and the two that had no rule at all were reachable from
    `host.frames` with a colleague's frame id.

    Three properties worth stating, because each was a defect:

    **Scoped by the root session, not the frame.** Ownership is recorded
    per session in `session_owners`; a child frame has no row of its own.
    Matching on `frame_id` therefore hid every child frame from its own
    owner while `frame_detail` — which takes any frame id — had no rule at
    all. `COALESCE(root_frame_id, frame_id)` is the session a row belongs
    to.

    **A global guest never widens.** `session_visible_to` returns False for
    any account whose *global* role is guest, before it consults
    `project_members`. This clause omitted that, so an admin adding a guest
    to a project with the default `member` row listed them every
    project-visibility session — which they were then 404'd from opening.
    A listing that names sessions the caller cannot open is the leak INV-13
    describes, not a cosmetic inconsistency.

    **A session with no owner row is admin-only.** Pre-team history and
    demo seeds have none, and "we do not know whose this is" must not
    resolve to "everyone's".

    Filtered in SQL rather than after the read: keyset pagination reports
    `has_more` from row counts, so a post-read filter turns a full page of
    hidden rows into a phantom end-of-list — and for `search` and
    `frame_detail` the rows carry cell code and stdout, which a post-read
    filter would have already loaded.

    **A delegate-frame key resolves through the frames table.** A delegated
    child's rows are keyed under the child's own delegate frame
    (`frame_id = root_frame_id = <child frame id>`), which never has a
    `session_owners` row — resolving the raw key against `session_owners`
    made every child-keyed row admin-only, invisible to the very owner whose
    session spawned it. The frames table stores each frame's fully-resolved
    session root, so one lookup maps a delegate-frame key to the parent
    session; a key with no frames-table entry keeps the raw-key rule above.
    """
    # `session_expr` for a table that is not frames-shaped: `artifacts` has a
    # `root_frame_id` and no `frame_id`, and the ⌘K search needs the same rule
    # over it. Defaulted rather than required, so every existing caller keeps
    # the frames spelling.
    key = session_expr or f"COALESCE({table}.root_frame_id, {table}.frame_id)"
    session = (
        "COALESCE((SELECT fr.root_frame_id FROM frames fr"
        f" WHERE fr.frame_id = {key}), {key})"
    )
    clause = (
        "(NOT EXISTS (SELECT 1 FROM users gu WHERE gu.id = ? AND gu.role = 'guest')"
        " AND EXISTS (SELECT 1 FROM session_owners so"
        f" WHERE so.session_id = {session} AND ("
        "so.user_id = ? OR ("
        "so.visibility = 'project' AND so.project_id IS NOT NULL"
        " AND EXISTS (SELECT 1 FROM project_members pm"
        " WHERE pm.project_id = so.project_id AND pm.user_id = ?"
        " AND pm.role = 'member')))))"
    )
    return clause, [user_id, user_id, user_id]


PROJECT_PAGE_DEFAULT = 100
PROJECT_PAGE_MAX = 100
PROJECT_Q_MAX_CODEPOINTS = 128


def project_filter_fingerprint(*, q: str, team_scope: str) -> str:
    """Stable digest of the filter identity a project-list cursor is bound to."""
    canonical = json.dumps(
        {"q": q, "team_scope": team_scope},
        separators=(",", ":"),
        sort_keys=True,
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def encode_project_cursor(
    *,
    last_active_at: int | None,
    project_id: str,
    fingerprint: str,
) -> str:
    """Opaque keyset cursor. Opaque on purpose: a client that parses it
    becomes coupled to the sort key."""
    stamp = None if last_active_at is None else int(last_active_at)
    raw = json.dumps(
        {"fp": fingerprint, "id": project_id, "t": stamp},
        separators=(",", ":"),
        sort_keys=True,
    ).encode("utf-8")
    return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")


def decode_project_cursor(
    value: str | None, *, fingerprint: str
) -> tuple[int | None, str] | None:
    """Return the keyset tuple, or raise ``ValueError``.

    A fingerprint mismatch is the same failure as a malformed payload: the
    caller changed ``q`` or principal scope, and reusing the old cursor must
    not walk a different listing. The HTTP adapter maps this to 400
    ``invalid_cursor`` rather than restarting at page one.
    """
    if not value:
        return None
    try:
        padded = value + "=" * (-len(value) % 4)
        payload = json.loads(base64.urlsafe_b64decode(padded).decode("utf-8"))
        if not isinstance(payload, dict):
            raise ValueError("cursor is not an object")
        if payload.get("fp") != fingerprint:
            raise ValueError("cursor filter mismatch")
        project_id = payload.get("id")
        if not isinstance(project_id, str) or not project_id:
            raise ValueError("missing project id")
        if "t" not in payload:
            raise ValueError("missing last_active_at")
        stamp = payload["t"]
        if stamp is not None:
            stamp = int(stamp)
            # SQLite binds a 64-bit integer. A forged cursor carrying a wider
            # value passed this decoder and then raised OverflowError at bind
            # time, which is not a ValueError and so surfaced as a 500 rather
            # than the documented 400 `invalid_cursor`.
            if not -(2**63) <= stamp < 2**63:
                raise ValueError("last_active_at out of range")
        return (stamp, project_id)
    except ValueError:
        raise
    except Exception as exc:  # noqa: BLE001
        raise ValueError(str(exc)) from exc


def _project_listing_filters(
    *,
    q: str | None,
    visible_to_user_id: str | None,
    before: tuple[int | None, str] | None = None,
    alias: str = "p",
    activity_alias: str = "a",
) -> tuple[list[str], list[Any]]:
    """WHERE conjuncts applied *before* ORDER BY / LIMIT.

    Team visibility is a SQL predicate, not a post-filter: keyset ``has_more``
    is observed from row counts, so filtering after LIMIT turns hidden rows
    into empty slots an attacker can count.
    """
    clauses: list[str] = []
    params: list[Any] = []
    query = (q or "").strip()
    if query:
        clauses.append(
            f"({alias}.name LIKE ? ESCAPE '\\' "
            f"OR {alias}.description LIKE ? ESCAPE '\\')"
        )
        pattern = like_contains(query)
        params.extend([pattern, pattern])
    if visible_to_user_id is not None:
        clauses.append(
            f"({alias}.project_id IN (SELECT project_id FROM project_members "
            f"WHERE user_id=?) OR {alias}.project_id IN "
            f"(SELECT project_id FROM session_owners WHERE user_id=? "
            f"AND project_id IS NOT NULL))"
        )
        params.extend([visible_to_user_id, visible_to_user_id])
    if before is not None:
        before_t, before_id = before
        if before_t is None:
            clauses.append(
                f"({activity_alias}.last_active_at IS NULL "
                f"AND {alias}.project_id < ?)"
            )
            params.append(before_id)
        else:
            clauses.append(
                f"({activity_alias}.last_active_at < ? OR "
                f"({activity_alias}.last_active_at = ? AND {alias}.project_id < ?) "
                f"OR {activity_alias}.last_active_at IS NULL)"
            )
            params.extend([before_t, before_t, before_id])
    return clauses, params


_ACTIVITY_JOIN = (
    "FROM projects p LEFT JOIN ("
    "SELECT project_id, COUNT(*) AS conversation_count, "
    "MAX(updated_at) AS last_active_at FROM frames "
    "WHERE parent_id IS NULL GROUP BY project_id"
    ") a ON a.project_id = p.project_id"
)


class FrameRepository:
    """Own the persisted conversation hierarchy on a Store connection.

    The repository shares ``Store``'s SQLite connection and re-entrant lock.
    Project and frame deletion remain aggregate operations because their legacy
    transaction deletes every row owned by that lifecycle boundary.
    """

    def __init__(
        self,
        connection: sqlite3.Connection,
        lock: Any,
        *,
        clock_ms: Callable[[], int],
        get_frame: Callable[[str], dict | None] | None = None,
        resolve_frame_scope: Callable[..., dict] | None = None,
        get_project: Callable[[str], dict | None] | None = None,
    ) -> None:
        self._connection = connection
        self._lock = lock
        self._clock_ms = clock_ms
        self._get_frame = get_frame
        self._resolve_scope = resolve_frame_scope
        self._get_project = get_project
        self._deletions = SessionDeletionRepository(connection, lock)

    # --- frames ------------------------------------------------------
    def new_frame(
        self,
        *,
        parent_id: str | None = None,
        project_id: str = "default",
        kind: str = "turn",
        name: str | None = None,
        model: str | None = None,
        depth: int = 0,
        status: str = "processing",
    ) -> str:
        frame_id = f"f-{uuid.uuid4().hex[:12]}"
        if parent_id is None:
            root = frame_id
        else:
            get_frame = self._get_frame or self.get_frame
            parent = get_frame(parent_id)
            if parent is None:
                # Preserve the legacy orphan fallback during delete/delegate
                # races: an orphan becomes its own root.
                root = frame_id
            else:
                resolve_scope = self._resolve_scope or self.resolve_frame_scope
                scope = resolve_scope(
                    parent_id,
                    fallback_project=project_id,
                )
                root = scope["root_frame_id"] or frame_id
                project_id = scope["project_id"]
        now = self._clock_ms()
        self._execute(
            "INSERT INTO frames(frame_id,parent_id,project_id,root_frame_id,kind,"
            "name,model,status,depth,created_at,updated_at) "
            "VALUES(?,?,?,?,?,?,?,?,?,?,?)",
            (
                frame_id,
                parent_id,
                project_id,
                root,
                kind,
                name,
                model,
                status,
                depth,
                now,
                now,
            ),
        )
        return frame_id

    def resolve_frame_scope(
        self,
        frame_id: str | None,
        *,
        fallback_project: str = "default",
    ) -> dict:
        """Resolve actor, root session, and root-owned project dynamically."""
        if not frame_id:
            return {
                "frame_id": frame_id,
                "root_frame_id": frame_id,
                "project_id": fallback_project,
            }
        with self._lock:
            frame = self._connection.execute(
                "SELECT frame_id,root_frame_id,project_id FROM frames "
                "WHERE frame_id=?",
                (frame_id,),
            ).fetchone()
            if not frame:
                return {
                    "frame_id": frame_id,
                    "root_frame_id": frame_id,
                    "project_id": fallback_project,
                }
            root_frame_id = frame["root_frame_id"] or frame["frame_id"]
            root = self._connection.execute(
                "SELECT project_id FROM frames WHERE frame_id=?",
                (root_frame_id,),
            ).fetchone()
        return {
            "frame_id": frame["frame_id"],
            "root_frame_id": root_frame_id,
            "project_id": (
                (root["project_id"] if root else None)
                or frame["project_id"]
                or fallback_project
            ),
        }

    def unpin_model(self, frame_id: str) -> None:
        """Drop one frame's model pin so the next send re-binds.

        The by-profile release above covers a deleted profile. This covers the
        other way the pin goes dangling with no click involved: the profile is
        still there and the bound *revision* is not — a database that predates
        the revision history, a rebuilt profile, or seeded builtins dropped on
        first open of an upgraded database. Same 409, same dead end.
        """
        with self._lock:
            self._connection.execute(
                "UPDATE frames SET model_profile_id=NULL, "
                "model_profile_revision=NULL WHERE frame_id=?",
                (str(frame_id),),
            )
            self._connection.commit()

    def release_model_binding(self, profile_id: str) -> int:
        """Unpin every frame bound to a model profile that no longer exists.

        Without this, deleting a profile permanently bricked every session
        pinned to it: `bind_model_revision` answers 409 "choose one to
        continue" whenever the bound profile is missing, and returns before
        reaching either of the two statements that write `model_profile_id` —
        so nothing in the product could choose. `PATCH /frames/{id}` allowlists
        name and task_summary, forking inherits the pin, and profile ids are
        random, so re-creating the profile under the same name did not help
        either. The session's history and artifacts stayed readable and it
        could never be sent to again.

        Clearing the pin drops the session into the path already written for
        frames that predate the pin: recover the configuration from the
        recorded model string, else adopt the active profile. That is a
        supported state, reached on every daemon upgrade.
        """
        profile_id = str(profile_id or "").strip()
        if not profile_id:
            return 0
        with self._lock:
            cursor = self._connection.execute(
                "UPDATE frames SET model_profile_id=NULL, "
                "model_profile_revision=NULL WHERE model_profile_id=?",
                (profile_id,),
            )
            self._connection.commit()
            return int(cursor.rowcount or 0)

    def update_frame(self, frame_id: str, **fields: Any) -> None:
        if not fields:
            return
        fields["updated_at"] = self._clock_ms()
        columns = ", ".join(f"{key}=?" for key in fields)
        self._execute(
            f"UPDATE frames SET {columns} WHERE frame_id=?",
            (*fields.values(), frame_id),
        )

    def add_frame_tokens(
        self,
        frame_id: str,
        *,
        input_tokens: int = 0,
        output_tokens: int = 0,
        cost_usd: float = 0.0,
    ) -> None:
        with self._lock:
            self._connection.execute(
                "UPDATE frames SET input_tokens=COALESCE(input_tokens,0)+?,"
                "output_tokens=COALESCE(output_tokens,0)+?,"
                "cost_usd=COALESCE(cost_usd,0)+?,updated_at=? WHERE frame_id=?",
                (
                    input_tokens,
                    output_tokens,
                    cost_usd,
                    self._clock_ms(),
                    frame_id,
                ),
            )
            self._connection.commit()

    # --- projects ----------------------------------------------------
    def create_project(
        self,
        *,
        name: str,
        description: str = "",
        context: str = "",
        project_id: str | None = None,
        is_example: bool = False,
    ) -> dict:
        project_id = project_id or f"proj_{uuid.uuid4().hex[:12]}"
        now = self._clock_ms()
        self._execute(
            "INSERT OR REPLACE INTO projects(project_id,name,description,context,"
            "is_example,created_at,updated_at) VALUES(?,?,?,?,?,?,?)",
            (
                project_id,
                name,
                description,
                context,
                1 if is_example else 0,
                now,
                now,
            ),
        )
        get_project = self._get_project or self.get_project
        return get_project(project_id) or {}

    def get_project(self, project_id: str) -> dict | None:
        with self._lock:
            row = self._connection.execute(
                "SELECT * FROM projects WHERE project_id=?",
                (project_id,),
            ).fetchone()
        return dict(row) if row else None

    def update_project(self, project_id: str, **fields: Any) -> None:
        if not fields:
            return
        fields["updated_at"] = self._clock_ms()
        columns = ", ".join(f"{key}=?" for key in fields)
        self._execute(
            f"UPDATE projects SET {columns} WHERE project_id=?",
            (*fields.values(), project_id),
        )

    def delete_project(self, project_id: str) -> dict:
        return self._deletions.delete_project(project_id)

    def project_session_ids(self, project_id: str) -> list[str]:
        return self._deletions.project_session_ids(project_id)

    def list_projects(
        self,
        *,
        q: str | None = None,
        limit: int | None = None,
        offset: int | None = None,
        before: tuple[int | None, str] | None = None,
        visible_to_user_id: str | None = None,
    ) -> list[dict]:
        """Return projects with conversation count and last activity.

        One JOIN aggregates ``conversation_count`` / ``last_active_at``;
        callers that also need the exact filtered total issue
        :meth:`count_projects` (two statements per page, never N+1).

        With no paging arguments this is the compatibility full dump, ordered
        by ``projects.updated_at`` as before. A ``limit`` / ``before`` /
        ``offset`` uses the keyset order ``(last_active_at DESC NULLS LAST,
        project_id DESC)``. Empty ``last_active_at`` (no root frames) sorts
        last. Visibility and ``q`` are WHERE conjuncts, not post-filters.
        """
        paging = limit is not None or offset is not None or before is not None
        clauses, params = _project_listing_filters(
            q=q,
            visible_to_user_id=visible_to_user_id,
            before=before if paging else None,
        )
        where = (" WHERE " + " AND ".join(clauses)) if clauses else ""
        if paging:
            order = (
                " ORDER BY (a.last_active_at IS NULL) ASC, "
                "a.last_active_at DESC, p.project_id DESC"
            )
        else:
            order = " ORDER BY p.updated_at DESC"
        sql = (
            "SELECT p.*, COALESCE(a.conversation_count, 0) AS conversation_count, "
            "COALESCE(a.last_active_at, p.updated_at) AS last_active_at, "
            "a.last_active_at AS last_active_raw " + _ACTIVITY_JOIN + where + order
        )
        bind: list[Any] = list(params)
        if limit is not None or offset is not None:
            # OFFSET is only grammatical after LIMIT; a negative LIMIT is
            # SQLite's "unbounded". Emitting the clause under ``limit`` alone
            # switched the order to keyset for an offset-only call and then
            # returned the whole table from row 0, with no error.
            sql += " LIMIT ?"
            bind.append(-1 if limit is None else max(1, int(limit)))
            if offset is not None:
                sql += " OFFSET ?"
                bind.append(max(0, int(offset)))
        with self._lock:
            rows = self._connection.execute(sql, tuple(bind)).fetchall()
        return [dict(row) for row in rows]

    def count_projects(
        self,
        *,
        q: str | None = None,
        visible_to_user_id: str | None = None,
    ) -> int:
        """Exact count after permission filter and ``q``, not after LIMIT."""
        clauses, params = _project_listing_filters(
            q=q,
            visible_to_user_id=visible_to_user_id,
        )
        where = (" WHERE " + " AND ".join(clauses)) if clauses else ""
        sql = "SELECT COUNT(*) AS n FROM projects p" + where
        with self._lock:
            row = self._connection.execute(sql, tuple(params)).fetchone()
        return int(row["n"] or 0)

    # --- messages ----------------------------------------------------
    def add_message(
        self,
        *,
        root_frame_id: str,
        branch_id: str | None = None,
        role: str,
        content: str,
        frame_id: str | None = None,
        metadata: dict | None = None,
        created_at: int | None = None,
    ) -> dict:
        now = created_at if created_at is not None else self._clock_ms()
        message_id = f"m-{uuid.uuid4().hex[:12]}"
        branch_id = branch_id or root_frame_id
        with self._lock:
            seq = self._connection.execute(
                "SELECT COALESCE(MAX(seq),-1)+1 AS s FROM messages "
                "WHERE root_frame_id=?",
                (root_frame_id,),
            ).fetchone()["s"]
            self._connection.execute(
                "INSERT INTO messages(message_id,root_frame_id,branch_id,frame_id,"
                "seq,role,content,metadata,created_at) VALUES(?,?,?,?,?,?,?,?,?)",
                (
                    message_id,
                    root_frame_id,
                    branch_id,
                    frame_id,
                    seq,
                    role,
                    content,
                    json.dumps(metadata, ensure_ascii=False) if metadata else None,
                    now,
                ),
            )
            self._connection.commit()
        return {
            "message_id": message_id,
            "root_frame_id": root_frame_id,
            "branch_id": branch_id,
            "seq": seq,
            "role": role,
            "content": content,
            "created_at": now,
        }

    def update_message_metadata(self, message_id: str, patch: dict) -> dict | None:
        """Merge keys into one message's metadata blob after it is written.

        The user message has to exist *before* its `@`-references are resolved:
        resolving may materialise a sibling session's file into this workspace,
        and the message row plus the fork checkpoint taken from it are the
        branch point that must be durable before anything writes. So the
        structured references arrive a moment late and are merged in rather
        than passed at INSERT.

        Merged, not replaced: the blob is shared with whatever else a caller
        stamps on a message, and an overwrite here would silently drop it.
        """
        with self._lock:
            row = self._connection.execute(
                "SELECT metadata FROM messages WHERE message_id=?",
                (message_id,),
            ).fetchone()
            if row is None:
                return None
            try:
                current = json.loads(row["metadata"] or "{}")
            except (TypeError, ValueError):
                current = {}
            if not isinstance(current, dict):
                current = {}
            current.update(dict(patch or {}))
            self._connection.execute(
                "UPDATE messages SET metadata=? WHERE message_id=?",
                (json.dumps(current, ensure_ascii=False), message_id),
            )
            self._connection.commit()
        return current

    def promote_candidate_message(
        self,
        *,
        message_id: str,
        root_frame_id: str,
        branch_id: str,
        frame_id: str | None,
        expected_content: str,
        content: str,
        metadata: Mapping[str, Any],
    ) -> dict:
        """CAS-promote one exact provisional assistant message.

        Stage 4 persists a canonical candidate before the reviewer runs.  A
        repair may later replace its text, but it must never guess which
        assistant row is newest.  Scope, role, candidate state, and the exact
        previous bytes are therefore checked in the same write transaction.
        """

        if not isinstance(message_id, str) or not message_id.strip():
            raise ValueError("message_id must be a non-empty string")
        if not isinstance(root_frame_id, str) or not root_frame_id.strip():
            raise ValueError("root_frame_id must be a non-empty string")
        if not isinstance(branch_id, str) or not branch_id.strip():
            raise ValueError("branch_id must be a non-empty string")
        if frame_id is not None and (
            not isinstance(frame_id, str) or not frame_id.strip()
        ):
            raise ValueError("frame_id must be a non-empty string")
        if not isinstance(expected_content, str):
            raise ValueError("candidate message expected content must be text")
        if not isinstance(content, str) or not content.strip():
            raise ValueError("promoted candidate content must be non-empty")
        if not isinstance(metadata, Mapping):
            raise ValueError("candidate verdict metadata must be an object")
        try:
            metadata_json = json.dumps(
                dict(metadata),
                ensure_ascii=False,
                sort_keys=True,
                separators=(",", ":"),
                allow_nan=False,
            )
            verdict_metadata = json.loads(metadata_json)
        except (TypeError, ValueError) as error:
            raise ValueError("candidate verdict metadata must be JSON-safe") from error
        if not isinstance(verdict_metadata, dict):  # pragma: no cover - round trip
            raise ValueError("candidate verdict metadata must be an object")
        verdict_digest_key = "candidate_verdict_metadata_sha256"
        if (
            "completion_delivery" in verdict_metadata
            or verdict_digest_key in verdict_metadata
        ):
            raise ValueError("candidate verdict metadata contains a reserved key")
        verdict = verdict_metadata.get("review_status")
        if verdict not in {
            "verified",
            "completed_with_issues",
            "review_unavailable",
        }:
            raise ValueError("candidate verdict metadata has no terminal review status")
        candidate_sha256 = hashlib.sha256(expected_content.encode("utf-8")).hexdigest()
        promoted_sha256 = hashlib.sha256(content.encode("utf-8")).hexdigest()
        if verdict_metadata.get("candidate_content_sha256") not in (
            None,
            candidate_sha256,
        ):
            raise ValueError("candidate verdict metadata digest changed")
        if verdict_metadata.get("reviewed_content_sha256") not in (
            None,
            promoted_sha256,
        ):
            raise ValueError("reviewed candidate metadata digest changed")
        verdict_metadata_sha256 = hashlib.sha256(
            metadata_json.encode("utf-8")
        ).hexdigest()

        with self._lock:
            if self._connection.in_transaction:
                raise RuntimeError(
                    "candidate promotion requires a clean SQLite transaction"
                )
            self._connection.execute("BEGIN IMMEDIATE")
            try:
                row = self._connection.execute(
                    "SELECT root_frame_id,branch_id,frame_id,role,content,metadata,"
                    "created_at,seq FROM messages WHERE message_id=?",
                    (message_id,),
                ).fetchone()
                if (
                    row is None
                    or row["root_frame_id"] != root_frame_id
                    or row["branch_id"] != branch_id
                    or row["frame_id"] != frame_id
                    or row["role"] != "assistant"
                ):
                    raise RuntimeError("candidate message scope changed")
                try:
                    current = json.loads(row["metadata"] or "{}")
                except (TypeError, ValueError) as error:
                    raise RuntimeError(
                        "candidate message metadata is invalid"
                    ) from error
                if not isinstance(current, dict):
                    raise RuntimeError("candidate message metadata is invalid")
                if "completion_delivery" in current:
                    raise RuntimeError(
                        "completion delivery candidates require the delivery CAS"
                    )

                if current.get("review_status") == "candidate":
                    if row["content"] != expected_content:
                        raise RuntimeError("candidate message content changed")
                    bound_candidate_sha256 = current.get("candidate_content_sha256")
                    if bound_candidate_sha256 not in (None, candidate_sha256):
                        raise RuntimeError("candidate message digest changed")
                    if verdict_digest_key in current:
                        raise RuntimeError("candidate verdict digest is invalid")
                    promoted = dict(current)
                    promoted.update(verdict_metadata)
                    promoted["candidate_content_sha256"] = candidate_sha256
                    promoted[verdict_digest_key] = verdict_metadata_sha256
                    encoded = json.dumps(
                        promoted,
                        ensure_ascii=False,
                        sort_keys=True,
                        separators=(",", ":"),
                        allow_nan=False,
                    )
                    cursor = self._connection.execute(
                        "UPDATE messages SET content=?,metadata=? WHERE message_id=? "
                        "AND root_frame_id=? AND branch_id=? AND frame_id IS ? "
                        "AND role='assistant' AND content=? AND metadata IS ?",
                        (
                            content,
                            encoded,
                            message_id,
                            root_frame_id,
                            branch_id,
                            frame_id,
                            expected_content,
                            row["metadata"],
                        ),
                    )
                    if cursor.rowcount != 1:
                        raise RuntimeError("candidate message promotion lost its CAS")
                    current = promoted
                elif (
                    row["content"] == content
                    and current.get("candidate_content_sha256") == candidate_sha256
                    and current.get(verdict_digest_key) == verdict_metadata_sha256
                    and all(
                        current.get(key) == value
                        for key, value in verdict_metadata.items()
                    )
                ):
                    # The exact retry of a promotion that already committed is
                    # a read.  The durable candidate digest prevents another
                    # caller laundering different expected bytes through it.
                    pass
                else:
                    raise RuntimeError("candidate message is not provisional")
                self._connection.commit()
            except Exception:
                if self._connection.in_transaction:
                    self._connection.rollback()
                raise
        return {
            "message_id": message_id,
            "root_frame_id": root_frame_id,
            "branch_id": branch_id,
            "frame_id": frame_id,
            "seq": int(row["seq"]),
            "role": "assistant",
            "content": content,
            "metadata": current,
            "created_at": int(row["created_at"]),
        }

    def list_messages(
        self,
        root_frame_id: str,
        *,
        branch_id: str | None = None,
        start: int = 0,
        limit: int | None = 300,
        before_seq: int | None = None,
        newest_first: bool = False,
    ) -> list[dict]:
        """Messages for one session, oldest-first by default.

        ``newest_first`` with ``before_seq`` is the pagination a conversation
        actually needs. Opening a 640-message session returned messages 0-299:
        the *oldest* page, with the newest 340 simply not present. A reader
        arriving at a long session wants the end of it, and then to walk
        backwards.

        `before_seq` is a keyset cursor, not an offset. Ordering newest-first
        and paging by offset would skew on every arriving message, because a
        new message shifts what "offset 50 from the newest" means — the exact
        problem the session list solved with `(created_at, frame_id)`. `seq` is
        already monotonic and unique per root frame, so it needs no tiebreaker.
        """
        where = "root_frame_id=?"
        params: list[Any] = [root_frame_id]
        if branch_id is not None:
            where += " AND branch_id=?"
            params.append(branch_id)
        if before_seq is not None:
            where += " AND seq<?"
            params.append(int(before_seq))
        order = " ORDER BY seq DESC" if newest_first else " ORDER BY seq ASC"
        suffix = ""
        if limit is not None:
            # OFFSET stays for the oldest-first callers that still use `start`.
            # It is meaningless alongside a keyset cursor, so a caller passing
            # both gets the cursor honoured and the offset ignored rather than
            # a silently wrong page.
            suffix = " LIMIT ?" + ("" if before_seq is not None else " OFFSET ?")
            params.append(max(0, int(limit)))
            if before_seq is None:
                params.append(max(0, int(start)))
        with self._lock:
            rows = self._connection.execute(
                "SELECT role,content,metadata,created_at,seq FROM messages WHERE "
                + where
                + order
                + suffix,
                tuple(params),
            ).fetchall()
        values = [dict(row) for row in rows]
        if limit is None and start and before_seq is None:
            return values[start:]
        return values

    def list_message_boundaries(
        self,
        root_frame_id: str,
        *,
        branch_id: str | None = None,
        start: int = 0,
        limit: int | None = 300,
    ) -> list[dict]:
        """Return public message identities plus exact fork proof, if present."""

        where = "m.root_frame_id=?"
        params: list[Any] = [root_frame_id]
        if branch_id is not None:
            where += " AND m.branch_id=?"
            params.append(branch_id)
        suffix = ""
        if limit is not None:
            suffix = " LIMIT ? OFFSET ?"
            params.extend((max(0, int(limit)), max(0, int(start))))
        with self._lock:
            rows = self._connection.execute(
                # `m.metadata` rides along because the structured artifact
                # references a message was sent with live there, and the
                # conversation route is the only reader that can restore them
                # on reopen. Without it the client had a `@name#v-id` string to
                # re-parse and no way to learn which version was actually read.
                "SELECT m.message_id,m.root_frame_id,m.branch_id,m.seq,m.role,"
                "m.content,m.metadata,m.created_at,(SELECT "
                "c.checkpoint_id FROM session_checkpoints AS c WHERE "
                "c.root_frame_id=m.root_frame_id AND c.source_kind='message' "
                "AND c.source_id=m.message_id LIMIT 1) AS fork_checkpoint_id "
                "FROM messages AS m WHERE " + where + " ORDER BY m.seq ASC" + suffix,
                tuple(params),
            ).fetchall()
        values = [dict(row) for row in rows]
        return values[start:] if limit is None and start else values

    def message_count(self, root_frame_id: str) -> int:
        with self._lock:
            row = self._connection.execute(
                "SELECT COUNT(*) AS n FROM messages WHERE root_frame_id=?",
                (root_frame_id,),
            ).fetchone()
        return row["n"] or 0

    def has_message_history(self) -> bool:
        """Whether any session on this install has ever held a message."""
        with self._lock:
            row = self._connection.execute("SELECT 1 FROM messages LIMIT 1").fetchone()
        return row is not None

    def has_execution_history(self) -> bool:
        """Whether any Cell on this install has ever been executed.

        A Notebook REPL cell is recorded here and never as a message, so this
        is the history an install used only through the Notebook has.
        """
        with self._lock:
            row = self._connection.execute(
                "SELECT 1 FROM execution_log LIMIT 1"
            ).fetchone()
        return row is not None

    def cell_count(self, root_frame_id: str) -> int:
        with self._lock:
            row = self._connection.execute(
                "SELECT COUNT(*) AS n FROM execution_log WHERE root_frame_id=?",
                (root_frame_id,),
            ).fetchone()
        return row["n"] or 0

    def latest_state_revision(self, root_frame_id: str) -> int:
        """Return the durable session revision cursor used for the next Cell.

        Indexed historical rows are authoritative.  A count fallback reserves
        ordinals for older unindexed rows without fabricating per-row revision
        metadata for them.
        """

        with self._lock:
            row = self._connection.execute(
                "SELECT (SELECT COUNT(*) FROM execution_log WHERE root_frame_id=?) "
                "AS n,(SELECT MAX(COALESCE(state_revision,cell_index,0)) FROM "
                "execution_log WHERE root_frame_id=?) AS logged_revision,"
                "(SELECT MAX(a.state_revision) FROM execution_attempts AS a "
                "JOIN action_groups AS g ON g.group_id=a.group_id "
                "WHERE g.root_frame_id=?) AS attempt_revision",
                (root_frame_id, root_frame_id, root_frame_id),
            ).fetchone()
        return max(
            int(row["n"] or 0),
            int(row["logged_revision"] or 0),
            int(row["attempt_revision"] or 0),
        )

    # --- semantic activity steps ------------------------------------
    def add_step(
        self,
        *,
        step_id: str,
        frame_id: str,
        kind: str,
        title: str | None = None,
        input: dict | None = None,
        status: str = "running",
    ) -> dict:
        now = self._clock_ms()
        with self._lock:
            seq = self._connection.execute(
                "SELECT COALESCE(MAX(seq),-1)+1 AS s FROM frame_steps "
                "WHERE frame_id=?",
                (frame_id,),
            ).fetchone()["s"]
            self._connection.execute(
                "INSERT INTO frame_steps(step_id,frame_id,seq,kind,title,input,"
                "output,status,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?)",
                (
                    step_id,
                    frame_id,
                    seq,
                    kind,
                    title,
                    (
                        json.dumps(input, ensure_ascii=False, default=str)
                        if input is not None
                        else None
                    ),
                    None,
                    status,
                    now,
                    now,
                ),
            )
            self._connection.commit()
        return {"step_id": step_id, "seq": seq, "created_at": now}

    def update_step(
        self,
        step_id: str,
        *,
        status: str | None = None,
        output: dict | None = None,
        title: str | None = None,
        summary: str | None = None,
    ) -> None:
        now = self._clock_ms()
        sets, params = [], []
        if status is not None:
            sets.append("status=?")
            params.append(status)
        if title is not None:
            sets.append("title=?")
            params.append(title)
        if summary is not None:
            sets.append("summary=?")
            params.append(summary)
        if output is not None:
            sets.append("output=?")
            params.append(json.dumps(output, ensure_ascii=False, default=str))
        sets.append("updated_at=?")
        params.append(now)
        params.append(step_id)
        with self._lock:
            self._connection.execute(
                f"UPDATE frame_steps SET {','.join(sets)} WHERE step_id=?",
                params,
            )
            self._connection.commit()

    def list_steps(
        self,
        frame_id: str,
        *,
        start: int = 0,
        limit: int = 800,
    ) -> list[dict]:
        with self._lock:
            rows = self._connection.execute(
                "SELECT step_id,seq,kind,title,summary,input,output,status,"
                "created_at FROM frame_steps WHERE frame_id=? ORDER BY seq ASC "
                "LIMIT ? OFFSET ?",
                (frame_id, limit, max(0, start)),
            ).fetchall()
        steps = []
        for row in rows:
            step = dict(row)
            for key in ("input", "output"):
                if step.get(key):
                    try:
                        step[key] = json.loads(step[key])
                    except (ValueError, TypeError):
                        pass
            steps.append(step)
        self._read_legacy_capture_environments(steps)
        return steps

    def _read_legacy_capture_environments(self, steps: list[dict]) -> None:
        """Read a pre-0.3.0 Cell capture step's runtime through its Cell.

        The Web "Saving ..." step was written with the Python kernel's label
        whatever language the Cell was, so an R Cell's card said "environment
        python" while the artifact's env snapshot for the same file said "r".
        New rows record ``language`` next to ``environment`` and are read as
        written. A row without it is read through the Cell that produced its
        artifact versions (or, failing that, the generation their snapshot
        names), on the one read every consumer shares -- the steps route, the
        review, the session package and the share projection -- rather than
        by rewriting what was stored.

        Only that shape is corrected: a Python-labelled row whose every version
        resolves to an R Cell. Anything unresolved or mixed is left as stored.
        """
        pending: dict[int, list[str]] = {}
        for position, step in enumerate(steps):
            data = step.get("input")
            output = step.get("output")
            if (
                step.get("kind") != "artifact"
                or not isinstance(data, dict)
                or not isinstance(output, dict)
                or "language" in data
                or not str(data.get("environment") or "").startswith("python")
            ):
                continue
            versions = [
                str(item["version_id"])
                for item in output.get("artifacts") or []
                if isinstance(item, dict) and item.get("version_id")
            ]
            if versions:
                pending[position] = versions
        if not pending:
            return
        wanted = sorted({version for group in pending.values() for version in group})
        r_labels: dict[str, str | None] = {}
        for offset in range(0, len(wanted), 500):
            chunk = wanted[offset : offset + 500]
            marks = ",".join("?" * len(chunk))
            with self._lock:
                try:
                    rows = self._connection.execute(
                        "SELECT v.version_id, x.language AS cell_language, "
                        "x.kernel_id, g.language AS generation_language "
                        "FROM artifact_versions v "
                        "LEFT JOIN execution_log x "
                        "ON x.producing_cell_id=v.producing_cell_id "
                        "LEFT JOIN env_snapshots s ON s.snapshot_id=v.env_snapshot_id "
                        "LEFT JOIN kernel_generations g "
                        "ON g.generation_id=s.generation_id "
                        f"WHERE v.version_id IN ({marks})",
                        tuple(chunk),
                    ).fetchall()
                except sqlite3.Error:
                    return
            for row in rows:
                languages = {
                    str(value).strip().lower()
                    for value in (row["cell_language"], row["generation_language"])
                    if value and str(value).strip()
                }
                if languages != {"r"}:
                    r_labels[row["version_id"]] = None
                    continue
                kernel_id = str(row["kernel_id"] or "").strip()
                r_labels[row["version_id"]] = (
                    kernel_id
                    if kernel_id == "r" or kernel_id.startswith("r — ")
                    else "r"
                )
        for position, versions in pending.items():
            labels = {r_labels.get(version) for version in versions}
            if None in labels:
                continue
            steps[position]["input"]["environment"] = (
                labels.pop() if len(labels) == 1 else "r"
            )

    def step_count(self, frame_id: str) -> int:
        with self._lock:
            row = self._connection.execute(
                "SELECT COUNT(*) AS n FROM frame_steps WHERE frame_id=?",
                (frame_id,),
            ).fetchone()
        return row["n"] or 0

    # --- frame browse/detail/search ----------------------------------
    def browse_frames(
        self,
        *,
        project_id: str | None = "default",
        status: str | None = None,
        roots_only: bool = True,
        limit: int = 50,
        before: tuple[int, str] | None = None,
        visible_to_user_id: str | None = None,
    ) -> list[dict]:
        """Newest-first page of frames.

        ``before`` is a keyset cursor — the ``(created_at, frame_id)`` of the
        last row of the previous page — not an offset. An offset would skip or
        repeat rows whenever a frame is created or deleted between pages, which
        for a session list is routine rather than exotic.

        The ``frame_id`` tiebreaker is what makes the cursor sound at all:
        ``created_at`` is a millisecond timestamp and two sessions created in
        the same millisecond are not rare (a script, a test, a fast fork). With
        ordering by timestamp alone their relative order is undefined, so a
        cursor could land in the middle of a tie and silently drop the rest of
        it.
        """
        clauses, params = [], []
        if project_id and project_id != "all":
            clauses.append("project_id=?")
            params.append(project_id)
        if status:
            clauses.append("status=?")
            params.append(status)
        if roots_only:
            clauses.append("parent_id IS NULL")
        if visible_to_user_id is not None:
            clause, clause_params = visible_session_clause(visible_to_user_id)
            clauses.append(clause)
            params.extend(clause_params)
        if before is not None:
            before_created, before_id = before
            clauses.append("(created_at < ? OR (created_at = ? AND frame_id < ?))")
            params.extend([before_created, before_created, before_id])
        where = (" WHERE " + " AND ".join(clauses)) if clauses else ""
        with self._lock:
            rows = self._connection.execute(
                "SELECT frame_id,parent_id,root_frame_id,project_id,kind,name,"
                "task_summary,model,status,depth,input_tokens,output_tokens,"
                "cost_usd,created_at,updated_at FROM frames"
                + where
                + " ORDER BY created_at DESC, frame_id DESC LIMIT ?",
                (*params, limit),
            ).fetchall()
        return [dict(row) for row in rows]

    def frame_detail(
        self,
        frame_id: str,
        *,
        page: int = 0,
        page_size: int = 50,
        visible_to_user_id: str | None = None,
    ) -> dict | None:
        """Return frame metadata, paged cells, and direct children.

        `visible_to_user_id` scopes by the same rule the listings use. It is
        applied to the *first* query, so a frame this caller may not see
        returns None before any cell code or stdout is read -- the rows this
        method returns are the most sensitive in the database.
        """
        scope_clause = ""
        scope_params: list = []
        if visible_to_user_id is not None:
            scope_clause, scope_params = visible_session_clause(visible_to_user_id)
            scope_clause = " AND " + scope_clause
        with self._lock:
            frame = self._connection.execute(
                f"SELECT * FROM frames WHERE frame_id=?{scope_clause}",
                (frame_id, *scope_params),
            ).fetchone()
            if frame is None:
                # Indistinguishable from "no such frame", deliberately.
                return None
            total = self._connection.execute(
                "SELECT COUNT(*) AS n FROM execution_log WHERE frame_id=?",
                (frame_id,),
            ).fetchone()["n"]
            cells = self._connection.execute(
                "SELECT producing_cell_id,cell_seq,origin,code,stdout,stderr,"
                "error,interrupted,wall_s,cpu_s,created_at FROM execution_log "
                "WHERE frame_id=? ORDER BY created_at ASC LIMIT ? OFFSET ?",
                (frame_id, page_size, page * page_size),
            ).fetchall()
            children = self._connection.execute(
                # frame_id breaks created_at ties so same-millisecond delegate
                # siblings keep one stable order (child export ordinals rely
                # on this being deterministic across store generations).
                "SELECT frame_id,kind,name,status,depth FROM frames "
                "WHERE parent_id=? ORDER BY created_at ASC, frame_id ASC",
                (frame_id,),
            ).fetchall()
        page_count = max(1, (total + page_size - 1) // page_size)
        return {
            "frame": dict(frame),
            "cells": [dict(cell) for cell in cells],
            "children": [dict(child) for child in children],
            "page": page,
            "page_size": page_size,
            "n_pages": page_count,
            "total_cells": total,
            "last_page": page >= page_count - 1,
        }

    def search_frames(
        self,
        pattern: str,
        *,
        project_id: str | None = "default",
        limit: int = 50,
        visible_to_user_id: str | None = None,
    ) -> list[dict]:
        """Regex-search frame names and cell code/stdout.

        `visible_to_user_id` narrows the *outer* query, so the per-row
        `SELECT code,stdout` below never runs for a session this caller may
        not see. That ordering is the whole point here: this method reads
        the code somebody wrote and the output it printed, and a filter
        applied to the returned matches would have read them first.

        Note `project_id="all"` drops the project clause entirely, which is
        exactly how `host.frames(pattern=..., project_id="all")` became a
        regex search over every tenant's cells.
        """
        regex = re.compile(pattern, re.IGNORECASE)
        clauses, params = [], []
        if project_id and project_id != "all":
            clauses.append("f.project_id=?")
            params.append(project_id)
        if visible_to_user_id is not None:
            clause, clause_params = visible_session_clause(
                visible_to_user_id, table="f"
            )
            clauses.append(clause)
            params.extend(clause_params)
        where = (" WHERE " + " AND ".join(clauses)) if clauses else ""
        with self._lock:
            rows = self._connection.execute(
                "SELECT DISTINCT f.frame_id,f.kind,f.name,f.status,f.depth,"
                "f.project_id,f.created_at FROM frames f "
                "LEFT JOIN execution_log e ON e.frame_id=f.frame_id"
                + where
                + " ORDER BY f.created_at DESC",
                tuple(params),
            ).fetchall()
            matches = []
            for row in rows:
                haystack = [row["name"] or ""]
                cells = self._connection.execute(
                    "SELECT code,stdout FROM execution_log WHERE frame_id=?",
                    (row["frame_id"],),
                ).fetchall()
                for cell in cells:
                    haystack.append(cell["code"] or "")
                    haystack.append(cell["stdout"] or "")
                if regex.search("\n".join(haystack)):
                    matches.append(dict(row))
                if len(matches) >= limit:
                    break
        return matches

    # --- execution log ----------------------------------------------
    def log_cell(
        self,
        *,
        frame_id: str | None,
        code: str,
        result: dict,
        origin: str = "agent",
        cell_seq: int | None = None,
        project_id: str = "default",
        root_frame_id: str | None = None,
        cell_index: int | None = None,
        state_revision: int | None = None,
        kernel_id: str = "python",
        language: str = "python",
        visibility: str | None = None,
        pin: bool = False,
        replay_policy: str | None = None,
        figures: list | None = None,
        files_read: list | None = None,
        files_written: list | None = None,
        generation_id: str | None = None,
    ) -> str:
        cell_id = result.get("id") or f"c-{uuid.uuid4().hex[:12]}"
        if visibility is None:
            visibility = default_visibility(origin)
        if visibility not in VISIBILITIES:
            raise ValueError(f"unknown Cell visibility: {visibility}")
        if type(pin) is not bool:
            raise TypeError("pin must be a boolean")
        if replay_policy is None:
            replay_policy = default_replay_policy(visibility)
        if replay_policy not in REPLAY_POLICIES:
            raise ValueError(f"unknown Cell replay_policy: {replay_policy}")
        dependencies = analyze_code(code, language)
        usage = result.get("usage") or {}
        status = (
            "interrupted"
            if result.get("interrupted")
            else ("error" if result.get("error") else "ok")
        )
        with self._lock:
            reserved = self._connection.execute(
                "SELECT state_revision FROM execution_attempts "
                "WHERE producing_cell_id=? AND state_revision IS NOT NULL "
                "ORDER BY attempt_ordinal DESC LIMIT 1",
                (cell_id,),
            ).fetchone()
        reserved_revision = (
            int(reserved["state_revision"]) if reserved is not None else None
        )
        if state_revision is None:
            state_revision = (
                reserved_revision if reserved_revision is not None else cell_index
            )
        elif reserved_revision is not None and state_revision != reserved_revision:
            raise ValueError("state_revision must match the durable execution attempt")
        latest_revision = (
            self.latest_state_revision(root_frame_id) if root_frame_id else 0
        )
        if state_revision is None and root_frame_id:
            state_revision = latest_revision + 1
        if state_revision is not None:
            if isinstance(state_revision, bool) or not isinstance(state_revision, int):
                raise TypeError("state_revision must be an integer")
            if state_revision < 1:
                raise ValueError("state_revision must be positive")
            if (
                root_frame_id
                and reserved_revision is None
                and state_revision <= latest_revision
            ):
                raise ValueError(
                    "state_revision must advance the session revision cursor"
                )
        # Execution history is an append-only audit record. A duplicate Cell ID
        # means the caller is attempting to overwrite an already-observed
        # execution, which must fail loudly instead of silently replacing its
        # source, output, error, provenance, or timestamp.
        self._execute(
            "INSERT INTO execution_log(producing_cell_id,frame_id,"
            "root_frame_id,project_id,cell_seq,cell_index,state_revision,"
            "kernel_id,language,"
            "status,origin,code,code_hash,visibility,pin,replay_policy,"
            "variable_reads,variable_writes,variable_deletes,"
            "mutation_uncertain,stdout,stderr,error,figures,files_read,"
            "files_written,interrupted,wall_s,cpu_s,peak_rss_kb,created_at,"
            "generation_id) "
            "VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
            (
                cell_id,
                frame_id,
                root_frame_id,
                project_id,
                cell_seq,
                cell_index,
                state_revision,
                kernel_id,
                language,
                status,
                origin,
                code,
                dependencies.code_hash,
                visibility,
                1 if pin else 0,
                replay_policy,
                json.dumps(dependencies.reads, ensure_ascii=False),
                json.dumps(dependencies.writes, ensure_ascii=False),
                json.dumps(dependencies.deletes, ensure_ascii=False),
                1 if dependencies.uncertain else 0,
                result.get("stdout"),
                result.get("stderr"),
                result.get("error"),
                json.dumps(figures or [], ensure_ascii=False),
                json.dumps(files_read or [], ensure_ascii=False),
                json.dumps(files_written or [], ensure_ascii=False),
                1 if result.get("interrupted") else 0,
                usage.get("wall_s"),
                usage.get("cpu_s"),
                usage.get("peak_rss_kb"),
                self._clock_ms(),
                generation_id,
            ),
        )
        return cell_id

    def list_cells(
        self, root_frame_id: str, *, branch_id: str | None = None
    ) -> list[dict]:
        """Return a session's notebook execution log oldest first."""
        branch_filter = ""
        params: list[Any] = [root_frame_id]
        if branch_id is not None:
            branch_filter = (
                " AND (EXISTS (SELECT 1 FROM execution_attempts AS ba "
                "JOIN action_groups AS bg ON bg.group_id=ba.group_id "
                "WHERE ba.producing_cell_id=e.producing_cell_id "
                "AND bg.root_frame_id=? AND bg.branch_id=?)"
            )
            params.extend((root_frame_id, branch_id))
            if branch_id == root_frame_id:
                branch_filter += (
                    " OR NOT EXISTS (SELECT 1 FROM execution_attempts AS legacy "
                    "WHERE legacy.producing_cell_id=e.producing_cell_id)"
                )
            branch_filter += ")"
        with self._lock:
            rows = self._connection.execute(
                "SELECT e.producing_cell_id,e.cell_index,e.state_revision,"
                "e.kernel_id,e.language,e.status,e.origin,e.code,e.stdout,"
                "e.code_hash,e.visibility,e.pin,e.replay_policy,"
                "e.variable_reads,e.variable_writes,e.variable_deletes,"
                "e.mutation_uncertain,e.stderr,e.error,e.figures,e.files_read,e.files_written,"
                "e.interrupted,"
                "e.cpu_s,e.peak_rss_kb,e.created_at,COALESCE((SELECT a.generation_id "
                "FROM execution_attempts AS a WHERE a.producing_cell_id="
                "e.producing_cell_id AND a.generation_id IS NOT NULL "
                "ORDER BY a.attempt_ordinal DESC LIMIT 1),e.generation_id) "
                "AS generation_id "
                "FROM execution_log AS e WHERE e.root_frame_id=? " + branch_filter + " "
                "ORDER BY COALESCE(e.state_revision,e.cell_index) ASC,"
                "e.created_at ASC,e.producing_cell_id ASC",
                tuple(params),
            ).fetchall()
        cells = []
        for row in rows:
            cell = dict(row)
            for key in (
                "figures",
                "files_read",
                "files_written",
                "variable_reads",
                "variable_writes",
                "variable_deletes",
            ):
                try:
                    cell[key] = json.loads(cell.get(key) or "[]")
                except (TypeError, ValueError):
                    cell[key] = []
            for key in (
                "variable_reads",
                "variable_writes",
                "variable_deletes",
            ):
                cell[key] = list(normalize_string_list(cell.get(key)))
            cell["pin"] = bool(cell.get("pin"))
            cell["mutation_uncertain"] = bool(cell.get("mutation_uncertain"))
            cell["interrupted"] = bool(cell.get("interrupted"))
            if cell.get("state_revision") is None:
                cell["state_revision"] = cell.get("cell_index")
            cells.append(cell)
        return cells

    def list_cell_outputs(self, root_frame_id: str) -> list[dict]:
        """Per-cell ``files_written``/``figures`` for one session, decoded.

        The submission-evidence gatherer reads only these two lists;
        ``list_cells`` materializes every cell's code and stdout (up to 1M
        chars per cell) plus a correlated attempts subquery — all discarded
        on that path, which runs while the kernel worker blocks on the
        host-call lock.
        """
        with self._lock:
            rows = self._connection.execute(
                "SELECT files_written,figures FROM execution_log "
                "WHERE root_frame_id=?",
                (root_frame_id,),
            ).fetchall()
        cells = []
        for row in rows:
            cell = dict(row)
            for key in ("files_written", "figures"):
                try:
                    cell[key] = json.loads(cell.get(key) or "[]")
                except (TypeError, ValueError):
                    cell[key] = []
            cells.append(cell)
        return cells

    def cell_detail(self, producing_cell_id: str) -> dict | None:
        # Aliased apart from the raw column: `e.*` now expands to a
        # `generation_id` of its own, and sqlite3.Row resolves a duplicated
        # name to the first (raw) column — which would shadow the
        # attempt-derived binding for Web cells.
        with self._lock:
            row = self._connection.execute(
                "SELECT e.*,COALESCE((SELECT a.generation_id "
                "FROM execution_attempts AS a "
                "WHERE a.producing_cell_id=e.producing_cell_id "
                "AND a.generation_id IS NOT NULL ORDER BY a.attempt_ordinal DESC "
                "LIMIT 1),e.generation_id) AS resolved_generation_id "
                "FROM execution_log AS e "
                "WHERE e.producing_cell_id=?",
                (producing_cell_id,),
            ).fetchone()
        if not row:
            return None
        cell = dict(row)
        cell["generation_id"] = cell.pop("resolved_generation_id")
        for key in (
            "figures",
            "files_read",
            "files_written",
            "variable_reads",
            "variable_writes",
            "variable_deletes",
        ):
            try:
                cell[key] = json.loads(cell.get(key) or "[]")
            except (TypeError, ValueError):
                cell[key] = []
        for key in (
            "variable_reads",
            "variable_writes",
            "variable_deletes",
        ):
            cell[key] = list(normalize_string_list(cell.get(key)))
        cell["pin"] = bool(cell.get("pin"))
        cell["mutation_uncertain"] = bool(cell.get("mutation_uncertain"))
        if cell.get("state_revision") is None:
            cell["state_revision"] = cell.get("cell_index")
        return cell

    def delete_frame(self, frame_id: str) -> dict[str, Any]:
        """Delete one complete root-session aggregate in a single transaction."""

        return self._deletions.delete_session(frame_id)

    def get_frame(self, frame_id: str) -> dict | None:
        with self._lock:
            row = self._connection.execute(
                "SELECT * FROM frames WHERE frame_id=?",
                (frame_id,),
            ).fetchone()
        return dict(row) if row else None

    def _execute(self, sql: str, params: tuple = ()) -> None:
        with self._lock:
            self._connection.execute(sql, params)
            self._connection.commit()


__all__ = [
    "FrameRepository",
    "PROJECT_PAGE_DEFAULT",
    "PROJECT_PAGE_MAX",
    "PROJECT_Q_MAX_CODEPOINTS",
    "decode_project_cursor",
    "encode_project_cursor",
    "project_filter_fingerprint",
]