bench/manager/core/events.py

"""Ingesting normalized events — the fixed contract every adapter speaks.

Adapters translate their vendor's payloads at the edge (see
adapters/*/emit*) and POST the normalized schema here:

    {"v": 1, "session": str, "kind": str, "summary": str,
     "file"?: str, "cmd"?: str, "detail"?: str, "ok"?: bool,
     "running"?: bool, "agent"?: str, "task"?: str}

Core sanitises, updates the session registry, persists a slim record per
session, and pushes to connected browsers over SSE. It never interprets a
vendor's tool vocabulary — that knowledge lives in the adapter.

Beside each session's event log sits its identity — who the session
belonged to, written whole (see state.persist_identity). Events say what
happened; the identity says whose, and it is the only part a restart
cannot recover from the stream.
"""

from __future__ import annotations

import json
import time

import config
import state
from taskfiles import NUMBER_RE

KINDS = {"session", "end", "idle", "edit", "read", "search", "command",
         "test", "check", "git", "plan", "subagent", "web", "report", "other"}


# In-memory copy of what each session's identity file already says, so the
# sidecar is rewritten only when what the board knows actually changes.
_WRITTEN: dict[str, dict] = {}


def session_label(meta: dict) -> str:
    """Who a session was — in three registers, because there are three
    different states and only one of them is the person.

    An agent's name comes from the live launch record while this board
    still holds it, and from the identity persisted with the session after
    a restart; `Agent` (or `Review`) is the honest fallback when the id is
    known but the name is not. `You` is said only of a session the board
    positively knows carried no agent — every live one, and every replayed
    one whose identity file records that. A log written before identities
    were recorded is none of those: it is unknown, and says so rather than
    claiming to have been you.
    """
    agent_id = meta.get("agentId") or ""
    if agent_id:
        record = state.AGENTS.get(agent_id) or {}
        task = meta.get("task") or ""
        num = NUMBER_RE.match(task)
        who = (record.get("name") or meta.get("agentName")
               or ("Review" if agent_id.startswith("review-") else "Agent"))
        return f"{who} · #{num.group(1)}" if num else who
    if meta.get("known"):
        return f"You · {meta['id'][:8]}"
    return f"Session · {meta['id'][:8]}"


def _txt(value, cap: int) -> str | None:
    return value[:cap] if isinstance(value, str) and value else None


def ingest_event(raw: dict) -> None:
    if not isinstance(raw, dict) or "kind" not in raw:
        return  # not a normalized event — adapters own translation
    sid = str(raw.get("session") or "unknown")
    kind = raw["kind"] if raw.get("kind") in KINDS else "other"

    event = {"ts": time.time(), "session": sid, "kind": kind,
             "summary": _txt(raw.get("summary"), 300) or kind}
    for key, cap in (("file", 300), ("cmd", 300), ("detail", 900)):
        value = _txt(raw.get(key), cap)
        if value:
            event[key] = value
    if isinstance(raw.get("ok"), bool):
        event["ok"] = raw["ok"]
    if raw.get("running") is True:
        event["running"] = True

    agent_id = _txt(raw.get("agent"), 120)
    task = _txt(raw.get("task"), 200)

    with state.LOCK:
        meta = state.SESSIONS.setdefault(sid, {
            "id": sid, "started": event["ts"], "count": 0,
            "agentId": None, "agentName": None, "agentModel": None,
            "task": None, "status": "active",
        })
        just_linked = False
        if agent_id:
            meta["agentId"] = agent_id
            record = state.AGENTS.get(agent_id)
            if record is not None:
                # The name and the model exist nowhere but this record, and
                # it may only have been registered after the child's first
                # event — so they are taken every time, not just on linking.
                meta["agentName"] = record.get("name")
                meta["agentModel"] = record.get("model")
                just_linked = record["session"] is None
                record["session"] = sid
                task = task or record["task"]
        if task:
            meta["task"] = task
        # An event reaching here is a session the board is watching live, so
        # it knows what it is looking at: an agent when one identified
        # itself, the person when none did.
        meta["known"] = True
        meta["last"] = event["ts"]
        meta["lastSummary"] = event["summary"]
        meta["lastKind"] = kind
        meta["status"] = {"end": "ended", "idle": "idle"}.get(kind, "active")
        meta["label"] = session_label(meta)
        identity = None
        if not event.get("running"):
            meta["count"] += 1
            state.EVENTS.setdefault(sid, []).append(event)
            del state.EVENTS[sid][:-config.EVENTS_CAP]
            identity = {"agentId": meta["agentId"], "name": meta["agentName"],
                        "model": meta["agentModel"], "task": meta["task"]}
            if identity == _WRITTEN.get(sid):
                identity = None
            else:
                _WRITTEN[sid] = identity
        meta_snapshot = dict(meta)

    if not event.get("running"):
        state.persist(f"{sid}.jsonl", event)
    if identity is not None:
        # written beside the log it belongs to, and only when it changed
        state.persist_identity(sid, identity)
    if just_linked:
        # the agent's card can now show its live line instead of "warming up"
        state.broadcast({"type": "agents"})
    state.broadcast({"type": "event", "event": event, "session": meta_snapshot})


def load_disk_sessions() -> None:
    """Rebuild session metadata from state/sessions/ so past sessions replay."""
    if not config.SESSIONS_DIR.is_dir():
        return
    for path in config.SESSIONS_DIR.glob("*.jsonl"):
        sid = path.stem
        if sid == "board" or sid in state.SESSIONS:
            continue
        try:
            lines = path.read_text(encoding="utf-8").splitlines()
            first = json.loads(lines[0])
            last = json.loads(lines[-1])
        except (OSError, json.JSONDecodeError, IndexError):
            continue
        # Who it was, if it was recorded. Absent = a log from before
        # identities were written; the label must not fill that gap in.
        identity = state.read_identity(sid)
        known = identity is not None
        identity = identity or {}
        meta = {
            "id": sid, "started": first.get("ts"), "last": last.get("ts"),
            "count": len(lines),
            "agentId": identity.get("agentId"),
            "agentName": identity.get("name"),
            "agentModel": identity.get("model"),
            "task": identity.get("task"),
            "known": known,
            "status": "ended", "lastSummary": last.get("summary"),
            "lastKind": last.get("kind"),
        }
        meta["label"] = session_label(meta)
        state.SESSIONS[sid] = meta

    board_log = config.SESSIONS_DIR / "board.jsonl"
    if board_log.is_file():
        try:
            lines = board_log.read_text(encoding="utf-8").splitlines()[-100:]
            state.BOARD_EVENTS.extend(json.loads(l) for l in lines)
        except (OSError, json.JSONDecodeError):
            pass


def session_events(sid: str) -> list[dict]:
    with state.LOCK:
        if sid in state.EVENTS:
            return list(state.EVENTS[sid])
    path = config.SESSIONS_DIR / f"{sid}.jsonl"
    if not path.is_file() or "/" in sid or ".." in sid:
        return []
    events = []
    try:
        for line in path.read_text(encoding="utf-8").splitlines()[-config.EVENTS_CAP:]:
            events.append(json.loads(line))
    except (OSError, json.JSONDecodeError):
        pass
    return events