chat/chat/state/meanwhile.py

"""Meanwhile-scene projection (T63).

A "meanwhile" scene is a 2-bot scene where ``present_set = {host_bot_id,
guest_bot_id}`` and "you" is absent. It runs alongside an active you-scene
(its parent) so bots can have private interactions whose outcome later
surfaces back to the you-scene as a pending digest.

The underlying ``scenes`` table (migration 0007) has no explicit ``status``
column; "active" is encoded as ``ended_at IS NULL`` and "closed" as
``ended_at IS NOT NULL``. This module preserves that convention and adds
two new columns introduced by migration 0011:

- ``present_set_kind`` — ``'you_host'`` (default) for normal scenes,
  ``'host_guest'`` for meanwhile child scenes.
- ``parent_scene_id`` — the you-scene a meanwhile child hangs off of.

Pending meanwhile digests live in their own table
(``meanwhile_digest_pending``) and are consumed when their summary is
surfaced in the next you-scene's prompt.
"""
from __future__ import annotations
import json
from sqlite3 import Connection

from chat.eventlog.projector import on
from chat.eventlog.log import Event


@on("meanwhile_scene_started")
def _apply_meanwhile_scene_started(conn: Connection, e: Event) -> None:
    """Insert a new scenes row for the meanwhile child.

    The caller supplies an explicit ``scene_id`` so subsequent events
    (close, digest) can reference it without round-tripping through
    ``lastrowid``.
    """
    p = e.payload
    conn.execute(
        "INSERT INTO scenes ("
        "id, chat_id, started_at, ended_at, significance, "
        "participants_json, present_set_kind, parent_scene_id"
        ") VALUES (?, ?, ?, NULL, 0, ?, 'host_guest', ?)",
        (
            p["scene_id"],
            p["chat_id"],
            p.get("started_at"),
            # participants_json mirrors the present_set: host + guest bot.
            # json.dumps ensures bot ids with quotes/backslashes can't corrupt the JSON literal.
            json.dumps([p["host_bot_id"], p["guest_bot_id"]]),
            p["parent_scene_id"],
        ),
    )


@on("meanwhile_scene_closed")
def _apply_meanwhile_scene_closed(conn: Connection, e: Event) -> None:
    """Mark the meanwhile scene closed by stamping ``ended_at``."""
    p = e.payload
    conn.execute(
        "UPDATE scenes SET ended_at = ? "
        "WHERE id = ? AND present_set_kind = 'host_guest' "
        "AND ended_at IS NULL",
        (p.get("closed_at"), p["scene_id"]),
    )


@on("meanwhile_digest_created")
def _apply_meanwhile_digest_created(conn: Connection, e: Event) -> None:
    """Queue a digest for surfacing to the next you-scene's prompt."""
    p = e.payload
    conn.execute(
        "INSERT INTO meanwhile_digest_pending (scene_id, chat_id, summary) "
        "VALUES (?, ?, ?)",
        (p["scene_id"], p["chat_id"], p["summary"]),
    )


@on("meanwhile_digest_consumed")
def _apply_meanwhile_digest_consumed(conn: Connection, e: Event) -> None:
    """Mark a pending digest as consumed (idempotent on re-projection)."""
    p = e.payload
    conn.execute(
        "UPDATE meanwhile_digest_pending SET consumed_at = ? "
        "WHERE id = ? AND consumed_at IS NULL",
        (p.get("consumed_at"), p["digest_id"]),
    )


def _scene_row_to_dict(row: tuple) -> dict:
    """Shape a meanwhile-scene row.

    ``status`` is derived from ``ended_at`` for callers that prefer the
    higher-level vocabulary; ``closed_at`` aliases ``ended_at`` for the
    same reason. The underlying column remains ``ended_at``.
    """
    ended_at = row[5]
    return {
        "id": row[0],
        "chat_id": row[1],
        "started_at": row[2],
        "present_set_kind": row[3],
        "parent_scene_id": row[4],
        "closed_at": ended_at,
        "status": "closed" if ended_at is not None else "active",
    }


def list_meanwhile_scenes(
    conn: Connection, chat_id: str, status: str = "active"
) -> list[dict]:
    """Return meanwhile scenes for ``chat_id`` filtered by derived status."""
    if status == "active":
        ended_clause = "s.ended_at IS NULL"
    elif status == "closed":
        ended_clause = "s.ended_at IS NOT NULL"
    else:
        raise ValueError(f"unknown status: {status!r}")
    rows = conn.execute(
        "SELECT s.id, s.chat_id, s.started_at, s.present_set_kind, "
        "s.parent_scene_id, s.ended_at "
        "FROM scenes s "
        "WHERE s.chat_id = ? AND s.present_set_kind = 'host_guest' "
        f"AND {ended_clause} "
        "ORDER BY s.id ASC",
        (chat_id,),
    ).fetchall()
    return [_scene_row_to_dict(r) for r in rows]


def get_parent_scene(conn: Connection, scene_id: int) -> dict | None:
    """Given a meanwhile scene id, return its parent (you-scene) row."""
    row = conn.execute(
        "SELECT s.id, s.chat_id, s.started_at, s.present_set_kind, "
        "s.parent_scene_id, s.ended_at "
        "FROM scenes s JOIN scenes m ON m.parent_scene_id = s.id "
        "WHERE m.id = ?",
        (scene_id,),
    ).fetchone()
    if row is None:
        return None
    return _scene_row_to_dict(row)


def list_pending_meanwhile_digests(
    conn: Connection, chat_id: str
) -> list[dict]:
    """Return digests for ``chat_id`` that haven't been consumed yet."""
    rows = conn.execute(
        "SELECT id, scene_id, chat_id, summary, created_at "
        "FROM meanwhile_digest_pending "
        "WHERE chat_id = ? AND consumed_at IS NULL "
        "ORDER BY id ASC",
        (chat_id,),
    ).fetchall()
    return [
        {
            "id": r[0],
            "scene_id": r[1],
            "chat_id": r[2],
            "summary": r[3],
            "created_at": r[4],
        }
        for r in rows
    ]