a291f3cb6e
Conversations that never edit a `projects/<slug>` file — service work, skills, scripts, docs, plain questions — used to fall out of every project-shaped view as "(unattributed)". They now land on a root project card backed by the repo root itself. - projects.py: ROOT_SLUG (`lab`), attributed_projects() (derived at read time, never written to the sidecar), project_dir()/repo_rel() so callers resolve the slug to REPO_DIR instead of assuming projects/<dir>, plus the gallery record and detail view (repo CLAUDE.md, GOAL.md, git history). A real ~/projects/lab dir would win over the synthesised record. - main.py: _conv_meta applies the fallback (so tags, project pages, graph and dashboards all see it), and the goals board + goal-work resolve the unit's dir through project_dir()/repo_rel(). - project_costs.py: same fallback, so the lab's spend rolls up on its card. - gitdiff.py: `project:lab` resolves to the superproject. - count_loc: the lab's LoC comes from `git ls-files` (tracked source only) — a directory walk would count services/media and the untracked data/ tree. - Frontend: house icon, dashed tinted frame and a `root` badge on the card and detail page; no .env editor; the tag's repo path reads `.`.
3874 lines
165 KiB
Python
3874 lines
165 KiB
Python
"""
|
||
ai-agent backend — a viewer/editor + analytics dashboard for the homelab's
|
||
Claude context.
|
||
|
||
Exposes the repo's CLAUDE.md files and the `.claude/` tree (skills, hooks,
|
||
settings) as a flat list of files with **real** Claude token counts and dollar
|
||
costs, lets them be edited, and surfaces skill-usage analytics mined from the
|
||
Claude Code transcripts.
|
||
|
||
Token counts and skill stats are maintained in a small SQLite store by a
|
||
background indexer (see indexer.py): token counts come from the count_tokens
|
||
endpoint and are only refreshed for files whose content changed, after a debounce.
|
||
The built React PWA is served from STATIC_DIR at the web root.
|
||
"""
|
||
|
||
import copy
|
||
import datetime
|
||
import io
|
||
import json
|
||
import os
|
||
import pathlib
|
||
import queue
|
||
import re
|
||
import shutil
|
||
import socket
|
||
import threading
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
import uuid as uuidlib
|
||
import zipfile
|
||
|
||
from fastapi import FastAPI, File, HTTPException, Request, UploadFile
|
||
from fastapi.responses import (FileResponse, JSONResponse, PlainTextResponse,
|
||
Response, StreamingResponse)
|
||
from pydantic import BaseModel
|
||
|
||
import activity as activity_mod
|
||
import agents as agents_mod
|
||
import artefacts as artefacts_mod
|
||
import conversations
|
||
import editsdiff
|
||
import envfile as envfile_mod
|
||
import gitdiff
|
||
import goal as goalmd
|
||
import memories as memories_mod
|
||
import models as models_mod
|
||
import openrouter as openrouter_mod
|
||
import cron as cron_mod
|
||
import forms as forms_mod
|
||
import notify as notify_mod
|
||
import notify_audio as notify_audio_mod
|
||
import notif_read as notif_read_mod
|
||
import plans as plans_mod
|
||
import project_costs as project_costs_mod
|
||
import projects as projects_mod
|
||
import scaffold as scaffold_mod
|
||
import schemas
|
||
import skills as skills_mod
|
||
import svc as svc_mod
|
||
import templates as templates_mod
|
||
import worktreediff
|
||
from db import Store
|
||
from deploy_status import DeployWatcher, read_status as read_deploy_status
|
||
from events import Hub, Watcher
|
||
from indexer import (INPUT_PRICE_PER_MTOK, MODEL, Indexer, _estimate_tokens,
|
||
_price)
|
||
from meta import MetaStore
|
||
from ui_state import UiStateStore, valid_key
|
||
|
||
WORKSPACE = pathlib.Path(os.environ.get("WORKSPACE", "/workspace")).resolve()
|
||
STATIC_DIR = pathlib.Path(os.environ.get("STATIC_DIR", "/app/static")).resolve()
|
||
# Screenshots captured by the `screenshot` skill (docs/screenshots/, mounted
|
||
# read-only). Served by /api/screenshot and shown inline in the thread viewer.
|
||
SCREENSHOTS_DIR = pathlib.Path(
|
||
os.environ.get("SCREENSHOTS_DIR", "/screenshots")).resolve()
|
||
SCREENSHOT_EXTS = {".jpeg", ".jpg", ".png", ".webp", ".gif"}
|
||
# The repo's root `data/` tree (mounted read-only at /workspace/data). Generated
|
||
# artifacts (images, renders, charts) land here and are served by /api/data-asset
|
||
# so a `Read` of an image under data/ can be previewed inline in the thread.
|
||
DATA_DIR = (WORKSPACE / "data")
|
||
# Claude Code's per-session scratchpad root (mounted read-only). A session's
|
||
# temp dir is <root>/<encoded-cwd>/<sessionId>/scratchpad/…, which is where a
|
||
# run's throwaway renders and screenshots land when they aren't worth committing
|
||
# to a project's .ai/artefacts/. Served by /api/scratchpad-asset so those still
|
||
# preview inline in the thread instead of showing "not previewable".
|
||
SCRATCHPAD_DIR = pathlib.Path(
|
||
os.environ.get("SCRATCHPAD_DIR", "/scratchpads")).resolve()
|
||
IMAGE_EXTS = {".jpeg", ".jpg", ".png", ".webp", ".gif", ".svg", ".bmp", ".avif"}
|
||
# Files uploaded from the composer (SpawnBox/ResumeBox). Stored inside the
|
||
# service's own writable data dir; served back by /api/upload-file and read by
|
||
# the spawned `claude -p` session. UPLOADS_DIR is the container path; the repo
|
||
# prefix is what the host-side session (cwd = repo root) sees, so the composer
|
||
# injects "<prefix>/<batch>/<name>" into the prompt for Claude to Read.
|
||
UPLOADS_DIR = pathlib.Path(
|
||
os.environ.get("UPLOADS_DIR", "/data/uploads")).resolve()
|
||
UPLOADS_REPO_PREFIX = os.environ.get(
|
||
"UPLOADS_REPO_PREFIX", "services/ai-agent/data/uploads").strip("/")
|
||
MAX_UPLOAD_BYTES = int(os.environ.get("MAX_UPLOAD_BYTES", 50 * 1024 * 1024))
|
||
# Custom avatar sets imported as a zip (sprite editor's "Export set"). Unpacked
|
||
# into the writable data volume and served at /avatars/custom/ so the roaming
|
||
# avatar can play a full custom emote set (not just a single strip URL). One
|
||
# active set at a time.
|
||
AVATARS_DIR = pathlib.Path(
|
||
os.environ.get("AVATARS_DIR", "/data/avatars")).resolve()
|
||
CUSTOM_AVATAR_DIR = AVATARS_DIR / "custom"
|
||
MAX_AVATAR_ZIP_BYTES = int(os.environ.get("MAX_AVATAR_ZIP_BYTES", 40 * 1024 * 1024))
|
||
_UNSAFE_NAME_RE = re.compile(r"[^A-Za-z0-9._-]+")
|
||
DB_PATH = os.environ.get("DB_PATH", "/data/ai-agent.db")
|
||
_transcripts = os.environ.get("TRANSCRIPTS_DIR", "/transcripts")
|
||
# Live (read-only) Claude Code transcripts that get imported into the archive.
|
||
SOURCE_DIR = pathlib.Path(_transcripts).resolve() if _transcripts else None
|
||
# Second live source: transcripts mirrored by the pi-harness runner
|
||
# (services/ai-agent/runner writes Claude-Code-schema JSONL there).
|
||
_pi_transcripts = os.environ.get("PI_TRANSCRIPTS_DIR", "/pi-transcripts")
|
||
PI_SOURCE_DIR = (pathlib.Path(_pi_transcripts).resolve()
|
||
if _pi_transcripts else None)
|
||
# Third live source: the in-container runner (RUNNER_IN_CONTAINER=1, see
|
||
# docker-entrypoint.sh). The Claude Code CLI baked into the image writes its
|
||
# transcripts under the runner's own $HOME, inside the container — watching that
|
||
# dir is what makes an in-container session stream into the viewer like any other.
|
||
# Unset on the homelab, where the runner is the host sidecar and its transcripts
|
||
# arrive through the read-only /transcripts mount instead.
|
||
_runner_transcripts = os.environ.get("RUNNER_TRANSCRIPTS_DIR", "")
|
||
RUNNER_SOURCE_DIR = (pathlib.Path(_runner_transcripts).resolve()
|
||
if _runner_transcripts else None)
|
||
# dict.fromkeys: order-preserving dedupe — pointing two of these at the same dir
|
||
# would otherwise import every transcript in it twice.
|
||
SOURCE_DIRS = list(dict.fromkeys(
|
||
d for d in (SOURCE_DIR, PI_SOURCE_DIR, RUNNER_SOURCE_DIR) if d))
|
||
# Persistent archive: imported transcripts live here and are indexed/read from
|
||
# it (so conversations Claude later prunes from the source survive).
|
||
ARCHIVE_DIR = pathlib.Path(
|
||
os.environ.get("ARCHIVE_DIR", "/data/transcripts")).resolve()
|
||
TRANSCRIPTS_DIR = ARCHIVE_DIR # the detail endpoint reads from the archive
|
||
# Sidecar of per-conversation action metadata edited by the skill scripts.
|
||
META_PATH = os.environ.get("META_PATH", "/data/conversations-meta.json")
|
||
# Cron jobs: scheduled agent sessions spawned from prompt files (see cron.py).
|
||
CRON_PATH = os.environ.get("CRON_PATH", "/data/cron-jobs.json")
|
||
# First-class notifications: webhook config + notification/ask log (notify.py).
|
||
NOTIFY_PATH = os.environ.get("NOTIFY_PATH", "/data/notifications.json")
|
||
FORMS_PATH = os.environ.get("FORMS_PATH", "/data/forms.json")
|
||
# Server-side store for the PWA's small client state (Settings config + the
|
||
# notification feed's seen/new bookkeeping) — moved off browser localStorage so
|
||
# it follows the user across browsers/devices. See ui_state.py.
|
||
UI_STATE_PATH = os.environ.get("UI_STATE_PATH", "/data/ui-state.json")
|
||
# A conversation whose last activity is within this window counts as "running".
|
||
RUNNING_WINDOW_SECS = float(os.environ.get("RUNNING_WINDOW_SECS", "600"))
|
||
# Host-side sidecar that actually launches `claude -p` (see services/ai-agent/
|
||
# sidecar/). The container reaches it over host.docker.internal.
|
||
SIDECAR_URL = os.environ.get("SIDECAR_URL", "http://host.docker.internal:8790")
|
||
SIDECAR_TOKEN = os.environ.get("SIDECAR_TOKEN", "")
|
||
|
||
# Directories never worth scanning (huge / generated / vendored).
|
||
IGNORE_DIRS = {
|
||
".git", "node_modules", ".forge", "dist", "build", "__pycache__",
|
||
".venv", "venv", ".cache", "history", "tmp", ".next", "coverage",
|
||
}
|
||
MD_EXTS = {".md", ".markdown"}
|
||
# Text-ish files worth surfacing from the repo's root `data/` tree (plans, logs,
|
||
# notes, kanban, widget templates). Binary/image files (screenshots, thumbnails)
|
||
# are skipped — the viewer is a text editor and can't render them.
|
||
DATA_TEXT_EXTS = {
|
||
".md", ".markdown", ".txt", ".json", ".yml", ".yaml", ".toml", ".ini",
|
||
".cfg", ".py", ".sh", ".js", ".ts", ".jinja", ".j2", ".log", ".csv",
|
||
}
|
||
|
||
app = FastAPI(title="ai-agent", docs_url=None, redoc_url=None)
|
||
|
||
|
||
# ── trusted-caller gate ──────────────────────────────────────────────────────
|
||
# The backend has no auth of its own: the browser reaches it through Traefik
|
||
# (behind Authelia) and host tooling through the 127.0.0.1-published port. But
|
||
# the container also sits on the shared `main` docker network, where any other
|
||
# container could hit it directly — and a PUT into `.claude/` (hooks) or a
|
||
# /api/spawn is host-level code execution. So every /api request must come from
|
||
# a trusted peer: Traefik (the Authelia-guarded edge), the docker gateway (how
|
||
# host-originated connections to the published port appear), localhost, or a
|
||
# caller presenting the shared INTERNAL_API_TOKEN. Only /api/health stays open
|
||
# (deploy probes). Static assets are public-shell only, so they stay open too.
|
||
INTERNAL_API_TOKEN = os.environ.get("INTERNAL_API_TOKEN", "")
|
||
# A read-ONLY API key: a caller presenting it (X-API-Key) may hit **safe**
|
||
# (GET/HEAD/OPTIONS) /api/* routes only — every mutating method is refused. This
|
||
# is what the desk phone holds to fetch unread notifications: it must never be
|
||
# able to spawn a session or PUT into .claude/ (that's host-level code exec).
|
||
# Comma-separated to allow more than one key. Empty ⇒ feature off.
|
||
READONLY_API_KEYS = frozenset(
|
||
k.strip() for k in os.environ.get("READONLY_API_KEY", "").split(",")
|
||
if k.strip())
|
||
# Methods a read-only key is allowed to use.
|
||
_SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
|
||
TRAEFIK_HOST = os.environ.get("TRAEFIK_HOST", "traefik")
|
||
_TRUSTED_TTL_S = 30.0
|
||
_trusted_lock = threading.Lock()
|
||
_trusted_cache: tuple[float, frozenset[str]] = (0.0, frozenset())
|
||
|
||
|
||
def _default_gateway_ips() -> set[str]:
|
||
"""The container's default-gateway IP(s) (/proc/net/route, IPv4)."""
|
||
ips: set[str] = set()
|
||
try:
|
||
for line in pathlib.Path("/proc/net/route").read_text().splitlines()[1:]:
|
||
f = line.split()
|
||
if len(f) >= 3 and f[1] == "00000000": # default route
|
||
ips.add(socket.inet_ntoa(bytes.fromhex(f[2])[::-1]))
|
||
except (OSError, ValueError):
|
||
pass
|
||
return ips
|
||
|
||
|
||
def _trusted_ips() -> frozenset[str]:
|
||
"""Gateway + Traefik IPs, re-resolved at most every _TRUSTED_TTL_S."""
|
||
global _trusted_cache
|
||
now = time.monotonic()
|
||
with _trusted_lock:
|
||
ts, ips = _trusted_cache
|
||
if now - ts < _TRUSTED_TTL_S and ips:
|
||
return ips
|
||
fresh = {"127.0.0.1", "::1"} | _default_gateway_ips()
|
||
try:
|
||
for info in socket.getaddrinfo(TRAEFIK_HOST, None):
|
||
fresh.add(info[4][0])
|
||
except OSError:
|
||
pass
|
||
out = frozenset(fresh)
|
||
with _trusted_lock:
|
||
_trusted_cache = (now, out)
|
||
return out
|
||
|
||
|
||
@app.middleware("http")
|
||
async def _trusted_caller_gate(request: Request, call_next):
|
||
path = request.url.path
|
||
if path.startswith("/api/") and path != "/api/health":
|
||
# A read-only API key downgrades the caller to safe methods, whatever the
|
||
# source IP — the phone reaches us over a "trusted" host loopback but is
|
||
# only allowed to read (fetch notifications), never to spawn/mutate. So
|
||
# this check comes FIRST and, when the key matches, it decides the request
|
||
# outright (a read-only key on a POST is refused, not silently upgraded by
|
||
# a trusted IP).
|
||
api_key = request.headers.get("x-api-key", "")
|
||
if api_key and api_key in READONLY_API_KEYS:
|
||
if request.method not in _SAFE_METHODS:
|
||
return JSONResponse(
|
||
{"detail": "read-only api key: method not allowed"},
|
||
status_code=403)
|
||
return await call_next(request)
|
||
|
||
client = request.client.host if request.client else ""
|
||
if client not in _trusted_ips():
|
||
tok = request.headers.get("x-internal-auth", "")
|
||
if not (INTERNAL_API_TOKEN and tok == INTERNAL_API_TOKEN):
|
||
return JSONResponse({"detail": "forbidden"}, status_code=403)
|
||
return await call_next(request)
|
||
|
||
|
||
def _r(model) -> dict:
|
||
"""Attach a response schema to a route for OpenAPI/codegen **only**.
|
||
|
||
Passed as ``responses=`` (not ``response_model=``) so FastAPI documents the
|
||
200 body in ``/openapi.json`` — which Orval turns into the frontend's types —
|
||
without validating or filtering the handler's actual return value. See
|
||
``schemas.py``."""
|
||
return {200: {"model": model}}
|
||
|
||
store = Store(DB_PATH)
|
||
meta_store = MetaStore(META_PATH)
|
||
ui_state = UiStateStore(UI_STATE_PATH)
|
||
cron_store = cron_mod.CronStore(CRON_PATH)
|
||
notify_store = notify_mod.NotifyStore(NOTIFY_PATH)
|
||
form_store = forms_mod.FormStore(FORMS_PATH)
|
||
# Server-side "read" ledger for the notification feed — lets the phone fetch
|
||
# *unread* notifications (listening ≠ reading; only the UI marks them read).
|
||
NOTIF_READ_PATH = os.environ.get("NOTIF_READ_PATH", "/data/notif-read.json")
|
||
notif_read_store = notif_read_mod.NotifReadStore(NOTIF_READ_PATH)
|
||
hub = Hub()
|
||
|
||
|
||
# ── file discovery ──────────────────────────────────────────────────────────
|
||
def _is_data_file(rel: pathlib.PurePosixPath) -> bool:
|
||
"""A text file under the repo's root `data/` tree (plans/logs/notes/etc.).
|
||
|
||
Surfaced read-only for visibility; binaries and the ignored `data/tmp/`
|
||
(scratch + secrets) are filtered out via DATA_TEXT_EXTS + IGNORE_DIRS."""
|
||
return (
|
||
len(rel.parts) > 1
|
||
and rel.parts[0] == "data"
|
||
and rel.suffix.lower() in DATA_TEXT_EXTS
|
||
)
|
||
|
||
|
||
def _is_context_file(rel: pathlib.PurePosixPath) -> bool:
|
||
"""A file we expose: any CLAUDE.md, anything under a `.claude/` tree, or a
|
||
text file under the repo's root `data/` tree."""
|
||
if rel.name == "CLAUDE.md":
|
||
return True
|
||
if ".claude" in rel.parts:
|
||
return True
|
||
return _is_data_file(rel)
|
||
|
||
|
||
def _is_editable(rel: pathlib.PurePosixPath) -> bool:
|
||
"""Text files we allow editing. Binary/lock files stay read-only, and the
|
||
`data/` tree is read-only (mounted :ro — surfaced for viewing only)."""
|
||
if _is_data_file(rel):
|
||
return False
|
||
return rel.suffix.lower() in {
|
||
".md", ".markdown", ".sh", ".py", ".js", ".mjs", ".ts", ".tsx",
|
||
".json", ".yml", ".yaml", ".txt", ".toml", ".cfg", ".ini", "",
|
||
}
|
||
|
||
|
||
def _kind(rel: pathlib.PurePosixPath) -> str:
|
||
name = rel.name
|
||
if name == "CLAUDE.md":
|
||
return "claude-md"
|
||
if name == "SKILL.md":
|
||
return "skill-md"
|
||
if rel.suffix in MD_EXTS:
|
||
return "markdown"
|
||
if rel.suffix in {".sh", ".py", ".js", ".mjs", ".ts", ".tsx"}:
|
||
return "script"
|
||
if rel.suffix in {".json", ".yml", ".yaml", ".toml", ".ini", ".cfg"}:
|
||
return "config"
|
||
return "other"
|
||
|
||
|
||
def _iter_files():
|
||
"""Yield (abs_path, rel_posix) for every exposed context file."""
|
||
for dirpath, dirnames, filenames in os.walk(WORKSPACE):
|
||
dirnames[:] = [d for d in dirnames if d not in IGNORE_DIRS]
|
||
for fn in filenames:
|
||
ap = pathlib.Path(dirpath) / fn
|
||
rel = pathlib.PurePosixPath(ap.relative_to(WORKSPACE).as_posix())
|
||
if _is_context_file(rel):
|
||
yield ap, rel
|
||
|
||
|
||
def _read_text(ap: pathlib.Path) -> str:
|
||
try:
|
||
return ap.read_text(encoding="utf-8")
|
||
except (UnicodeDecodeError, OSError):
|
||
return ""
|
||
|
||
|
||
indexer = Indexer(store, WORKSPACE, SOURCE_DIRS, ARCHIVE_DIR,
|
||
_iter_files, _read_text)
|
||
watcher = Watcher(SOURCE_DIRS, meta_store, indexer, hub)
|
||
# Surface the blue-green deploy script's progress snapshot as a live SSE banner.
|
||
deploy_watcher = DeployWatcher(hub)
|
||
|
||
|
||
# ── entry assembly ──────────────────────────────────────────────────────────
|
||
def _tokens_for(path: str, text: str) -> tuple[int, float, bool]:
|
||
"""Pull the stored real token count/cost; fall back to a cheap estimate."""
|
||
row = store.get_file(path)
|
||
if row and row["tokens"] is not None:
|
||
return row["tokens"], row["cost"] or 0.0, bool(row["estimated"])
|
||
est = _estimate_tokens(text)
|
||
return est, _price(est), True
|
||
|
||
|
||
def _file_entry(ap: pathlib.Path, rel: pathlib.PurePosixPath, with_content: bool):
|
||
content = _read_text(ap)
|
||
is_md = rel.suffix.lower() in MD_EXTS
|
||
words = len(content.split())
|
||
chars = len(content)
|
||
tokens, cost, estimated = _tokens_for(str(rel), content)
|
||
try:
|
||
bytes_ = ap.stat().st_size
|
||
except OSError:
|
||
bytes_ = chars
|
||
entry = {
|
||
"path": str(rel),
|
||
"name": rel.name,
|
||
"ext": rel.suffix,
|
||
"kind": _kind(rel),
|
||
"isMarkdown": is_md,
|
||
"bytes": bytes_,
|
||
"words": words,
|
||
"chars": chars,
|
||
"tokens": tokens,
|
||
"cost": cost,
|
||
"estimated": estimated,
|
||
"editable": _is_editable(rel),
|
||
}
|
||
if with_content:
|
||
entry["content"] = content
|
||
return entry
|
||
|
||
|
||
# ── path safety ─────────────────────────────────────────────────────────────
|
||
def _resolve(rel_path: str) -> tuple[pathlib.Path, pathlib.PurePosixPath]:
|
||
rel = pathlib.PurePosixPath(rel_path)
|
||
if rel.is_absolute() or ".." in rel.parts:
|
||
raise HTTPException(400, "invalid path")
|
||
ap = (WORKSPACE / rel).resolve()
|
||
try:
|
||
ap.relative_to(WORKSPACE)
|
||
except ValueError:
|
||
raise HTTPException(400, "path escapes workspace")
|
||
if not _is_context_file(rel):
|
||
raise HTTPException(403, "not an exposed context file")
|
||
return ap, rel
|
||
|
||
|
||
# ── API ─────────────────────────────────────────────────────────────────────
|
||
@app.get("/api/bundle", responses=_r(schemas.Bundle))
|
||
def bundle():
|
||
"""The whole file catalog — metadata only, served from the index DB.
|
||
|
||
This used to read every context file's content per request (a 16 MB,
|
||
12-second response the app shell fetched on every load). File *content* now
|
||
loads lazily through ``GET /api/file`` when a file is opened; the words /
|
||
chars / token stats come from the ``files`` table, which the indexer keeps
|
||
fresh (recomputed only when a file's content actually changes)."""
|
||
files = []
|
||
for path, r in store.all_files().items():
|
||
rel = pathlib.PurePosixPath(path)
|
||
tokens = r["tokens"]
|
||
estimated = bool(r["estimated"])
|
||
if tokens is None:
|
||
# not counted yet — estimate off the stored char count (~chars/4)
|
||
tokens = ((r["chars"] or 0) + 3) // 4
|
||
estimated = True
|
||
files.append({
|
||
"path": path,
|
||
"name": rel.name,
|
||
"ext": rel.suffix,
|
||
"kind": _kind(rel),
|
||
"isMarkdown": bool(r["is_markdown"]),
|
||
"bytes": r["size"] or 0,
|
||
"words": r["words"] or 0,
|
||
"chars": r["chars"] or 0,
|
||
"tokens": tokens,
|
||
"cost": r["cost"] or _price(tokens),
|
||
"estimated": estimated,
|
||
"editable": _is_editable(rel),
|
||
})
|
||
files.sort(key=lambda f: f["path"])
|
||
totals = {
|
||
"files": len(files),
|
||
"mdFiles": sum(1 for f in files if f["isMarkdown"]),
|
||
"words": sum(f["words"] for f in files),
|
||
"chars": sum(f["chars"] for f in files),
|
||
"tokens": sum(f["tokens"] for f in files),
|
||
"cost": sum(f["cost"] for f in files),
|
||
"estimated": sum(1 for f in files if f["estimated"]),
|
||
}
|
||
return {"workspace": str(WORKSPACE.name), "totals": totals, "files": files,
|
||
"pricing": {"model": MODEL, "inputPerMTok": INPUT_PRICE_PER_MTOK}}
|
||
|
||
|
||
@app.get("/api/file", responses=_r(schemas.FileEntry))
|
||
def get_file(path: str):
|
||
ap, rel = _resolve(path)
|
||
if not ap.is_file():
|
||
raise HTTPException(404, "not found")
|
||
return _file_entry(ap, rel, with_content=True)
|
||
|
||
|
||
class SaveBody(BaseModel):
|
||
path: str
|
||
content: str
|
||
|
||
|
||
@app.put("/api/file", responses=_r(schemas.FileEntry))
|
||
def put_file(body: SaveBody):
|
||
ap, rel = _resolve(body.path)
|
||
if not _is_editable(rel):
|
||
raise HTTPException(403, "not editable")
|
||
if not ap.exists():
|
||
raise HTTPException(404, "not found")
|
||
ap.write_text(body.content, encoding="utf-8")
|
||
# Refresh on-disk metadata immediately (force the discovery walk so a
|
||
# brand-new file lands in the catalog at once); the real token count
|
||
# refreshes after the debounce.
|
||
indexer.scan_files(force_walk=True)
|
||
return _file_entry(ap, rel, with_content=True)
|
||
|
||
|
||
# ── UI state (Settings config + notification bookkeeping) ────────────────────
|
||
# The PWA persists its small Zustand `persist` stores here instead of the
|
||
# browser's localStorage, so config and the notification feed's seen/new state
|
||
# follow the user across browsers/devices. The value is the store's opaque
|
||
# serialized JSON blob; the backend just holds the string (see ui_state.py).
|
||
|
||
class UiStateBody(BaseModel):
|
||
value: str
|
||
|
||
|
||
class NotifSeenBody(BaseModel):
|
||
# Notification ids to mark read/seed; `all` marks/seeds the whole feed.
|
||
ids: list[str] | None = None
|
||
all: bool = False
|
||
|
||
|
||
@app.get("/api/ui-state/{key}", responses=_r(schemas.UiStateValue))
|
||
def get_ui_state(key: str):
|
||
if not valid_key(key):
|
||
raise HTTPException(400, "bad key")
|
||
return {"value": ui_state.get(key)}
|
||
|
||
|
||
@app.put("/api/ui-state/{key}", responses=_r(schemas.OkResponse))
|
||
def put_ui_state(key: str, body: UiStateBody):
|
||
if not valid_key(key):
|
||
raise HTTPException(400, "bad key")
|
||
ui_state.set(key, body.value)
|
||
return {"ok": True}
|
||
|
||
|
||
@app.delete("/api/ui-state/{key}", responses=_r(schemas.OkResponse))
|
||
def delete_ui_state(key: str):
|
||
if not valid_key(key):
|
||
raise HTTPException(400, "bad key")
|
||
ui_state.delete(key)
|
||
return {"ok": True}
|
||
|
||
|
||
def _context_by_skill(files: dict) -> dict[str, dict]:
|
||
"""Group the indexed files' tokens/cost by the skill that owns them.
|
||
|
||
A skill's *context cost* is what loading its whole dir
|
||
(``.claude/skills/<name>/…``) into a session costs once. Shared by the
|
||
dashboard rollup and the skills catalog."""
|
||
by_skill: dict[str, dict] = {}
|
||
for path, r in files.items():
|
||
parts = path.split("/")
|
||
if ".claude" in parts:
|
||
i = parts.index(".claude")
|
||
if len(parts) > i + 2 and parts[i + 1] == "skills":
|
||
name = parts[i + 2]
|
||
e = by_skill.setdefault(name, {"tokens": 0, "cost": 0.0})
|
||
e["tokens"] += r["tokens"] or 0
|
||
e["cost"] += r["cost"] or 0.0
|
||
return by_skill
|
||
|
||
|
||
@app.get("/api/dashboard", responses=_r(schemas.Dashboard))
|
||
def dashboard():
|
||
"""Skill-usage analytics + cost rollups for the homepage."""
|
||
files = store.all_files()
|
||
|
||
total_tokens = sum((r["tokens"] or 0) for r in files.values())
|
||
total_cost = sum((r["cost"] or 0.0) for r in files.values())
|
||
estimated = sum(1 for r in files.values() if r["estimated"])
|
||
|
||
by_skill = _context_by_skill(files)
|
||
|
||
skills = []
|
||
for row in store.all_skills():
|
||
ctx = by_skill.get(row["name"])
|
||
ctx_tokens = ctx["tokens"] if ctx else None
|
||
ctx_cost = ctx["cost"] if ctx else None
|
||
skills.append({
|
||
"name": row["name"],
|
||
"count": row["count"],
|
||
"lastUsed": row["last_used"],
|
||
"contextTokens": ctx_tokens,
|
||
"contextCost": ctx_cost,
|
||
# estimated total spend = cost of loading the skill once × times used
|
||
"estSpend": (ctx_cost * row["count"]) if ctx_cost is not None else None,
|
||
})
|
||
|
||
total_invocations = sum(s["count"] for s in skills)
|
||
skill_spend = sum(s["estSpend"] for s in skills if s["estSpend"] is not None)
|
||
|
||
top_files = sorted(
|
||
({"path": p, "tokens": r["tokens"] or 0, "cost": r["cost"] or 0.0,
|
||
"estimated": bool(r["estimated"])}
|
||
for p, r in files.items()),
|
||
key=lambda f: f["cost"], reverse=True,
|
||
)[:12]
|
||
|
||
return {
|
||
"pricing": {"model": MODEL, "inputPerMTok": INPUT_PRICE_PER_MTOK},
|
||
"totals": {
|
||
"files": len(files),
|
||
"tokens": total_tokens,
|
||
"cost": total_cost,
|
||
"estimated": estimated,
|
||
"skills": len(skills),
|
||
"invocations": total_invocations,
|
||
"skillSpend": skill_spend,
|
||
},
|
||
"skills": skills,
|
||
"topFiles": top_files,
|
||
}
|
||
|
||
|
||
# ── activity dashboard ──────────────────────────────────────────────────────
|
||
# The rollup walks every summary's pre-computed `activeBins`, so it is cheap —
|
||
# but the whole archive's worth of it is still a few thousand cells, and the
|
||
# page refetches on every SSE list ping. Memoized on (summaries version,
|
||
# timezone), same pattern as _skill_usage.
|
||
_activity_cache: tuple[tuple[int, int], dict] | None = None
|
||
_activity_lock = threading.Lock()
|
||
|
||
|
||
@app.get("/api/activity", responses=_r(schemas.ActivityResponse))
|
||
def activity_rollup(tzOffset: int = 0):
|
||
"""When the agents ran: per day, per 5-minute bin per model, per model.
|
||
|
||
``tzOffset`` is the caller's minutes east of UTC (the browser's
|
||
``-new Date().getTimezoneOffset()``); days and histogram slots are bucketed
|
||
in that local time. Clamped to ±16 h — a nonsense offset would otherwise
|
||
shift every day silently.
|
||
"""
|
||
tz = max(-960, min(960, int(tzOffset)))
|
||
global _activity_cache
|
||
key = (store.summaries_version, tz)
|
||
with _activity_lock:
|
||
if _activity_cache and _activity_cache[0] == key:
|
||
return _activity_cache[1]
|
||
out = activity_mod.rollup(store.all_summaries(), tz)
|
||
with _activity_lock:
|
||
_activity_cache = (key, out)
|
||
return out
|
||
|
||
|
||
# ── skills catalog ──────────────────────────────────────────────────────────
|
||
# The per-skill usage rollup walks every transcript's skills blob and joins it to
|
||
# the conversation summaries; memoized on the summaries version so repeat
|
||
# requests between changes are a dict lookup (same pattern as _project_costs).
|
||
_skill_usage_cache: tuple[int, dict] | None = None
|
||
_skill_usage_lock = threading.Lock()
|
||
|
||
|
||
def _skill_usage() -> dict:
|
||
global _skill_usage_cache
|
||
key = store.summaries_version
|
||
with _skill_usage_lock:
|
||
if _skill_usage_cache and _skill_usage_cache[0] == key:
|
||
return _skill_usage_cache[1]
|
||
usage = skills_mod.usage_by_skill(
|
||
store.all_summaries(), store.all_transcript_skill_rows())
|
||
with _skill_usage_lock:
|
||
_skill_usage_cache = (key, usage)
|
||
return usage
|
||
|
||
|
||
@app.get("/api/skills", responses=_r(schemas.SkillsResponse))
|
||
def skills_list():
|
||
"""Card metadata + usage/cost stats for every skill, last-used first.
|
||
|
||
The list is the **union** of the skills on disk and the ones the transcripts
|
||
saw invoked: a built-in or plugin skill (``code-review``, ``artifact-design``)
|
||
has no ``SKILL.md`` in this repo but still earned its calls, so it appears
|
||
with ``sourceKind: "builtin"`` and no editor link.
|
||
"""
|
||
usage = _skill_usage()
|
||
ctx = _context_by_skill(store.all_files())
|
||
|
||
def stats(name: str) -> dict:
|
||
u = usage.get(name) or {}
|
||
c = ctx.get(name)
|
||
count = u.get("count", 0)
|
||
ctx_cost = c["cost"] if c else None
|
||
return {
|
||
"count": count,
|
||
"lastUsed": u.get("lastUsed"),
|
||
"conversations": u.get("conversations", 0),
|
||
"contextTokens": c["tokens"] if c else None,
|
||
"contextCost": ctx_cost,
|
||
# what invoking it has plausibly cost: loading its context, per call
|
||
"estSpend": (ctx_cost * count) if ctx_cost is not None else None,
|
||
# total spend of the conversations that used it (coarse — a session
|
||
# is credited in full to every skill it invoked)
|
||
"sessionCost": u.get("sessionCost", 0.0),
|
||
"sessionTokens": u.get("sessionTokens", 0),
|
||
}
|
||
|
||
items = [{**s, **stats(s["name"])} for s in skills_mod.catalog()]
|
||
known = {s["name"] for s in items}
|
||
for name in sorted(usage.keys() - known):
|
||
items.append({
|
||
"name": name, "title": name, "description": "",
|
||
"dir": None, "path": None,
|
||
"source": "builtin", "sourceKind": "builtin",
|
||
"files": 0, "updatedAt": None,
|
||
**stats(name),
|
||
})
|
||
|
||
# Last-used first (never-used skills sink to the bottom, alphabetical) — the
|
||
# frontend re-sorts by the same field, this just makes the payload sane.
|
||
items.sort(key=lambda s: (s["lastUsed"] or "", s["name"]), reverse=True)
|
||
|
||
return {
|
||
"skills": items,
|
||
"totals": {
|
||
"skills": len(items),
|
||
"invocations": sum(s["count"] for s in items),
|
||
"used": sum(1 for s in items if s["count"] > 0),
|
||
"contextCost": sum(s["contextCost"] or 0.0 for s in items),
|
||
"estSpend": sum(s["estSpend"] or 0.0 for s in items),
|
||
},
|
||
}
|
||
|
||
|
||
def _auto_state(ended_at: str | None) -> str:
|
||
"""``running`` if the last recorded activity is recent, else ``finished``."""
|
||
if not ended_at:
|
||
return "finished"
|
||
try:
|
||
dt = datetime.datetime.fromisoformat(ended_at.replace("Z", "+00:00"))
|
||
except ValueError:
|
||
return "finished"
|
||
now = datetime.datetime.now(datetime.timezone.utc)
|
||
age = (now - dt).total_seconds()
|
||
return "running" if 0 <= age < RUNNING_WINDOW_SECS else "finished"
|
||
|
||
|
||
def _outlived_exit(exited_at: str | None, ended_at: str | None) -> bool:
|
||
"""True when the transcript's last activity is clearly *after* the recorded
|
||
process exit (>10s slack) — the session kept going in a run the sidecar
|
||
never saw (a terminal or remote-control turn), so the exit-time "finished"
|
||
stamp is stale and must not be trusted."""
|
||
if not exited_at or not ended_at:
|
||
return False
|
||
try:
|
||
exited = datetime.datetime.fromisoformat(exited_at.replace("Z", "+00:00"))
|
||
ended = datetime.datetime.fromisoformat(ended_at.replace("Z", "+00:00"))
|
||
except ValueError:
|
||
return False
|
||
return (ended - exited).total_seconds() > 10
|
||
|
||
|
||
def _resolve_state(summary: dict, m: dict, notified: str | None) -> str:
|
||
"""Decide a conversation's lifecycle state.
|
||
|
||
The ``worktree``/``select-project`` skills stamp ``state=running`` but
|
||
nothing ever clears it, so finished conversations get stuck "running". We
|
||
treat an explicit ``running`` as trustworthy only while activity is recent;
|
||
once a conversation has both sent a notification and ended its last message
|
||
with the ``DONE`` marker (or the `complete` skill's ``COMPLETED`` marker —
|
||
see ``_COMPLETED_RE``; that pass continues the same session, so its own
|
||
reply becomes the transcript's last message and overwrites ``doneMarker``),
|
||
it is finished regardless of the recency window.
|
||
The exit watcher (``_watch_sidecar_runs``) stamps ``finished`` the moment a
|
||
sidecar-launched run's process dies, so the badge doesn't sit on "running"
|
||
for the rest of the recency window when the prompt never said DONE.
|
||
A run whose transcript ends on a ``[Request interrupted by user]`` marker is
|
||
reported as ``interrupted`` (a terminal state, so it wins over recency and
|
||
over a stale ``running`` stamp); one ending on a ``No response requested.``
|
||
reply is reported as ``paused``. A run that parked itself waiting on a
|
||
trigger (a trailing ``ScheduleWakeup``/``Monitor`` — see the parser's
|
||
``waiting_trailing``) is ``paused`` too, instead of sitting on a "running"
|
||
badge while nothing happens. Other explicit states (e.g. ``review``) are
|
||
always honoured.
|
||
"""
|
||
auto = _auto_state(summary.get("endedAt")) # running if recent, else finished
|
||
# A trailing interrupt marker in the transcript is authoritative: the run
|
||
# was stopped by the user (Esc, or the interrupt button in this viewer).
|
||
if summary.get("interruptedByUser"):
|
||
return "interrupted"
|
||
# A trailing "No response requested." reply (a queued/resume prompt that had
|
||
# nothing to answer) leaves the run paused rather than finished.
|
||
if summary.get("pausedByUser"):
|
||
return "paused"
|
||
if (summary.get("doneMarker") or summary.get("completedMarker")) and notified:
|
||
return "finished"
|
||
explicit = m.get("state")
|
||
# An exit-time "finished" is trusted only while the transcript hasn't
|
||
# moved past that exit; later activity means the session continued outside
|
||
# the sidecar — fall back to the recency heuristic instead.
|
||
if explicit == "finished" and _outlived_exit(m.get("exitedAt"),
|
||
summary.get("endedAt")):
|
||
explicit = None
|
||
if explicit in (None, "running"):
|
||
# Claude scheduled a wake-up / blocked on a Monitor and the turn ended:
|
||
# the run is idle until the trigger fires, however recent its activity.
|
||
if summary.get("waitingForTrigger"):
|
||
return "paused"
|
||
return auto # stale/absent "running" → fall back to recency
|
||
return explicit
|
||
|
||
|
||
def _merge_worktrees(*sources) -> list[dict]:
|
||
"""Union worktree records from several sources, keyed by directory basename.
|
||
|
||
Each source is a list of ``{name, dir, createdAt, removedAt}`` (from the
|
||
transcript parse and/or the live sidecar). Same ``dir`` → one entry: earliest
|
||
``createdAt`` and latest ``removedAt`` win, and a real short ``name`` beats a
|
||
fallback that equals the dir. Ordered by creation (unknown-created last)."""
|
||
by_dir: dict[str, dict] = {}
|
||
for src in sources:
|
||
for e in src or []:
|
||
d = (e.get("dir") or e.get("name") or "").strip()
|
||
if not d:
|
||
continue
|
||
cur = by_dir.get(d)
|
||
if cur is None:
|
||
by_dir[d] = {"name": e.get("name") or d, "dir": d,
|
||
"createdAt": e.get("createdAt"),
|
||
"removedAt": e.get("removedAt")}
|
||
continue
|
||
nm = e.get("name")
|
||
if nm and nm != d and (not cur["name"] or cur["name"] == d):
|
||
cur["name"] = nm
|
||
for k, pick in (("createdAt", min), ("removedAt", max)):
|
||
vals = [v for v in (cur.get(k), e.get(k)) if v]
|
||
cur[k] = pick(vals) if vals else None
|
||
return sorted(by_dir.values(), key=lambda w: (w.get("createdAt") is None,
|
||
w.get("createdAt") or ""))
|
||
|
||
|
||
def _conv_title(summary: dict) -> str | None:
|
||
"""The conversation's display title.
|
||
|
||
A session can rename itself (``conv-meta title "…"``) once it knows what it
|
||
is actually doing — the goal-keeper does this after picking its item, so its
|
||
card reads "Add a Work button to the Goals page" rather than the generic
|
||
spawn prompt it was launched with. That sidecar title wins over the parsed
|
||
one (Claude's ``ai-title``, else the first user message)."""
|
||
sid = summary.get("sessionId") or ""
|
||
return meta_store.peek(sid).get("title") or summary.get("title")
|
||
|
||
|
||
def _conv_meta(summary: dict) -> dict:
|
||
"""Merge a conversation's parsed auto fields with its action sidecar."""
|
||
sid = summary.get("sessionId") or ""
|
||
m = meta_store.peek(sid) # read-only — never mutated below
|
||
projects = list(summary.get("projectsAuto") or [])
|
||
for p in (m.get("projects") or []):
|
||
if p not in projects:
|
||
projects.append(p)
|
||
# No project at all ⇒ the conversation worked on the lab itself (a service,
|
||
# a skill, a script, a doc, a question). Attribute it to the root project so
|
||
# it stops falling out of every project-shaped view. Derived, never stored.
|
||
projects = projects_mod.attributed_projects(projects)
|
||
# Homelab services touched by the conversation (mirrors ``projects``).
|
||
services = list(summary.get("servicesAuto") or [])
|
||
for s in (m.get("services") or []):
|
||
if s not in services:
|
||
services.append(s)
|
||
# Lifecycle stamps: the live sidecar (written by the skills) is authoritative;
|
||
# where it's empty we backfill from evidence mined out of the transcript.
|
||
auto = summary.get("lifecycleAuto") or {}
|
||
stamp = lambda k: m.get(k) or auto.get(k) # noqa: E731
|
||
notified = stamp("notified")
|
||
return {
|
||
"state": _resolve_state(summary, m, notified),
|
||
# The custom sidecar title override, when set (the resolved top-level
|
||
# `title` already prefers it) — lets the UI tell a renamed card from an
|
||
# auto-titled one and offer "reset to auto title" in the rename dialog.
|
||
"title": m.get("title") or None,
|
||
"projects": projects,
|
||
"services": services,
|
||
# Git worktrees the conversation created/removed (with created/removed
|
||
# timestamps), merging the transcript parse with anything the worktree
|
||
# skill recorded live in the sidecar.
|
||
"worktrees": _merge_worktrees(summary.get("worktreesAuto"),
|
||
m.get("worktrees")),
|
||
"committed": stamp("committed"),
|
||
"pushed": stamp("pushed"),
|
||
"merged": stamp("merged"),
|
||
"deployed": stamp("deployed"),
|
||
"notified": notified,
|
||
"notifications": m.get("notifications") or [],
|
||
# Hidden-from-list flag, toggled from the conversation card context menu.
|
||
"archived": bool(m.get("archived")),
|
||
# Which agent harness runs this session ("claude"/"pi"). Stamped by
|
||
# /api/spawn; conversations predating the field (or spawned from a
|
||
# terminal) fall back to "claude" client-side.
|
||
"harness": m.get("harness"),
|
||
# Source conversation + message ordinal when this one is a fork.
|
||
"forkedFrom": m.get("forkedFrom"),
|
||
# The cron job that spawned this session ({id, name}), if any — the
|
||
# viewer tags the conversation and links back to Settings → Cron.
|
||
"cron": m.get("cron"),
|
||
# The agent definition this session was launched from ({name, at}) by
|
||
# the agent page's "Run now" — the manual sibling of `cron`, and what
|
||
# puts the run in that agent's history.
|
||
"agentRun": m.get("agentRun"),
|
||
# When the `complete` skill's end-of-task pass ran on this session:
|
||
# its reply signs off with COMPLETED instead of DONE (the parser's
|
||
# ``completedMarker``), so the marker on the last message *is* the
|
||
# record — no sidecar stamp to keep in sync. The viewer turns it into
|
||
# the COMPLETED seal on the card.
|
||
"completedAt": (summary.get("endedAt")
|
||
if summary.get("completedMarker") else None),
|
||
# The conversation's published summary, written by that same pass
|
||
# ({markdown, at}) and rendered under the thread (see scaffold.py).
|
||
"summary": m.get("summary") or None,
|
||
}
|
||
|
||
|
||
def _conv_id(path: str) -> str:
|
||
"""Transcript path → the rel id the detail endpoint / UI use."""
|
||
try:
|
||
return str(pathlib.Path(path).relative_to(TRANSCRIPTS_DIR)) \
|
||
if TRANSCRIPTS_DIR else pathlib.Path(path).name
|
||
except ValueError:
|
||
return pathlib.Path(path).name
|
||
|
||
|
||
# ── subagent (sidechain) linkage ─────────────────────────────────────────────
|
||
# Subagent transcripts live at `<convId>/subagents/agent-<id>.jsonl` next to the
|
||
# parent `<convId>.jsonl`. They are imported + parsed like any transcript (flagged
|
||
# `isSidechain` by the parser) but hidden from the top-level lists; instead each is
|
||
# linked from the parent's Task/Agent card and its usage is rolled into the parent.
|
||
def _is_sidechain_path(abs_path: str) -> bool:
|
||
return "subagents" in pathlib.Path(abs_path).parts
|
||
|
||
|
||
def _parent_path_of(abs_path: str) -> str | None:
|
||
"""Given a subagent transcript path, the parent `<convId>.jsonl` archive path."""
|
||
parts = pathlib.Path(abs_path).parts
|
||
if "subagents" not in parts:
|
||
return None
|
||
idx = parts.index("subagents")
|
||
conv_dir = pathlib.Path(*parts[:idx]) # …/<convId>
|
||
return str(conv_dir.with_suffix(".jsonl")) # …/<convId>.jsonl
|
||
|
||
|
||
def _rollup_children(summary: dict, children: list[dict]) -> dict:
|
||
"""Fold each subagent's aggregate usage into the parent's totals and its
|
||
`agent` activity bucket, so a conversation's cost includes the sub-agents it
|
||
spawned. Returns a shallow copy; leaves the stored summary untouched."""
|
||
if not children:
|
||
return summary
|
||
s = dict(summary)
|
||
usage = dict(summary.get("usage") or conversations.zero_usage())
|
||
by_tool = {k: dict(v) for k, v in (summary.get("byTool") or {}).items()}
|
||
by_tool_sub = {k: {sk: dict(sv) for sk, sv in v.items()}
|
||
for k, v in (summary.get("byToolSub") or {}).items()}
|
||
add_tokens = add_output = 0
|
||
add_cost = 0.0
|
||
for c in children:
|
||
cu = c.get("usage") or {}
|
||
for k in usage:
|
||
usage[k] = (usage.get(k) or 0) + (cu.get(k) or 0)
|
||
add_tokens += c.get("tokens") or 0
|
||
add_cost += c.get("cost") or 0.0
|
||
add_output += cu.get("output") or 0
|
||
agent = by_tool.get("agent") or {"tokens": 0, "cost": 0.0, "output": 0}
|
||
by_tool["agent"] = {"tokens": (agent.get("tokens") or 0) + add_tokens,
|
||
"cost": (agent.get("cost") or 0.0) + add_cost,
|
||
"output": (agent.get("output") or 0) + add_output}
|
||
# Mirror the rollup into the drill-down: a "subagents" sub of the agent bucket.
|
||
asub = by_tool_sub.setdefault("agent", {})
|
||
sa = asub.get("subagents") or {"tokens": 0, "cost": 0.0, "output": 0}
|
||
asub["subagents"] = {"tokens": (sa.get("tokens") or 0) + add_tokens,
|
||
"cost": (sa.get("cost") or 0.0) + add_cost,
|
||
"output": (sa.get("output") or 0) + add_output}
|
||
s["usage"] = usage
|
||
s["byTool"] = by_tool
|
||
s["byToolSub"] = by_tool_sub
|
||
s["tokens"] = (summary.get("tokens") or 0) + add_tokens
|
||
s["cost"] = (summary.get("cost") or 0.0) + add_cost
|
||
s["subagentCount"] = len(children)
|
||
s["subagentTokens"] = add_tokens
|
||
s["subagentCost"] = add_cost
|
||
return s
|
||
|
||
|
||
def _subagent_meta(sub_jsonl: pathlib.Path) -> dict:
|
||
"""Read the `agent-<id>.meta.json` sitting next to a subagent transcript."""
|
||
meta_f = sub_jsonl.with_name(sub_jsonl.stem + ".meta.json")
|
||
try:
|
||
return json.loads(meta_f.read_text(encoding="utf-8"))
|
||
except (OSError, ValueError):
|
||
return {}
|
||
|
||
|
||
# meta.json files are written once at spawn and never change, so list endpoints
|
||
# may cache them (only non-empty reads — an unwritten file may appear later).
|
||
_SUBAGENT_META_CACHE: dict[str, dict] = {}
|
||
|
||
|
||
def _subagent_meta_cached(sub_jsonl: pathlib.Path) -> dict:
|
||
key = str(sub_jsonl)
|
||
hit = _SUBAGENT_META_CACHE.get(key)
|
||
if hit is not None:
|
||
return hit
|
||
mj = _subagent_meta(sub_jsonl)
|
||
if mj:
|
||
_SUBAGENT_META_CACHE[key] = mj
|
||
return mj
|
||
|
||
|
||
def _running_agents(summary: dict, state: str | None = None) -> list[dict]:
|
||
"""The subagents a conversation still has in flight.
|
||
|
||
An open Task call only means "running" while the conversation itself is: a
|
||
run that was interrupted (or died) mid-agent leaves its call open forever,
|
||
and so does every transcript written before the parser learned to close a
|
||
background agent on its `<task-notification>`. Nothing is running there."""
|
||
ra = summary.get("runningAgents") or []
|
||
if not ra:
|
||
return []
|
||
st = state if state is not None else _conv_meta(summary).get("state")
|
||
return ra if st == "running" else []
|
||
|
||
|
||
def _is_deploying(summary: dict, state: str | None = None) -> bool:
|
||
"""True while this conversation has a deploy command in flight (issued, no
|
||
tool_result yet) — gated the same way as `_running_agents`: a dangling call
|
||
left open by an interrupted/dead run isn't actually deploying anymore."""
|
||
if not summary.get("deploying"):
|
||
return False
|
||
st = state if state is not None else _conv_meta(summary).get("state")
|
||
return st == "running"
|
||
|
||
|
||
def _sidechain_state(ap: pathlib.Path) -> str | None:
|
||
"""A subagent's own lifecycle state: it runs exactly as long as the parent's
|
||
Task call is still open. `None` if `ap` isn't a subagent transcript.
|
||
|
||
Its `_conv_meta` state can't be trusted: a sidechain's sessionId is the
|
||
*parent's*, so the sidecar lookup returns the parent's (possibly stale)
|
||
state, and the recency fallback would keep a just-finished subagent "running"
|
||
for the whole recency window."""
|
||
pp = _is_sidechain_path(str(ap)) and _parent_path_of(str(ap))
|
||
if not pp:
|
||
return None
|
||
psum = store.get_summary(pp) or {}
|
||
tuid = _subagent_meta(ap).get("toolUseId")
|
||
open_calls = {a.get("toolUseId") for a in _running_agents(psum)}
|
||
return "running" if tuid and tuid in open_calls else "finished"
|
||
|
||
|
||
def _attach_subagents(ap: pathlib.Path, data: dict) -> dict:
|
||
"""Enrich a parsed parent thread with its subagents: link each Task/Agent card
|
||
to the child transcript it spawned (via meta.json `toolUseId`) and roll the
|
||
children's usage into the parent totals. If `ap` is itself a subagent, attach a
|
||
`parentConversation` back-link instead. Returns the (mutated) data."""
|
||
# A subagent detail view: link back up to the parent + the exact Task card.
|
||
if _is_sidechain_path(str(ap)):
|
||
pp = _parent_path_of(str(ap))
|
||
if pp:
|
||
psum = store.get_summary(pp) or {}
|
||
mj = _subagent_meta(ap)
|
||
data["parentConversation"] = {
|
||
"id": _conv_id(pp),
|
||
"title": _conv_title(psum),
|
||
"toolUseId": mj.get("toolUseId"),
|
||
"agentType": mj.get("agentType"),
|
||
"description": mj.get("description"),
|
||
}
|
||
return data
|
||
# A parent view. Mark every Task/Agent card with whether its agent is still
|
||
# running (the parser tracks the open calls as `runningAgents`) — including
|
||
# the ones whose subagent transcript hasn't been written yet, which is
|
||
# exactly the moment the card most needs to say "running".
|
||
open_calls = {a.get("toolUseId") for a in (data.get("runningAgents") or [])}
|
||
for it in data.get("thread") or []:
|
||
if it.get("name") in ("Task", "Agent") and it.get("toolUseId"):
|
||
it["agentRunning"] = it["toolUseId"] in open_calls
|
||
# Gather the children from the sibling `subagents/` dir.
|
||
sub_dir = ap.with_suffix("") / "subagents"
|
||
if not sub_dir.is_dir():
|
||
return data
|
||
link_by_tooluse: dict[str, dict] = {}
|
||
infos_by_id: dict[str, dict] = {}
|
||
child_summaries: list[dict] = []
|
||
for meta_f in sorted(sub_dir.glob("*.meta.json")):
|
||
stem = meta_f.name[:-len(".meta.json")] # agent-<id>
|
||
child = sub_dir / f"{stem}.jsonl"
|
||
csum = store.get_summary(str(child)) or {}
|
||
mj = _subagent_meta(child)
|
||
tuid = mj.get("toolUseId")
|
||
info = {"id": _conv_id(str(child)),
|
||
"agentType": mj.get("agentType"),
|
||
"description": mj.get("description"),
|
||
"title": csum.get("title"), "model": csum.get("model"),
|
||
"tokens": csum.get("tokens"), "cost": csum.get("cost"),
|
||
"messages": csum.get("messages"),
|
||
"startedAt": csum.get("startedAt"),
|
||
"endedAt": csum.get("endedAt"),
|
||
"state": "running" if tuid in open_calls else "finished"}
|
||
infos_by_id[info["id"]] = info
|
||
if mj.get("toolUseId"):
|
||
link_by_tooluse[mj["toolUseId"]] = info
|
||
if csum:
|
||
child_summaries.append(csum)
|
||
# Link each Task/Agent card to its child and collect the subagents in the
|
||
# order they were spawned (thread order), so the detail view can list them.
|
||
ordered: list[dict] = []
|
||
seen: set[str] = set()
|
||
for it in data.get("thread") or []:
|
||
tuid = it.get("toolUseId")
|
||
info = tuid and link_by_tooluse.get(tuid)
|
||
if info:
|
||
it["subagent"] = info
|
||
# A subagent's spend is rolled into the parent's total and its `agent`
|
||
# bucket below, but it was billed to the *child's* transcript — so no
|
||
# parent thread item carries it, and the cost-growth chart (which sums
|
||
# thread items) would report less than the total. Credit it to the card
|
||
# that spawned it: the turn that actually caused the spend.
|
||
child_cost = info.get("cost") or 0.0
|
||
if child_cost > 0:
|
||
it["turnCost"] = (it.get("turnCost") or 0.0) + child_cost
|
||
it["turnTokens"] = (it.get("turnTokens") or 0) + (info.get("tokens") or 0)
|
||
# It's all downstream agent work, whatever the card's own turn drove.
|
||
it["turnBucket"] = "agent"
|
||
if info["id"] not in seen:
|
||
ordered.append(info)
|
||
seen.add(info["id"])
|
||
for cid, info in infos_by_id.items(): # any orphans (no matching card)
|
||
if cid not in seen:
|
||
ordered.append(info)
|
||
seen.add(cid)
|
||
# No card claimed this child's cost above, but the rollup still counts
|
||
# it — park it on an invisible carrier so the chart stays whole.
|
||
if (info.get("cost") or 0.0) > 0 and data.get("thread") is not None:
|
||
data["thread"].append({
|
||
"role": "assistant", "kind": "turn", "first": False, "out": 0,
|
||
"turnCost": info["cost"], "turnTokens": info.get("tokens") or 0,
|
||
"turnBucket": "agent",
|
||
})
|
||
data["subagents"] = ordered
|
||
# Roll subagent usage into the parent totals / `agent` bucket.
|
||
rolled = _rollup_children(data, child_summaries)
|
||
for k in ("usage", "byTool", "byToolSub", "tokens", "cost",
|
||
"subagentCount", "subagentTokens", "subagentCost"):
|
||
if k in rolled:
|
||
data[k] = rolled[k]
|
||
return data
|
||
|
||
|
||
_skills_by_path_cache: tuple[int, dict[str, dict[str, int]]] | None = None
|
||
|
||
|
||
def _skills_by_path() -> dict[str, dict[str, int]]:
|
||
"""transcript path → ``{skill: calls}``, decoded once per store version.
|
||
|
||
The indexer keeps a transcript's skill map in its **own column** (it feeds
|
||
the catalog rollup), not inside the summary blob — so a stored summary has
|
||
no `skills` key at all, and this is the join that puts it back.
|
||
"""
|
||
global _skills_by_path_cache
|
||
hit = _skills_by_path_cache
|
||
if hit and hit[0] == store.summaries_version:
|
||
return hit[1]
|
||
out: dict[str, dict[str, int]] = {}
|
||
for path, raw in store.all_transcript_skill_rows():
|
||
try:
|
||
contrib = json.loads(raw or "{}")
|
||
except ValueError:
|
||
continue
|
||
counts = {name: d.get("count") or 0 for name, d in contrib.items()}
|
||
if counts:
|
||
out[path] = counts
|
||
_skills_by_path_cache = (store.summaries_version, out)
|
||
return out
|
||
|
||
|
||
def _skills_used(paths: list[str], extra: dict | None = None) -> list[str]:
|
||
"""Skill names used across `paths`, busiest first.
|
||
|
||
Callers pass a conversation *and its subagent transcripts*: a `Skill(...)`
|
||
run by a spawned Explore is work this conversation caused, and the child is
|
||
never listed on its own — so hiding its skills would lose them entirely.
|
||
`extra` folds in a freshly parsed ``{skill: {count, …}}`` map for a
|
||
transcript the store may not have re-read yet (the detail view parses live).
|
||
"""
|
||
by_path = _skills_by_path()
|
||
counts: dict[str, int] = {}
|
||
for p in paths:
|
||
for name, n in by_path.get(p, {}).items():
|
||
counts[name] = counts.get(name, 0) + n
|
||
for name, d in (extra or {}).items():
|
||
counts[name] = counts.get(name, 0) + (d.get("count") or 0)
|
||
return sorted(counts, key=lambda n: (-counts[n], n))
|
||
|
||
|
||
_agents_by_conv_cache: tuple[tuple[int, int], dict[str, list[str]]] | None = None
|
||
|
||
|
||
def _agents_by_conversation() -> dict[str, list[str]]:
|
||
"""conversation id → the agent types that ran in it, busiest first.
|
||
|
||
Built from the same run list the agents catalog counts (`_agent_runs`) —
|
||
not from a second pass over the `Task` calls — so a conversation card can
|
||
never disagree with the agent page. That list already resolves both origins:
|
||
a subagent run is keyed to its *parent* conversation (where its Task card
|
||
lives), and a cron/manual agent session to the conversation it *is*.
|
||
"""
|
||
global _agents_by_conv_cache
|
||
key = (store.summaries_version, meta_store.version)
|
||
hit = _agents_by_conv_cache
|
||
if hit and hit[0] == key:
|
||
return hit[1]
|
||
counts: dict[str, dict[str, int]] = {}
|
||
for r in _agent_runs():
|
||
cid, name = r.get("conversationId"), r.get("agent")
|
||
if not cid or not name:
|
||
continue
|
||
c = counts.setdefault(cid, {})
|
||
c[name] = c.get(name, 0) + 1
|
||
out = {cid: sorted(c, key=lambda n: (-c[n], n)) for cid, c in counts.items()}
|
||
_agents_by_conv_cache = (key, out)
|
||
return out
|
||
|
||
|
||
def _conv_card(path: str, s: dict) -> dict:
|
||
"""Compact conversation reference used when cross-linking from a memory."""
|
||
return {
|
||
"id": _conv_id(path),
|
||
"sessionId": s.get("sessionId"),
|
||
"title": _conv_title(s),
|
||
"model": s.get("model"),
|
||
"endedAt": s.get("endedAt"),
|
||
"cost": s.get("cost"),
|
||
"tokens": s.get("tokens"),
|
||
"meta": _conv_meta(s),
|
||
}
|
||
|
||
|
||
# Card fields every list view needs. `usage`/`byToolSub` are heavy per-card and
|
||
# only the analytics dashboard reads them — it asks with ``full=1``.
|
||
_CARD_KEYS = ("sessionId", "project", "gitBranch", "model", "models", "efforts",
|
||
"title", "userTurns", "assistantTurns", "messages", "startedAt", "endedAt",
|
||
"tokens", "cost", "byTool", "subagentCount",
|
||
"subagentTokens", "subagentCost", "tasks")
|
||
_CARD_KEYS_FULL = _CARD_KEYS + ("usage", "byToolSub")
|
||
|
||
|
||
def _dedup_by_session(summaries: list[tuple[str, dict]]) -> list[tuple[str, dict]]:
|
||
"""Collapse transcript files that share a ``sessionId`` into one.
|
||
|
||
Resuming a session in a different working directory (a worktree, or
|
||
``claude --resume`` from another dir) makes Claude Code write the
|
||
continuation to a *new* cwd-encoded folder under the SAME sessionId — so
|
||
one logical conversation lands on disk as several ``<cwd>/<sid>.jsonl``
|
||
files. Keyed on the file path, each became its own card and the
|
||
conversation showed up two (or more) times. Keep only the file with the
|
||
most recent activity (latest ``endedAt``, then most messages, then longest
|
||
path as a stable tiebreak) — that's the fully-resumed transcript. Files
|
||
without a sessionId are never collapsed (nothing links them)."""
|
||
best: dict[str, tuple[str, dict]] = {}
|
||
passthrough: list[tuple[str, dict]] = []
|
||
for path, s in summaries:
|
||
sid = s.get("sessionId")
|
||
if not sid:
|
||
passthrough.append((path, s))
|
||
continue
|
||
cur = best.get(sid)
|
||
if cur is None or (
|
||
(s.get("endedAt") or "", s.get("messages") or 0, path)
|
||
> (cur[1].get("endedAt") or "", cur[1].get("messages") or 0, cur[0])
|
||
):
|
||
best[sid] = (path, s)
|
||
return list(best.values()) + passthrough
|
||
|
||
|
||
def _conversation_cards(full: bool) -> list[dict]:
|
||
"""Every non-sidechain conversation as a card dict, unsorted.
|
||
|
||
Summaries come from the store's shared decode cache and are **never
|
||
mutated** here — rollups copy, and per-card extras land on the card."""
|
||
summaries = store.all_summaries()
|
||
# Group subagent summaries under their parent so we can roll their usage up.
|
||
children: dict[str, list[dict]] = {}
|
||
child_paths: dict[str, list[str]] = {}
|
||
for path, s in summaries:
|
||
if s.get("isSidechain"):
|
||
pp = _parent_path_of(path)
|
||
if pp:
|
||
children.setdefault(pp, []).append(s)
|
||
child_paths.setdefault(pp, []).append(path)
|
||
keys = _CARD_KEYS_FULL if full else _CARD_KEYS
|
||
# A session resumed across working dirs has one transcript file per cwd —
|
||
# collapse those so the conversation shows up once, not once per copy.
|
||
out = []
|
||
agents_by_conv = _agents_by_conversation()
|
||
for path, s in _dedup_by_session(
|
||
[(p, s) for p, s in summaries if not s.get("isSidechain")]
|
||
):
|
||
if not s.get("messages"):
|
||
continue # skip empty transcripts
|
||
skills_used = _skills_used([path, *child_paths.get(path, [])])
|
||
s = _rollup_children(s, children.get(path, []))
|
||
m = _conv_meta(s)
|
||
cid = _conv_id(path)
|
||
out.append({"id": cid, **{k: s.get(k) for k in keys},
|
||
"title": _conv_title(s),
|
||
"runningAgents": _running_agents(s, m.get("state")),
|
||
"deploying": _is_deploying(s, m.get("state")),
|
||
"skillsUsed": skills_used,
|
||
"agentsUsed": agents_by_conv.get(cid, []),
|
||
"meta": m})
|
||
return out
|
||
|
||
|
||
@app.get("/api/conversations", responses=_r(schemas.ConversationsResponse))
|
||
def conversations_list(limit: int = 50, before: str | None = None,
|
||
archived: str = "false", full: bool = False,
|
||
sessions: str = ""):
|
||
"""List parsed transcripts (newest first) for the conversation viewer.
|
||
|
||
Paginated: ``limit`` caps the page (0 = everything), ``before`` is the
|
||
``endedAt`` cursor of the previous page's last card, and ``archived``
|
||
filters hidden conversations server-side (``false`` — the default view —
|
||
excludes them; ``all`` includes them). ``full=1`` adds the per-card
|
||
``usage``/``byToolSub`` blocks the analytics dashboard aggregates.
|
||
|
||
``sessions`` (comma-separated session ids) resolves a specific handful of
|
||
conversations instead of a page — what the home page's "waiting for
|
||
feedback" cards use to join a pending form/ask back to the session that
|
||
posted it, without pulling the whole (~MB) list for one title. An explicit
|
||
id lookup ignores the archived filter and the cursor."""
|
||
cards = _conversation_cards(full)
|
||
if sessions:
|
||
want = {s.strip() for s in sessions.split(",") if s.strip()}
|
||
cards = [c for c in cards if c.get("sessionId") in want]
|
||
return {"conversations": cards, "count": len(cards),
|
||
"nextBefore": None,
|
||
"pricing": {"model": MODEL,
|
||
"inputPerMTok": INPUT_PRICE_PER_MTOK}}
|
||
if archived != "all":
|
||
cards = [c for c in cards if not c["meta"].get("archived")]
|
||
cards.sort(key=lambda c: c.get("endedAt") or "", reverse=True)
|
||
total = len(cards)
|
||
if before:
|
||
cards = [c for c in cards if (c.get("endedAt") or "") < before]
|
||
next_before = None
|
||
if limit and limit > 0 and len(cards) > limit:
|
||
cards = cards[:limit]
|
||
next_before = cards[-1].get("endedAt")
|
||
return {"conversations": cards, "count": total, "nextBefore": next_before,
|
||
"pricing": {"model": MODEL, "inputPerMTok": INPUT_PRICE_PER_MTOK}}
|
||
|
||
|
||
@app.get("/api/subagents", responses=_r(schemas.SubagentsResponse))
|
||
def subagents_list():
|
||
"""Every subagent (sidechain) transcript as a flat lite ref carrying its
|
||
parent conversation id. The top-level conversation list hides sidechains;
|
||
this is the one place they're enumerated — the graph view's node source.
|
||
All from the in-process summaries cache (+ cached meta.json reads)."""
|
||
out = []
|
||
open_by_parent: dict[str, set] = {}
|
||
for path, s in store.all_summaries():
|
||
if not s.get("isSidechain") or not s.get("messages"):
|
||
continue
|
||
pp = _parent_path_of(path)
|
||
if not pp:
|
||
continue
|
||
# One open-Task lookup per parent, shared by all its children.
|
||
if pp not in open_by_parent:
|
||
psum = store.get_summary(pp) or {}
|
||
open_by_parent[pp] = {a.get("toolUseId")
|
||
for a in _running_agents(psum)}
|
||
mj = _subagent_meta_cached(pathlib.Path(path))
|
||
tuid = mj.get("toolUseId")
|
||
out.append({
|
||
"id": _conv_id(path),
|
||
"parentId": _conv_id(pp),
|
||
"agentType": mj.get("agentType"),
|
||
"description": mj.get("description"),
|
||
"title": s.get("title"),
|
||
"tokens": s.get("tokens"),
|
||
"cost": s.get("cost"),
|
||
"state": "running" if tuid and tuid in open_by_parent[pp]
|
||
else "finished",
|
||
})
|
||
return {"subagents": out}
|
||
|
||
|
||
@app.get("/api/conversation-summary",
|
||
responses=_r(schemas.ConversationSummary))
|
||
def conversation_summary(id: str):
|
||
"""One conversation's list card — what an SSE `transcript` ping patches
|
||
into the cached list, so a live turn costs a few KB instead of a full
|
||
list refetch."""
|
||
ap = _resolve_transcript(id)
|
||
s = store.get_summary(str(ap))
|
||
if not s:
|
||
raise HTTPException(404, "summary not indexed yet")
|
||
children: list[dict] = []
|
||
paths = [str(ap)]
|
||
sub_dir = ap.with_suffix("") / "subagents"
|
||
if sub_dir.is_dir():
|
||
for child in sorted(sub_dir.glob("*.jsonl")):
|
||
paths.append(str(child))
|
||
cs = store.get_summary(str(child))
|
||
if cs:
|
||
children.append(cs)
|
||
skills_used = _skills_used(paths)
|
||
s = _rollup_children(s, children)
|
||
m = _conv_meta(s)
|
||
return {"id": id, **{k: s.get(k) for k in _CARD_KEYS},
|
||
"title": _conv_title(s),
|
||
"runningAgents": _running_agents(s, m.get("state")),
|
||
"deploying": _is_deploying(s, m.get("state")),
|
||
"skillsUsed": skills_used,
|
||
"agentsUsed": _agents_by_conversation().get(id, []),
|
||
"meta": m}
|
||
|
||
|
||
@app.get("/api/notifications", responses=_r(schemas.NotificationsResponse))
|
||
def notifications_feed():
|
||
"""Feed source for the notification history: every conversation that
|
||
recorded a push, with just the fields the feed needs. Replaces walking the
|
||
whole conversation list client-side."""
|
||
out = []
|
||
for path, s in store.all_summaries():
|
||
if not s.get("messages") or s.get("isSidechain"):
|
||
continue
|
||
sid = s.get("sessionId") or ""
|
||
notes = meta_store.peek(sid).get("notifications") or []
|
||
if not notes:
|
||
continue
|
||
m = _conv_meta(s)
|
||
out.append({"id": _conv_id(path), "title": _conv_title(s),
|
||
"endedAt": s.get("endedAt"),
|
||
"projects": m.get("projects") or [],
|
||
"services": m.get("services") or [],
|
||
"notifications": notes})
|
||
out.sort(key=lambda c: c.get("endedAt") or "", reverse=True)
|
||
return {"conversations": out, "count": len(out)}
|
||
|
||
|
||
# ── flat notification feed (server-side mirror of the frontend's buildFeed) ──
|
||
# The PWA flattens /api/notifications + /api/notify-log into one list of feed
|
||
# items (lib/notifications.ts). The phone can't run that, so we do the same
|
||
# flattening here — same stable ids — for the unread endpoint and the read
|
||
# ledger. Keep the two in sync: id scheme is `<convId>#<at>#<index>` for
|
||
# conv-linked pushes and `log#<id>` for hub-only entries.
|
||
_DEDUP_WINDOW_MS = 120_000
|
||
|
||
|
||
def _parse_iso_ms(at: str | None) -> float | None:
|
||
if not at:
|
||
return None
|
||
try:
|
||
s = at.replace("Z", "+00:00")
|
||
return datetime.datetime.fromisoformat(s).timestamp() * 1000.0
|
||
except (ValueError, AttributeError):
|
||
return None
|
||
|
||
|
||
def _flat_notifications() -> list[dict]:
|
||
"""Every notification as a flat, de-duplicated, newest-first feed item —
|
||
the exact set the PWA shows, so "unread" means the same on the phone."""
|
||
feed: list[dict] = []
|
||
for path, s in store.all_summaries():
|
||
if not s.get("messages") or s.get("isSidechain"):
|
||
continue
|
||
sid = s.get("sessionId") or ""
|
||
notes = meta_store.peek(sid).get("notifications") or []
|
||
if not notes:
|
||
continue
|
||
m = _conv_meta(s)
|
||
cid = _conv_id(path)
|
||
title = _conv_title(s)
|
||
last_seen: dict[str, float] = {}
|
||
for i, n in enumerate(notes):
|
||
key = (f"{n.get('title') or ''} {n.get('body') or ''} "
|
||
f"{n.get('type') or ''} {n.get('url') or ''}")
|
||
t = _parse_iso_ms(n.get("at"))
|
||
prev = last_seen.get(key)
|
||
if prev is not None and (t is None or abs(t - prev) <= _DEDUP_WINDOW_MS):
|
||
continue # duplicate push within the conversation
|
||
if t is not None:
|
||
last_seen[key] = t
|
||
feed.append({
|
||
"id": f"{cid}#{n.get('at') or ''}#{i}",
|
||
"convId": cid, "convTitle": title,
|
||
"title": n.get("title") or "", "body": n.get("body") or "",
|
||
"type": n.get("type") or "", "url": n.get("url") or "",
|
||
"at": n.get("at"),
|
||
"projects": m.get("projects") or [],
|
||
"services": m.get("services") or [],
|
||
"kind": "notify",
|
||
"spoken": n.get("spoken") or "",
|
||
})
|
||
# Merge the hub log: a log entry that matches a conv push (title+body within
|
||
# the dedup window) is a duplicate and skipped; hub-only entries are added.
|
||
for e in notify_store.log(limit=500):
|
||
t = _parse_iso_ms(e.get("at"))
|
||
dup = False
|
||
for n in feed:
|
||
if n["title"] == (e.get("title") or "") and \
|
||
n["body"] == (e.get("body") or ""):
|
||
nt = _parse_iso_ms(n.get("at"))
|
||
if t is None or nt is None or abs(nt - t) <= _DEDUP_WINDOW_MS:
|
||
dup = True
|
||
break
|
||
if dup:
|
||
continue
|
||
feed.append({
|
||
"id": f"log#{e.get('id')}",
|
||
"convId": "", "convTitle": "",
|
||
"title": e.get("title") or "", "body": e.get("body") or "",
|
||
"type": e.get("type") or "", "url": e.get("url") or "",
|
||
"at": e.get("at"),
|
||
"projects": [], "services": [],
|
||
"kind": e.get("kind") or "notify",
|
||
})
|
||
feed.sort(key=lambda n: n.get("at") or "", reverse=True)
|
||
return feed
|
||
|
||
|
||
@app.get("/api/notifications/unread",
|
||
responses=_r(schemas.UnreadNotificationsResponse))
|
||
def notifications_unread(limit: int = 20):
|
||
"""Unread notifications (not yet opened in the UI), newest first.
|
||
|
||
Pure read: fetching this NEVER marks anything read — the phone can list and
|
||
read these aloud without "consuming" them; only the UI (or an explicit
|
||
``POST /api/notifications/seen``) advances the read ledger. This is the
|
||
endpoint the desk phone hits with its read-only API key.
|
||
"""
|
||
feed = _flat_notifications()
|
||
unread_ids = set(notif_read_store.unread([n["id"] for n in feed]))
|
||
items = [n for n in feed if n["id"] in unread_ids]
|
||
total = len(items)
|
||
if limit and limit > 0:
|
||
items = items[:limit]
|
||
return {"notifications": items, "count": len(items), "total": total}
|
||
|
||
|
||
class NotifAudioBody(BaseModel):
|
||
"""What a notification sounded like: its type (→ jingle) and spoken line."""
|
||
type: str | None = None
|
||
spoken: str | None = None
|
||
# Read instead of `spoken` when the push carried none (most of the log —
|
||
# --spoken only became mandatory recently).
|
||
title: str | None = None
|
||
message: str | None = None
|
||
|
||
|
||
@app.post("/api/notifications/audio")
|
||
def notification_audio(body: NotifAudioBody):
|
||
"""Render a notification's sound as a WAV so the UI can replay it.
|
||
|
||
The desk phone plays a per-type jazz jingle and then reads the French
|
||
`spoken` line; the notification card's play button asks for those same
|
||
bytes. Rendering happens in the phone service (one synthesizer, one voice —
|
||
see notify_audio.py). A push with no spoken line still gets a voice: its own
|
||
title + message, minus the bits written only for a screen.
|
||
"""
|
||
spoken = (body.spoken or "").strip() or notify_audio_mod.fallback_speech(
|
||
body.title or "", body.message or "")
|
||
try:
|
||
wav = notify_audio_mod.render(spoken, body.type or "")
|
||
except notify_audio_mod.RenderError as e:
|
||
raise HTTPException(502, str(e))
|
||
return Response(content=wav, media_type="audio/wav",
|
||
headers={"Cache-Control": "private, max-age=86400"})
|
||
|
||
|
||
@app.post("/api/notifications/seen", responses=_r(schemas.OkResponse))
|
||
def notifications_mark_seen(body: NotifSeenBody):
|
||
"""Mark notification ids as read (opened/dismissed in the UI). Mutating —
|
||
refused to a read-only API-key caller by the gate, so the phone can't use
|
||
it. Passing no ids with ``all=true`` marks the whole current feed read."""
|
||
ids = list(body.ids or [])
|
||
if body.all:
|
||
ids += [n["id"] for n in _flat_notifications()]
|
||
count = notif_read_store.mark_seen(ids)
|
||
return {"ok": True, "seen": count}
|
||
|
||
|
||
@app.post("/api/notifications/seed", responses=_r(schemas.OkResponse))
|
||
def notifications_seed(body: NotifSeenBody):
|
||
"""Adopt the given ids (or the whole current feed) as the read baseline —
|
||
the first-sync op so history isn't reported as unread."""
|
||
ids = list(body.ids or [])
|
||
if body.all or not ids:
|
||
ids += [n["id"] for n in _flat_notifications()]
|
||
count = notif_read_store.seed(ids)
|
||
return {"ok": True, "seen": count}
|
||
|
||
|
||
@app.get("/api/search-index", responses=_r(schemas.SearchIndexResponse))
|
||
def search_index():
|
||
"""Full-text search corpus: every conversation's user/assistant text blocks.
|
||
|
||
Shipped once to the client, which runs a fuzzy search (Fuse.js) over it. Each
|
||
message carries its ordinal `i`, which matches the `mi` on the detail thread's
|
||
text items, so a hit can deep-link to `/conversation/<id>?m=<i>` and scroll to
|
||
the exact message."""
|
||
out = []
|
||
for path, s in store.all_summaries():
|
||
if not s.get("messages"):
|
||
continue
|
||
if s.get("isSidechain"):
|
||
continue # subagent messages are reachable via the parent, not listed
|
||
msgs = s.get("searchMsgs") or []
|
||
if not msgs:
|
||
continue
|
||
out.append({
|
||
"id": _conv_id(path),
|
||
"title": _conv_title(s),
|
||
"project": s.get("project"),
|
||
"model": s.get("model"),
|
||
"endedAt": s.get("endedAt"),
|
||
# serve-time clip: search + snippets don't need more, and this
|
||
# corpus ships to the client in one payload
|
||
"msgs": [{**m, "text": (m.get("text") or "")[:500]} for m in msgs],
|
||
})
|
||
out.sort(key=lambda c: c.get("endedAt") or "", reverse=True)
|
||
return {"conversations": out, "count": len(out)}
|
||
|
||
|
||
# Recently parsed full threads, keyed by path and validated on (mtime, size).
|
||
# A parse of a large transcript costs hundreds of ms; re-opening a conversation
|
||
# (or a meta-only SSE ping refetching the open one) shouldn't pay it again.
|
||
# Served as a deep copy — the endpoint decorates the dict in place.
|
||
_detail_cache: dict[str, tuple[tuple[float, int], dict]] = {}
|
||
_DETAIL_CACHE_CAP = 4
|
||
_detail_lock = threading.Lock()
|
||
|
||
|
||
def _parse_full_cached(ap: pathlib.Path) -> dict:
|
||
key = str(ap)
|
||
try:
|
||
st = ap.stat()
|
||
sig = (st.st_mtime, st.st_size)
|
||
except OSError:
|
||
sig = None
|
||
with _detail_lock:
|
||
hit = _detail_cache.get(key)
|
||
if hit and sig and hit[0] == sig:
|
||
_detail_cache[key] = _detail_cache.pop(key) # LRU bump
|
||
return copy.deepcopy(hit[1])
|
||
data = conversations.parse_conversation(ap, full=True)
|
||
if sig:
|
||
with _detail_lock:
|
||
_detail_cache[key] = (sig, copy.deepcopy(data))
|
||
while len(_detail_cache) > _DETAIL_CACHE_CAP:
|
||
_detail_cache.pop(next(iter(_detail_cache)))
|
||
return data
|
||
|
||
|
||
@app.get("/api/conversation", responses=_r(schemas.ConversationDetail))
|
||
def conversation_detail(id: str):
|
||
"""Full parsed thread (tools collapsible client-side) + per-turn usage."""
|
||
ap = _resolve_transcript(id)
|
||
data = _parse_full_cached(ap)
|
||
# The raw `{skill: {count, last}}` map is indexer-internal; the thread only
|
||
# needs the names it used, and those roll the subagents' calls in. This
|
||
# parse is fresher than the store (a live turn re-parses here first), so the
|
||
# conversation's own map comes off `data` and only the children are joined.
|
||
sub_dir = ap.with_suffix("") / "subagents"
|
||
kid_paths = ([str(c) for c in sorted(sub_dir.glob("*.jsonl"))]
|
||
if sub_dir.is_dir() else [])
|
||
data["skillsUsed"] = _skills_used(kid_paths, extra=data.get("skills"))
|
||
data["agentsUsed"] = _agents_by_conversation().get(id, [])
|
||
data.pop("skills", None)
|
||
data["id"] = id
|
||
data["title"] = _conv_title(data)
|
||
data["meta"] = _conv_meta(data)
|
||
sub_state = _sidechain_state(ap)
|
||
if sub_state:
|
||
data["meta"]["state"] = sub_state
|
||
# Resolve the state first: which Task calls count as running depends on it.
|
||
data["runningAgents"] = _running_agents(data, data["meta"].get("state"))
|
||
data["deploying"] = _is_deploying(data, data["meta"].get("state"))
|
||
_attach_subagents(ap, data)
|
||
# Cross-link the memories this conversation read (recalled) and created.
|
||
read_slugs = set(data.get("memoriesRead") or [])
|
||
data["memoriesRead"] = [m for m in memories_mod.list_memories()
|
||
if m["slug"] in read_slugs]
|
||
data["memoriesCreated"] = memories_mod.by_origin(data.get("sessionId"))
|
||
# Plans this conversation authored (the `plan` skill stamps the planning
|
||
# session id / conversationIds into each plan's frontmatter).
|
||
sid = data.get("sessionId")
|
||
data["plans"] = ([p for p in plans_mod.list_plans()
|
||
if sid in plans_mod.linked_session_ids(p)] if sid else [])
|
||
# Generated artefacts (screenshots, renders…) the conversation's projects/
|
||
# services hold for this session — <dir>/.ai/artefacts/<date>/<sid>/<file>.
|
||
data["artefacts"] = artefacts_mod.list_for_conversation(
|
||
sid or "", data["meta"].get("projects") or [],
|
||
data["meta"].get("services") or [])
|
||
return data
|
||
|
||
|
||
# ── diff view: per-conversation commits + commit / conversation diffs ────────
|
||
def _resolve_transcript(id: str) -> pathlib.Path:
|
||
"""Validate a conversation id and return its archived transcript path."""
|
||
if not TRANSCRIPTS_DIR:
|
||
raise HTTPException(404, "transcripts not mounted")
|
||
rel = pathlib.PurePosixPath(id)
|
||
if rel.is_absolute() or ".." in rel.parts or rel.suffix != ".jsonl":
|
||
raise HTTPException(400, "invalid id")
|
||
ap = (TRANSCRIPTS_DIR / id).resolve()
|
||
try:
|
||
ap.relative_to(TRANSCRIPTS_DIR)
|
||
except ValueError:
|
||
raise HTTPException(400, "id escapes transcripts")
|
||
if not ap.is_file():
|
||
raise HTTPException(404, "not found")
|
||
return ap
|
||
|
||
|
||
def _summary_and_meta(id: str) -> tuple[str, dict, dict]:
|
||
"""(sessionId, summary, meta sidecar) for a conversation id."""
|
||
ap = _resolve_transcript(id)
|
||
summary = store.get_summary(str(ap)) or conversations.parse_conversation(ap, full=False)
|
||
sid = summary.get("sessionId") or ""
|
||
return sid, summary, meta_store.get(sid)
|
||
|
||
|
||
@app.get("/api/conversation-commits", responses=_r(schemas.ConversationCommits))
|
||
def conversation_commits(id: str):
|
||
"""The commits a conversation produced (cached), tagged with their repo."""
|
||
sid, summary, meta = _summary_and_meta(id)
|
||
data = gitdiff.conversation_commits(store, summary, meta, sid)
|
||
return {"id": id, **data}
|
||
|
||
|
||
@app.get("/api/commit-diff", responses=_r(schemas.DiffResult))
|
||
def commit_diff(repo: str, sha: str):
|
||
"""Structured per-file hunk diff of a single commit within a repo token."""
|
||
out = gitdiff.commit_diff(repo, sha)
|
||
if out is None:
|
||
raise HTTPException(404, "commit not found")
|
||
return out
|
||
|
||
|
||
# Content-type for image blobs served in the diff view (the only kind the diff
|
||
# viewer requests). Anything else falls back to octet-stream.
|
||
_IMAGE_CT = {
|
||
".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
|
||
".gif": "image/gif", ".webp": "image/webp", ".bmp": "image/bmp",
|
||
".avif": "image/avif", ".svg": "image/svg+xml", ".ico": "image/x-icon",
|
||
}
|
||
|
||
|
||
@app.get("/api/diff-blob")
|
||
def diff_blob(repo: str, path: str, sha: str | None = None, wt: str | None = None):
|
||
"""Raw bytes of a file inside a diff — the image side rendered in the viewer.
|
||
|
||
``repo`` is a diff repo token (``project:`` / ``service:`` / ``super:_``),
|
||
``path`` is repo-relative. With ``sha`` the file is read at that rev (e.g. the
|
||
commit's tip for an added image, or ``<sha>^`` for a deleted one); without it
|
||
the current working-tree copy is served (conversation / uncommitted diffs).
|
||
``wt`` names a worktree directory — the file is then read from that worktree
|
||
instead of the canonical checkout (see worktreediff)."""
|
||
data = (worktreediff.blob_bytes(wt, path) if wt
|
||
else gitdiff.blob_bytes(repo, path, sha=sha))
|
||
if data is None:
|
||
raise HTTPException(404, "blob not found")
|
||
ext = pathlib.PurePosixPath(path).suffix.lower()
|
||
ct = _IMAGE_CT.get(ext, "application/octet-stream")
|
||
# SVG can carry scripts; serving it inline as image/svg+xml is fine for <img>
|
||
# but nosniff + no inline-disposition keeps it from being treated as a page.
|
||
return Response(content=data, media_type=ct,
|
||
headers={"X-Content-Type-Options": "nosniff",
|
||
"Cache-Control": "public, max-age=3600"})
|
||
|
||
|
||
# Replaying a transcript's edits re-reads the whole file; during a live run the
|
||
# panel refetches on every SSE ping, so cache per (path, mtime, size, commits).
|
||
_udiff_cache: dict[str, tuple[tuple, dict]] = {}
|
||
_UDIFF_CACHE_CAP = 8
|
||
_udiff_lock = threading.Lock()
|
||
|
||
|
||
@app.get("/api/conversation-uncommitted-diff", responses=_r(schemas.DiffResult))
|
||
def conversation_uncommitted_diff(id: str, stats: bool = False):
|
||
"""The conversation's current not-yet-committed changes.
|
||
|
||
When it worked in a git worktree that still exists, this is a **real**
|
||
``git diff HEAD`` (+ untracked files) read from that worktree — see
|
||
worktreediff. Otherwise (or for edits made outside it) the changes are
|
||
replayed from the transcript's accumulated edits/creates/deletes (editsdiff).
|
||
``stats`` strips the hunks — the detail panel only needs paths and +/− counts.
|
||
"""
|
||
sid, summary, meta = _summary_and_meta(id)
|
||
ap = _resolve_transcript(id)
|
||
data = gitdiff.conversation_commits(store, summary, meta, sid)
|
||
wt_out = worktreediff.worktree_diff(summary, meta,
|
||
summary.get("title") or "", stats=stats)
|
||
try:
|
||
st = ap.stat()
|
||
sig = (st.st_mtime, st.st_size,
|
||
tuple(c.get("sha") for c in data["commits"]))
|
||
except OSError:
|
||
sig = None
|
||
out = None
|
||
with _udiff_lock:
|
||
hit = _udiff_cache.get(str(ap))
|
||
if hit and sig and hit[0] == sig:
|
||
out = hit[1]
|
||
if out is None:
|
||
out = editsdiff.uncommitted_diff(ap, summary.get("title") or "",
|
||
data["commits"])
|
||
if sig:
|
||
with _udiff_lock:
|
||
_udiff_cache[str(ap)] = (sig, out)
|
||
while len(_udiff_cache) > _UDIFF_CACHE_CAP:
|
||
_udiff_cache.pop(next(iter(_udiff_cache)))
|
||
if stats:
|
||
out = {**out, "files": [{**f, "hunks": []} for f in out["files"]]}
|
||
if wt_out is None:
|
||
return out
|
||
# A live worktree is authoritative for its own repo, so the replay only
|
||
# contributes what it saw *elsewhere* (e.g. a stray edit in the main
|
||
# checkout) — otherwise reverted or already-committed files would linger.
|
||
covered = {w["repo"] for w in wt_out["worktrees"]}
|
||
extra = [f for f in out["files"] if f.get("repo") not in covered]
|
||
if not extra:
|
||
return wt_out
|
||
files = wt_out["files"] + extra
|
||
wts = wt_out["worktrees"]
|
||
return {**wt_out, "files": files, "approx": True,
|
||
"additions": sum(f["additions"] for f in files),
|
||
"deletions": sum(f["deletions"] for f in files),
|
||
"subtitle": worktreediff.subtitle(wts, len(files), len(extra))}
|
||
|
||
|
||
@app.get("/api/conversation-diff", responses=_r(schemas.DiffResult))
|
||
def conversation_diff(id: str):
|
||
"""Aggregated net diff across all of a conversation's commits."""
|
||
sid, summary, meta = _summary_and_meta(id)
|
||
data = gitdiff.conversation_commits(store, summary, meta, sid)
|
||
by_repo: dict[str, list[str]] = {}
|
||
for c in data["commits"]:
|
||
by_repo.setdefault(c["repo"], []).append(c["sha"])
|
||
files: list[dict] = []
|
||
add = dele = 0
|
||
for token, shas in by_repo.items():
|
||
agg = gitdiff.aggregate_diff(token, shas, title="", subtitle="")
|
||
if agg:
|
||
files += agg["files"]
|
||
add += agg["additions"]
|
||
dele += agg["deletions"]
|
||
title = summary.get("title") or "Conversation changes"
|
||
n = len(data["commits"])
|
||
return {
|
||
"title": title,
|
||
"subtitle": f"{n} commit{'s' if n != 1 else ''} · {len(files)} file{'s' if len(files) != 1 else ''}",
|
||
"files": files, "additions": add, "deletions": dele,
|
||
"commits": data["commits"], "approx": True,
|
||
}
|
||
|
||
|
||
# ── conversation scaffold + published summary (see scaffold.py) ─────────────
|
||
# The algorithmic half of the `complete` skill: everything about a conversation
|
||
# can be aggregated rather than reasoned about, handed to the model as markdown
|
||
# with two sections left blank. `conv-scaffold` (scripts/) is the CLI wrapper.
|
||
|
||
@app.get("/api/conversations/{id:path}/scaffold",
|
||
response_class=PlainTextResponse)
|
||
def conversation_scaffold(id: str):
|
||
"""The completion scaffold for one conversation, as markdown."""
|
||
sid, _, meta = _summary_and_meta(id)
|
||
path = _resolve_transcript(id)
|
||
# A full parse: the merged tool sections need the thread, which the cached
|
||
# rollup summary deliberately doesn't carry.
|
||
full = conversations.parse_conversation(path, full=True)
|
||
return scaffold_mod.build(_conv_id(str(path)), sid, full, meta, store)
|
||
|
||
|
||
class SummaryBody(BaseModel):
|
||
markdown: str
|
||
|
||
|
||
@app.put("/api/conversations/{id:path}/summary", responses=_r(schemas.OkResponse))
|
||
def conversation_summary_put(id: str, body: SummaryBody):
|
||
"""Publish a filled-in scaffold as the conversation's summary. Rejects an
|
||
untouched one — an unedited scaffold is a report of nothing."""
|
||
sid, _, _ = _summary_and_meta(id)
|
||
md = (body.markdown or "").strip()
|
||
if not md:
|
||
raise HTTPException(400, "markdown is required")
|
||
if not scaffold_mod.is_filled(md):
|
||
raise HTTPException(422, "the Summary section is still the scaffold's "
|
||
"placeholder — fill it in before publishing")
|
||
now = datetime.datetime.now(datetime.timezone.utc) \
|
||
.strftime("%Y-%m-%dT%H:%M:%SZ")
|
||
meta_store.update(sid, {"summary": {
|
||
"markdown": scaffold_mod.clean_for_publish(md), "at": now}})
|
||
hub.publish({"type": "meta", "id": sid})
|
||
return {"ok": True}
|
||
|
||
|
||
# ── conversation action metadata (edited by skills, streamed to the UI) ──────
|
||
class MetaPatch(BaseModel):
|
||
id: str # Claude Code session id (or "<proj>/<sid>.jsonl")
|
||
# Display title the session chose for itself (`conv-meta title "…"`), used
|
||
# instead of the parsed one. Lets an agent spawned with a generic prompt
|
||
# rename its card once it knows what it's actually building.
|
||
title: str | None = None
|
||
projects: list[str] | str | None = None
|
||
services: list[str] | str | None = None
|
||
state: str | None = None
|
||
committed: str | None = None
|
||
pushed: str | None = None
|
||
merged: str | None = None
|
||
deployed: str | None = None
|
||
notified: str | None = None
|
||
notification: dict | None = None
|
||
worktree: dict | None = None # a single worktree create/remove event to fold in
|
||
archived: bool | None = None # hide/unhide from the default conversation list
|
||
|
||
|
||
def _meta_key(raw: str) -> str:
|
||
"""Accept a bare session id or a transcript id; store under the session id."""
|
||
name = pathlib.PurePosixPath(raw).name
|
||
return name[:-6] if name.endswith(".jsonl") else name
|
||
|
||
|
||
@app.get("/api/conversation-meta", responses=_r(schemas.ConversationMetaResponse))
|
||
def conversation_meta():
|
||
"""The raw action-metadata sidecar (keyed by session id)."""
|
||
return {"meta": meta_store.all()}
|
||
|
||
|
||
@app.post("/api/conversation-meta", responses=_r(schemas.MetaUpdateResult))
|
||
def update_conversation_meta(patch: MetaPatch):
|
||
body = patch.model_dump(exclude_none=True)
|
||
cid = _meta_key(body.pop("id"))
|
||
if not cid:
|
||
raise HTTPException(400, "missing id")
|
||
merged = meta_store.update(cid, body)
|
||
hub.publish({"type": "meta", "id": cid})
|
||
return {"id": cid, "meta": merged}
|
||
|
||
|
||
@app.post("/api/conversations/archive-old",
|
||
responses=_r(schemas.ArchiveOldResult))
|
||
def archive_old_conversations():
|
||
"""Archive (hide) every conversation that ended before the start of yesterday.
|
||
|
||
Bulk cleanup for the drawer: keeps today's and yesterday's transcripts
|
||
visible and folds everything older behind the "show archived" toggle."""
|
||
# endedAt is a UTC ISO timestamp; compare against a UTC start-of-yesterday so
|
||
# the day boundary lines up (lexicographic works at day granularity).
|
||
today = datetime.datetime.now(datetime.timezone.utc).date()
|
||
cutoff = datetime.datetime.combine(
|
||
today - datetime.timedelta(days=1), datetime.time.min).isoformat()
|
||
n = 0
|
||
for path, s in store.all_summaries():
|
||
if not s.get("messages") or s.get("isSidechain"):
|
||
continue
|
||
ended = s.get("endedAt") or ""
|
||
if ended and ended < cutoff:
|
||
sid = s.get("sessionId") or _meta_key(_conv_id(path))
|
||
if sid and not meta_store.get(sid).get("archived"):
|
||
meta_store.update(sid, {"archived": True})
|
||
n += 1
|
||
hub.publish({"type": "meta"})
|
||
return {"archived": n, "cutoff": cutoff}
|
||
|
||
|
||
# ── spawn / resume / interrupt `claude -p` sessions via the host sidecar ──────
|
||
def _sidecar_call(path: str, payload: dict) -> dict:
|
||
"""POST to the host sidecar (bearer auth) and return its JSON reply.
|
||
|
||
Raises 503 when the sidecar isn't configured, or 502 when it's unreachable
|
||
or answers with an error — so every proxy endpoint reports failures the
|
||
same way."""
|
||
if not SIDECAR_TOKEN:
|
||
raise HTTPException(503, "sidecar not configured (no SIDECAR_TOKEN)")
|
||
data = json.dumps(payload).encode()
|
||
req = urllib.request.Request(
|
||
f"{SIDECAR_URL}{path}", data=data, method="POST",
|
||
headers={"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {SIDECAR_TOKEN}"})
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=15) as r:
|
||
return json.loads(r.read().decode())
|
||
except urllib.error.HTTPError as e:
|
||
detail = e.read().decode(errors="replace")[:500]
|
||
raise HTTPException(502, f"sidecar error {e.code}: {detail}")
|
||
except (urllib.error.URLError, OSError, ValueError) as e:
|
||
raise HTTPException(502, f"sidecar unreachable: {e}")
|
||
|
||
|
||
def _valid_sid(raw: str) -> str:
|
||
sid = (raw or "").strip()
|
||
try:
|
||
uuidlib.UUID(sid)
|
||
except ValueError:
|
||
raise HTTPException(400, "sessionId must be a valid UUID")
|
||
return sid
|
||
|
||
|
||
# ── run-exit detection ────────────────────────────────────────────────────────
|
||
# The sidecar knows exactly when a run's process dies (its pidfiles), so the
|
||
# viewer doesn't have to guess "finished" from the DONE marker or wait out the
|
||
# recency window. A background thread polls the sidecar's live-session list and
|
||
# stamps a conversation finished the moment its run vanishes from it.
|
||
|
||
# Sessions with an in-flight /resume or /interrupt: those deliberately stop or
|
||
# replace the run, so the exit watcher must not race the vanish and stamp
|
||
# "finished" over the state the endpoint is about to write.
|
||
_sidecar_busy: set[str] = set()
|
||
|
||
|
||
def _sidecar_live_sids() -> set[str] | None:
|
||
"""Session ids with a live process behind them, per the sidecar's
|
||
``/sessions`` (which also reaps finished runs' pidfiles as it answers).
|
||
None = sidecar unconfigured/unreachable — the caller must treat that as
|
||
"unknown", never as "nothing is running"."""
|
||
if not SIDECAR_TOKEN:
|
||
return None
|
||
req = urllib.request.Request(
|
||
f"{SIDECAR_URL}/sessions",
|
||
headers={"Authorization": f"Bearer {SIDECAR_TOKEN}"})
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=3) as r:
|
||
out = json.loads(r.read().decode())
|
||
return {s["sessionId"] for s in out.get("sessions", [])
|
||
if s.get("sessionId")}
|
||
except (urllib.error.URLError, OSError, ValueError, KeyError):
|
||
return None
|
||
|
||
|
||
def _mark_run_exited(sid: str) -> None:
|
||
"""A sidecar-launched run's process is gone: stamp the conversation
|
||
``finished`` (with the exit time) unless an endpoint/skill already moved it
|
||
to another state. The transcript's own trailing markers still outrank the
|
||
stamp in ``_resolve_state`` (interrupted, DONE+notified, …), and
|
||
``exitedAt`` lets it be distrusted if the session later continues outside
|
||
the sidecar (see ``_outlived_exit``)."""
|
||
if meta_store.get(sid).get("state") not in (None, "running"):
|
||
return
|
||
now = datetime.datetime.now(datetime.timezone.utc) \
|
||
.strftime("%Y-%m-%dT%H:%M:%SZ")
|
||
meta_store.update(sid, {"state": "finished", "exitedAt": now})
|
||
hub.publish({"type": "meta", "id": sid})
|
||
|
||
|
||
def _watch_sidecar_runs(interval: float = 2.0) -> None:
|
||
"""Poll the sidecar's live-session list; a session id vanishing from it
|
||
means its process exited — flip the conversation to ``finished`` within a
|
||
tick instead of after the recency window.
|
||
|
||
The first successful poll only sets the baseline (a fresh backend can't
|
||
tell "exited while we were down" from "never sidecar-launched" — those
|
||
stay on the recency fallback). Ids with an in-flight resume/interrupt are
|
||
deferred a tick so the endpoint's own stamp wins; if the resume then
|
||
failed, the id is still absent next tick and gets flipped then."""
|
||
prev: set[str] | None = None
|
||
while True:
|
||
time.sleep(interval)
|
||
try:
|
||
cur = _sidecar_live_sids()
|
||
if cur is None:
|
||
continue # unknown — keep the baseline, flip nothing
|
||
if prev is not None:
|
||
for sid in prev - cur:
|
||
if sid in _sidecar_busy:
|
||
cur.add(sid) # decide next tick
|
||
continue
|
||
_mark_run_exited(sid)
|
||
prev = cur
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
@app.get("/api/models", responses=_r(schemas.ModelsResponse))
|
||
def list_models():
|
||
"""The models a session can be spawned/resumed on (a tag each in the composer).
|
||
|
||
Fetched from the Anthropic API and cached in-process (see models.py), so the
|
||
picker follows new releases without a redeploy. `default` is what a run uses
|
||
when no model tag is picked."""
|
||
return {"models": models_mod.list_models(),
|
||
"default": models_mod.DEFAULT_MODEL}
|
||
|
||
|
||
@app.get("/api/models/openrouter",
|
||
responses=_r(schemas.OpenRouterModelsResponse))
|
||
def list_openrouter_models():
|
||
"""Every model a pi.dev session can run on OpenRouter, cheapest first.
|
||
|
||
Not part of `/api/models` on purpose: that endpoint is the composer's short
|
||
pickable list (one chip per Claude family + the curated pi entries), while
|
||
this is a few hundred models the model browser searches through. Prices
|
||
($/Mtok in / out / cache), context window and parameter size all come from
|
||
APPE's catalogue — the homelab's model-data source of truth (openrouter.py).
|
||
"""
|
||
return {"models": openrouter_mod.catalogue(),
|
||
"source": openrouter_mod.CATALOGUE_URL,
|
||
"generatedAt": openrouter_mod.generated_at()}
|
||
|
||
|
||
class SpawnBody(BaseModel):
|
||
prompt: str
|
||
model: str | None = None
|
||
# Which agent harness runs the session: "claude" (the claude CLI, default)
|
||
# or "pi" (the pi.dev runner — see services/ai-agent/runner/).
|
||
harness: str | None = None
|
||
# Whether the session thinks. False ⇒ the sidecar runs the harness with
|
||
# extended thinking off (`--thinking disabled` / `--thinking off`); unset ⇒
|
||
# each CLI keeps its own default. Both harnesses support it.
|
||
thinking: bool | None = None
|
||
# How hard a claude run works a turn (`--effort`: low|medium|high|xhigh|max).
|
||
# Claude-only — it's what the composer shows instead of the thinking toggle
|
||
# for claude models; unset ⇒ the CLI's own default.
|
||
effort: str | None = None
|
||
|
||
|
||
class ResumeBody(BaseModel):
|
||
sessionId: str
|
||
prompt: str
|
||
model: str | None = None
|
||
cwd: str | None = None
|
||
harness: str | None = None # unset ⇒ the harness the session started on
|
||
thinking: bool | None = None # unset ⇒ what the session last ran with
|
||
effort: str | None = None # unset ⇒ what the session last ran with
|
||
|
||
|
||
def _send_message(prompt: str, *, session_id: str | None = None,
|
||
model: str | None = None, cwd: str | None = None,
|
||
harness: str | None = None,
|
||
thinking: bool | None = None,
|
||
effort: str | None = None) -> dict:
|
||
"""The one send path behind both composers.
|
||
|
||
With no ``session_id`` this starts a new conversation: we mint the UUID here
|
||
so the frontend can start watching for the transcript to sync (~2s) and
|
||
redirect to it without waiting for the run. With one, it continues that
|
||
session — and a continue *always* takes the prompt: if a run is still live
|
||
(or parked on a ScheduleWakeup/Monitor), the sidecar stops it first (SIGINT
|
||
the group, wait for it to die) and then launches the ``--resume`` with the
|
||
new turn, so sending into a "running"/"paused" conversation restarts it on
|
||
the new message. Either way the actual agent process (`claude -p` or the pi
|
||
runner, per ``harness``) is launched by the host sidecar (the container
|
||
can't reach the host's authenticated CLIs), and we stamp ``running``
|
||
immediately so the UI flips before the first turn has synced."""
|
||
prompt = (prompt or "").strip()
|
||
if not prompt:
|
||
raise HTTPException(400, "empty prompt")
|
||
if session_id:
|
||
sid = _valid_sid(session_id)
|
||
# A conversation resumes on the harness it started on — and with the
|
||
# thinking setting it last ran with — unless the caller overrides. Both
|
||
# were stamped into the metadata sidecar by the spawn.
|
||
prev = meta_store.get(sid)
|
||
harness = (harness or prev.get("harness") or "claude").lower()
|
||
if thinking is None:
|
||
thinking = prev.get("thinking")
|
||
if effort is None:
|
||
effort = prev.get("effort")
|
||
# A force-resume swaps the session's process; shield the swap from the
|
||
# exit watcher so the old run's vanish isn't stamped "finished" over
|
||
# the "running" written below.
|
||
_sidecar_busy.add(sid)
|
||
try:
|
||
out = _sidecar_call("/resume", {"sessionId": sid, "prompt": prompt,
|
||
"model": model, "cwd": cwd,
|
||
"harness": harness,
|
||
"thinking": thinking,
|
||
"effort": effort, "force": True})
|
||
if out.get("reused"):
|
||
# force=True makes the sidecar stop a live run instead of
|
||
# answering `reused`; only a sidecar predating the flag still
|
||
# can. The turn was dropped either way — a 409, not a success,
|
||
# so the composer says so instead of waiting on an answer that
|
||
# isn't coming.
|
||
meta_store.update(sid, {"state": "running"})
|
||
hub.publish({"type": "meta", "id": sid})
|
||
raise HTTPException(409, "this session is still running and "
|
||
"the sidecar is too old to stop it — "
|
||
"restart the claude-sidecar service")
|
||
# Stamp running (and re-stamp harness + thinking) before lifting the
|
||
# shield — the UI flips before the transcript syncs. A None thinking
|
||
# is dropped by the merge, so the session keeps what it had.
|
||
meta_store.update(sid, {"state": "running", "harness": harness,
|
||
"thinking": thinking, "effort": effort})
|
||
finally:
|
||
_sidecar_busy.discard(sid)
|
||
else:
|
||
sid = str(uuidlib.uuid4())
|
||
harness = (harness or "claude").lower()
|
||
out = _sidecar_call("/spawn", {"prompt": prompt, "sessionId": sid,
|
||
"model": model, "harness": harness,
|
||
"thinking": thinking, "effort": effort})
|
||
# A fresh id can't be in the exit watcher's baseline — no shield needed.
|
||
meta_store.update(sid, {"state": "running", "harness": harness,
|
||
"thinking": thinking, "effort": effort})
|
||
hub.publish({"type": "meta", "id": sid})
|
||
return {"sessionId": sid, "pid": out.get("pid")}
|
||
|
||
|
||
@app.post("/api/spawn", responses=_r(schemas.SpawnResult))
|
||
def spawn_conversation(body: SpawnBody):
|
||
"""Kick off a new headless agent session on the host (see _send_message)."""
|
||
return _send_message(body.prompt, model=body.model, harness=body.harness,
|
||
thinking=body.thinking, effort=body.effort)
|
||
|
||
|
||
@app.post("/api/resume", responses=_r(schemas.SpawnResult))
|
||
def resume_conversation(body: ResumeBody):
|
||
"""Continue an existing conversation with a new user turn (see _send_message).
|
||
|
||
Runs ``claude -p <prompt> --resume <sessionId>`` in the session's original
|
||
cwd; resume reuses the session id, so Claude appends the new turns to the
|
||
same transcript the viewer is already watching and the open conversation
|
||
streams the continuation live. A still-running session is stopped first —
|
||
sending a message always wins."""
|
||
return _send_message(body.prompt, session_id=body.sessionId,
|
||
model=body.model, cwd=body.cwd, harness=body.harness,
|
||
thinking=body.thinking, effort=body.effort)
|
||
|
||
|
||
class ForkBody(BaseModel):
|
||
id: str # source conversation id (<slug>/<sid>.jsonl)
|
||
mi: int # ordinal of the user message being edited (fork point)
|
||
prompt: str # the edited message — becomes the fork's next turn
|
||
model: str | None = None
|
||
harness: str | None = None # unset ⇒ the harness the source ran on
|
||
|
||
|
||
@app.post("/api/fork", responses=_r(schemas.SpawnResult))
|
||
def fork_conversation(body: ForkBody):
|
||
"""Branch a new conversation off an existing one at a user message.
|
||
|
||
The new session's history is the source transcript up to (excluding) user
|
||
message ``mi``; ``prompt`` (the edited message) is then sent as its next
|
||
turn. Neither CLI can truncate at a message (`claude --fork-session` only
|
||
clones whole sessions), so the host sidecar does transcript surgery: it
|
||
copies the first ``cutLine`` lines of the source session file under a
|
||
fresh session id and resumes that — see sidecar ``/fork``. ``cutText`` and
|
||
``cutUserOrd`` let the sidecar/runner verify they're slicing the same
|
||
message this viewer showed (the archive is an append-only copy of the
|
||
live file, so the line indices match; the canary catches drift)."""
|
||
prompt = (body.prompt or "").strip()
|
||
if not prompt:
|
||
raise HTTPException(400, "empty prompt")
|
||
ap = _resolve_transcript(body.id)
|
||
src_sid, summary, src_meta = _summary_and_meta(body.id)
|
||
if not src_sid:
|
||
raise HTTPException(409, "source conversation has no session id")
|
||
cut = conversations.find_cut_line(ap, body.mi)
|
||
if cut is None:
|
||
raise HTTPException(404, "fork point not found — message ordinal "
|
||
"missing or not a user message")
|
||
if cut["line"] == 0:
|
||
raise HTTPException(400, "nothing before the first message — start a "
|
||
"new conversation instead")
|
||
harness = (body.harness or src_meta.get("harness") or "claude").lower()
|
||
sid = str(uuidlib.uuid4())
|
||
out = _sidecar_call("/fork", {
|
||
"sessionId": sid, "srcSessionId": src_sid,
|
||
"cwd": summary.get("cwd"),
|
||
"cutLine": cut["line"], "cutText": cut["text"][:120],
|
||
"cutUserOrd": cut["userOrd"],
|
||
"prompt": prompt, "model": body.model, "harness": harness})
|
||
meta_store.update(sid, {
|
||
"state": "running", "harness": harness,
|
||
"forkedFrom": {"id": body.id, "sessionId": src_sid, "mi": body.mi}})
|
||
hub.publish({"type": "meta", "id": sid})
|
||
return {"sessionId": sid, "pid": out.get("pid")}
|
||
|
||
|
||
class InterruptBody(BaseModel):
|
||
sessionId: str
|
||
|
||
|
||
@app.post("/api/interrupt", responses=_r(schemas.InterruptResult))
|
||
def interrupt_conversation(body: InterruptBody):
|
||
"""Stop a running Claude session this viewer spawned.
|
||
|
||
Proxies to the host sidecar, which sends the run's process group a SIGINT —
|
||
exactly like pressing Esc in a terminal. Claude aborts the turn and records a
|
||
``[Request interrupted by user]`` marker in the transcript, which the parser
|
||
then surfaces as the ``interrupted`` state. We also stamp the metadata
|
||
immediately so the UI flips before that marker has synced."""
|
||
sid = _valid_sid(body.sessionId)
|
||
# Shield the kill from the exit watcher: the vanish it causes must resolve
|
||
# to the "interrupted" stamped below, not a generic "finished".
|
||
_sidecar_busy.add(sid)
|
||
try:
|
||
try:
|
||
out = _sidecar_call("/interrupt", {"sessionId": sid})
|
||
except HTTPException as e:
|
||
# 404 from the sidecar = no live process for this session (it
|
||
# already exited, or was started from a terminal) — surfaced as a
|
||
# 502 by _sidecar_call. A stale "running" badge over a dead run is
|
||
# exactly what the user is clearing here, so stamp it interrupted
|
||
# anyway.
|
||
if "sidecar error 404" not in str(e.detail):
|
||
raise
|
||
out = {"note": "no live process — marked interrupted"}
|
||
meta_store.update(sid, {"state": "interrupted"})
|
||
finally:
|
||
_sidecar_busy.discard(sid)
|
||
hub.publish({"type": "meta", "id": sid})
|
||
return {"sessionId": sid, "ok": True, **out}
|
||
|
||
|
||
# ── cron jobs: scheduled agent sessions (see cron.py) ────────────────────────
|
||
# sessionId → conversation id, for linking a job's run history to the viewer.
|
||
# Memoized on the summaries version (rebuilding walks ~900 summaries).
|
||
_sid_map_cache: tuple[int, dict] | None = None
|
||
_sid_map_lock = threading.Lock()
|
||
|
||
|
||
def _session_conv_map() -> dict:
|
||
global _sid_map_cache
|
||
v = store.summaries_version
|
||
with _sid_map_lock:
|
||
if _sid_map_cache and _sid_map_cache[0] == v:
|
||
return _sid_map_cache[1]
|
||
m: dict[str, str] = {}
|
||
for path, s in store.all_summaries():
|
||
sid = s.get("sessionId")
|
||
if sid and not _is_sidechain_path(path):
|
||
m[sid] = _conv_id(path)
|
||
with _sid_map_lock:
|
||
_sid_map_cache = (v, m)
|
||
return m
|
||
|
||
|
||
def _cron_job_out(job: dict) -> dict:
|
||
"""A stored job + the computed fields the UI shows: the next fire time, the
|
||
run config read out of its prompt file's frontmatter, and the run history
|
||
enriched with conversation ids (newest first)."""
|
||
out = dict(job)
|
||
out["nextRun"] = (cron_mod.next_run(job.get("schedule") or "")
|
||
if job.get("enabled") else None)
|
||
out["runConfig"] = cron_mod.run_config(_cron_prompt_text(job))
|
||
sid2conv = _session_conv_map()
|
||
out["runs"] = [{"at": at, "sessionId": sid,
|
||
"conversationId": sid2conv.get(sid)}
|
||
for at, sid in sorted((job.get("history") or {}).items(),
|
||
reverse=True)]
|
||
return out
|
||
|
||
|
||
def _cron_prompt_path(job: dict) -> tuple[pathlib.Path, str]:
|
||
rel = cron_mod.valid_prompt_file(job.get("promptFile") or "")
|
||
return (WORKSPACE / rel), rel
|
||
|
||
|
||
def _cron_prompt_text(job: dict) -> str:
|
||
"""The job's prompt file verbatim (frontmatter included), or ``""`` if it
|
||
isn't readable — a missing file is reported when the job fires, not here."""
|
||
try:
|
||
ap, _ = _cron_prompt_path(job)
|
||
return ap.read_text(encoding="utf-8")
|
||
except (OSError, ValueError):
|
||
return ""
|
||
|
||
|
||
def _fire_cron_job(job: dict) -> dict:
|
||
"""Run a cron job once: read its prompt file, spawn a session on the
|
||
harness/model/parameters its frontmatter declares, record the run in the
|
||
job's history and tag the conversation with the job."""
|
||
_, rel = _cron_prompt_path(job)
|
||
raw = _cron_prompt_text(job)
|
||
cfg = cron_mod.run_config(raw)
|
||
prompt = cron_mod.strip_frontmatter(raw).strip()
|
||
if not prompt:
|
||
cron_store.record_run(job["id"], "",
|
||
status=f"error: prompt file {rel} missing/empty")
|
||
hub.publish({"type": "cron"})
|
||
raise HTTPException(409, f"prompt file {rel} is missing or empty")
|
||
# Frontmatter wins; the store's `model` is the pre-frontmatter fallback.
|
||
out = _send_message(prompt, model=cfg["model"] or job.get("model"),
|
||
harness=cfg["harness"], thinking=cfg["thinking"],
|
||
effort=cfg["effort"])
|
||
sid = out["sessionId"]
|
||
meta_store.update(sid, {"cron": {"id": job["id"],
|
||
"name": job.get("name")}})
|
||
cron_store.record_run(job["id"], sid, status="ok")
|
||
hub.publish({"type": "cron"})
|
||
hub.publish({"type": "meta", "id": sid})
|
||
return out
|
||
|
||
|
||
def _scheduler_fire(job: dict) -> None:
|
||
"""The scheduler-thread wrapper: outcomes land in the store (surfaced as
|
||
the job's lastStatus), never raise into the loop."""
|
||
try:
|
||
_fire_cron_job(job)
|
||
except HTTPException as e:
|
||
cron_store.record_run(job["id"], "", status=f"error: {e.detail}")
|
||
hub.publish({"type": "cron"})
|
||
|
||
|
||
cron_scheduler = cron_mod.CronScheduler(cron_store, _scheduler_fire)
|
||
|
||
|
||
class CronCreateBody(BaseModel):
|
||
name: str
|
||
schedule: str
|
||
promptFile: str | None = None # default: .claude/agents/<slug>.md
|
||
enabled: bool = True
|
||
model: str | None = None
|
||
|
||
|
||
class CronPatchBody(BaseModel):
|
||
name: str | None = None
|
||
schedule: str | None = None
|
||
promptFile: str | None = None
|
||
enabled: bool | None = None
|
||
model: str | None = None
|
||
|
||
|
||
class CronPromptBody(BaseModel):
|
||
content: str
|
||
|
||
|
||
@app.get("/api/cron", responses=_r(schemas.CronJobsResponse))
|
||
def cron_jobs():
|
||
"""Every cron job, with its computed next fire time and run history."""
|
||
return {"jobs": [_cron_job_out(j) for j in cron_store.list()]}
|
||
|
||
|
||
@app.post("/api/cron", responses=_r(schemas.CronJob))
|
||
def cron_create(body: CronCreateBody):
|
||
"""Create a job. The prompt file is created (a stub carrying the default
|
||
run-config frontmatter) if it doesn't exist yet, so the editor always opens
|
||
on something and the harness/model/effort keys are there to edit."""
|
||
name = (body.name or "").strip()
|
||
if not name:
|
||
raise HTTPException(400, "name is required")
|
||
prompt_file = body.promptFile or \
|
||
f"{cron_mod.PROMPT_DIR}/{cron_mod.slugify(name)}.md"
|
||
try:
|
||
job = cron_store.create(name, body.schedule, prompt_file,
|
||
enabled=body.enabled, model=body.model)
|
||
except ValueError as e:
|
||
raise HTTPException(400, str(e))
|
||
ap, _ = _cron_prompt_path(job)
|
||
if not ap.exists():
|
||
try:
|
||
ap.parent.mkdir(parents=True, exist_ok=True)
|
||
ap.write_text(
|
||
cron_mod.frontmatter_block(cron_mod.DEFAULT_RUN_CONFIG)
|
||
+ f"\n# {name}\n\n", encoding="utf-8")
|
||
indexer.scan_files(force_walk=True)
|
||
except OSError:
|
||
pass # surfaced as missing/empty when the job fires
|
||
hub.publish({"type": "cron"})
|
||
return _cron_job_out(job)
|
||
|
||
|
||
@app.put("/api/cron/{job_id}", responses=_r(schemas.CronJob))
|
||
def cron_update(job_id: str, body: CronPatchBody):
|
||
try:
|
||
job = cron_store.update(job_id, body.model_dump())
|
||
except ValueError as e:
|
||
raise HTTPException(400, str(e))
|
||
if not job:
|
||
raise HTTPException(404, "cron job not found")
|
||
hub.publish({"type": "cron"})
|
||
return _cron_job_out(job)
|
||
|
||
|
||
@app.delete("/api/cron/{job_id}", responses=_r(schemas.OkResponse))
|
||
def cron_delete(job_id: str):
|
||
"""Remove a job. Its prompt file is left in place (it's repo content)."""
|
||
if not cron_store.delete(job_id):
|
||
raise HTTPException(404, "cron job not found")
|
||
hub.publish({"type": "cron"})
|
||
return {"ok": True}
|
||
|
||
|
||
@app.post("/api/cron/{job_id}/run", responses=_r(schemas.SpawnResult))
|
||
def cron_run_now(job_id: str):
|
||
"""Fire a job immediately (manual run — works even when disabled)."""
|
||
job = cron_store.get(job_id)
|
||
if not job:
|
||
raise HTTPException(404, "cron job not found")
|
||
return _fire_cron_job(job)
|
||
|
||
|
||
@app.get("/api/cron/{job_id}/prompt", responses=_r(schemas.CronPromptFile))
|
||
def cron_prompt(job_id: str):
|
||
"""The job's prompt file content (the editable text in Settings → Cron)."""
|
||
job = cron_store.get(job_id)
|
||
if not job:
|
||
raise HTTPException(404, "cron job not found")
|
||
ap, rel = _cron_prompt_path(job)
|
||
try:
|
||
return {"path": rel, "content": ap.read_text(encoding="utf-8"),
|
||
"exists": True}
|
||
except OSError:
|
||
return {"path": rel, "content": "", "exists": False}
|
||
|
||
|
||
@app.put("/api/cron/{job_id}/prompt", responses=_r(schemas.CronPromptFile))
|
||
def cron_prompt_save(job_id: str, body: CronPromptBody):
|
||
job = cron_store.get(job_id)
|
||
if not job:
|
||
raise HTTPException(404, "cron job not found")
|
||
ap, rel = _cron_prompt_path(job)
|
||
try:
|
||
ap.parent.mkdir(parents=True, exist_ok=True)
|
||
ap.write_text(body.content, encoding="utf-8")
|
||
except OSError as e:
|
||
raise HTTPException(500, f"could not write {rel}: {e}")
|
||
indexer.scan_files(force_walk=True)
|
||
hub.publish({"type": "cron"})
|
||
return {"path": rel, "content": body.content, "exists": True}
|
||
|
||
|
||
# ── agents catalog ──────────────────────────────────────────────────────────
|
||
# `agents.py` owns the disk scan and the arithmetic; assembling the run list
|
||
# needs the summary store, the metadata sidecar and the cron store, so it lives
|
||
# here — the same split the skills catalog uses. Sits after the cron section
|
||
# because a scheduled agent's runs come out of `cron_store`.
|
||
_agent_runs_cache: tuple[tuple[int, int], list[dict]] | None = None
|
||
_agent_runs_lock = threading.Lock()
|
||
|
||
|
||
def _build_agent_runs() -> list[dict]:
|
||
"""Every recorded run of an agent, newest first, in one flat shape.
|
||
|
||
Two very different things are being unified here (see `agents.py`): a
|
||
**subagent** run, which is a sidechain transcript whose `agent-*.meta.json`
|
||
names the agent type, and a **session** run, which is an ordinary top-level
|
||
conversation that happened to be spawned *from* an agent file by cron or by
|
||
the agent page's "Run now". Both carry a real cost, because both are whole
|
||
transcripts — that's the whole point of measuring agents rather than skills.
|
||
|
||
A subagent run's `conversationId` is its **parent** (where its Task card
|
||
lives, and the unit "used in N conversations" counts); `subConversationId`
|
||
opens the child transcript itself. For a session run the two are the same
|
||
conversation, so only the former is set.
|
||
"""
|
||
by_file = {a["path"]: a["name"] for a in agents_mod.catalog()}
|
||
runs: list[dict] = []
|
||
sessions: dict[str, tuple[str, dict]] = {}
|
||
|
||
for path, s in store.all_summaries():
|
||
if _is_sidechain_path(path):
|
||
name = _subagent_meta_cached(pathlib.Path(path)).get("agentType")
|
||
if not name:
|
||
continue # a Task call with no resolved type (pre-meta.json)
|
||
parent = _parent_path_of(path)
|
||
runs.append({
|
||
"agent": name,
|
||
"origin": "task",
|
||
"at": s.get("endedAt") or s.get("startedAt"),
|
||
"cost": s.get("cost") or 0.0,
|
||
"tokens": s.get("tokens") or 0,
|
||
"model": s.get("model"),
|
||
"title": s.get("title"),
|
||
"conversationId": _conv_id(parent) if parent else None,
|
||
"subConversationId": _conv_id(path),
|
||
})
|
||
elif s.get("sessionId"):
|
||
sessions[s["sessionId"]] = (path, s)
|
||
|
||
def session_run(name: str, origin: str, sid: str, at: str | None) -> dict:
|
||
"""A top-level run, filled in from its transcript when one has synced —
|
||
a just-spawned session has none yet, so it shows up on the recorded
|
||
timestamp with zero spend rather than not at all."""
|
||
path, s = sessions.get(sid, (None, {}))
|
||
return {
|
||
"agent": name,
|
||
"origin": origin,
|
||
"at": s.get("endedAt") or s.get("startedAt") or at,
|
||
"cost": s.get("cost") or 0.0,
|
||
"tokens": s.get("tokens") or 0,
|
||
"model": s.get("model"),
|
||
"title": s.get("title"),
|
||
"conversationId": _conv_id(path) if path else None,
|
||
"subConversationId": None,
|
||
}
|
||
|
||
for job in cron_store.list():
|
||
name = by_file.get((job.get("promptFile") or "").lstrip("/"))
|
||
if not name:
|
||
continue # a job whose prompt file isn't an agent definition
|
||
for at, sid in (job.get("history") or {}).items():
|
||
if sid:
|
||
runs.append(session_run(name, "cron", sid, at))
|
||
|
||
for sid, m in meta_store.all().items():
|
||
ar = m.get("agentRun") or {}
|
||
if ar.get("name"):
|
||
runs.append(session_run(ar["name"], "manual", sid, ar.get("at")))
|
||
|
||
runs.sort(key=lambda r: r["at"] or "", reverse=True)
|
||
return runs
|
||
|
||
|
||
def _agent_runs() -> list[dict]:
|
||
"""Memoized `_build_agent_runs` — a full pass walks every summary, so the
|
||
list/detail pages (which the SSE bus refetches on every transcript write)
|
||
share one build per store version."""
|
||
global _agent_runs_cache
|
||
key = (store.summaries_version, meta_store.version)
|
||
with _agent_runs_lock:
|
||
if _agent_runs_cache and _agent_runs_cache[0] == key:
|
||
return _agent_runs_cache[1]
|
||
out = _build_agent_runs()
|
||
with _agent_runs_lock:
|
||
_agent_runs_cache = (key, out)
|
||
return out
|
||
|
||
|
||
def _agent_card(entry: dict | None, name: str, usage: dict, files: dict) -> dict:
|
||
"""One agent card: its disk record (or a built-in stub) + its run stats.
|
||
|
||
A built-in type (`Explore`, `general-purpose`, `Plan`, …) has no file in
|
||
this repo but has still earned its runs, so — exactly like a built-in skill
|
||
— it appears with `sourceKind: "builtin"` and no editor link.
|
||
"""
|
||
base = entry or {
|
||
"name": name, "title": name, "description": "",
|
||
"path": None, "dir": None,
|
||
"source": "builtin", "sourceKind": "builtin",
|
||
"tools": [], "bytes": 0, "updatedAt": None,
|
||
"harness": None, "model": None, "effort": None, "thinking": None,
|
||
}
|
||
u = usage.get(name) or {}
|
||
row = files.get(base.get("path") or "")
|
||
return {
|
||
**base,
|
||
"runs": u.get("runs", 0),
|
||
"taskRuns": u.get("taskRuns", 0),
|
||
"sessionRuns": u.get("sessionRuns", 0),
|
||
"lastRun": u.get("lastRun"),
|
||
"conversations": u.get("conversations", 0),
|
||
# Real spend: an agent run is a whole conversation, so unlike a skill's
|
||
# `estSpend` this is measured, not context-size × calls.
|
||
"cost": u.get("cost", 0.0),
|
||
"tokens": u.get("tokens", 0),
|
||
"avgCost": u.get("avgCost", 0.0),
|
||
# What the definition itself costs to load into a session.
|
||
"contextTokens": row["tokens"] if row else None,
|
||
"contextCost": row["cost"] if row else None,
|
||
}
|
||
|
||
|
||
@app.get("/api/agents", responses=_r(schemas.AgentsResponse))
|
||
def agents_list():
|
||
"""Card metadata + run/cost stats for every agent, last-run first.
|
||
|
||
The **union** of the agent definitions on disk and the types the transcripts
|
||
saw run — same shape as `/api/skills`, but the money here is real spend.
|
||
"""
|
||
runs = _agent_runs()
|
||
usage = agents_mod.rollup(runs)
|
||
files = store.all_files()
|
||
|
||
items = [_agent_card(a, a["name"], usage, files)
|
||
for a in agents_mod.catalog()]
|
||
known = {a["name"] for a in items}
|
||
for name in sorted(usage.keys() - known):
|
||
items.append(_agent_card(None, name, usage, files))
|
||
|
||
items.sort(key=lambda a: (a["lastRun"] or "", a["name"]), reverse=True)
|
||
|
||
return {
|
||
"agents": items,
|
||
"totals": {
|
||
"agents": len(items),
|
||
"used": sum(1 for a in items if a["runs"] > 0),
|
||
"runs": sum(a["runs"] for a in items),
|
||
"cost": sum(a["cost"] for a in items),
|
||
"tokens": sum(a["tokens"] for a in items),
|
||
},
|
||
}
|
||
|
||
|
||
def _agent_entry(name: str) -> dict | None:
|
||
return next((a for a in agents_mod.catalog() if a["name"] == name), None)
|
||
|
||
|
||
def _agent_cron_job(path: str | None) -> dict | None:
|
||
"""The cron job scheduled on this agent's file, if one exists — what turns
|
||
the detail page's "Schedule" button into a "next run in 3h" line."""
|
||
if not path:
|
||
return None
|
||
for job in cron_store.list():
|
||
if (job.get("promptFile") or "").lstrip("/") == path:
|
||
return _cron_job_out(job)
|
||
return None
|
||
|
||
|
||
# How many runs the detail page's history list shows. The rollup counts them
|
||
# all; only the tail is worth shipping.
|
||
AGENT_RUN_LIMIT = 40
|
||
|
||
|
||
@app.get("/api/agents/{name}", responses=_r(schemas.AgentDetail))
|
||
def agent_detail(name: str):
|
||
"""One agent: its definition, its stats, its recent runs and its schedule.
|
||
|
||
404s only when the name is neither on disk nor in any transcript — an agent
|
||
that exists but has never run is a perfectly good (empty) page.
|
||
"""
|
||
entry = _agent_entry(name)
|
||
runs = [r for r in _agent_runs() if r["agent"] == name]
|
||
if entry is None and not runs:
|
||
raise HTTPException(404, "agent not found")
|
||
|
||
usage = agents_mod.rollup(runs)
|
||
card = _agent_card(entry, name, usage, store.all_files())
|
||
|
||
content = ""
|
||
if entry:
|
||
try:
|
||
content = (WORKSPACE / entry["path"]).read_text(
|
||
encoding="utf-8", errors="replace")
|
||
except OSError:
|
||
content = ""
|
||
|
||
# Schedulable = cron would accept this path as a prompt file. Nested
|
||
# definitions (`.claude/agents/hooks/post-task.md`) and project-local ones
|
||
# aren't, so the UI can say why instead of offering a button that 400s.
|
||
can_schedule = False
|
||
if entry:
|
||
try:
|
||
cron_mod.valid_prompt_file(entry["path"])
|
||
can_schedule = True
|
||
except ValueError:
|
||
can_schedule = False
|
||
|
||
return {
|
||
"agent": card,
|
||
"content": content,
|
||
"runs": runs[:AGENT_RUN_LIMIT],
|
||
"daily": agents_mod.daily(runs),
|
||
"cron": _agent_cron_job(entry["path"] if entry else None),
|
||
"canSchedule": can_schedule,
|
||
}
|
||
|
||
|
||
class AgentRunBody(BaseModel):
|
||
# An extra instruction appended under the definition — the "what should it
|
||
# do this time?" box on the agent page. Empty ⇒ run the definition as-is,
|
||
# exactly like a cron firing does.
|
||
prompt: str | None = None
|
||
|
||
|
||
@app.post("/api/agents/{name}/run", responses=_r(schemas.SpawnResult))
|
||
def agent_run(name: str, body: AgentRunBody):
|
||
"""Run an agent now, as a top-level session — the agent page's "Run now".
|
||
|
||
Same machinery as a cron firing (`_fire_cron_job`): the definition's body is
|
||
the prompt and its frontmatter picks the harness/model/effort, so a "Run
|
||
now" and a scheduled run are the same run. The session is stamped
|
||
`agentRun` in the metadata sidecar, which is both what puts it in this
|
||
agent's history and what makes the conversation card say where it came from.
|
||
"""
|
||
entry = _agent_entry(name)
|
||
if entry is None:
|
||
raise HTTPException(404, "agent not found")
|
||
try:
|
||
raw = (WORKSPACE / entry["path"]).read_text(encoding="utf-8")
|
||
except OSError:
|
||
raise HTTPException(409, f"could not read {entry['path']}")
|
||
cfg = cron_mod.run_config(raw)
|
||
prompt = cron_mod.strip_frontmatter(raw).strip()
|
||
if not prompt:
|
||
raise HTTPException(409, f"{entry['path']} has no prompt body")
|
||
extra = (body.prompt or "").strip()
|
||
if extra:
|
||
prompt = f"{prompt}\n\n---\n\n**This run's task:** {extra}"
|
||
|
||
out = _send_message(prompt, model=cfg["model"], harness=cfg["harness"],
|
||
thinking=cfg["thinking"], effort=cfg["effort"])
|
||
sid = out["sessionId"]
|
||
meta_store.update(sid, {"agentRun": {"name": name,
|
||
"at": notify_mod._now_iso()}})
|
||
hub.publish({"type": "meta", "id": sid})
|
||
return out
|
||
|
||
|
||
# Claude Code's own hooks, read out of the workspace's .claude/settings.json.
|
||
# Read-only on purpose: writing that file from the container is host-level code
|
||
# execution (see the trusted-caller note at the top of this module), so the
|
||
# page shows them and points at the repo for edits.
|
||
CLAUDE_SETTINGS_REL = ".claude/settings.json"
|
||
|
||
|
||
@app.get("/api/claude-hooks", responses=_r(schemas.ClaudeHooksResponse))
|
||
def claude_hooks():
|
||
"""The native Claude Code hooks registered in the workspace repo, flattened
|
||
to one entry per (event, matcher, command). A missing or unparseable
|
||
settings file is an empty list, not an error (standalone has no repo)."""
|
||
ap = WORKSPACE / CLAUDE_SETTINGS_REL
|
||
try:
|
||
data = json.loads(ap.read_text(encoding="utf-8")) or {}
|
||
except (OSError, ValueError):
|
||
return {"path": CLAUDE_SETTINGS_REL, "exists": ap.exists(),
|
||
"entries": []}
|
||
entries = []
|
||
blocks = data.get("hooks")
|
||
for event, matchers in (blocks or {}).items():
|
||
for m in (matchers if isinstance(matchers, list) else []):
|
||
if not isinstance(m, dict):
|
||
continue
|
||
for h in (m.get("hooks") or []):
|
||
if not isinstance(h, dict):
|
||
continue
|
||
entries.append({"event": str(event),
|
||
"matcher": m.get("matcher") or None,
|
||
"type": h.get("type") or "command",
|
||
"command": h.get("command") or ""})
|
||
return {"path": CLAUDE_SETTINGS_REL, "exists": True, "entries": entries}
|
||
|
||
|
||
|
||
# ── first-class notifications: webhooks + notify/ask (see notify.py) ─────────
|
||
# This backend is the notification source of truth: the notify-done skill
|
||
# POSTs pushes and asks here; they are recorded in /data/notifications.json and
|
||
# forwarded to the configured webhooks — normally the Home Assistant webhook
|
||
# registered by the `ai_agent` custom integration (integrations/home-assistant/),
|
||
# which renders the phone notification and POSTs ask answers back.
|
||
class WebhooksBody(BaseModel):
|
||
webhooks: list[dict]
|
||
|
||
|
||
class NotifyBody(BaseModel):
|
||
title: str
|
||
message: str
|
||
type: str | None = None
|
||
url: str | None = None
|
||
image: str | None = None
|
||
icon: str | None = None
|
||
color: str | None = None
|
||
channel: str | None = None
|
||
importance: str | None = None
|
||
vibrationPattern: str | None = None
|
||
tag: str | None = None
|
||
persistent: bool | None = None
|
||
actions: list[dict] | None = None
|
||
# A phone-only line to read aloud (the phone webhook's TTS prefers it over
|
||
# title/message). Forwarded verbatim to webhooks; ignored by the HA screen
|
||
# notification. See services/phone/bridge/app.py.
|
||
spoken: str | None = None
|
||
sessionId: str | None = None
|
||
|
||
|
||
class AskBody(BaseModel):
|
||
question: str
|
||
options: list[str]
|
||
type: str | None = None
|
||
url: str | None = None
|
||
image: str | None = None
|
||
icon: str | None = None
|
||
color: str | None = None
|
||
timeoutSecs: int | None = None
|
||
sessionId: str | None = None
|
||
|
||
|
||
class AskAnswerBody(BaseModel):
|
||
index: int | None = None
|
||
label: str | None = None
|
||
|
||
|
||
@app.get("/api/webhooks", responses=_r(schemas.WebhooksResponse))
|
||
def webhooks_get():
|
||
return {"webhooks": notify_store.webhooks()}
|
||
|
||
|
||
@app.put("/api/webhooks", responses=_r(schemas.WebhooksResponse))
|
||
def webhooks_put(body: WebhooksBody):
|
||
try:
|
||
saved = notify_store.set_webhooks(body.webhooks)
|
||
except ValueError as e:
|
||
raise HTTPException(400, str(e))
|
||
hub.publish({"type": "webhooks"})
|
||
return {"webhooks": saved}
|
||
|
||
|
||
@app.post("/api/webhooks/{webhook_id}/test", responses=_r(schemas.NotifyResult))
|
||
def webhook_test(webhook_id: str):
|
||
"""Send a test push through one webhook (regardless of enabled/events)."""
|
||
target = next((w for w in notify_store.webhooks()
|
||
if w.get("id") == webhook_id), None)
|
||
if not target:
|
||
raise HTTPException(404, "unknown webhook")
|
||
nid = "ntf_" + uuidlib.uuid4().hex[:12]
|
||
delivered = notify_mod.forward([target], {
|
||
"event": "notify", "id": nid, "title": "ai-agent test",
|
||
"message": f"Webhook '{target['name']}' is wired up.",
|
||
"type": "success", "url": "https://ai-agent.lab.gabvdl.xyz"})
|
||
return {"ok": all(d["ok"] for d in delivered), "id": nid,
|
||
"delivered": delivered}
|
||
|
||
|
||
@app.post("/api/notify", responses=_r(schemas.NotifyResult))
|
||
def notify_push(body: NotifyBody):
|
||
"""Record a push notification and forward it to the subscribed webhooks."""
|
||
nid = "ntf_" + uuidlib.uuid4().hex[:12]
|
||
payload = {"event": "notify", "id": nid,
|
||
**{k: v for k, v in body.model_dump().items()
|
||
if v not in (None, "")}}
|
||
payload.pop("sessionId", None)
|
||
targets = notify_store.targets("notify")
|
||
delivered = notify_mod.forward(targets, payload)
|
||
notify_store.record({
|
||
"id": nid, "kind": "notify", "title": body.title,
|
||
"body": body.message, "type": body.type or "", "url": body.url or "",
|
||
"at": notify_mod._now_iso(), "sessionId": body.sessionId or "",
|
||
"delivered": delivered,
|
||
# Kept (not just forwarded) so the UI can replay what the phone said.
|
||
"spoken": body.spoken or "",
|
||
})
|
||
hub.publish({"type": "notification", "id": nid})
|
||
return {"ok": bool(delivered) and all(d["ok"] for d in delivered),
|
||
"id": nid, "delivered": delivered}
|
||
|
||
|
||
@app.post("/api/ask", responses=_r(schemas.NotifyResult))
|
||
def ask_push(body: AskBody):
|
||
"""Create an ask (choice question), forward it to the webhooks, return its
|
||
id. The HA integration renders the tappable notification and POSTs the
|
||
answer back to /api/ask/{id}/answer; callers long-poll GET /api/ask/{id}."""
|
||
options = [o.strip() for o in (body.options or []) if o and o.strip()]
|
||
if not body.question.strip() or not 2 <= len(options) <= 3:
|
||
raise HTTPException(400, "need a question and 2-3 options")
|
||
ask = notify_store.create_ask(
|
||
body.question.strip(), options, type_=body.type or "",
|
||
url=body.url or "", session_id=body.sessionId or "",
|
||
timeout_s=body.timeoutSecs or notify_mod.DEFAULT_ASK_TIMEOUT_S)
|
||
payload = {"event": "ask", "id": ask["id"], "question": ask["body"],
|
||
"options": options,
|
||
**{k: v for k, v in body.model_dump().items()
|
||
if k in ("type", "url", "image", "icon", "color")
|
||
and v not in (None, "")}}
|
||
targets = notify_store.targets("ask")
|
||
delivered = notify_mod.forward(targets, payload)
|
||
notify_store.set_delivered(ask["id"], delivered)
|
||
hub.publish({"type": "notification", "id": ask["id"]})
|
||
return {"ok": bool(delivered) and all(d["ok"] for d in delivered),
|
||
"id": ask["id"], "delivered": delivered}
|
||
|
||
|
||
@app.get("/api/ask/{ask_id}", responses=_r(schemas.AskRecord))
|
||
def ask_get(ask_id: str, waitSecs: float = 0):
|
||
"""The ask's current state; ``waitSecs`` long-polls for the answer."""
|
||
ask = (notify_store.wait_for_answer(ask_id, waitSecs) if waitSecs > 0
|
||
else notify_store.ask_status(ask_id))
|
||
if not ask:
|
||
raise HTTPException(404, "unknown ask")
|
||
return ask
|
||
|
||
|
||
@app.post("/api/ask/{ask_id}/answer", responses=_r(schemas.AskRecord))
|
||
def ask_answer(ask_id: str, body: AskAnswerBody):
|
||
"""Callback for the HA integration: record which button was tapped."""
|
||
ask = notify_store.answer_ask(ask_id, body.index, body.label)
|
||
if not ask:
|
||
raise HTTPException(404, "unknown ask")
|
||
hub.publish({"type": "notification", "id": ask_id})
|
||
return ask
|
||
|
||
|
||
@app.get("/api/notify-log", responses=_r(schemas.NotifyLogResponse))
|
||
def notify_log(limit: int = 200):
|
||
"""The first-class notification/ask log, newest first."""
|
||
entries = notify_store.log(limit=limit)
|
||
return {"notifications": entries, "count": len(entries)}
|
||
|
||
|
||
# ── forms: rich structured questions answered in the PWA (forms.py) ──────────
|
||
class FormCreateBody(BaseModel):
|
||
title: str
|
||
fields: list[dict]
|
||
description: str | None = None
|
||
type: str | None = None
|
||
url: str | None = None
|
||
sessionId: str | None = None
|
||
|
||
|
||
class FormSubmitBody(BaseModel):
|
||
answers: dict
|
||
|
||
|
||
@app.post("/api/forms", responses=_r(schemas.FormRecord))
|
||
def form_create(body: FormCreateBody):
|
||
"""Create a form (the ask-form skill's entry point) and return it.
|
||
|
||
The caller then sends its own push notification linking to the
|
||
conversation with ``?form=<id>`` and long-polls GET /api/forms/{id}."""
|
||
if not body.title.strip():
|
||
raise HTTPException(400, "need a title")
|
||
try:
|
||
fields = forms_mod.valid_fields(body.fields)
|
||
except ValueError as e:
|
||
raise HTTPException(400, str(e))
|
||
form = form_store.create(
|
||
body.title.strip(), fields, description=(body.description or "").strip(),
|
||
type_=body.type or "", url=body.url or "",
|
||
session_id=body.sessionId or "")
|
||
hub.publish({"type": "form", "id": form["id"],
|
||
"sessionId": form["sessionId"]})
|
||
return form
|
||
|
||
|
||
@app.get("/api/forms", responses=_r(schemas.FormsResponse))
|
||
def forms_list(session: str = "", limit: int = 100):
|
||
"""Forms, newest first; ``session`` filters to one conversation."""
|
||
items = form_store.list(session_id=session or None, limit=limit)
|
||
return {"forms": items, "count": len(items)}
|
||
|
||
|
||
@app.get("/api/forms/{form_id}", responses=_r(schemas.FormRecord))
|
||
def form_get(form_id: str, waitSecs: float = 0):
|
||
"""The form's current state; ``waitSecs`` long-polls for a submit."""
|
||
form = (form_store.wait(form_id, waitSecs) if waitSecs > 0
|
||
else form_store.get(form_id))
|
||
if not form:
|
||
raise HTTPException(404, "unknown form")
|
||
return form
|
||
|
||
|
||
@app.post("/api/forms/{form_id}/submit", responses=_r(schemas.FormRecord))
|
||
def form_submit(form_id: str, body: FormSubmitBody):
|
||
"""Record the user's answers (from the PWA's form card / modal)."""
|
||
try:
|
||
form = form_store.submit(form_id, body.answers)
|
||
except ValueError as e:
|
||
raise HTTPException(400, str(e))
|
||
if not form:
|
||
raise HTTPException(404, "unknown form")
|
||
hub.publish({"type": "form", "id": form_id,
|
||
"sessionId": form.get("sessionId") or ""})
|
||
return form
|
||
|
||
|
||
@app.post("/api/forms/{form_id}/cancel", responses=_r(schemas.FormRecord))
|
||
def form_cancel(form_id: str):
|
||
"""Dismiss a form unanswered (the card's Cancel button, or the agent
|
||
giving up on the wait)."""
|
||
form = form_store.cancel(form_id)
|
||
if not form:
|
||
raise HTTPException(404, "unknown form")
|
||
hub.publish({"type": "form", "id": form_id,
|
||
"sessionId": form.get("sessionId") or ""})
|
||
return form
|
||
|
||
|
||
# ── projects gallery (the repo's projects/ dir) ──────────────────────────────
|
||
# Cost rollups walk every summary; memoized on the (summaries, sidecar)
|
||
# versions so repeat requests between changes are a dict lookup.
|
||
_costs_cache: tuple[tuple[int, int], dict] | None = None
|
||
_costs_lock = threading.Lock()
|
||
|
||
|
||
def _project_costs() -> dict:
|
||
global _costs_cache
|
||
key = (store.summaries_version, meta_store.version)
|
||
with _costs_lock:
|
||
if _costs_cache and _costs_cache[0] == key:
|
||
return _costs_cache[1]
|
||
costs = project_costs_mod.costs_by_project(
|
||
store.all_summaries(), meta_store.all())
|
||
with _costs_lock:
|
||
_costs_cache = (key, costs)
|
||
return costs
|
||
|
||
|
||
@app.get("/api/projects", responses=_r(schemas.ProjectsResponse))
|
||
def projects_list():
|
||
"""Card metadata for every directory under projects/ (newest commit first).
|
||
|
||
Each card carries a compact ``costs`` rollup (total $/tokens + conversation
|
||
count) aggregated from every conversation attributed to the project."""
|
||
costs = _project_costs()
|
||
items = projects_mod.list_projects()
|
||
for it in items:
|
||
agg = costs.get(it["dir"])
|
||
it["costs"] = {
|
||
"cost": agg["cost"] if agg else 0.0,
|
||
"tokens": agg["tokens"] if agg else 0,
|
||
"conversations": agg["conversations"] if agg else 0,
|
||
}
|
||
return {"projects": items}
|
||
|
||
|
||
@app.get("/api/project", responses=_r(schemas.ProjectDetail))
|
||
def project_detail(slug: str):
|
||
"""Full metadata + key-file contents (README/CLAUDE.md/index meta/og) for one project.
|
||
|
||
Also attaches a ``costs`` block: total/mean/median spend across the project's
|
||
conversations, a per-token-type breakdown, and mean cost per line of code."""
|
||
detail = projects_mod.project_detail(slug)
|
||
if detail is None:
|
||
raise HTTPException(404, "project not found")
|
||
agg = _project_costs().get(slug) or project_costs_mod._aggregate([])
|
||
loc = projects_mod.count_loc(slug)
|
||
agg = {**agg, "loc": loc,
|
||
"meanCostPerLoc": (agg["cost"] / loc) if loc else 0.0}
|
||
detail["costs"] = agg
|
||
return detail
|
||
|
||
|
||
def _project_env_payload(slug: str, entry: pathlib.Path) -> dict:
|
||
env_path = entry / ".env"
|
||
ef = envfile_mod.ProjectEnvFile.load(str(env_path))
|
||
return {
|
||
"slug": slug,
|
||
"exists": env_path.is_file(),
|
||
"vars": [
|
||
{
|
||
"key": ent["key"],
|
||
"value": None if ent["secret"] else ent["value"],
|
||
"secret": ent["secret"],
|
||
"hasValue": bool(ent["value"]),
|
||
}
|
||
for ent in ef.entries()
|
||
],
|
||
}
|
||
|
||
|
||
@app.get("/api/project-env", responses=_r(schemas.ProjectEnvResponse))
|
||
def project_env(slug: str):
|
||
"""Key/value view of a project's `.env`. Vars below a `# --- secrets ---`
|
||
comment are secrets: their keys are listed but values never leave disk."""
|
||
entry = projects_mod._safe_entry(slug)
|
||
if entry is None:
|
||
raise HTTPException(404, "project not found")
|
||
return _project_env_payload(slug, entry)
|
||
|
||
|
||
class EnvSetItem(BaseModel):
|
||
key: str
|
||
# None = keep the value already on disk (used to move a var between
|
||
# sections without the UI ever having seen a secret's value).
|
||
value: str | None = None
|
||
secret: bool = False
|
||
|
||
|
||
class EnvSaveBody(BaseModel):
|
||
slug: str
|
||
set: list[EnvSetItem] = []
|
||
unset: list[str] = []
|
||
|
||
|
||
@app.put("/api/project-env", responses=_r(schemas.ProjectEnvResponse))
|
||
def project_env_save(body: EnvSaveBody):
|
||
"""Apply key/value edits to a project's `.env`, preserving comments, order
|
||
and the secrets section. Unsets run first, then sets."""
|
||
entry = projects_mod._safe_entry(body.slug)
|
||
if entry is None:
|
||
raise HTTPException(404, "project not found")
|
||
for item in body.set:
|
||
if not envfile_mod.KEY_RE.match(item.key):
|
||
raise HTTPException(400, f"invalid env key: {item.key!r}")
|
||
ef = envfile_mod.ProjectEnvFile.load(str(entry / ".env"))
|
||
for key in body.unset:
|
||
ef.unset(key)
|
||
current = {e["key"]: e["value"] for e in ef.entries()}
|
||
for item in body.set:
|
||
value = item.value if item.value is not None else current.get(item.key, "")
|
||
ef.set_var(item.key, value, secret=item.secret)
|
||
ef.save()
|
||
return _project_env_payload(body.slug, entry)
|
||
|
||
|
||
# Raster formats that are safe to render inline on the app origin. Anything
|
||
# else served from user/repo-supplied bytes (SVG can carry scripts, HTML is
|
||
# HTML) goes out as a download so it can never script against the API's cookies.
|
||
_INLINE_IMAGE_EXTS = {".jpeg", ".jpg", ".png", ".webp", ".gif", ".bmp", ".avif"}
|
||
|
||
# Video containers safe to play inline on the app origin: a media file can't
|
||
# script against the API's cookies the way an SVG/HTML can, so it need not go
|
||
# out as an attachment. FileResponse honours Range requests (Starlette), so the
|
||
# <video> element can seek.
|
||
_INLINE_VIDEO_EXTS = {".mp4", ".webm", ".mov", ".m4v", ".ogv"}
|
||
|
||
|
||
def _safe_file_response(p: pathlib.Path) -> FileResponse:
|
||
headers = {"X-Content-Type-Options": "nosniff"}
|
||
ext = p.suffix.lower()
|
||
if ext not in _INLINE_IMAGE_EXTS and ext not in _INLINE_VIDEO_EXTS:
|
||
headers["Content-Disposition"] = f'attachment; filename="{p.name}"'
|
||
return FileResponse(p, headers=headers)
|
||
|
||
|
||
@app.get("/api/project-asset")
|
||
def project_asset(slug: str, path: str):
|
||
"""Serve a whitelisted image asset (e.g. og-image) from a project dir."""
|
||
p = projects_mod.project_asset_path(slug, path)
|
||
if p is None:
|
||
raise HTTPException(404, "asset not found")
|
||
return _safe_file_response(p)
|
||
|
||
|
||
@app.get("/api/artefact")
|
||
def artefact_asset(kind: str, slug: str, path: str):
|
||
"""Serve one generated artefact from a project/service's `.ai/artefacts/`
|
||
tree (``path`` is ``<date>/<sessionId>/<file>``, relative to that tree)."""
|
||
p = artefacts_mod.artefact_path(kind, slug, path)
|
||
if p is None:
|
||
raise HTTPException(404, "artefact not found")
|
||
return _safe_file_response(p)
|
||
|
||
|
||
@app.get("/api/screenshot")
|
||
def screenshot_asset(path: str):
|
||
"""Serve a captured screenshot image from docs/screenshots/ (read-only).
|
||
|
||
``path`` is the image's location relative to the screenshots root
|
||
(``<project>/<name>_<ts>.jpeg``). A leading ``docs/screenshots/`` prefix,
|
||
if present, is stripped so the raw path printed by the skill also works.
|
||
"""
|
||
rel = path.replace("\\", "/").lstrip("/")
|
||
for pref in ("docs/screenshots/", "screenshots/"):
|
||
if rel.startswith(pref):
|
||
rel = rel[len(pref):]
|
||
rel_p = pathlib.Path(rel)
|
||
if not rel or rel_p.is_absolute() or ".." in rel_p.parts:
|
||
raise HTTPException(400, "bad path")
|
||
p = (SCREENSHOTS_DIR / rel_p).resolve()
|
||
try:
|
||
p.relative_to(SCREENSHOTS_DIR)
|
||
except ValueError:
|
||
raise HTTPException(400, "bad path")
|
||
if not (p.is_file() and p.suffix.lower() in SCREENSHOT_EXTS):
|
||
raise HTTPException(404, "screenshot not found")
|
||
return _safe_file_response(p)
|
||
|
||
|
||
@app.get("/api/data-asset")
|
||
def data_asset(path: str):
|
||
"""Serve an image or video artifact from the repo's root `data/` tree (read-only).
|
||
|
||
``path`` is the file's location relative to ``data/`` (a leading ``data/``
|
||
prefix, if present, is stripped so the raw repo-relative path also works).
|
||
Used to preview media the agent ``Read`` from under ``data/`` inline in the
|
||
conversation viewer — see the CLAUDE.md convention to write artifacts there.
|
||
"""
|
||
rel = path.replace("\\", "/").lstrip("/")
|
||
if rel.startswith("data/"):
|
||
rel = rel[len("data/"):]
|
||
rel_p = pathlib.Path(rel)
|
||
if not rel or rel_p.is_absolute() or ".." in rel_p.parts:
|
||
raise HTTPException(400, "bad path")
|
||
p = (DATA_DIR / rel_p).resolve()
|
||
try:
|
||
p.relative_to(DATA_DIR)
|
||
except ValueError:
|
||
raise HTTPException(400, "bad path")
|
||
if not (p.is_file() and p.suffix.lower() in (IMAGE_EXTS | _INLINE_VIDEO_EXTS)):
|
||
raise HTTPException(404, "data asset not found")
|
||
return _safe_file_response(p)
|
||
|
||
|
||
@app.get("/api/scratchpad-asset")
|
||
def scratchpad_asset(path: str):
|
||
"""Serve an image or video from Claude Code's session scratchpad tree.
|
||
|
||
``path`` is relative to the scratchpad root — ``<encoded-cwd>/<sessionId>/
|
||
scratchpad/<file>`` — so a run's throwaway media previews inline in the
|
||
thread without having to be copied into a project's ``.ai/artefacts/``.
|
||
The mount is read-only and only media extensions are served.
|
||
"""
|
||
rel = path.replace("\\", "/").lstrip("/")
|
||
rel_p = pathlib.Path(rel)
|
||
if not rel or rel_p.is_absolute() or ".." in rel_p.parts:
|
||
raise HTTPException(400, "bad path")
|
||
p = (SCRATCHPAD_DIR / rel_p).resolve()
|
||
try:
|
||
p.relative_to(SCRATCHPAD_DIR)
|
||
except ValueError:
|
||
raise HTTPException(400, "bad path")
|
||
if not (p.is_file() and p.suffix.lower() in (IMAGE_EXTS | _INLINE_VIDEO_EXTS)):
|
||
raise HTTPException(404, "scratchpad asset not found")
|
||
return _safe_file_response(p)
|
||
|
||
|
||
# ── composer file uploads ────────────────────────────────────────────────────
|
||
# The composer (SpawnBox / ResumeBox) uploads files here; each batch lands in a
|
||
# random subdir under UPLOADS_DIR. The endpoint returns, per file, the
|
||
# repo-relative path to inject into the prompt (so the host-side `claude -p`
|
||
# session can Read it) and a serve URL the frontend renders inline in the thread.
|
||
def _safe_upload_name(name: str) -> str:
|
||
base = pathlib.PurePosixPath((name or "").replace("\\", "/")).name
|
||
base = _UNSAFE_NAME_RE.sub("_", base).strip("._")
|
||
return (base or "file")[:120]
|
||
|
||
|
||
def _upload_rel(path: str) -> pathlib.Path:
|
||
"""Validate a client-supplied uploads path and return it relative to
|
||
UPLOADS_DIR, tolerating a leading repo prefix (the injected prompt form)."""
|
||
rel = path.replace("\\", "/").lstrip("/")
|
||
prefix = UPLOADS_REPO_PREFIX + "/"
|
||
if rel.startswith(prefix):
|
||
rel = rel[len(prefix):]
|
||
rel_p = pathlib.Path(rel)
|
||
if not rel or rel_p.is_absolute() or ".." in rel_p.parts:
|
||
raise HTTPException(400, "bad path")
|
||
p = (UPLOADS_DIR / rel_p).resolve()
|
||
try:
|
||
p.relative_to(UPLOADS_DIR)
|
||
except ValueError:
|
||
raise HTTPException(400, "bad path")
|
||
return p
|
||
|
||
|
||
@app.post("/api/upload", responses=_r(schemas.UploadResult))
|
||
async def upload_files(files: list[UploadFile] = File(...)):
|
||
"""Store one or more composer attachments and describe where they landed.
|
||
|
||
Each upload is written to ``UPLOADS_DIR/<batch>/<safe-name>``. The returned
|
||
``repoPath`` is what the composer injects into the prompt (the session reads
|
||
it from the repo working dir); ``url`` is the backend serve route the thread
|
||
viewer renders the file from."""
|
||
if not files:
|
||
raise HTTPException(400, "no files")
|
||
batch = uuidlib.uuid4().hex[:12]
|
||
dest = UPLOADS_DIR / batch
|
||
dest.mkdir(parents=True, exist_ok=True)
|
||
out = []
|
||
for f in files:
|
||
name = _safe_upload_name(f.filename or "file")
|
||
target = dest / name
|
||
i = 1
|
||
while target.exists():
|
||
target = dest / f"{target.stem}-{i}{pathlib.Path(name).suffix}"
|
||
i += 1
|
||
size = 0
|
||
try:
|
||
with open(target, "wb") as w:
|
||
while chunk := await f.read(1 << 20):
|
||
size += len(chunk)
|
||
if size > MAX_UPLOAD_BYTES:
|
||
raise HTTPException(
|
||
413, f"{name} exceeds {MAX_UPLOAD_BYTES // (1<<20)} MB")
|
||
w.write(chunk)
|
||
except HTTPException:
|
||
target.unlink(missing_ok=True)
|
||
raise
|
||
rel = f"{batch}/{target.name}"
|
||
out.append({
|
||
"name": target.name,
|
||
"size": size,
|
||
"contentType": f.content_type or "application/octet-stream",
|
||
"repoPath": f"{UPLOADS_REPO_PREFIX}/{rel}",
|
||
"url": f"/api/upload-file?path={rel}",
|
||
})
|
||
return {"files": out}
|
||
|
||
|
||
@app.get("/api/upload-file")
|
||
def upload_file_asset(path: str):
|
||
"""Serve a previously uploaded composer attachment (raw bytes, inline)."""
|
||
p = _upload_rel(path)
|
||
if not p.is_file():
|
||
raise HTTPException(404, "upload not found")
|
||
return _safe_file_response(p)
|
||
|
||
|
||
# ── custom avatar set (a sprite-editor "Export set" zip) ──────────────────────
|
||
# The zip is one square-cell strip per clip plus a manifest.json in the Hemp
|
||
# Henry shape ({clip: {cell, fps, frames, loop, src}}). We validate it, unpack a
|
||
# clean copy into the writable data volume, and serve the strips inline so the
|
||
# roaming avatar plays a full custom emote set. Only strips referenced by the
|
||
# manifest are kept, and only as safe raster images — the zip never lands as-is.
|
||
_AVATAR_STRIP_EXTS = {".webp", ".png", ".gif"}
|
||
|
||
|
||
def _read_custom_manifest() -> dict:
|
||
"""The stored custom manifest with each src rewritten to its serve URL, or
|
||
{} when no set is imported."""
|
||
mf = CUSTOM_AVATAR_DIR / "manifest.json"
|
||
if not mf.is_file():
|
||
return {}
|
||
try:
|
||
raw = json.loads(mf.read_text())
|
||
except Exception:
|
||
return {}
|
||
out: dict = {}
|
||
for clip, meta in (raw or {}).items():
|
||
if not isinstance(meta, dict):
|
||
continue
|
||
src = str(meta.get("src") or f"{clip}.webp")
|
||
name = pathlib.PurePosixPath(src.replace("\\", "/")).name
|
||
if not (CUSTOM_AVATAR_DIR / name).is_file():
|
||
continue
|
||
out[clip] = {**meta, "src": f"/api/avatar-asset?path={name}"}
|
||
return out
|
||
|
||
|
||
@app.get("/api/avatar/manifest")
|
||
def avatar_manifest():
|
||
"""The imported custom avatar set's manifest (empty {} if none), with each
|
||
clip src pointing at /api/avatar-asset so the frontend can render it."""
|
||
return _read_custom_manifest()
|
||
|
||
|
||
@app.get("/api/avatar-asset")
|
||
def avatar_asset(path: str):
|
||
"""Serve one strip from the imported custom avatar set (inline image)."""
|
||
name = pathlib.PurePosixPath((path or "").replace("\\", "/")).name
|
||
ext = pathlib.Path(name).suffix.lower()
|
||
if not name or ext not in _AVATAR_STRIP_EXTS:
|
||
raise HTTPException(400, "bad path")
|
||
p = (CUSTOM_AVATAR_DIR / name).resolve()
|
||
try:
|
||
p.relative_to(CUSTOM_AVATAR_DIR)
|
||
except ValueError:
|
||
raise HTTPException(400, "bad path")
|
||
if not p.is_file():
|
||
raise HTTPException(404, "avatar asset not found")
|
||
# Pin the image content-type: the slim container's mimetypes DB can lack the
|
||
# .webp mapping, and with nosniff a text/plain response won't render as an
|
||
# image (breaking the sprite background). All _AVATAR_STRIP_EXTS are rasters.
|
||
media = {".webp": "image/webp", ".png": "image/png", ".gif": "image/gif"}[ext]
|
||
return FileResponse(p, media_type=media, headers={"X-Content-Type-Options": "nosniff"})
|
||
|
||
|
||
@app.post("/api/avatar/import")
|
||
async def avatar_import(file: UploadFile = File(...)):
|
||
"""Import a custom avatar set from a sprite-editor "Export set" zip.
|
||
|
||
Validates the zip has a manifest.json plus one strip per referenced clip,
|
||
then unpacks a clean copy into CUSTOM_AVATAR_DIR (replacing any previous
|
||
set) and returns the served manifest. Rejects anything that isn't a small
|
||
zip of a manifest + raster strips — no path traversal, no oversized members,
|
||
nothing executable ever written."""
|
||
raw = b""
|
||
while chunk := await file.read(1 << 20):
|
||
raw += chunk
|
||
if len(raw) > MAX_AVATAR_ZIP_BYTES:
|
||
raise HTTPException(413, f"zip exceeds {MAX_AVATAR_ZIP_BYTES // (1 << 20)} MB")
|
||
if not raw:
|
||
raise HTTPException(400, "empty upload")
|
||
|
||
try:
|
||
zf = zipfile.ZipFile(io.BytesIO(raw))
|
||
except zipfile.BadZipFile:
|
||
raise HTTPException(400, "not a valid zip")
|
||
|
||
# Index members by basename (tolerate a wrapping top-level folder).
|
||
members: dict[str, zipfile.ZipInfo] = {}
|
||
total = 0
|
||
for info in zf.infolist():
|
||
if info.is_dir():
|
||
continue
|
||
name = pathlib.PurePosixPath(info.filename.replace("\\", "/")).name
|
||
if not name or name.startswith("."):
|
||
continue
|
||
total += info.file_size
|
||
if total > MAX_AVATAR_ZIP_BYTES:
|
||
raise HTTPException(413, "zip contents too large")
|
||
members[name] = info
|
||
|
||
if "manifest.json" not in members:
|
||
raise HTTPException(400, "zip is missing manifest.json")
|
||
try:
|
||
manifest = json.loads(zf.read(members["manifest.json"].filename))
|
||
except Exception:
|
||
raise HTTPException(400, "manifest.json is not valid JSON")
|
||
if not isinstance(manifest, dict) or not manifest:
|
||
raise HTTPException(400, "manifest.json must be a non-empty object of clips")
|
||
|
||
# Validate every referenced strip is present and a safe raster image.
|
||
clean: dict = {}
|
||
strips: dict[str, bytes] = {}
|
||
for clip, meta in manifest.items():
|
||
if not isinstance(clip, str) or not _UNSAFE_NAME_RE.sub("", clip):
|
||
raise HTTPException(400, f"bad clip name: {clip!r}")
|
||
if not isinstance(meta, dict):
|
||
raise HTTPException(400, f"clip {clip!r}: entry must be an object")
|
||
src = str(meta.get("src") or f"{clip}.webp")
|
||
name = pathlib.PurePosixPath(src.replace("\\", "/")).name
|
||
if pathlib.Path(name).suffix.lower() not in _AVATAR_STRIP_EXTS:
|
||
raise HTTPException(400, f"clip {clip!r}: src must be a webp/png/gif strip")
|
||
if name not in members:
|
||
raise HTTPException(400, f"clip {clip!r}: strip {name} not in zip")
|
||
try:
|
||
frames = int(meta.get("frames") or 0)
|
||
cell = int(meta.get("cell") or 0)
|
||
except (TypeError, ValueError):
|
||
raise HTTPException(400, f"clip {clip!r}: frames/cell must be numbers")
|
||
strips[name] = zf.read(members[name].filename)
|
||
clean[clip] = {
|
||
"cell": cell,
|
||
"fps": int(meta.get("fps") or 6) or 6,
|
||
"frames": max(1, frames),
|
||
"loop": bool(meta.get("loop", True)),
|
||
"src": name,
|
||
}
|
||
|
||
# Write a clean set atomically-ish: build in a temp dir, then swap.
|
||
AVATARS_DIR.mkdir(parents=True, exist_ok=True)
|
||
tmp = AVATARS_DIR / f".custom-{uuidlib.uuid4().hex[:8]}"
|
||
tmp.mkdir(parents=True, exist_ok=True)
|
||
try:
|
||
for name, data in strips.items():
|
||
(tmp / name).write_bytes(data)
|
||
(tmp / "manifest.json").write_text(json.dumps(clean, indent=2, sort_keys=True) + "\n")
|
||
if CUSTOM_AVATAR_DIR.exists():
|
||
shutil.rmtree(CUSTOM_AVATAR_DIR)
|
||
tmp.rename(CUSTOM_AVATAR_DIR)
|
||
except Exception:
|
||
shutil.rmtree(tmp, ignore_errors=True)
|
||
raise
|
||
return {"manifest": _read_custom_manifest(), "clips": sorted(clean.keys())}
|
||
|
||
|
||
@app.delete("/api/avatar/custom")
|
||
def avatar_clear():
|
||
"""Remove the imported custom avatar set."""
|
||
if CUSTOM_AVATAR_DIR.exists():
|
||
shutil.rmtree(CUSTOM_AVATAR_DIR, ignore_errors=True)
|
||
return {"ok": True}
|
||
|
||
|
||
# ── services catalog (the repo's services/ dir — read-only, no Docker) ────────
|
||
@app.get("/api/services", responses=_r(schemas.ServicesResponse))
|
||
def services_list():
|
||
"""Card metadata for every service dir (containers/urls/auth, newest first)."""
|
||
return {"services": svc_mod.list_services()}
|
||
|
||
|
||
@app.get("/api/service", responses=_r(schemas.ServiceDetail))
|
||
def service_detail(slug: str):
|
||
"""Full parse of one service: containers, Traefik routers, env keys, tree."""
|
||
detail = svc_mod.service_detail(slug)
|
||
if detail is None:
|
||
raise HTTPException(404, "service not found")
|
||
return detail
|
||
|
||
|
||
@app.get("/api/service-file")
|
||
def service_file(slug: str, path: str):
|
||
"""A single whitelisted text file's contents from a service dir (512 KB cap)."""
|
||
txt = svc_mod.service_file(slug, path)
|
||
if txt is None:
|
||
raise HTTPException(404, "file not found or not previewable")
|
||
return PlainTextResponse(txt)
|
||
|
||
|
||
# ── goals board (every GOAL.md in the repo, with who's working on what) ───────
|
||
def _goal_activity() -> dict[tuple[str, str], dict]:
|
||
"""Per-unit conversation activity, keyed by ``(kind, dir)``.
|
||
|
||
``agents`` counts the sessions *currently running* on that unit — the badge
|
||
the Goals page shows so you can see, at a glance, that three agents are
|
||
already on `brain` and none on `mail`. ``lastAt`` is the newest
|
||
conversation's end time; the board sorts on it, so the units you last
|
||
worked on come first.
|
||
|
||
A conversation is attributed exactly as everywhere else in this app: the
|
||
union of the transcript-mined ``projectsAuto``/``servicesAuto`` and the
|
||
manual sidecar lists (see ``_conv_meta``). One conversation can therefore
|
||
count toward several units, which is correct — an agent editing both a
|
||
project and its service really is working on both.
|
||
"""
|
||
out: dict[tuple[str, str], dict] = {}
|
||
for card in _conversation_cards(full=False):
|
||
m = card.get("meta") or {}
|
||
ended = card.get("endedAt") or ""
|
||
running = m.get("state") == "running"
|
||
for kind, key in (("project", "projects"), ("service", "services")):
|
||
for d in (m.get(key) or []):
|
||
a = out.setdefault((kind, d),
|
||
{"agents": 0, "conversations": 0,
|
||
"lastAt": None, "running": []})
|
||
a["conversations"] += 1
|
||
if not a["lastAt"] or ended > a["lastAt"]:
|
||
a["lastAt"] = ended
|
||
if running:
|
||
a["agents"] += 1
|
||
a["running"].append({"id": card["id"],
|
||
"title": card.get("title") or "",
|
||
"startedAt": card.get("startedAt")})
|
||
return out
|
||
|
||
|
||
@app.get("/api/goals", responses=_r(schemas.GoalsResponse))
|
||
def goals_list():
|
||
"""Every project/service that keeps a GOAL.md, as a board.
|
||
|
||
One card per goal-bearing unit: its checklist progress, the version
|
||
milestones off its ``## Horizons``, the themed wishlist groups, the
|
||
goal-keeper's in-flight ``## Being worked on`` claims, and how many agent
|
||
sessions are running on it right now.
|
||
|
||
Sorted by most-recent conversation (the unit you last touched is the one
|
||
you most likely want to push), with never-worked-on goals last.
|
||
"""
|
||
activity = _goal_activity()
|
||
goals = []
|
||
for kind, resolve, items in (
|
||
# The root project lives at the repo root, not under projects/ — let the
|
||
# module resolve each slug rather than assuming <root>/<dir>.
|
||
("project", projects_mod.project_dir, projects_mod.list_projects()),
|
||
("service", lambda d: svc_mod.SERVICES_DIR / d, svc_mod.list_services()),
|
||
):
|
||
for it in items:
|
||
unit = resolve(it["dir"])
|
||
board = goalmd.goal_board(unit) if unit else None
|
||
if board is None:
|
||
continue # no GOAL.md — not on the board
|
||
act = activity.get((kind, it["dir"])) or {}
|
||
goals.append({
|
||
"kind": kind,
|
||
"dir": it["dir"],
|
||
"name": it["name"],
|
||
"description": it.get("description") or "",
|
||
"url": (it.get("url") if kind == "project"
|
||
else (it.get("urls") or [None])[0]),
|
||
"updatedAt": it.get("updatedAt"),
|
||
**board,
|
||
"agents": act.get("agents", 0),
|
||
"runningConversations": act.get("running", []),
|
||
"conversations": act.get("conversations", 0),
|
||
"lastConversationAt": act.get("lastAt"),
|
||
})
|
||
goals.sort(key=lambda g: g["lastConversationAt"] or "", reverse=True)
|
||
return {"goals": goals}
|
||
|
||
|
||
class GoalWorkBody(BaseModel):
|
||
kind: str # "project" | "service"
|
||
dir: str # the unit's directory name
|
||
item: str | None = None # a specific checklist item to work on (the detail
|
||
# page's per-item Work button); None ⇒ let the
|
||
# agent pick one (the Goals board's Work button)
|
||
|
||
|
||
@app.post("/api/goal-work", responses=_r(schemas.SpawnResult))
|
||
def goal_work(body: GoalWorkBody):
|
||
"""Spawn a goal-keeper session pointed at one unit's GOAL.md.
|
||
|
||
The "Work" button on the Goals page (no ``item``) and the per-item Work
|
||
button on a detail page's checklist (``item`` set). Same machinery as any
|
||
other spawn (`_send_message` → host sidecar), the difference being the
|
||
prompt: it invokes the `goal-keeper` agent scoped to a single directory
|
||
instead of letting it choose among every GOAL.md in the repo, and it tells
|
||
the agent to claim its item in ``## Being worked on`` first — which is what
|
||
stops two agents launched from this page landing on the same checkbox. When
|
||
``item`` is given the agent is told to work *that* item rather than pick one.
|
||
"""
|
||
if body.kind not in ("project", "service"):
|
||
raise HTTPException(400, "kind must be 'project' or 'service'")
|
||
resolve = (projects_mod.project_dir if body.kind == "project"
|
||
else svc_mod._safe_entry)
|
||
unit = resolve(body.dir)
|
||
if unit is None or not (unit / goalmd.GOAL_FILE).is_file():
|
||
raise HTTPException(404, "no GOAL.md for that unit")
|
||
# `projects/<dir>` / `services/<dir>`, except the root project — whose
|
||
# GOAL.md is the repo's own, so it is named `.` (see projects.repo_rel).
|
||
rel = (projects_mod.repo_rel(body.dir) if body.kind == "project"
|
||
else f"services/{body.dir}")
|
||
item = (body.item or "").strip()
|
||
if item:
|
||
# The per-item Work button: the user already chose the checkbox, so the
|
||
# agent works exactly that one instead of surveying the wishlist.
|
||
pick = (
|
||
f"Use the goal-keeper agent to push `{rel}` forward by completing "
|
||
f"this specific item from its GOAL.md:\n\n {item}\n\n"
|
||
f"Read `{rel}/GOAL.md` for the full context around it. If that exact "
|
||
f"item is already listed under its `## Being worked on` section, "
|
||
f"stop — another agent has it — and say so.\n\n"
|
||
)
|
||
else:
|
||
pick = (
|
||
f"Use the goal-keeper agent to push `{rel}` forward by one item.\n\n"
|
||
f"Read `{rel}/GOAL.md` and pick ONE unchecked wishlist item — the "
|
||
f"smallest high-leverage step you can finish end-to-end this "
|
||
f"session. Skip anything already listed under its `## Being worked "
|
||
f"on` section: another agent has claimed it.\n\n"
|
||
)
|
||
prompt = (
|
||
pick +
|
||
f"Before you start, claim your item by appending a bullet to that "
|
||
f"`## Being worked on` section (create it after the North star if it "
|
||
f"doesn't exist), in the "
|
||
f"form `- [{body.dir}] <the item> — @<UTC ISO timestamp>`. Commit that "
|
||
f"claim on its own so other agents see it immediately. When the item "
|
||
f"lands, tick its checkbox and remove your claim line in the same "
|
||
f"commit.\n\n"
|
||
f"Then — still before you create a worktree — rename this conversation "
|
||
f"to the item you claimed, so its card says what you're building "
|
||
f"instead of repeating this prompt: `conv-meta title \"<the feature, a "
|
||
f"handful of words, imperative>\"`. Leave the project out of the title; "
|
||
f"the card is already tagged with it.\n\n"
|
||
f"Follow the repo conventions (worktree, commit skill, notify-done)."
|
||
)
|
||
return _send_message(prompt)
|
||
|
||
|
||
# ── memories (the assistant's persistent file-based memory for this repo) ─────
|
||
@app.get("/api/memories", responses=_r(schemas.MemoriesResponse))
|
||
def memories_list():
|
||
"""All memory summaries (projects + links derived), newest-updated first."""
|
||
return {"memories": memories_mod.list_memories()}
|
||
|
||
|
||
@app.get("/api/memory", responses=_r(schemas.MemoryDetail))
|
||
def memory_detail(slug: str):
|
||
"""One memory + its body, with the conversations that created / read it."""
|
||
mem = memories_mod.memory_detail(slug)
|
||
if mem is None:
|
||
raise HTTPException(404, "memory not found")
|
||
origin = mem.get("originSessionId")
|
||
created_in, read_in = [], []
|
||
created_sids: set[str] = set()
|
||
for path, s in store.all_summaries():
|
||
if not s.get("messages"):
|
||
continue
|
||
sid = s.get("sessionId")
|
||
if sid and sid == origin:
|
||
created_in.append(_conv_card(path, s))
|
||
created_sids.add(sid)
|
||
for path, s in store.all_summaries():
|
||
if not s.get("messages"):
|
||
continue
|
||
# don't repeat the originating conversation under "read"
|
||
if s.get("sessionId") in created_sids:
|
||
continue
|
||
if slug in (s.get("memoriesRead") or []):
|
||
read_in.append(_conv_card(path, s))
|
||
created_in.sort(key=lambda c: c.get("endedAt") or "", reverse=True)
|
||
read_in.sort(key=lambda c: c.get("endedAt") or "", reverse=True)
|
||
mem["createdIn"] = created_in
|
||
mem["readIn"] = read_in
|
||
return mem
|
||
|
||
|
||
# ── plans (data/plans/, written by the `plan` skill — on-machine, not tracked) ─
|
||
def _duration_ms(started: str | None, ended: str | None) -> int | None:
|
||
if not started or not ended:
|
||
return None
|
||
try:
|
||
a = datetime.datetime.fromisoformat(started.replace("Z", "+00:00"))
|
||
b = datetime.datetime.fromisoformat(ended.replace("Z", "+00:00"))
|
||
return max(0, int((b - a).total_seconds() * 1000))
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
@app.get("/api/plans", responses=_r(schemas.PlansResponse))
|
||
def plans_list():
|
||
"""All plan summaries (data/plans/*.md), newest-updated first."""
|
||
return {"plans": plans_mod.list_plans()}
|
||
|
||
|
||
@app.get("/api/plan", responses=_r(schemas.PlanDetail))
|
||
def plan_detail(slug: str):
|
||
"""One plan + body, with the ACTUAL cost/time/tokens joined from the linked
|
||
conversation transcripts (shown next to the plan's own estimate)."""
|
||
plan = plans_mod.plan_detail(slug)
|
||
if plan is None:
|
||
raise HTTPException(404, "plan not found")
|
||
# sessionId → newest transcript summary for it.
|
||
by_sid: dict[str, tuple[str, dict]] = {}
|
||
for path, s in store.all_summaries():
|
||
sid = s.get("sessionId")
|
||
if not sid or not s.get("messages"):
|
||
continue
|
||
prev = by_sid.get(sid)
|
||
if prev is None or (s.get("endedAt") or "") > (prev[1].get("endedAt") or ""):
|
||
by_sid[sid] = (path, s)
|
||
convs, cost, tokens, dur = [], 0.0, 0, 0
|
||
for sid in plans_mod.linked_session_ids(plan):
|
||
hit = by_sid.get(sid)
|
||
if not hit:
|
||
continue
|
||
path, s = hit
|
||
card = _conv_card(path, s)
|
||
card["startedAt"] = s.get("startedAt")
|
||
card["durationMs"] = _duration_ms(s.get("startedAt"), s.get("endedAt"))
|
||
convs.append(card)
|
||
cost += s.get("cost") or 0.0
|
||
tokens += s.get("tokens") or 0
|
||
dur += card["durationMs"] or 0
|
||
plan["conversations"] = convs
|
||
plan["actual"] = {"cost": cost, "tokens": tokens,
|
||
"durationMs": dur or None} if convs else None
|
||
return plan
|
||
|
||
|
||
# ── project scaffolding templates (projects/templates/) ──────────────────────
|
||
@app.get("/api/templates", responses=_r(schemas.TemplatesResponse))
|
||
def templates_list():
|
||
"""List every scaffold under projects/templates/ with stack tags + stats."""
|
||
return {"templates": templates_mod.list_templates()}
|
||
|
||
|
||
@app.get("/api/template", responses=_r(schemas.TemplateDetail))
|
||
def template_detail(name: str):
|
||
"""One template: README, npm scripts, deps and a nested file tree."""
|
||
return templates_mod.template_detail(name)
|
||
|
||
|
||
@app.get("/api/template-file")
|
||
def template_file(name: str, path: str):
|
||
"""A single text file's contents from within a template (512 KB cap)."""
|
||
return PlainTextResponse(templates_mod.template_file(name, path))
|
||
|
||
|
||
@app.get("/api/events")
|
||
def events():
|
||
"""Server-Sent Events: ``meta`` / ``transcript`` change pings for the UI."""
|
||
q = hub.subscribe()
|
||
|
||
def gen():
|
||
try:
|
||
yield "retry: 3000\n\n"
|
||
while True:
|
||
try:
|
||
data = q.get(timeout=15)
|
||
yield f"data: {data}\n\n"
|
||
except queue.Empty:
|
||
yield ": ping\n\n" # keep proxies from closing the stream
|
||
finally:
|
||
hub.unsubscribe(q)
|
||
|
||
return StreamingResponse(
|
||
gen(), media_type="text/event-stream",
|
||
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no",
|
||
"Connection": "keep-alive"})
|
||
|
||
|
||
@app.get("/api/deploy-status", responses=_r(schemas.DeployStatus))
|
||
def deploy_status():
|
||
"""Current blue-green deploy snapshot (or ``null`` when none is in flight).
|
||
|
||
Written by ``services/ai-agent/deploy.sh`` into the data dir; polled by the
|
||
:class:`DeployWatcher`, which pings a ``deploy`` SSE event on any change.
|
||
"""
|
||
return read_deploy_status()
|
||
|
||
|
||
@app.get("/api/health")
|
||
def health():
|
||
return {"ok": True}
|
||
|
||
|
||
@app.on_event("startup")
|
||
def _startup():
|
||
# seed file metadata so the first request has data, then index in background
|
||
try:
|
||
indexer.scan_files()
|
||
except Exception:
|
||
pass
|
||
indexer.start()
|
||
watcher.start()
|
||
deploy_watcher.start()
|
||
# Run-exit detection: flip "running" conversations to "finished" the
|
||
# moment their sidecar-launched process dies (see _watch_sidecar_runs).
|
||
threading.Thread(target=_watch_sidecar_runs, daemon=True).start()
|
||
# Cron jobs: fire scheduled agent sessions (see cron.py).
|
||
cron_scheduler.start()
|
||
# OpenRouter catalogue (prices + sizes from APPE): pulled off the request
|
||
# path so the first model-browser open and the first priced transcript
|
||
# already have it (openrouter.py).
|
||
openrouter_mod.warm()
|
||
|
||
|
||
# ── static PWA + SPA fallback (declared last so /api/* wins) ─────────────────
|
||
# Everything under /assets/ carries a content hash in its name, so it can never
|
||
# go stale and is cached forever. The rest of the shell — index.html, the
|
||
# service worker, the manifest — keeps its name across deploys, so it must be
|
||
# revalidated on every request or a client can pin itself to an old bundle.
|
||
_IMMUTABLE = "public, max-age=31536000, immutable"
|
||
_REVALIDATE = "no-cache"
|
||
|
||
|
||
def _static_response(path: pathlib.Path) -> FileResponse:
|
||
hashed = path.parent.name == "assets" and path.parent.parent == STATIC_DIR
|
||
cache = _IMMUTABLE if hashed else _REVALIDATE
|
||
return FileResponse(path, headers={"Cache-Control": cache})
|
||
|
||
|
||
# Serve built assets when they exist, otherwise hand back index.html so the
|
||
# client-side router can resolve deep links like /_/.claude/skills/...
|
||
@app.get("/{full_path:path}")
|
||
def spa(full_path: str):
|
||
index = STATIC_DIR / "index.html"
|
||
if not index.is_file():
|
||
return JSONResponse({"detail": "frontend not built"}, status_code=200)
|
||
if full_path:
|
||
candidate = (STATIC_DIR / full_path).resolve()
|
||
try:
|
||
candidate.relative_to(STATIC_DIR)
|
||
except ValueError:
|
||
candidate = index # path traversal attempt → fall back
|
||
if candidate.is_dir() and (candidate / "index.html").is_file():
|
||
# Static sub-sites shipped in public/ (e.g. /avatars/hemp-henry/)
|
||
# get their own index instead of the SPA shell.
|
||
candidate = candidate / "index.html"
|
||
if candidate.is_file():
|
||
return _static_response(candidate)
|
||
return _static_response(index)
|