_utils/_share_contract.py

"""
Security contract for server-backed conversation shares.

The browser is untrusted.  Share requests carry structured conversation data and
an allowlisted representation id; callers never choose response MIME types or
submit rendered HTML for the server to host.
"""

from __future__ import annotations

import hashlib
import hmac
import html
import json
import math
import re
import secrets
from typing import Any
from urllib.parse import urlsplit, urlunsplit

SHARE_SCHEMA_VERSION = "2.1"
SHARE_ACCEPTED_SCHEMA_VERSIONS = frozenset({"2.0", "2.1"})
SHARE_FORMATS: dict[str, tuple[str, str]] = {
    "html": ("text/html; charset=utf-8", ".html"),
    "json": ("application/json; charset=utf-8", ".json"),
    "txt": ("text/plain; charset=utf-8", ".txt"),
    "yaml": ("application/yaml", ".yaml"),
    "toml": ("application/toml", ".toml"),
}


def share_artifact_filename(fmt: str) -> str:
    """
    Return the stable human-facing filename for a Global Share artifact.

    The public read capability is intentionally excluded from the filename.
    """
    fmt = validate_share_format(fmt)
    _mime, ext = SHARE_FORMATS[fmt]
    return f"ai-conversation-global-share-{fmt}{ext}"


MAX_SHARE_RECORDS = 1000
MAX_SHARE_TEXT_CHARS = 200_000
MAX_SHARE_METADATA_CHARS = 2048
MAX_SHARE_RESOURCES_PER_MESSAGE = 512
MAX_SHARE_RESOURCES_TOTAL = 4096
MAX_SAFE_INTEGER = 9_007_199_254_740_991

_SHARE_ID_RE = re.compile(
    r"^(?:[0-9a-f]{32}|[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})$"
)


class ShareValidationError(ValueError):
    """Raised when an untrusted share snapshot violates the public contract."""


def _bounded_string(
    value: Any, *, limit: int, field: str, nullable: bool = True
) -> str | None:
    if value is None and nullable:
        return None
    if not isinstance(value, str):
        raise ShareValidationError(
            f"{field} must be a string" + (" or null" if nullable else "")
        )
    if len(value) > limit:
        raise ShareValidationError(f"{field} is too long")
    return value


def _bounded_int(value: Any, *, field: str, nullable: bool = True) -> int | None:
    if value is None and nullable:
        return None
    if isinstance(value, bool) or not isinstance(value, int):
        raise ShareValidationError(
            f"{field} must be an integer" + (" or null" if nullable else "")
        )
    return value


def _safe_scalar(value: Any, *, field: str) -> str | int | float | bool | None:
    if value is None or isinstance(value, (str, bool, int)):
        if isinstance(value, str) and len(value) > MAX_SHARE_METADATA_CHARS:
            raise ShareValidationError(f"{field} is too long")
        return value
    if isinstance(value, float) and math.isfinite(value):
        return value
    raise ShareValidationError(f"{field} must be a finite primitive value")


def sanitize_share_page_url(value: Any) -> str:
    """Return an HTTP(S) source URL without credentials, query, or fragment."""
    if not isinstance(value, str) or not value:
        return ""
    if len(value) > 8192:  # ruff: ignore[magic-value-comparison]
        return ""
    try:
        parts = urlsplit(value)
    except ValueError:
        return ""
    if parts.scheme.lower() not in {"http", "https"} or not parts.hostname:
        return ""
    host = parts.hostname
    if ":" in host and not host.startswith("["):
        host = f"[{host}]"
    try:
        port = parts.port
    except ValueError:
        return ""
    if port is not None:
        host = f"{host}:{port}"
    return urlunsplit((parts.scheme.lower(), host, parts.path or "/", "", ""))


_RESOURCE_KINDS = frozenset(
    {
        "text",
        "image",
        "vector_image",
        "audio",
        "video",
        "data",
        "file",
        "replay",
        "page",
        "pdf",
        "archive",
    }
)
_RESOURCE_DELIVERIES = frozenset({"context", "raw", "not_sent"})
_RESOURCE_INTENTS = frozenset({"", "auto", "raw", "extract", "context"})


def _resource_count(
    value: Any, *, field: str, maximum: int = MAX_SHARE_RESOURCES_PER_MESSAGE
) -> int:
    if (
        isinstance(value, bool)
        or not isinstance(value, int)
        or value < 0
        or value > maximum
    ):
        raise ShareValidationError(
            f"{field} must be an integer between 0 and {maximum}"
        )
    return value


def _resource_bool(value: Any, *, field: str) -> bool:
    if not isinstance(value, bool):
        raise ShareValidationError(f"{field} must be a boolean")
    return value


def _safe_relative_path(value: Any, *, field: str) -> str:
    text = _bounded_string(value, limit=1024, field=field) or ""
    if not text:
        return ""
    text = text.replace("\\", "/")
    if text.startswith("/") or re.match(r"^[A-Za-z]:", text):
        return ""
    parts = [part for part in text.split("/") if part not in {"", "."}]
    if any(
        part == ".."  # lint
        or any(ord(ch) < 32 for ch in part)  # ruff: ignore[magic-value-comparison]
        for part in parts
    ):
        return ""
    return "/".join(parts)


def _safe_resource_name(value: Any, *, field: str, limit: int = 512) -> str:
    text = _bounded_string(value, limit=limit, field=field, nullable=False) or ""
    return (
        "".join(
            ch  # lint
            for ch in text  # lint
            if ord(ch) >= 32  # ruff: ignore[magic-value-comparison]
        )
        .replace("/", "_")
        .replace("\\", "_")[:limit]
    )


def _canonical_resource_item(
    raw: Any, record_index: int, item_index: int, *, allow_source_urls: bool
) -> dict[str, Any]:
    prefix = f"records[{record_index}].resources.items[{item_index}]"
    if not isinstance(raw, dict):
        raise ShareValidationError(f"{prefix} must be an object")
    kind = _bounded_string(
        raw.get("kind"), limit=32, field=f"{prefix}.kind", nullable=False
    )
    if kind not in _RESOURCE_KINDS:
        raise ShareValidationError(f"{prefix}.kind is not allowed")
    delivery = _bounded_string(
        raw.get("delivery"), limit=32, field=f"{prefix}.delivery", nullable=False
    )
    if delivery not in _RESOURCE_DELIVERIES:
        raise ShareValidationError(f"{prefix}.delivery is not allowed")
    intent = (
        _bounded_string(raw.get("intent"), limit=32, field=f"{prefix}.intent") or ""
    )
    if intent not in _RESOURCE_INTENTS:
        raise ShareValidationError(f"{prefix}.intent is not allowed")
    size = _bounded_int(raw.get("size"), field=f"{prefix}.size", nullable=False)
    line_count = _bounded_int(
        raw.get("lineCount"), field=f"{prefix}.lineCount", nullable=False
    )
    if size is None or size < 0 or size > MAX_SAFE_INTEGER:
        raise ShareValidationError(f"{prefix}.size is out of range")
    if (
        line_count is None  # lint
        or line_count < 0  # lint
        or line_count > 1_000_000  # ruff: ignore[magic-value-comparison]
    ):
        raise ShareValidationError(f"{prefix}.lineCount is out of range")
    badge = (
        _bounded_string(
            raw.get("badge"), limit=12, field=f"{prefix}.badge", nullable=False
        )
        or "FILE"
    )
    badge = re.sub(r"[^A-Za-z0-9+._-]", "", badge).upper()[:12] or "FILE"
    modality = (
        _bounded_string(raw.get("modality"), limit=32, field=f"{prefix}.modality")
        or kind
    )
    modality = re.sub(r"[^A-Za-z0-9_-]", "", modality)[:32]
    source_kind = (
        _bounded_string(raw.get("sourceKind"), limit=24, field=f"{prefix}.sourceKind")
        or ""
    )
    source_kind = re.sub(r"[^A-Za-z0-9_-]", "", source_kind)[:24]
    archive_name = (
        _bounded_string(
            raw.get("archiveName"), limit=512, field=f"{prefix}.archiveName"
        )
        or ""
    )
    if archive_name:
        archive_name = _safe_resource_name(archive_name, field=f"{prefix}.archiveName")
    item = {
        "name": _safe_resource_name(raw.get("name"), field=f"{prefix}.name"),
        "badge": badge,
        "kind": kind,
        "size": size,
        "lineCount": line_count,
        "included": _resource_bool(raw.get("included"), field=f"{prefix}.included"),
        "localOnly": _resource_bool(raw.get("localOnly"), field=f"{prefix}.localOnly"),
        "delivery": delivery,
        "modality": modality,
        "intent": intent,
        "replay": _resource_bool(raw.get("replay"), field=f"{prefix}.replay"),
        "boundedExcerpt": _resource_bool(
            raw.get("boundedExcerpt"), field=f"{prefix}.boundedExcerpt"
        ),
        "status": (
            _bounded_string(raw.get("status"), limit=96, field=f"{prefix}.status") or ""
        ),
        "type": (
            _bounded_string(raw.get("type"), limit=120, field=f"{prefix}.type") or ""
        ),
        "relativePath": _safe_relative_path(
            raw.get("relativePath"), field=f"{prefix}.relativePath"
        ),
        "sourceKind": source_kind,
        "archiveName": archive_name,
    }
    if kind == "page":
        role = (
            _bounded_string(
                raw.get("contextRole"), limit=16, field=f"{prefix}.contextRole"
            )
            or "pinned"
        )
        item["contextRole"] = "current" if role == "current" else "pinned"
        item["sourceUrl"] = (
            sanitize_share_page_url(raw.get("sourceUrl")) if allow_source_urls else ""
        )
    return item


def _resource_aggregates(items: list[dict[str, Any]]) -> dict[str, int]:
    return {
        "includedCount": sum(1 for item in items if item["included"]),
        "localOnlyCount": sum(1 for item in items if item["localOnly"]),
        "contextCount": sum(1 for item in items if item["delivery"] == "context"),
        "rawCount": sum(1 for item in items if item["delivery"] == "raw"),
        "notSentCount": sum(1 for item in items if item["delivery"] == "not_sent"),
        "pageCount": sum(1 for item in items if item["kind"] == "page"),
        "replayCount": sum(
            1 for item in items if item["kind"] == "replay" or item["replay"]
        ),
        "totalBytes": min(MAX_SAFE_INTEGER, sum(item["size"] for item in items)),
    }


def _canonical_resources(
    raw: Any, record_index: int, *, allow_source_urls: bool
) -> dict[str, Any]:
    prefix = f"records[{record_index}].resources"
    if not isinstance(raw, dict):
        raise ShareValidationError(f"{prefix} must be an object")
    raw_items = raw.get("items")
    if not isinstance(raw_items, list):
        raise ShareValidationError(f"{prefix}.items must be an array")
    if len(raw_items) > MAX_SHARE_RESOURCES_PER_MESSAGE:
        raise ShareValidationError(f"{prefix}.items contains too many resources")
    items = [
        _canonical_resource_item(
            item, record_index, i, allow_source_urls=allow_source_urls
        )
        for i, item in enumerate(raw_items)
    ]
    derived = _resource_aggregates(items)
    total = _resource_count(
        raw.get("totalCount", len(items)), field=f"{prefix}.totalCount"
    )
    if total < len(items):
        raise ShareValidationError(f"{prefix}.totalCount cannot be smaller than items")

    def merged(name: str) -> int:
        value = _resource_count(raw.get(name, derived[name]), field=f"{prefix}.{name}")
        return max(derived[name], min(total, value))

    total_bytes = _bounded_int(
        raw.get("totalBytes", derived["totalBytes"]),
        field=f"{prefix}.totalBytes",
        nullable=False,
    )
    if total_bytes is None or total_bytes < 0 or total_bytes > MAX_SAFE_INTEGER:
        raise ShareValidationError(f"{prefix}.totalBytes is out of range")
    omitted = _resource_count(
        raw.get("omittedCount", max(0, total - len(items))),
        field=f"{prefix}.omittedCount",
    )
    omitted = max(omitted, total - len(items))
    complete = raw.get("complete")
    if not isinstance(complete, bool):
        raise ShareValidationError(f"{prefix}.complete must be a boolean")
    version = _resource_count(
        raw.get("version", 2), field=f"{prefix}.version", maximum=2
    )
    if version != 2:  # ruff: ignore[magic-value-comparison]
        raise ShareValidationError(f"{prefix}.version must be 2")
    return {
        "version": 2,
        "totalCount": total,
        "includedCount": merged("includedCount"),
        "localOnlyCount": merged("localOnlyCount"),
        "contextCount": merged("contextCount"),
        "rawCount": merged("rawCount"),
        "notSentCount": merged("notSentCount"),
        "pageCount": merged("pageCount"),
        "replayCount": merged("replayCount"),
        "totalBytes": max(derived["totalBytes"], total_bytes),
        "itemCount": len(items),
        "omittedCount": omitted,
        "complete": complete and omitted == 0 and total == len(items),
        "items": items,
    }


def _canonical_record(
    raw: Any, index: int, safe_page: str | None, session_id: str | None
) -> dict[str, Any]:
    if not isinstance(raw, dict):
        raise ShareValidationError(f"records[{index}] must be an object")
    role = raw.get("role")
    if role not in {"user", "assistant", "error"}:
        raise ShareValidationError(f"records[{index}].role is not allowed")
    text = _bounded_string(
        raw.get("text"),
        limit=MAX_SHARE_TEXT_CHARS,
        field=f"records[{index}].text",
        nullable=False,
    )
    turn_index = _bounded_int(
        raw.get("turn_index"), field=f"records[{index}].turn_index", nullable=False
    )
    message_index = _bounded_int(
        raw.get("message_index"),
        field=f"records[{index}].message_index",
        nullable=False,
    )
    ts = _bounded_int(raw.get("ts"), field=f"records[{index}].ts")
    ts_iso = _bounded_string(
        raw.get("ts_iso"), limit=128, field=f"records[{index}].ts_iso"
    )
    if raw.get("resources") is not None and role != "user":
        raise ShareValidationError(
            f"records[{index}].resources is allowed only for user messages"
        )

    return {
        "turn_index": turn_index,
        "message_index": message_index,
        "role": role,
        "text": text,
        "ts": ts,
        "ts_iso": ts_iso,
        "model_id": _bounded_string(
            raw.get("model_id"),
            limit=MAX_SHARE_METADATA_CHARS,
            field=f"records[{index}].model_id",
        ),
        "model_provider": _bounded_string(
            raw.get("model_provider"),
            limit=MAX_SHARE_METADATA_CHARS,
            field=f"records[{index}].model_provider",
        ),
        "model_name": _bounded_string(
            raw.get("model_name"),
            limit=MAX_SHARE_METADATA_CHARS,
            field=f"records[{index}].model_name",
        ),
        "feedback_rating_value": _safe_scalar(
            raw.get("feedback_rating_value"),
            field=f"records[{index}].feedback_rating_value",
        ),
        "feedback_rating_label": _bounded_string(
            raw.get("feedback_rating_label"),
            limit=MAX_SHARE_METADATA_CHARS,
            field=f"records[{index}].feedback_rating_label",
        ),
        "feedback_message": _bounded_string(
            raw.get("feedback_message"),
            limit=MAX_SHARE_TEXT_CHARS,
            field=f"records[{index}].feedback_message",
        ),
        "resources": (
            _canonical_resources(
                raw.get("resources"), index, allow_source_urls=bool(safe_page)
            )
            if raw.get("resources") is not None and role == "user"
            else None
        ),
        # Never trust duplicated per-record identity/page claims from the client;
        # bind them to the canonical session values reconstructed above.
        "session_id": session_id,
        "page_url": safe_page,
    }


def _build_turns(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
    turns: list[dict[str, Any]] = []
    current: dict[str, Any] | None = None
    for row in records:
        if row["role"] == "user":
            current = {
                "turn_index": row["turn_index"],
                "user": {
                    "text": row["text"],
                    "ts": row["ts"],
                    "ts_iso": row["ts_iso"],
                    "resources": row.get("resources"),
                },
                "assistant": None,
            }
            turns.append(current)
        elif (
            row["role"] == "assistant"
            and current is not None
            and current["assistant"] is None
        ):
            current["assistant"] = {
                "text": row["text"],
                "ts": row["ts"],
                "ts_iso": row["ts_iso"],
                "model_id": row["model_id"],
                "model_provider": row["model_provider"],
                "model_name": row["model_name"],
                "feedback_rating_value": row["feedback_rating_value"],
                "feedback_rating_label": row["feedback_rating_label"],
                "feedback_message": row["feedback_message"],
            }
    return turns


def canonicalize_share_snapshot(raw: Any) -> dict[str, Any]:
    """Validate and reconstruct the allowlisted canonical schema-v2.1 Share snapshot."""
    if not isinstance(raw, dict):
        raise ShareValidationError("snapshot must be an object")
    if raw.get("schema_version") not in SHARE_ACCEPTED_SCHEMA_VERSIONS:
        raise ShareValidationError(
            "snapshot.schema_version must be one of: '2.0', '2.1'"
        )

    raw_session = raw.get("session")
    if not isinstance(raw_session, dict):
        raise ShareValidationError("snapshot.session must be an object")
    session_id = _bounded_string(raw_session.get("id"), limit=256, field="session.id")
    safe_page = sanitize_share_page_url(raw_session.get("page_url")) or None
    session = {
        "id": session_id,
        "page_url": safe_page,
        "page_title": _bounded_string(
            raw_session.get("page_title"), limit=2048, field="session.page_title"
        ),
        "assistant_name": (
            _bounded_string(
                raw_session.get("assistant_name"),
                limit=256,
                field="session.assistant_name",
            )
            or "AI Assistant"
        ),
        "exported_at": _bounded_int(
            raw_session.get("exported_at"), field="session.exported_at"
        ),
        "exported_at_iso": _bounded_string(
            raw_session.get("exported_at_iso"),
            limit=128,
            field="session.exported_at_iso",
        ),
    }

    raw_records = raw.get("records")
    if not isinstance(raw_records, list) or not raw_records:
        raise ShareValidationError("snapshot.records must be a non-empty array")
    if len(raw_records) > MAX_SHARE_RECORDS:
        raise ShareValidationError("snapshot.records contains too many messages")
    records = [
        _canonical_record(row, i, safe_page, session_id)
        for i, row in enumerate(raw_records)
    ]
    if (
        sum((row.get("resources") or {}).get("itemCount", 0) for row in records)
        > MAX_SHARE_RESOURCES_TOTAL
    ):
        raise ShareValidationError(
            "snapshot.records contains too many resource metadata rows"
        )

    # Never accept caller-supplied turns/unknown root data as trusted.  Turns are
    # a derived view of validated records and are rebuilt server-side.
    return {
        "schema_version": SHARE_SCHEMA_VERSION,
        "session": session,
        "turns": _build_turns(records),
        "records": records,
    }


def validate_share_format(value: Any) -> str:
    if not isinstance(value, str) or value not in SHARE_FORMATS:
        raise ShareValidationError("format must be one of: html, json, txt, yaml, toml")
    return value


def _resource_summary(manifest: dict[str, Any] | None) -> str:
    if not manifest or not manifest.get("totalCount"):
        return ""
    return (
        f"Resources used for this question: {manifest['totalCount']}"
        f" · context {manifest['contextCount']}"
        f" · raw {manifest['rawCount']}"
        f" · not-sent {manifest['notSentCount']}"
    )


def _resource_text_lines(manifest: dict[str, Any] | None) -> list[str]:
    if not manifest or not manifest.get("totalCount"):
        return []
    lines = [f"[{_resource_summary(manifest)}]"]
    for item in manifest.get("items") or []:
        line = f"- [{item.get('badge') or 'FILE'}] {item.get('name') or 'file'}"
        if item.get("status"):
            line += f" — {item['status']}"
        lines.append(line)
    if manifest.get("omittedCount"):
        count = manifest["omittedCount"]
        lines.append(
            f"- … {count} resource metadata row{'s' if count != 1 else ''} omitted from restored state"
        )
    return lines


def _resource_html(manifest: dict[str, Any] | None) -> str:
    if not manifest or not manifest.get("totalCount"):
        return ""
    rows: list[str] = []
    for item in manifest.get("items") or []:
        status = (
            f'<span class="resource-status">{html.escape(str(item.get("status") or ""))}</span>'
            if item.get("status")
            else ""
        )
        rows.append(
            '<li class="resource-card">'
            f'<span class="resource-badge">{html.escape(str(item.get("badge") or "FILE"))}</span>'
            f'<span class="resource-name">{html.escape(str(item.get("name") or "file"))}</span>{status}</li>'
        )
    if manifest.get("omittedCount"):
        count = manifest["omittedCount"]
        rows.append(
            f'<li class="resource-card omitted">… {count} resource metadata row{"s" if count != 1 else ""} omitted</li>'
        )
    return (
        '<section class="resources"><div class="resource-summary">'
        + html.escape(_resource_summary(manifest))
        + '</div><ul class="resource-list">'
        + "".join(rows)
        + "</ul></section>"
    )


def _render_html(snapshot: dict[str, Any]) -> str:
    session = snapshot["session"]
    assistant_name = html.escape(str(session.get("assistant_name") or "AI Assistant"))
    page_title = html.escape(str(session.get("page_title") or "Shared conversation"))
    page_url = str(session.get("page_url") or "")
    source = ""
    if page_url:
        escaped_url = html.escape(page_url, quote=True)
        source = f'<p class="source">Source: <a href="{escaped_url}" rel="noopener noreferrer">{escaped_url}</a></p>'

    messages: list[str] = []
    for row in snapshot["records"]:
        role = row["role"]
        label = (
            "You"
            if role == "user"
            else ("Error" if role == "error" else assistant_name)
        )
        text = html.escape(str(row.get("text") or ""))
        cls = (
            "user" if role == "user" else ("error" if role == "error" else "assistant")
        )
        meta_parts: list[str] = []
        if row.get("ts_iso"):
            meta_parts.append(html.escape(str(row["ts_iso"])))
        if row.get("model_name"):
            meta_parts.append(html.escape(str(row["model_name"])))
        if row.get("model_provider"):
            meta_parts.append(html.escape(str(row["model_provider"])))
        if (
            row.get("feedback_rating_label")
            or row.get("feedback_rating_value") is not None
        ):
            rating = str(
                row.get("feedback_rating_label") or row.get("feedback_rating_value")
            )
            if row.get("feedback_rating_value") is not None and row.get(
                "feedback_rating_label"
            ):
                rating += f" ({row['feedback_rating_value']})"
            meta_parts.append("Rating: " + html.escape(rating))
        if row.get("feedback_message"):
            meta_parts.append("Feedback: " + html.escape(str(row["feedback_message"])))
        meta = f'<div class="meta">{" · ".join(meta_parts)}</div>' if meta_parts else ""
        resources = _resource_html(row.get("resources")) if role == "user" else ""
        messages.append(
            f'<article class="msg {cls}"><div class="role">{label}</div>'
            f"{resources}<pre>{text}</pre>{meta}</article>"
        )

    # No scripts and no remote resources.  CSP on the HTTP response is the
    # primary policy; this meta tag protects downloaded/copied representations.
    return (
        """<!doctype html>
<html lang="en"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<meta name="referrer" content="no-referrer">
<meta http-equiv="Content-Security-Policy" content="default-src 'none'; style-src 'unsafe-inline'; img-src data:; base-uri 'none'; form-action 'none'">
<title>Shared AI conversation</title><style>
:root{font-family:system-ui,sans-serif;color-scheme:light dark}body{margin:0;background:Canvas;color:CanvasText}.wrap{max-width:850px;margin:auto;padding:24px}.head{border-bottom:1px solid color-mix(in srgb,CanvasText 20%,transparent);padding-bottom:16px}.source{overflow-wrap:anywhere}.source a{color:inherit}.msg{margin:18px 0;padding:14px;border:1px solid color-mix(in srgb,CanvasText 18%,transparent);border-radius:12px}.msg.user{margin-left:10%}.msg.error{border-style:dashed}.role{font-weight:700;margin-bottom:8px}.resources{margin:0 0 10px}.resource-summary{font-size:.78rem;opacity:.7}.resource-list{list-style:none;padding:0;margin:6px 0;display:grid;gap:4px}.resource-card{display:flex;gap:6px;flex-wrap:wrap;font-size:.78rem}.resource-badge{font-weight:700}.resource-status{opacity:.65}.msg pre{white-space:pre-wrap;overflow-wrap:anywhere;font:inherit;margin:0}.meta{opacity:.65;font-size:.8rem;margin-top:8px}
</style></head><body><main class="wrap"><header class="head"><h1>"""
        + assistant_name
        + " — Shared conversation</h1><p>"
        + page_title
        + "</p>"
        + source
        + "</header>"
        + "".join(messages)
        + "</main></body></html>"
    )


def _render_text(snapshot: dict[str, Any]) -> str:
    lines: list[str] = []
    session = snapshot["session"]
    lines.append(
        f"{session.get('assistant_name') or 'AI Assistant'} — Shared conversation"
    )
    lines.append(f"Schema: {snapshot.get('schema_version') or SHARE_SCHEMA_VERSION}")
    if session.get("page_title"):
        lines.append(f"Page title: {session['page_title']}")
    if session.get("page_url"):
        lines.append(f"Source: {session['page_url']}")
    if session.get("exported_at_iso"):
        lines.append(f"Exported: {session['exported_at_iso']}")
    lines.append("")
    for row in snapshot["records"]:
        role = row["role"]
        label = (
            "USER" if role == "user" else ("ERROR" if role == "error" else "ASSISTANT")
        )
        meta: list[str] = []
        if row.get("ts_iso"):
            meta.append(str(row["ts_iso"]))
        if role in {"assistant", "error"} and (
            row.get("model_name") or row.get("model_id")
        ):
            model = str(row.get("model_name") or row.get("model_id"))
            if row.get("model_provider"):
                model += f" · {row['model_provider']}"
            meta.append(model)
        lines.append(f"[{label}]" + (f"  [{' · '.join(meta)}]" if meta else ""))
        if role == "user":
            lines.extend(_resource_text_lines(row.get("resources")))
        if role in {"assistant", "error"} and (
            row.get("feedback_rating_label")
            or row.get("feedback_rating_value") is not None
        ):
            bits = []
            if row.get("feedback_rating_label"):
                bits.append(str(row["feedback_rating_label"]))
            if row.get("feedback_rating_value") is not None:
                bits.append(str(row["feedback_rating_value"]))
            lines.append(f"[Rating: {' · '.join(bits)}]")
        if role in {"assistant", "error"} and row.get("feedback_message"):
            lines.append(f"[Feedback: {row['feedback_message']}]")
        lines.extend([str(row.get("text") or ""), ""])
    return "\n".join(lines).rstrip() + "\n"


def _yaml_scalar(value: Any) -> str:
    if value is None:
        return "null"
    if isinstance(value, bool):
        return "true" if value else "false"
    if isinstance(value, (int, float)) and not isinstance(value, bool):
        if isinstance(value, float) and not math.isfinite(value):
            return "null"
        return str(value)
    return json.dumps(str(value), ensure_ascii=False)


def _yaml_value(value: Any, indent: int = 0) -> str:
    pad = " " * indent
    if isinstance(value, list):
        if not value:
            return pad + "[]"
        rows: list[str] = []
        for item in value:
            if isinstance(item, (dict, list)):
                child = _yaml_value(item, indent + 2).splitlines()
                rows.append(pad + "- " + child[0][indent + 2 :])
                rows.extend(child[1:])
            else:
                rows.append(pad + "- " + _yaml_scalar(item))
        return "\n".join(rows)
    if isinstance(value, dict):
        if not value:
            return pad + "{}"
        rows = []
        for key, item in value.items():
            qkey = json.dumps(str(key), ensure_ascii=False)
            if isinstance(item, (dict, list)):
                rows.append(f"{pad}{qkey}:\n{_yaml_value(item, indent + 2)}")
            else:
                rows.append(f"{pad}{qkey}: {_yaml_scalar(item)}")
        return "\n".join(rows)
    return pad + _yaml_scalar(value)


def _render_yaml(snapshot: dict[str, Any]) -> str:
    return _yaml_value(snapshot) + "\n"


def _toml_scalar(value: Any) -> str | None:
    if isinstance(value, str):
        return json.dumps(value, ensure_ascii=False)
    if isinstance(value, bool):
        return "true" if value else "false"
    if isinstance(value, (int, float)) and not isinstance(value, bool):
        if isinstance(value, float) and not math.isfinite(value):
            return None
        return str(value)
    return None


def _toml_fields(
    lines: list[str], obj: dict[str, Any], omit: set[str] | None = None
) -> None:
    skipped = omit or set()
    for key, value in obj.items():
        if key in skipped or value is None:
            continue
        rendered = _toml_scalar(value)
        if rendered is not None:
            lines.append(f"{key} = {rendered}")


def _toml_resources(
    lines: list[str], table: str, manifest: dict[str, Any] | None
) -> None:
    if not manifest or not manifest.get("totalCount"):
        return
    lines.append(f"[{table}]")
    _toml_fields(lines, manifest, {"items"})
    for item in manifest.get("items") or []:
        lines.append(f"[[{table}.items]]")
        _toml_fields(lines, item)


def _render_toml(snapshot: dict[str, Any]) -> str:
    lines = [
        "# AI Assistant conversation export",
        "# schema v2.1 semantics: omitted optional values represent null",
        f"schema_version = {json.dumps(str(snapshot.get('schema_version') or SHARE_SCHEMA_VERSION), ensure_ascii=False)}",
        "",
        "[session]",
    ]
    _toml_fields(lines, snapshot.get("session") or {})
    for turn in snapshot.get("turns") or []:
        lines.extend(["", "[[turns]]"])
        if turn.get("turn_index") is not None:
            lines.append(f"turn_index = {turn['turn_index']}")
        if isinstance(turn.get("user"), dict):
            lines.append("[turns.user]")
            _toml_fields(lines, turn["user"], {"resources"})
            _toml_resources(
                lines, "turns.user.resources", turn["user"].get("resources")
            )
        if isinstance(turn.get("assistant"), dict):
            lines.append("[turns.assistant]")
            _toml_fields(lines, turn["assistant"])
    for record in snapshot.get("records") or []:
        lines.extend(["", "[[records]]"])
        _toml_fields(lines, record, {"resources"})
        _toml_resources(lines, "records.resources", record.get("resources"))
    return "\n".join(lines) + "\n"


def render_share(snapshot: dict[str, Any], fmt: str) -> tuple[str, str, str]:
    """Render a validated snapshot using a server-owned representation."""
    fmt = validate_share_format(fmt)
    mime, ext = SHARE_FORMATS[fmt]
    if fmt == "html":
        content = _render_html(snapshot)
    elif fmt == "json":
        content = json.dumps(snapshot, ensure_ascii=False, indent=2) + "\n"
    elif fmt == "yaml":
        content = _render_yaml(snapshot)
    elif fmt == "toml":
        content = _render_toml(snapshot)
    else:
        content = _render_text(snapshot)
    return content, mime, ext


def render_share_viewer_shell(
    read_path: str = "/v1/share/read", download_path: str = "/v1/share/download"
) -> str:
    """
    Return the fixed-path public Share viewer.

    The public read capability remains in ``location.hash`` and is sent to the
    fixed read endpoint only in a JSON request body. The shell renders all
    conversation values with DOM ``textContent`` and never injects untrusted HTML.
    """
    read_path_json = json.dumps(str(read_path), ensure_ascii=False)
    download_path_json = json.dumps(str(download_path), ensure_ascii=False)
    template = r"""<!doctype html><html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><meta name="referrer" content="no-referrer"><title>Shared AI conversation</title><style>:root{font-family:system-ui,sans-serif;color-scheme:light dark}body{margin:0;background:Canvas;color:CanvasText}.wrap{max-width:850px;margin:auto;padding:24px}.head{border-bottom:1px solid currentColor;padding-bottom:16px}.source{overflow-wrap:anywhere}.source a{color:inherit}.msg{margin:18px 0;padding:14px;border:1px solid currentColor;border-radius:12px}.msg.user{margin-left:10%}.msg.error{border-style:dashed}.role{font-weight:700;margin-bottom:8px}.meta{margin-top:8px;font-size:.82rem;opacity:.72;overflow-wrap:anywhere}.resources{margin:0 0 10px;padding:10px;border:1px solid color-mix(in srgb,currentColor 28%,transparent);border-radius:9px}.resource-summary{font-size:.8rem;opacity:.76;margin-bottom:6px}.resource-list{display:grid;gap:6px}.resource-card{display:flex;gap:8px;align-items:baseline;flex-wrap:wrap;font-size:.82rem}.resource-badge{font-size:.7rem;font-weight:700;border:1px solid currentColor;border-radius:5px;padding:1px 5px}.resource-source{color:inherit}.feedback{margin-top:8px;padding:8px 10px;border-left:3px solid currentColor;white-space:pre-wrap;overflow-wrap:anywhere}.artifact{display:flex;gap:10px;align-items:center;justify-content:space-between;flex-wrap:wrap;margin:0 0 18px;padding:10px 12px;border:1px solid currentColor;border-radius:10px}.artifact-name{font-size:.82rem;opacity:.8;overflow-wrap:anywhere}.download{font:inherit;padding:7px 11px;border:1px solid currentColor;border-radius:8px;background:Canvas;color:CanvasText;cursor:pointer}.download[disabled]{opacity:.55;cursor:wait}pre{white-space:pre-wrap;overflow-wrap:anywhere;font:inherit;margin:0}.error-note{border:1px dashed currentColor;padding:14px;border-radius:12px}</style></head><body><main id="app" class="wrap"><p>Loading shared conversation…</p></main><script>(()=>{'use strict';const app=document.getElementById('app');const fail=(m)=>{app.replaceChildren();const p=document.createElement('p');p.className='error-note';p.textContent=m;app.appendChild(p);};let raw=(location.hash||'').slice(1);if(raw.startsWith('share='))raw=raw.slice(6);try{raw=decodeURIComponent(raw)}catch(_e){}if(!/^(?:[0-9a-f]{32}|[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})$/i.test(raw)){fail('This Share link is invalid or incomplete.');return;}const readJson=async(r)=>{const max=4*1024*1024;const h=r.headers&&r.headers.get?r.headers.get('content-length'):null;if(h!=null&&String(h).trim()!==''){if(!/^\d+$/.test(String(h).trim())||Number(h)>max)throw new Error('Share response is too large.');}if(!r.body||typeof r.body.getReader!=='function'||typeof TextDecoder!=='function')throw new Error('Bounded Share reader unavailable.');const rd=r.body.getReader(),dec=new TextDecoder(),parts=[];let n=0;try{for(;;){const x=await rd.read();if(x.done)break;const v=x.value||new Uint8Array(0);n+=Number(v.byteLength||v.length||0);if(n>max)throw new Error('Share response is too large.');parts.push(dec.decode(v,{stream:true}));}parts.push(dec.decode());}catch(e){try{await rd.cancel()}catch(_e){}throw e;}finally{try{rd.releaseLock()}catch(_e){}}return JSON.parse(parts.join(''));};const readDownload=async(r)=>{const max=8*1024*1024;const h=r.headers&&r.headers.get?r.headers.get('content-length'):null;if(h!=null&&String(h).trim()!==''){if(!/^\d+$/.test(String(h).trim())||Number(h)>max)throw new Error('Shared artifact is too large to download.');}if(!r.body||typeof r.body.getReader!=='function'||typeof TextDecoder!=='function')throw new Error('Bounded Share download reader unavailable.');const rd=r.body.getReader(),dec=new TextDecoder(),parts=[];let n=0;try{for(;;){const x=await rd.read();if(x.done)break;const v=x.value||new Uint8Array(0);n+=Number(v.byteLength||v.length||0);if(n>max)throw new Error('Shared artifact is too large to download.');parts.push(dec.decode(v,{stream:true}));}parts.push(dec.decode());}catch(e){try{await rd.cancel()}catch(_e){}throw e;}finally{try{rd.releaseLock()}catch(_e){}}return parts.join('');};fetch(__READ_PATH__,{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({shareId:raw}),cache:'no-store',credentials:'omit',redirect:'error',referrerPolicy:'no-referrer'}).then(async r=>{if(!r.ok){throw new Error(r.status===410?'This Share has expired.':r.status===404?'This Share is unavailable.':'Could not load this Share.');}return await readJson(r);}).then(data=>{app.replaceChildren();const fmt=String(data.format||'txt').toLowerCase();const ext={json:'.json',html:'.html',txt:'.txt',yaml:'.yaml',toml:'.toml'}[fmt]||'.txt';const fallback='ai-conversation-global-share-'+fmt+ext;const requested=String(data.filename||'');const filename=/^ai-conversation-global-share-(?:json\.json|html\.html|txt\.txt|yaml\.yaml|toml\.toml)$/.test(requested)?requested:fallback;const bar=document.createElement('section');bar.className='artifact';bar.setAttribute('aria-label','Shared artifact');const artifactName=document.createElement('span');artifactName.className='artifact-name';artifactName.textContent='Global Share · '+fmt.toUpperCase()+' · '+filename;const download=document.createElement('button');download.type='button';download.className='download';download.textContent='Download '+fmt.toUpperCase();download.addEventListener('click',async()=>{if(download.disabled)return;download.disabled=true;const prior=download.textContent;download.textContent='Downloading…';try{const r=await fetch(__DOWNLOAD_PATH__,{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({shareId:raw}),cache:'no-store',credentials:'omit',redirect:'error',referrerPolicy:'no-referrer'});if(!r.ok)throw new Error(r.status===410?'This Share has expired.':r.status===404?'This Share is unavailable.':'Could not download this Share.');const text=await readDownload(r);const type=String((r.headers&&r.headers.get&&r.headers.get('content-type'))||data.mimeType||'text/plain;charset=utf-8');const blob=new Blob([text],{type});const objectUrl=URL.createObjectURL(blob);try{const a=document.createElement('a');a.href=objectUrl;a.download=filename;a.rel='noopener';a.style.display='none';document.body.appendChild(a);a.click();a.remove();}finally{setTimeout(()=>URL.revokeObjectURL(objectUrl),0);}}catch(e){fail(e&&e.message?e.message:'Could not download this Share.');return;}finally{download.disabled=false;download.textContent=prior;}});bar.append(artifactName,download);app.appendChild(bar);if(data.format==='html'&&data.snapshot&&data.snapshot.session&&Array.isArray(data.snapshot.records)){const snap=data.snapshot;const h=document.createElement('header');h.className='head';const h1=document.createElement('h1');h1.textContent=(snap.session.assistant_name||'AI Assistant')+' — Shared conversation';h.appendChild(h1);if(snap.session.page_title){const p=document.createElement('p');p.textContent=snap.session.page_title;h.appendChild(p);}if(snap.session.page_url){const p=document.createElement('p');p.className='source';p.append('Source: ');const a=document.createElement('a');a.href=snap.session.page_url;a.rel='noopener noreferrer';a.referrerPolicy='no-referrer';a.textContent=snap.session.page_url;p.appendChild(a);h.appendChild(p);}app.appendChild(h);const appendMeta=(article,row)=>{const bits=[];if(row.ts_iso)bits.push(String(row.ts_iso));const model=String(row.model_name||row.model_id||'');const provider=String(row.model_provider||'');if(model)bits.push(provider?model+' · '+provider:model);else if(provider)bits.push(provider);if(row.feedback_rating_label!=null||row.feedback_rating_value!=null){let rating='Rating: '+String(row.feedback_rating_label||'rated');if(row.feedback_rating_value!=null)rating+=' ('+String(row.feedback_rating_value)+')';bits.push(rating);}if(bits.length){const meta=document.createElement('div');meta.className='meta';meta.textContent=bits.join(' · ');article.appendChild(meta);}if(row.feedback_message){const note=document.createElement('div');note.className='feedback';note.textContent='Feedback: '+String(row.feedback_message);article.appendChild(note);}};const appendResources=(article,manifest)=>{if(!manifest||!Array.isArray(manifest.items)||Number(manifest.totalCount||0)<=0)return;const section=document.createElement('section');section.className='resources';section.setAttribute('aria-label','Resources used for this question');const summary=document.createElement('div');summary.className='resource-summary';summary.textContent='Resources: '+String(manifest.totalCount||manifest.items.length)+' · context '+String(manifest.contextCount||0)+' · raw '+String(manifest.rawCount||0)+' · not-sent '+String(manifest.notSentCount||0);section.appendChild(summary);const list=document.createElement('div');list.className='resource-list';for(const item of manifest.items){const card=document.createElement('div');card.className='resource-card';const badge=document.createElement('span');badge.className='resource-badge';badge.textContent=String(item.badge||'FILE');const label=document.createElement('span');label.textContent=String(item.name||'Resource')+(item.status?' — '+String(item.status):'');card.append(badge,label);const source=String(item.sourceUrl||'');if(/^https?:\/\//i.test(source)){const a=document.createElement('a');a.className='resource-source';a.href=source;a.rel='noopener noreferrer';a.referrerPolicy='no-referrer';a.textContent='Source';card.appendChild(a);}list.appendChild(card);}section.appendChild(list);article.appendChild(section);};for(const row of snap.records){const article=document.createElement('article');article.className='msg '+(row.role==='user'?'user':row.role==='error'?'error':'assistant');const role=document.createElement('div');role.className='role';role.textContent=row.role==='user'?'You':row.role==='error'?'Error':(snap.session.assistant_name||'AI Assistant');article.appendChild(role);if(row.role==='user')appendResources(article,row.resources);const pre=document.createElement('pre');pre.textContent=String(row.text||'');article.appendChild(pre);appendMeta(article,row);app.appendChild(article);}return;}const pre=document.createElement('pre');pre.textContent=String(data.content||'');app.appendChild(pre);}).catch(e=>fail(e&&e.message?e.message:'Could not load this Share.'));})();</script></body></html>"""
    return template.replace("__READ_PATH__", read_path_json).replace(
        "__DOWNLOAD_PATH__", download_path_json
    )


def generate_edit_token() -> str:
    return secrets.token_urlsafe(32)


def hash_edit_token(token: str) -> str:
    return hashlib.sha256(token.encode("utf-8")).hexdigest()


def verify_edit_token(token: str, expected_hash: str) -> bool:
    if not token or not expected_hash:
        return False
    candidate = hash_edit_token(token)
    return hmac.compare_digest(candidate, expected_hash)


def valid_share_id(value: str) -> bool:
    return bool(_SHARE_ID_RE.fullmatch(value or ""))