Remove unused engine runtime from Space

src/small_cuts/engine/__init__.py

@@ -1,9 +0,0 @@
-"""Home-node narration engine — mobile-facing WebSocket session (Team Inference).
-
-Optional install: `uv sync --extra engine`. Nothing in the existing app path
-imports this package, so the Space/UI keep working without the extra.
-"""
-
-from .app import build_engine_app
-
-__all__ = ["build_engine_app"]

src/small_cuts/engine/__main__.py

@@ -1,33 +0,0 @@
-"""Run the engine: `uv run python -m small_cuts.engine`."""
-
-import contextlib
-import os
-import signal
-
-import uvicorn
-
-from .app import build_engine_app
-
-
-def main() -> None:
-    from small_cuts import narrator
-
-    def shutdown(signum, frame):  # noqa: ARG001
-        with contextlib.suppress(Exception):
-            backend = narrator.get_backend()
-            close = getattr(backend, "close", None)
-            if close is not None:
-                close()
-        raise SystemExit(0)
-
-    signal.signal(signal.SIGTERM, shutdown)
-    uvicorn.run(
-        build_engine_app(),
-        host=os.environ.get("SMALL_CUTS_ENGINE_HOST", "0.0.0.0"),
-        port=int(os.environ.get("SMALL_CUTS_ENGINE_PORT", "8077")),
-        ws_max_size=int(os.environ.get("SMALL_CUTS_ENGINE_WS_MAX_SIZE", str(64 * 1024 * 1024))),
-    )
-
-
-if __name__ == "__main__":
-    main()

src/small_cuts/engine/app.py

@@ -1,141 +0,0 @@
-"""App factory for the home-node narration engine (Team Inference).
-
-One WebSocket per wearing session (`/v1/session`), plus the viewer-facing
-side of docs/contracts: the scene library REST API (D6) and the live SSE
-scene stream with Last-Event-ID resume (D7).
-"""
-
-from __future__ import annotations
-
-import asyncio
-import json
-from collections.abc import AsyncIterator
-from typing import Annotated, Any, Literal
-
-from fastapi import FastAPI, HTTPException, Query, Request, WebSocket
-from fastapi.responses import FileResponse, StreamingResponse
-from pydantic import BaseModel
-
-from .library import SceneLibrary
-from .session import EngineState, SceneSink, SessionRunner
-
-SSE_HEARTBEAT_S = 15.0
-
-Visibility = Literal["private", "shared", "public"]
-
-
-class VisibilityPatch(BaseModel):
-    visibility: Visibility
-
-
-def _sse_event(event: dict[str, Any]) -> str:
-    data = f"data: {json.dumps(event)}\n\n"
-    seq = event.get("seq")
-    # No seq -> ephemeral event (error frames): no id line, so it never becomes
-    # a Last-Event-ID resume cursor and is never expected in replay.
-    return data if seq is None else f"id: {seq}\n{data}"
-
-
-def _last_event_id(raw: str | None) -> int | None:
-    """SSE resume cursor; absent or unparsable means a fresh connect, live only."""
-    if raw is None:
-        return None
-    try:
-        return int(raw)
-    except ValueError:
-        return None
-
-
-async def scene_event_stream(
-    library: SceneLibrary, last_event_id: int | None, heartbeat_s: float
-) -> AsyncIterator[str]:
-    """SSE body: replay seq > Last-Event-ID, then live events; pings while idle.
-
-    Live events without a seq (pipeline errors) pass straight through —
-    they are ephemeral and never part of replay.
-    """
-    queue = library.subscribe()
-    try:
-        last_seq = -1
-        if last_event_id is not None:
-            last_seq = last_event_id
-            for scene in library.scenes_since(last_event_id):
-                last_seq = scene["seq"]
-                yield _sse_event(scene)
-        while True:
-            try:
-                event = await asyncio.wait_for(queue.get(), timeout=heartbeat_s)
-            except asyncio.TimeoutError:
-                yield ": ping\n\n"
-                continue
-            # Invariant: the library publishes each seq exactly once, so live
-            # events never need dedupe against each other. `last_seq` stays
-            # frozen at the replay boundary — it only filters scenes that were
-            # both replayed and queued (stored before replay, published after
-            # subscribe). It must NOT advance here: store() commits seq under
-            # the lock in a worker thread but publishes later on the loop, so
-            # concurrent sessions can publish out of seq order, and a moving
-            # cursor would drop the lower seq forever.
-            seq = event.get("seq")
-            if seq is not None and seq <= last_seq:
-                continue
-            yield _sse_event(event)
-    finally:
-        library.unsubscribe(queue)
-
-
-def build_engine_app(
-    scene_sink: SceneSink | None = None,
-    library: SceneLibrary | None = None,
-    sse_heartbeat_s: float = SSE_HEARTBEAT_S,
-) -> FastAPI:
-    """Engine app: session socket + scene library + SSE stream, per docs/contracts.
-
-    By default scenes are persisted to a `SceneLibrary` (root from
-    `SMALL_CUTS_LIBRARY_DIR`); pass `library` to inject one, or `scene_sink`
-    to replace the sink entirely (tests).
-    """
-    lib = library if library is not None else SceneLibrary()
-    sink = scene_sink if scene_sink is not None else lib
-    # Errors fan out to the viewer stream too (D9): the timeline shows failures.
-    state = EngineState(sink=sink, error_sink=lib.publish_event)
-    app = FastAPI(title="small-cuts engine")
-    app.state.library = lib
-
-    @app.websocket("/v1/session")
-    async def session(websocket: WebSocket) -> None:
-        await websocket.accept()
-        await SessionRunner(websocket, state).run()
-
-    @app.get("/v1/scenes")
-    def list_scenes(
-        session: str | None = None,
-        visibility: Visibility | None = None,
-        limit: Annotated[int, Query(ge=1, le=1000)] = 100,
-    ) -> dict[str, list[dict[str, Any]]]:
-        return {"scenes": lib.list_scenes(session_id=session, visibility=visibility, limit=limit)}
-
-    @app.get("/v1/scenes/stream")
-    async def stream_scenes(request: Request) -> StreamingResponse:
-        resume_from = _last_event_id(request.headers.get("last-event-id"))
-        return StreamingResponse(
-            scene_event_stream(lib, resume_from, sse_heartbeat_s),
-            media_type="text/event-stream",
-            headers={"Cache-Control": "no-cache"},
-        )
-
-    @app.patch("/v1/scenes/{scene_id}")
-    def set_visibility(scene_id: str, patch: VisibilityPatch) -> dict[str, Any]:
-        scene = lib.set_visibility(scene_id, patch.visibility)
-        if scene is None:
-            raise HTTPException(status_code=404, detail=f"unknown scene {scene_id}")
-        return scene
-
-    @app.get("/media/{scene_id}/{filename}")
-    def media(scene_id: str, filename: str) -> FileResponse:
-        path = lib.media_path(scene_id, filename)
-        if path is None:  # unknown name, traversal attempt, or missing file
-            raise HTTPException(status_code=404, detail="no such media")
-        return FileResponse(path)
-
-    return app

src/small_cuts/engine/library.py

@@ -1,356 +0,0 @@
-"""Engine-side scene library: filesystem media + sqlite index (D6).
-
-The real `SceneSink`. Every successful narration is persisted — frame JPEG,
-title card, voice WAV, one sqlite row — and fanned out to live SSE
-subscribers (D7). Blocking writes run in a worker thread via
-`asyncio.to_thread`; the publish happens back on the event loop. Stored
-entries and live events share the same NarratedScene shape, per
-docs/contracts/narrated-scene.schema.json.
-"""
-
-from __future__ import annotations
-
-import asyncio
-import contextlib
-import json
-import os
-import sqlite3
-import sys
-import threading
-from pathlib import Path
-from typing import Any
-
-from PIL import Image
-
-from small_cuts import narrator, tts
-from small_cuts.frames import pick_key_frame
-from small_cuts.title_card import derive_title, render_title_card
-
-from .session import CONTRACT_VERSION, _wav_bytes
-
-DEFAULT_ROOT = "~/.small-cuts/library"
-OWNER = "carlos"  # v1 engines are single-user; the field is reserved for multi-user
-VISIBILITIES = ("private", "shared", "public")
-MEDIA_FILES = ("frame.jpg", "card.webp", "voice.wav", "clip.mp4")
-CLIP_MP4_FPS = 12
-CLIP_BLEND_STEPS = 1
-H264_MIN_DIMENSION = 2
-POSTER_JPEG_QUALITY = 90
-RGB_MODE = "RGB"
-VIDEO_PIXEL_FORMAT = "yuv420p"
-PRIMARY_VIDEO_CODEC = "libx264"
-FALLBACK_VIDEO_CODEC = "h264"
-
-_SCHEMA = """\
-CREATE TABLE IF NOT EXISTS scenes (
-    scene_id    TEXT PRIMARY KEY,
-    seq         INTEGER NOT NULL UNIQUE,
-    moment_id   TEXT NOT NULL,
-    session_id  TEXT NOT NULL,
-    captured_at TEXT NOT NULL,
-    created_at  TEXT NOT NULL,
-    style_key   TEXT NOT NULL,
-    title       TEXT NOT NULL,
-    narration   TEXT NOT NULL,
-    visibility  TEXT NOT NULL DEFAULT 'private',
-    owner       TEXT NOT NULL,
-    engine      TEXT NOT NULL
-)"""
-
-_INSERT = """\
-INSERT INTO scenes (scene_id, seq, moment_id, session_id, captured_at, created_at,
-                    style_key, title, narration, visibility, owner, engine)
-VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"""
-
-
-class SceneLibrary:
-    """Scene store + in-process pub/sub. The instance itself is the async SceneSink.
-
-    Layout: `<root>/library.sqlite3` + `<root>/media/<scene_id>/{frame.jpg,
-    card.webp, voice.wav}`. One sqlite connection, guarded by a lock: the
-    sink writes from worker threads, queries come from request handlers.
-    """
-
-    def __init__(self, root: str | Path | None = None) -> None:
-        base = root or os.environ.get("SMALL_CUTS_LIBRARY_DIR") or DEFAULT_ROOT
-        self.root = Path(base).expanduser().resolve()
-        self.media_dir = self.root / "media"
-        self.media_dir.mkdir(parents=True, exist_ok=True)
-        self._lock = threading.Lock()  # guards the connection and seq allocation
-        self._db = sqlite3.connect(self.root / "library.sqlite3", check_same_thread=False)
-        self._db.row_factory = sqlite3.Row
-        with self._lock, self._db:
-            # WAL + busy_timeout: viewer reads don't block sink writes, and a
-            # briefly locked database waits instead of raising immediately.
-            self._db.execute("PRAGMA journal_mode=WAL")
-            self._db.execute("PRAGMA busy_timeout=5000")
-            self._db.execute(_SCHEMA)
-        self._subscribers: list[asyncio.Queue[dict[str, Any]]] = []
-
-    # -- the sink -----------------------------------------------------------------
-
-    async def __call__(self, scene: dict[str, Any]) -> None:
-        """SceneSink entry point: persist off the event loop, then publish.
-
-        A failed store (disk full, sqlite error) must not be silent data loss:
-        the mobile client already received its SceneAudio, so log to stderr and
-        fan an error ControlFrame to the viewer stream — the timeline stays
-        honest. `_hand_to_sink`'s suppression remains the last-resort backstop.
-        """
-        try:
-            narrated = await asyncio.to_thread(self.store, scene)
-        except Exception as exc:
-            print(
-                f"small_cuts.engine: library write failed for scene {scene['scene_id']}: {exc!r}",
-                file=sys.stderr,
-            )
-            self.publish_event(
-                {
-                    "contract_version": CONTRACT_VERSION,
-                    "kind": "error",
-                    "moment_id": scene["moment_id"],
-                    "error": {
-                        "stage": "storage",
-                        "code": "library_write_failed",
-                        "message": str(exc)[:300],
-                        "retryable": False,
-                    },
-                }
-            )
-            return
-        self.publish_event(narrated)
-
-    def publish_event(self, payload: dict[str, Any]) -> None:
-        """Fan any event (stored scene or ControlFrame error) to live subscribers.
-
-        Events without a seq (errors) are EPHEMERAL: not persisted, not in Last-Event-ID replay.
-        """
-        for queue in list(self._subscribers):
-            queue.put_nowait(payload)
-
-    def store(self, scene: dict[str, Any]) -> dict[str, Any]:
-        """Persist media + index row (blocking); returns the stored NarratedScene."""
-        scene_id: str = scene["scene_id"]
-        narration: str = scene["narration"]
-        style_key: str = scene["style_key"]
-        title = _stored_title(scene.get("title"), narration)
-
-        scene_dir = self.media_dir / scene_id
-        scene_dir.mkdir(parents=True, exist_ok=True)
-        clip_frames = scene.get("clip_frames") or []
-        poster = pick_key_frame(clip_frames) if clip_frames else scene["image"]
-        poster.convert(RGB_MODE).save(scene_dir / "frame.jpg", "JPEG", quality=POSTER_JPEG_QUALITY)
-        if len(clip_frames) >= 2:
-            try:
-                _write_clip_mp4(scene_dir / "clip.mp4", clip_frames)
-            except Exception as exc:
-                print(
-                    f"small_cuts.engine: clip write failed for scene {scene_id}: {exc!r}",
-                    file=sys.stderr,
-                )
-        render_title_card(title, style_key).save(scene_dir / "card.webp", "WEBP")
-        (scene_dir / "voice.wav").write_bytes(_wav_bytes(scene["audio"], scene["sample_rate"]))
-
-        narrator_backend = narrator.get_backend()
-        tts_backend = tts.get_tts_backend()
-        engine = {
-            "narrator_model": narrator_backend.model_id,
-            "narrator_backend": narrator_backend.name,
-            "tts_model": tts_backend.model_id,
-            "latency_ms": scene["latency_ms"],
-        }
-        with self._lock, self._db:
-            # max+1 under the lock: monotonic across the process AND across restarts.
-            seq = self._db.execute("SELECT COALESCE(MAX(seq), -1) + 1 FROM scenes").fetchone()[0]
-            self._db.execute(
-                _INSERT,
-                (
-                    scene_id,
-                    seq,
-                    scene["moment_id"],
-                    scene["session_id"],
-                    scene["captured_at"],
-                    scene["created_at"],
-                    style_key,
-                    title,
-                    narration,
-                    "private",
-                    OWNER,
-                    json.dumps(engine),
-                ),
-            )
-        stored = self.get(scene_id)
-        assert stored is not None  # the row was just inserted
-        return stored
-
-    # -- queries ---------------------------------------------------------------------
-
-    def to_narrated_scene(self, row: sqlite3.Row) -> dict[str, Any]:
-        """Contract-valid NarratedScene (1.1.0) for one stored row."""
-        scene_id = row["scene_id"]
-        media = {
-            "frame_url": f"/media/{scene_id}/frame.jpg",
-            "card_url": f"/media/{scene_id}/card.webp",
-            "audio_url": f"/media/{scene_id}/voice.wav",
-        }
-        if (self.media_dir / scene_id / "clip.mp4").is_file():
-            media["clip_url"] = f"/media/{scene_id}/clip.mp4"
-        return {
-            "contract_version": CONTRACT_VERSION,
-            "scene_id": scene_id,
-            "moment_id": row["moment_id"],
-            "session_id": row["session_id"],
-            "captured_at": row["captured_at"],
-            "created_at": row["created_at"],
-            "style_key": row["style_key"],
-            "title": row["title"],
-            "narration": row["narration"],
-            "visibility": row["visibility"],
-            "seq": row["seq"],
-            "owner": row["owner"],
-            "media": media,
-            "engine": json.loads(row["engine"]),
-        }
-
-    def list_scenes(
-        self,
-        session_id: str | None = None,
-        visibility: str | None = None,
-        limit: int = 100,
-    ) -> list[dict[str, Any]]:
-        """Newest bounded window, returned in scene chronology for the viewer."""
-        clauses, params = [], []
-        if session_id is not None:
-            clauses.append("session_id = ?")
-            params.append(session_id)
-        if visibility is not None:
-            clauses.append("visibility = ?")
-            params.append(visibility)
-        where = f" WHERE {' AND '.join(clauses)}" if clauses else ""
-        query = (
-            "SELECT * FROM ("
-            f"SELECT * FROM scenes{where} ORDER BY seq DESC LIMIT ?"
-            ") ORDER BY captured_at, seq"
-        )
-        with self._lock:
-            rows = self._db.execute(query, (*params, limit)).fetchall()
-        return [self.to_narrated_scene(row) for row in rows]
-
-    def get(self, scene_id: str) -> dict[str, Any] | None:
-        with self._lock:
-            row = self._db.execute(
-                "SELECT * FROM scenes WHERE scene_id = ?", (scene_id,)
-            ).fetchone()
-        return self.to_narrated_scene(row) if row is not None else None
-
-    def set_visibility(self, scene_id: str, visibility: str) -> dict[str, Any] | None:
-        """The viewer's only write (D7). Returns the updated scene, or None if unknown."""
-        if visibility not in VISIBILITIES:
-            raise ValueError(f"Unknown visibility {visibility!r}; expected one of {VISIBILITIES}")
-        with self._lock, self._db:
-            updated = self._db.execute(
-                "UPDATE scenes SET visibility = ? WHERE scene_id = ?", (visibility, scene_id)
-            ).rowcount
-        return self.get(scene_id) if updated else None
-
-    def scenes_since(self, seq: int) -> list[dict[str, Any]]:
-        """Scenes with seq > `seq`, ordered by seq — the SSE Last-Event-ID replay."""
-        with self._lock:
-            rows = self._db.execute(
-                "SELECT * FROM scenes WHERE seq > ? ORDER BY seq", (seq,)
-            ).fetchall()
-        return [self.to_narrated_scene(row) for row in rows]
-
-    def media_path(self, scene_id: str, filename: str) -> Path | None:
-        """Resolve a media file, or None: unknown name, traversal, or missing file."""
-        if filename not in MEDIA_FILES:
-            return None
-        path = (self.media_dir / scene_id / filename).resolve()
-        if not path.is_relative_to(self.media_dir):  # traversal via scene_id
-            return None
-        return path if path.is_file() else None
-
-    # -- pub/sub ----------------------------------------------------------------------
-
-    def subscribe(self) -> asyncio.Queue[dict[str, Any]]:
-        """New-scene feed for one SSE connection; pair with `unsubscribe`."""
-        queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
-        self._subscribers.append(queue)
-        return queue
-
-    def unsubscribe(self, queue: asyncio.Queue[dict[str, Any]]) -> None:
-        with contextlib.suppress(ValueError):
-            self._subscribers.remove(queue)
-
-    def close(self) -> None:
-        with self._lock:
-            self._db.close()
-
-
-def _write_clip_mp4(
-    path: Path,
-    frames: list[Image.Image],
-    fps: int = CLIP_MP4_FPS,
-    blend_steps: int = CLIP_BLEND_STEPS,
-) -> None:
-    """Render a small browser-playable MP4 from sampled POV frames."""
-    import av
-
-    rgb_frames = [frame.convert(RGB_MODE) for frame in frames]
-    width, height = rgb_frames[0].size
-    # H.264/yuv420p expects even dimensions. Preserve portrait aspect and only
-    # shave one pixel if needed; capture frames are already downscaled upstream.
-    width = max(H264_MIN_DIMENSION, width - (width % 2))
-    height = max(H264_MIN_DIMENSION, height - (height % 2))
-    encode_frames = _smooth_clip_frames(rgb_frames, blend_steps=blend_steps, size=(width, height))
-
-    container = av.open(str(path), "w")
-    try:
-        stream = container.add_stream(PRIMARY_VIDEO_CODEC, rate=fps)
-    except Exception:
-        stream = container.add_stream(FALLBACK_VIDEO_CODEC, rate=fps)
-    stream.width = width
-    stream.height = height
-    stream.pix_fmt = VIDEO_PIXEL_FORMAT
-
-    try:
-        for image in encode_frames:
-            frame = av.VideoFrame.from_image(image)
-            for packet in stream.encode(frame):
-                container.mux(packet)
-        for packet in stream.encode():
-            container.mux(packet)
-    finally:
-        container.close()
-
-
-def _smooth_clip_frames(
-    frames: list[Image.Image],
-    blend_steps: int = CLIP_BLEND_STEPS,
-    size: tuple[int, int] | None = None,
-) -> list[Image.Image]:
-    """Insert tiny cross-dissolve frames so sampled POV clips do not hard-cut."""
-    if not frames:
-        return []
-    prepared = []
-    for image in frames:
-        image = image.convert(RGB_MODE)
-        if size is not None and image.size != size:
-            image = image.resize(size, Image.Resampling.LANCZOS)
-        prepared.append(image)
-    if blend_steps <= 0 or len(prepared) < 2:
-        return prepared
-
-    smoothed = [prepared[0]]
-    for previous, current in zip(prepared, prepared[1:], strict=False):
-        for step in range(1, blend_steps + 1):
-            alpha = step / (blend_steps + 1)
-            smoothed.append(Image.blend(previous, current, alpha))
-        smoothed.append(current)
-    return smoothed
-
-
-def _stored_title(raw_title: object, narration: str) -> str:
-    if isinstance(raw_title, str) and raw_title.strip():
-        return derive_title(raw_title, max_len=80)
-    return derive_title(narration, max_len=80)

src/small_cuts/engine/read_gate.py

@@ -1,102 +0,0 @@
-"""Read-only public gate for the live-demo engine surface.
-
-The capture/write socket stays private on Tailnet. This app is the origin behind the public
-Cloudflare Tunnel hostname and only proxies viewer/library reads through to the local engine.
-"""
-
-from __future__ import annotations
-
-import os
-from collections.abc import AsyncIterator
-from contextlib import asynccontextmanager
-
-import httpx
-from fastapi import FastAPI, Request
-from fastapi.responses import PlainTextResponse, Response, StreamingResponse
-from starlette.background import BackgroundTask
-
-ORIGIN_ENV = "SMALL_CUTS_ORIGIN_ENGINE_URL"
-DEFAULT_ORIGIN = "http://127.0.0.1:8077"
-BLOCKED_TEXT = "small-cuts public endpoint is read-only\n"
-HOP_BY_HOP_HEADERS = {
-    "connection",
-    "keep-alive",
-    "proxy-authenticate",
-    "proxy-authorization",
-    "te",
-    "trailer",
-    "transfer-encoding",
-    "upgrade",
-}
-
-
-def is_public_read_allowed(method: str, path: str) -> bool:
-    if method.upper() != "GET":
-        return False
-    return path in ("/v1/scenes", "/v1/scenes/stream") or path.startswith("/media/")
-
-
-def _forward_headers(headers: httpx.Headers | dict[str, str]) -> dict[str, str]:
-    return {
-        key: value
-        for key, value in headers.items()
-        if key.lower() not in HOP_BY_HOP_HEADERS and key.lower() != "host"
-    }
-
-
-def _origin_url(origin_url: str, request: Request) -> str:
-    url = f"{origin_url.rstrip('/')}{request.url.path}"
-    return f"{url}?{request.url.query}" if request.url.query else url
-
-
-def build_read_gate_app(origin_url: str | None = None) -> FastAPI:
-    origin = (origin_url or os.environ.get(ORIGIN_ENV) or DEFAULT_ORIGIN).rstrip("/")
-
-    @asynccontextmanager
-    async def lifespan(app: FastAPI) -> AsyncIterator[None]:
-        app.state.client = httpx.AsyncClient(timeout=None)
-        try:
-            yield
-        finally:
-            await app.state.client.aclose()
-
-    app = FastAPI(title="small-cuts public read gate", lifespan=lifespan)
-
-    @app.api_route(
-        "/{path:path}",
-        methods=["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
-        response_model=None,
-    )
-    async def public_gate(path: str, request: Request) -> Response:
-        if not is_public_read_allowed(request.method, request.url.path):
-            return PlainTextResponse(BLOCKED_TEXT, status_code=403)
-
-        client: httpx.AsyncClient = request.app.state.client
-        upstream = await client.send(
-            client.build_request(
-                "GET",
-                _origin_url(origin, request),
-                headers=_forward_headers(request.headers),
-            ),
-            stream=True,
-        )
-
-        async def body() -> AsyncIterator[bytes]:
-            async for chunk in upstream.aiter_raw():
-                yield chunk
-
-        async def close() -> None:
-            await upstream.aclose()
-
-        return StreamingResponse(
-            body(),
-            status_code=upstream.status_code,
-            headers=_forward_headers(upstream.headers),
-            media_type=upstream.headers.get("content-type"),
-            background=BackgroundTask(close),
-        )
-
-    return app
-
-
-app = build_read_gate_app()

src/small_cuts/engine/session.py

@@ -1,452 +0,0 @@
-"""Mobile-facing WebSocket session for the narration engine.
-
-MomentEnvelope in, ControlFrame + SceneAudio out, per docs/contracts.
-Backpressure is D8 (queue depth <= 1, coalesce-to-newest); freshness is D9
-(`play_by` = created_at + 60 s). Pipeline failures become error frames —
-the socket itself never crashes on a bad moment.
-"""
-
-from __future__ import annotations
-
-import asyncio
-import base64
-import contextlib
-import inspect
-import io
-import json
-import sys
-import time
-import uuid
-import wave
-from collections import OrderedDict
-from collections.abc import Callable
-from dataclasses import dataclass, field
-from datetime import datetime, timedelta, timezone
-from pathlib import Path
-from typing import Any
-
-import jsonschema
-import numpy as np
-from fastapi import WebSocket, WebSocketDisconnect
-from PIL import Image
-
-from small_cuts import narrator, tts
-from small_cuts.styles import DEFAULT_STYLE_KEY
-
-CONTRACT_VERSION = "1.1.0"
-PLAY_BY_SECONDS = 60
-MAX_FRAME_SIDE = 1024  # contract cap: decoded longest side <= 1024 px (moment.schema.json)
-SEEN_MOMENTS_CAP = 4096  # a day of moments is far less
-_CONTRACTS = Path(__file__).resolve().parents[3] / "docs" / "contracts"
-
-SceneSink = Callable[[dict[str, Any]], Any]
-"""Receives every successful scene; Task 2 plugs the library/SSE fan-out here."""
-
-
-def _validator(name: str) -> jsonschema.Draft202012Validator:
-    return jsonschema.Draft202012Validator(json.loads((_CONTRACTS / name).read_text()))
-
-
-_MOMENT = _validator("moment.schema.json")
-_SCENE_AUDIO = _validator("scene-audio.schema.json")
-_BACKGROUND_STORAGE_TASKS: set[asyncio.Task[None]] = set()
-
-
-def _noop_sink(scene: dict[str, Any]) -> None:
-    return None
-
-
-class MomentIdLRU:
-    """Bounded dedupe set: insertion-ordered, oldest ids evicted past `cap`."""
-
-    def __init__(self, cap: int = SEEN_MOMENTS_CAP) -> None:
-        self._cap = cap
-        self._ids: OrderedDict[str, None] = OrderedDict()
-
-    def __contains__(self, moment_id: object) -> bool:
-        return moment_id in self._ids
-
-    def add(self, moment_id: str) -> None:
-        self._ids[moment_id] = None
-        self._ids.move_to_end(moment_id)
-        while len(self._ids) > self._cap:
-            self._ids.popitem(last=False)
-
-    def discard(self, moment_id: str) -> None:
-        self._ids.pop(moment_id, None)
-
-
-@dataclass
-class EngineState:
-    """Process-lifetime state shared across session sockets."""
-
-    sink: SceneSink = _noop_sink
-    error_sink: SceneSink | None = None  # receives every error ControlFrame (viewer fan-out, D9)
-    seen_moment_ids: MomentIdLRU = field(default_factory=MomentIdLRU)
-
-
-@dataclass
-class _Queued:
-    envelope: dict[str, Any]
-    queued_at: float
-
-
-class _ValidationFailure(Exception):
-    """Post-admission validation failure (undecodable or over-cap frame); never retryable."""
-
-    def __init__(self, code: str, message: str) -> None:
-        super().__init__(message)
-        self.code = code
-
-
-def _log_worker_failure(task: asyncio.Task) -> None:
-    """A drain-task bug must fail loudly, not strand moments as unretrieved exceptions."""
-    if task.cancelled():
-        return
-    exc = task.exception()
-    if exc is not None:
-        print(f"small_cuts.engine: session worker task crashed: {exc!r}", file=sys.stderr)
-
-
-def _retain_background_storage(task: asyncio.Task[None]) -> None:
-    """Keep shielded scene storage alive after the client WebSocket is gone."""
-    _BACKGROUND_STORAGE_TASKS.add(task)
-    task.add_done_callback(_BACKGROUND_STORAGE_TASKS.discard)
-
-
-class SessionRunner:
-    """One connected capture app: admission, the single queue slot, the pipeline."""
-
-    def __init__(self, ws: WebSocket, state: EngineState) -> None:
-        self._ws = ws
-        self._state = state
-        self._send_lock = asyncio.Lock()
-        self._pending: _Queued | None = None
-        self._worker: asyncio.Task | None = None
-        self._processing = False
-        self._last_status: tuple[bool, int] | None = None
-
-    async def run(self) -> None:
-        try:
-            while True:
-                message = await self._ws.receive()
-                if message["type"] == "websocket.disconnect":
-                    break
-                text = message.get("text")
-                if text is None:  # binary frame: not in the contract, but don't drop the socket
-                    await self._send_ack(None, "rejected", "binary frames not supported")
-                    continue
-                await self._admit(text)
-        except WebSocketDisconnect:
-            pass
-        finally:
-            if self._worker is not None:
-                self._worker.cancel()
-
-    # -- admission (every envelope gets exactly one ack) ----------------------
-
-    async def _admit(self, raw: str) -> None:
-        try:
-            envelope = json.loads(raw)
-        except json.JSONDecodeError as exc:
-            await self._send_ack(None, "rejected", f"invalid JSON: {exc}")
-            return
-        moment_id = envelope.get("moment_id") if isinstance(envelope, dict) else None
-        if not isinstance(moment_id, str):
-            moment_id = None
-        error = jsonschema.exceptions.best_match(_MOMENT.iter_errors(envelope))
-        if error is not None:
-            await self._send_ack(moment_id, "rejected", error.message)
-            return
-        if moment_id in self._state.seen_moment_ids:
-            await self._send_ack(moment_id, "duplicate")
-            return
-        self._state.seen_moment_ids.add(moment_id)
-
-        queued = _Queued(envelope, time.perf_counter())
-        if not self._processing:
-            self._processing = True
-            await self._send_ack(moment_id, "accepted")
-            self._worker = asyncio.create_task(self._drain(queued))
-            self._worker.add_done_callback(_log_worker_failure)
-        elif self._pending is None:
-            self._pending = queued
-            await self._send_ack(moment_id, "accepted")
-        else:  # D8: replace the un-started moment; stale narration is worse than none
-            dropped = self._pending
-            self._pending = queued
-            await self._send_ack(dropped.envelope["moment_id"], "dropped_coalesced")
-            await self._send_ack(moment_id, "accepted")
-        await self._emit_status()
-
-    # -- processing ------------------------------------------------------------
-
-    async def _drain(self, queued: _Queued) -> None:
-        current: _Queued | None = queued
-        try:
-            while current is not None:
-                await self._process(current)
-                current, self._pending = self._pending, None
-                await self._emit_status()
-        finally:
-            self._processing = False
-        await self._emit_status()  # skipped on cancellation: the socket is gone
-
-    async def _process(self, item: _Queued) -> None:
-        envelope = item.envelope
-        moment_id: str = envelope["moment_id"]
-        context = envelope.get("context") or {}
-        style_key = context.get("style_key") or DEFAULT_STYLE_KEY
-        started = time.perf_counter()
-        queue_ms = _ms(started - item.queued_at)
-        stage = "narration"
-        try:
-            image, narration = await asyncio.to_thread(
-                _decode_and_narrate,
-                envelope,
-                style_key,
-                context.get("user_hint", ""),
-            )
-            narration_ms = _ms(time.perf_counter() - started)
-            stage = "tts"
-            tts_started = time.perf_counter()
-            speech = await asyncio.to_thread(tts.speak, narration.text)
-            audio_b64 = base64.b64encode(_wav_bytes(speech.audio, speech.sample_rate)).decode()
-            tts_ms = _ms(time.perf_counter() - tts_started)
-
-            stage = "storage"  # the outgoing SceneAudio is the engine's stored artifact
-            created_at = datetime.now(timezone.utc)
-            payload = {
-                "contract_version": CONTRACT_VERSION,
-                "scene_id": str(uuid.uuid4()),
-                "moment_id": moment_id,
-                "created_at": created_at.isoformat(),
-                "play_by": (created_at + timedelta(seconds=PLAY_BY_SECONDS)).isoformat(),
-                "format": "wav_complete",
-                "audio_b64": audio_b64,
-                "sample_rate": speech.sample_rate,
-                "narration": narration.text,
-            }
-            _SCENE_AUDIO.validate(payload)  # outgoing drift becomes an error frame, never silence
-            await self._send_json(payload)
-        except _ValidationFailure as exc:
-            # The resend would fail the same way, but dedupe only what produced a scene.
-            self._state.seen_moment_ids.discard(moment_id)
-            await self._send_error(moment_id, "validation", exc, code=exc.code, retryable=False)
-            return
-        except Exception as exc:
-            # Drop the id so a client resend is genuinely re-processed (honest retryable).
-            self._state.seen_moment_ids.discard(moment_id)
-            retryable = stage in ("narration", "tts")
-            code = "scene_audio_schema_drift" if stage == "storage" else None
-            await self._send_error(moment_id, stage, exc, code=code, retryable=retryable)
-            return
-
-        storage_task = asyncio.create_task(
-            self._finish_scene_storage(
-                envelope=envelope,
-                image=image,
-                scene_audio=payload,
-                narration_text=narration.text,
-                title=narration.title,
-                speech=speech,
-                style_key=style_key,
-                queue_ms=queue_ms,
-                narration_ms=narration_ms,
-                tts_ms=tts_ms,
-            )
-        )
-        _retain_background_storage(storage_task)
-        try:
-            await asyncio.shield(storage_task)
-        except asyncio.CancelledError:
-            storage_task.add_done_callback(_log_worker_failure)
-            raise
-
-    async def _finish_scene_storage(
-        self,
-        *,
-        envelope: dict[str, Any],
-        image: Image.Image,
-        scene_audio: dict[str, Any],
-        narration_text: str,
-        title: str,
-        speech: tts.Speech,
-        style_key: str,
-        queue_ms: int,
-        narration_ms: int,
-        tts_ms: int,
-    ) -> None:
-        clip_frames = await asyncio.to_thread(
-            _decode_clip_frames_for_storage, envelope, image, scene_audio["scene_id"]
-        )
-        await self._hand_to_sink(
-            self._state.sink,
-            {
-                "scene_id": scene_audio["scene_id"],
-                "moment_id": envelope["moment_id"],
-                "session_id": envelope["session_id"],
-                "captured_at": envelope["captured_at"],
-                "created_at": scene_audio["created_at"],
-                "style_key": style_key,
-                "title": title,
-                "narration": narration_text,
-                "image": image,
-                "clip_frames": clip_frames,
-                "audio": speech.audio,
-                "sample_rate": speech.sample_rate,
-                "latency_ms": {
-                    "queue": queue_ms,
-                    "narration": narration_ms,
-                    "tts": tts_ms,
-                    "total": queue_ms + narration_ms + tts_ms,
-                },
-            },
-        )
-
-    async def _hand_to_sink(self, sink: SceneSink | None, payload: dict[str, Any]) -> None:
-        if sink is None:
-            return
-        with contextlib.suppress(Exception):  # a sink bug must not kill the session
-            result = sink(payload)
-            if inspect.isawaitable(result):
-                await result
-
-    # -- outbound frames ---------------------------------------------------------
-
-    async def _send_ack(self, moment_id: str | None, result: str, detail: str = "") -> None:
-        ack: dict[str, Any] = {"result": result}
-        if detail:
-            ack["detail"] = detail[:200]
-        await self._send_json(
-            {
-                "contract_version": CONTRACT_VERSION,
-                "kind": "ack",
-                "moment_id": moment_id,
-                "ack": ack,
-            }
-        )
-
-    async def _send_error(
-        self,
-        moment_id: str,
-        stage: str,
-        exc: Exception,
-        *,
-        retryable: bool,
-        code: str | None = None,
-    ) -> None:
-        frame = {
-            "contract_version": CONTRACT_VERSION,
-            "kind": "error",
-            "moment_id": moment_id,
-            "error": {
-                "stage": stage,
-                "code": (code or type(exc).__name__)[:60],
-                "message": str(exc)[:300],
-                "retryable": retryable,
-            },
-        }
-        await self._send_json(frame)
-        # D9 honest timeline: the same failure fans out to the viewer stream.
-        await self._hand_to_sink(self._state.error_sink, frame)
-
-    async def _emit_status(self) -> None:
-        snapshot = (self._processing, int(self._pending is not None))
-        if snapshot == self._last_status:
-            return
-        self._last_status = snapshot
-        await self._send_json(
-            {
-                "contract_version": CONTRACT_VERSION,
-                "kind": "status",
-                "moment_id": None,
-                "status": {"busy": snapshot[0], "queue_depth": snapshot[1]},
-            }
-        )
-
-    async def _send_json(self, payload: dict[str, Any]) -> None:
-        text = json.dumps(payload)  # serialization bugs must surface, not be swallowed
-        async with self._send_lock:
-            with contextlib.suppress(Exception):  # client gone mid-send; run() closes out
-                await self._ws.send_text(text)
-
-
-def _decode_and_narrate(
-    envelope: dict[str, Any], style_key: str, scene_hint: str
-) -> tuple[Image.Image, narrator.Narration]:
-    """Decode the selected frame + narrate it in one worker-thread hop."""
-    try:
-        selected = _decode_frame(envelope["frames"][0])
-        _validate_frame_size(selected)
-    except _ValidationFailure:
-        raise
-    except Exception as exc:
-        raise _ValidationFailure("frame_decode_failed", f"undecodable frame: {exc}") from exc
-    return (
-        selected,
-        narrator.narrate(selected, style_key=style_key, scene_hint=scene_hint),
-    )
-
-
-def _decode_frame(frame: dict[str, Any]) -> Image.Image:
-    data = base64.b64decode(frame["jpeg_b64"])
-    image = Image.open(io.BytesIO(data))
-    image.load()
-    return image
-
-
-def _validate_frame_size(image: Image.Image) -> None:
-    longest = max(image.size)
-    if longest > MAX_FRAME_SIDE:
-        raise _ValidationFailure(
-            "frame_exceeds_cap",
-            f"decoded longest side {longest} px exceeds the {MAX_FRAME_SIDE} px contract cap",
-        )
-
-
-def _decode_clip_frames(envelope: dict[str, Any], selected: Image.Image) -> list[Image.Image]:
-    decoded: list[tuple[int, int, Image.Image]] = [
-        (int(envelope["frames"][0].get("ts_offset_ms", 0)), 0, selected)
-    ]
-    for index, frame in enumerate(envelope["frames"][1:], start=1):
-        image = _decode_frame(frame)
-        _validate_frame_size(image)
-        decoded.append((int(frame.get("ts_offset_ms", index)), index, image))
-    return [image for _, _, image in sorted(decoded, key=lambda item: (item[0], item[1]))]
-
-
-def _decode_clip_frames_for_storage(
-    envelope: dict[str, Any], selected: Image.Image, scene_id: str
-) -> list[Image.Image]:
-    """Decode viewer-only supplemental frames after SceneAudio is already sent."""
-    if len(envelope["frames"]) < 2:
-        return [selected]
-    try:
-        return _decode_clip_frames(envelope, selected)
-    except Exception as exc:
-        print(
-            f"small_cuts.engine: clip frame decode failed for scene {scene_id}: {exc!r}",
-            file=sys.stderr,
-        )
-        return [selected]
-
-
-def _wav_bytes(audio: np.ndarray, sample_rate: int) -> bytes:
-    buffer = io.BytesIO()
-    try:
-        import soundfile
-
-        soundfile.write(buffer, audio, sample_rate, format="WAV", subtype="PCM_16")
-    except ImportError:
-        pcm = (np.clip(audio, -1.0, 1.0) * 32767.0).astype("<i2")
-        with wave.open(buffer, "wb") as wav:
-            wav.setnchannels(1)
-            wav.setsampwidth(2)
-            wav.setframerate(sample_rate)
-            wav.writeframes(pcm.tobytes())
-    return buffer.getvalue()
-
-
-def _ms(seconds: float) -> int:
-    return max(0, round(seconds * 1000))