feat(ui,api): strip board redesign - island, rail by day, context view, server search, vault graph

src/beaver_gateway/conversations/context.py

@@ -0,0 +1,201 @@
+"""What a conversation's model holds: prompt granules, skills, tools, files touched."""
+
+from __future__ import annotations
+
+import re
+from collections import Counter
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from collections.abc import Iterable, Mapping
+
+    from beaver_gateway.agents.claude import ClaudeAgent
+    from beaver_gateway.conversations.kinds import Kind
+
+__all__ = ["compose", "files_touched", "granules", "skill_sets"]
+
+CHARS_PER_TOKEN = 3.2
+READ_TOOLS = frozenset({"Read", "NotebookRead"})
+WRITE_TOOLS = frozenset({"Write", "Edit", "MultiEdit", "NotebookEdit"})
+SEARCH_TOOLS = frozenset({"Glob", "Grep"})
+PATH_KEYS = ("file_path", "notebook_path", "path", "file", "filename")
+_FRONTMATTER = re.compile(r"^---\s*\n(.*?)\n---", re.DOTALL)
+
+
+def _tokens(chars: int) -> int:
+    return round(chars / CHARS_PER_TOKEN)
+
+
+def _relative(path: str, cwd: Path | None) -> str:
+    if cwd is None:
+        return path
+    try:
+        return str(Path(path).relative_to(cwd))
+    except ValueError:
+        return path
+
+
+def _tool_uses(entries: Iterable[Mapping[str, Any]]) -> Iterable[tuple[str, Any, str]]:
+    for entry in entries:
+        message = entry.get("message")
+        if entry.get("type") != "assistant" or not isinstance(message, dict):
+            continue
+        content = message.get("content")
+        if not isinstance(content, list):
+            continue
+        stamp = str(entry.get("timestamp") or "")
+        for block in content:
+            if isinstance(block, dict) and block.get("type") == "tool_use":
+                yield str(block.get("name") or "?"), block.get("input"), stamp
+
+
+def files_touched(
+    entries: Iterable[Mapping[str, Any]], *, cwd: Path | None = None
+) -> tuple[list[dict[str, Any]], dict[str, int]]:
+    """Files named in tool inputs, with how they were touched, plus tool counts."""
+    files: dict[str, dict[str, Any]] = {}
+    counts: Counter[str] = Counter()
+    for name, tool_input, stamp in _tool_uses(entries):
+        counts[name] += 1
+        if not isinstance(tool_input, dict):
+            continue
+        raw = next(
+            (v for k in PATH_KEYS if isinstance(v := tool_input.get(k), str) and v),
+            None,
+        )
+        if raw is None or (name in SEARCH_TOOLS and "file_path" not in tool_input):
+            continue
+        key = _relative(raw, cwd)
+        row = files.setdefault(
+            key, {"path": key, "reads": 0, "writes": 0, "other": 0, "last_at": stamp}
+        )
+        if name in READ_TOOLS:
+            row["reads"] += 1
+        elif name in WRITE_TOOLS:
+            row["writes"] += 1
+        else:
+            row["other"] += 1
+        row["last_at"] = max(row["last_at"], stamp)
+    ordered = sorted(files.values(), key=lambda r: r["last_at"], reverse=True)
+    return ordered, dict(counts)
+
+
+def granules(agent: ClaudeAgent, kind: Kind) -> dict[str, Any]:
+    """The system prompt as its sources: tag, path, size, token estimate."""
+    sources = agent.prompt_for(kind) if agent.serves(kind) else None
+    rows: list[dict[str, Any]] = []
+    if sources is None:
+        chars = len(agent.system_prompt)
+        if chars:
+            rows.append(
+                {
+                    "tag": None,
+                    "path": None,
+                    "bytes": chars,
+                    "tokens_est": _tokens(chars),
+                }
+            )
+    else:
+        for source in sources:
+            tag, raw = source if isinstance(source, tuple) else (None, source)
+            path = Path(raw)
+            try:
+                chars = len(path.read_text(encoding="utf-8"))
+            except OSError:
+                chars = 0
+            rows.append(
+                {
+                    "tag": tag,
+                    "path": _relative(str(path), agent.cwd),
+                    "bytes": chars,
+                    "tokens_est": _tokens(chars),
+                }
+            )
+    total = sum(r["bytes"] for r in rows)
+    return {"bytes": total, "tokens_est": _tokens(total), "granules": rows}
+
+
+def _skill_meta(skill_md: Path) -> dict[str, str]:
+    try:
+        text = skill_md.read_text(encoding="utf-8")
+    except OSError:
+        return {}
+    match = _FRONTMATTER.match(text)
+    meta: dict[str, str] = {}
+    if match:
+        for line in match.group(1).splitlines():
+            key, sep, value = line.partition(":")
+            if sep:
+                meta[key.strip()] = value.strip()
+    return meta
+
+
+def skill_sets(agent: ClaudeAgent, kind: Kind) -> list[dict[str, Any]]:
+    """Skill-set directories the kind loads, with each skill's name and description."""
+    out: list[dict[str, Any]] = []
+    for directory in agent.skills_for(kind):
+        skills = []
+        try:
+            children = sorted(p for p in directory.iterdir() if p.is_dir())
+        except OSError:
+            children = []
+        for child in children:
+            skill_md = child / "SKILL.md"
+            if skill_md.is_file():
+                meta = _skill_meta(skill_md)
+                skills.append(
+                    {
+                        "name": meta.get("name") or child.name,
+                        "description": meta.get("description", ""),
+                        "path": _relative(str(child), agent.cwd),
+                    }
+                )
+        out.append(
+            {
+                "set": directory.name,
+                "path": _relative(str(directory), agent.cwd),
+                "skills": skills,
+            }
+        )
+    return out
+
+
+def compose(
+    agent: ClaudeAgent,
+    kind: Kind,
+    entries: Iterable[Mapping[str, Any]],
+    *,
+    context_tokens: int,
+) -> dict[str, Any]:
+    files, counts = files_touched(entries, cwd=agent.cwd)
+    prompt = granules(agent, kind)
+    skills = skill_sets(agent, kind)
+    skills_chars = sum(
+        len(s["name"]) + len(s["description"])
+        for group in skills
+        for s in group["skills"]
+    )
+    return {
+        "agent": {
+            "name": agent.name,
+            "model": agent.model,
+            "effort": agent.options.effort,
+        },
+        "kind": kind,
+        "context_tokens": context_tokens,
+        "prompt": prompt,
+        "skills": skills,
+        "skills_tokens_est": _tokens(skills_chars),
+        "tools": {
+            "allowed": list(agent.options.tools) if agent.options.tools else None,
+            "disallowed": list(agent.options.disallowed_tools),
+            "gateway": list(agent.gateway_tools),
+            "mcps": [m.name for m in agent.expose_mcps],
+        },
+        "files": files,
+        "tool_counts": counts,
+        "history_tokens_est": max(
+            0, context_tokens - prompt["tokens_est"] - _tokens(skills_chars)
+        ),
+    }