"""

darwin-deck  - Live recursive agent tree event bus

Ports: HTTP 9893, WebSocket 9892

Source: Claude Code agent transcripts under /tmp/claude-0/-shared-projects/<session>/tasks/*.output

Contract: /shared/projects/darwin/deck/CONTRACTS.md §1

"""

from __future__ import annotations

import asyncio

import json

import os

import re

import time

import uuid

from datetime import datetime, timezone

from pathlib import Path

from typing import Any

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request

from fastapi.middleware.cors import CORSMiddleware

from fastapi.responses import JSONResponse

from pydantic import BaseModel, Field

TASK_ROOTS = [Path("/tmp/claude-0/-shared-projects")]

SCAN_INTERVAL_S = 2.0

MAX_AGENT_AGE_H = 48

LIVE_WINDOW_S = 180.0

STRUCTURAL_MODELS = {"system", "orchestrator", "host"}

def _parse_ts(ts: str | None) -> float:

    """Parse an ISO-8601 timestamp to a UTC epoch. Returns 0.0 on failure."""

    if not ts:

        return 0.0

    try:

        s = ts.replace("Z", "+00:00")

        dt = datetime.fromisoformat(s)

        if dt.tzinfo is None:

            dt = dt.replace(tzinfo=timezone.utc)

        return dt.timestamp()

    except Exception:

        return 0.0

class DeckEvent(BaseModel):

    ts: str = Field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

    agent_id: str

    parent_id: str | None = None

    type: str

    label: str = ""

    model: str = "sonnet"

    status: str = "running"

    tool: str | None = None

    tokens: int = 0

    host: str = "ct215"

    session: str = ""

class AgentState:

    def __init__(self, agent_id: str, session: str):

        self.agent_id = agent_id

        self.session = session

        self.parent_id: str | None = None

        self.label: str = ""

        self.model: str = "sonnet"

        self.status: str = "running"

        self.tokens: int = 0

        self.host: str = "ct215"

        self.spawned_at: str = datetime.now(timezone.utc).isoformat()

        self.last_activity: str = datetime.now(timezone.utc).isoformat()

        self.current_tool: str | None = None

        self.task_count: int = 0

        self.last_activity_ts: float = 0.0

    def is_structural(self) -> bool:

        """Synthetic grouping nodes (root / per-session / per-host) - kept in the

        tree for layout but never counted as live agents."""

        return self.model in STRUCTURAL_MODELS or self.agent_id == "ct215-root"

    def effective_status(self, now: float | None = None) -> str:

        """Real-time status. An agent is 'running' only if it has shown activity

        within LIVE_WINDOW_S AND has not hit a terminal marker. Stale 'running'

        agents (old transcripts that never wrote a clean return) collapse to

        'done'. Structural grouping nodes are never counted as active."""

        if self.is_structural():

            return self.status

        if self.status in ("done", "error"):

            return self.status

        now = time.time() if now is None else now

        if (now - self.last_activity_ts) > LIVE_WINDOW_S:

            return "done"

        return self.status

    def to_tree_node(self) -> dict:

        return {

            "agent_id": self.agent_id,

            "parent_id": self.parent_id,

            "label": self.label or self.agent_id[:8],

            "model": self.model,

            "status": self.effective_status(),

            "tokens": self.tokens,

            "host": self.host,

            "session": self.session,

            "spawned_at": self.spawned_at,

            "last_activity": self.last_activity,

            "current_tool": self.current_tool,

            "task_count": self.task_count,

        }

class TreeStore:

    """In-memory agent tree. Thread-safe enough for single async loop."""

    def __init__(self):

        self.agents: dict[str, AgentState] = {}

        root = AgentState("ct215-root", "system")

        root.label = "DARWIN"

        root.model = "system"

        root.status = "running"

        root.host = "ct215"

        root.spawned_at = datetime.now(timezone.utc).isoformat()

        root.last_activity = datetime.now(timezone.utc).isoformat()

        self.agents["ct215-root"] = root

    def upsert(self, agent_id: str, session: str) -> AgentState:

        if agent_id not in self.agents:

            self.agents[agent_id] = AgentState(agent_id, session)

        return self.agents[agent_id]

    def get_tree(self) -> dict:

        now = time.time()

        nodes = [a.to_tree_node() for a in self.agents.values()]

        active = sum(

            1 for a in self.agents.values()

            if not a.is_structural() and a.effective_status(now) == "running"

        )

        total_tokens = sum(a.tokens for a in self.agents.values())

        return {

            "ts": datetime.now(timezone.utc).isoformat(),

            "nodes": nodes,

            "active_count": active,

            "total_tokens": total_tokens,

        }

    def health_summary(self) -> dict:

        now = time.time()

        real = [a for a in self.agents.values() if not a.is_structural()]

        eff = {a.agent_id: a.effective_status(now) for a in real}

        active = [a for a in real if eff[a.agent_id] == "running"]

        done = [a for a in real if eff[a.agent_id] == "done"]

        errors = [a for a in real if eff[a.agent_id] == "error"]

        return {

            "active_agents": len(active),

            "done_agents": len(done),

            "error_agents": len(errors),

            "total_agents": len(self.agents),

            "total_tokens": sum(a.tokens for a in self.agents.values()),

            "ok": True,

        }

class ConnectionManager:

    def __init__(self):

        self._connections: list[WebSocket] = []

    async def connect(self, ws: WebSocket):

        await ws.accept()

        self._connections.append(ws)

    def disconnect(self, ws: WebSocket):

        self._connections.discard(ws) if hasattr(self._connections, 'discard') else None

        if ws in self._connections:

            self._connections.remove(ws)

    async def broadcast(self, event: dict):

        msg = json.dumps(event)

        dead: list[WebSocket] = []

        for ws in list(self._connections):

            try:

                await ws.send_text(msg)

            except Exception:

                dead.append(ws)

        for ws in dead:

            self.disconnect(ws)

class TranscriptParser:

    """

    Tails .output JSONL files in task dirs.

    Each line is a JSON object with keys: type, agentId, parentUuid, timestamp,

    message.{role, content, usage}, sessionId.

    Normalises to DeckEvent per contract §1.

    """

    def __init__(self, store: TreeStore, manager: ConnectionManager):

        self.store = store

        self.manager = manager

        self._offsets: dict[str, int] = {}

        self._seen_agents: set[str] = set()

        self._file_mtime: float = time.time()

    def _all_task_files(self) -> list[Path]:

        files = []

        for root in TASK_ROOTS:

            if not root.exists():

                continue

            for session_dir in root.iterdir():

                if not session_dir.is_dir():

                    continue

                tasks_dir = session_dir / "tasks"

                if tasks_dir.is_dir():

                    for f in tasks_dir.glob("*.output"):

                        files.append(f)

        return files

    async def poll(self):

        """Called every SCAN_INTERVAL_S. Reads new lines from all transcript files."""

        events_to_broadcast: list[dict] = []

        for fpath in self._all_task_files():

            offset = self._offsets.get(str(fpath), 0)

            try:

                st = fpath.stat()

                size = st.st_size

                fmtime = st.st_mtime

            except OSError:

                continue

            if size <= offset:

                continue

            self._file_mtime = fmtime

            new_events = []

            try:

                with fpath.open("rb") as f:

                    f.seek(offset)

                    chunk = f.read(size - offset)

                    new_offset = offset + len(chunk)

                for raw_line in chunk.split(b"\n"):

                    raw_line = raw_line.strip()

                    if not raw_line:

                        continue

                    try:

                        obj = json.loads(raw_line)

                        ev = self._parse_line(obj, fpath)

                        if ev:

                            new_events.append(ev)

                    except (json.JSONDecodeError, Exception):

                        continue

                self._offsets[str(fpath)] = new_offset

            except OSError:

                continue

            events_to_broadcast.extend(new_events)

        for ev in events_to_broadcast:

            await self.manager.broadcast(ev)

    def _parse_line(self, obj: dict, fpath: Path) -> dict | None:

        """Parse one JSONL line into a normalized Deck event dict."""

        line_type = obj.get("type", "")

        agent_id = obj.get("agentId", "")

        session_id = obj.get("sessionId", "") or fpath.parent.parent.name

        if not agent_id or line_type not in ("user", "assistant", "attachment"):

            return None

        agent = self.store.upsert(agent_id, session_id)

        agent.last_activity_ts = max(agent.last_activity_ts, getattr(self, "_file_mtime", time.time()))

        if agent_id not in self._seen_agents:

            self._seen_agents.add(agent_id)

            ts = obj.get("timestamp", datetime.now(timezone.utc).isoformat())

            if line_type == "user":

                content = obj.get("message", {}).get("content", "")

                if isinstance(content, str):

                    label = content[:80].strip().replace("\n", " ")

                elif isinstance(content, list) and content:

                    first = content[0]

                    if isinstance(first, dict):

                        label = str(first.get("text", ""))[:80].replace("\n", " ")

                    else:

                        label = str(first)[:80]

                else:

                    label = ""

                agent.label = label or agent_id[:12]

                agent.spawned_at = ts

            agent.parent_id = f"session-{session_id[:8]}"

            self._ensure_session_node(session_id)

            agent.last_activity = ts

            event = {

                "ts": ts,

                "agent_id": agent_id,

                "parent_id": agent.parent_id,

                "type": "spawned",

                "label": agent.label,

                "model": agent.model,

                "status": "running",

                "tokens": 0,

                "host": "ct215",

                "session": session_id,

            }

            return event

        if line_type == "assistant":

            ts = obj.get("timestamp", datetime.now(timezone.utc).isoformat())

            msg = obj.get("message", {})

            usage = msg.get("usage", {})

            total = (

                usage.get("input_tokens", 0)

                + usage.get("cache_read_input_tokens", 0)

                + usage.get("cache_creation_input_tokens", 0)

                + usage.get("output_tokens", 0)

            )

            agent.tokens = max(agent.tokens, total)

            agent.last_activity = ts

            content = msg.get("content", [])

            stop_reason = msg.get("stop_reason", "")

            if isinstance(content, list):

                for c in content:

                    if not isinstance(c, dict):

                        continue

                    if c.get("type") == "tool_use":

                        tool_name = c.get("name", "")

                        agent.current_tool = tool_name

                        agent.task_count += 1

                        if tool_name == "Agent":

                            inp = c.get("input", {})

                            child_desc = inp.get("description", inp.get("prompt", "")[:60])

                            event = {

                                "ts": ts,

                                "agent_id": agent_id,

                                "parent_id": agent.parent_id,

                                "type": "child_spawned",

                                "label": agent.label,

                                "model": agent.model,

                                "status": "running",

                                "tool": f"Agent:{child_desc[:40]}",

                                "tokens": agent.tokens,

                                "host": "ct215",

                                "session": session_id,

                            }

                            return event

                        event = {

                            "ts": ts,

                            "agent_id": agent_id,

                            "parent_id": agent.parent_id,

                            "type": "tool_call",

                            "label": agent.label,

                            "model": agent.model,

                            "status": "running",

                            "tool": tool_name,

                            "tokens": agent.tokens,

                            "host": "ct215",

                            "session": session_id,

                        }

                        return event

            if stop_reason == "end_turn":

                agent.status = "done"

                agent.current_tool = None

                event = {

                    "ts": ts,

                    "agent_id": agent_id,

                    "parent_id": agent.parent_id,

                    "type": "returned",

                    "label": agent.label,

                    "model": agent.model,

                    "status": "done",

                    "tokens": agent.tokens,

                    "host": "ct215",

                    "session": session_id,

                }

                return event

            if agent.tokens > 0:

                event = {

                    "ts": ts,

                    "agent_id": agent_id,

                    "parent_id": agent.parent_id,

                    "type": "token_tick",

                    "label": agent.label,

                    "model": agent.model,

                    "status": "running",

                    "tokens": agent.tokens,

                    "host": "ct215",

                    "session": session_id,

                }

                return event

        return None

    def _ensure_session_node(self, session_id: str):

        """Create a session-level node that groups agents for this session."""

        node_id = f"session-{session_id[:8]}"

        if node_id not in self.store.agents:

            session_node = AgentState(node_id, session_id)

            session_node.label = f"session:{session_id[:8]}"

            session_node.model = "orchestrator"

            session_node.status = "running"

            session_node.parent_id = "ct215-root"

            self.store.agents[node_id] = session_node

            self.store.agents["ct215-root"].task_count += 1

def ensure_host_node(store: "TreeStore", host: str) -> str:

    """Create (once) a host-level grouping node under the root for non-ct215 hosts.

    Returns the host node id so cross-host agents can nest under it when they

    don't carry an explicit parent_id.

    """

    if not host or host == "ct215":

        return "ct215-root"

    node_id = f"host-{host}"

    if node_id not in store.agents:

        hn = AgentState(node_id, "system")

        hn.label = host.upper()

        hn.model = "host"

        hn.status = "running"

        hn.host = host

        hn.parent_id = "ct215-root"

        store.agents[node_id] = hn

        store.agents["ct215-root"].task_count += 1

    return node_id

app = FastAPI(title="darwin-deck", version="1.0.0")

app.add_middleware(

    CORSMiddleware,

    allow_origins=["*"],

    allow_methods=["*"],

    allow_headers=["*"],

)

store = TreeStore()

ws_manager = ConnectionManager()

parser = TranscriptParser(store, ws_manager)

@app.on_event("startup")

async def startup():

    asyncio.create_task(poll_loop())

async def poll_loop():

    while True:

        try:

            await parser.poll()

        except Exception as e:

            print(f"[deck] poll error: {e}")

        await asyncio.sleep(SCAN_INTERVAL_S)

@app.get("/health")

async def health():

    return store.health_summary()

@app.get("/tree")

async def get_tree():

    return store.get_tree()

@app.post("/event")

async def ingest_event(request: Request):

    """Generic ingest - any agent or process can POST a contract-schema event."""

    try:

        body = await request.json()

    except Exception:

        return JSONResponse({"error": "invalid json"}, status_code=400)

    agent_id = body.get("agent_id", "")

    if not agent_id:

        return JSONResponse({"error": "agent_id required"}, status_code=400)

    session = body.get("session", "external")

    agent = store.upsert(agent_id, session)

    if body.get("label"):

        agent.label = body["label"]

    if body.get("model"):

        agent.model = body["model"]

    if body.get("status"):

        agent.status = body["status"]

    if body.get("tokens"):

        agent.tokens = max(agent.tokens, int(body["tokens"]))

    host = body.get("host") or agent.host

    agent.host = host

    if body.get("tool"):

        agent.current_tool = body["tool"]

    if body.get("type") in ("tool_call", "child_spawned"):

        agent.task_count += 1

    parent_id = body.get("parent_id")

    if parent_id:

        if parent_id not in store.agents:

            stub = store.upsert(parent_id, session)

            stub.host = host

            stub.parent_id = ensure_host_node(store, host)

        agent.parent_id = parent_id

    elif not agent.parent_id:

        agent.parent_id = ensure_host_node(store, host)

    ev_ts = body.get("ts", datetime.now(timezone.utc).isoformat())

    agent.last_activity = ev_ts

    ev_epoch = _parse_ts(ev_ts)

    if ev_epoch <= 0:

        ev_epoch = time.time()

    agent.last_activity_ts = max(agent.last_activity_ts, ev_epoch)

    await ws_manager.broadcast(body)

    return {"ok": True, "agent_id": agent_id}

@app.websocket("/stream")

async def websocket_stream(ws: WebSocket):

    await ws_manager.connect(ws)

    try:

        snapshot = store.get_tree()

        await ws.send_text(json.dumps({"type": "snapshot", **snapshot}))

        while True:

            try:

                await asyncio.wait_for(ws.receive_text(), timeout=30)

            except asyncio.TimeoutError:

                await ws.send_text(json.dumps({"type": "ping", "ts": datetime.now(timezone.utc).isoformat()}))

    except WebSocketDisconnect:

        pass

    except Exception:

        pass

    finally:

        ws_manager.disconnect(ws)

if __name__ == "__main__":

    import uvicorn

    uvicorn.run("deck:app", host="0.0.0.0", port=9893, log_level="info")