diff --git a/AGENTS.md b/AGENTS.md index 339639fd..8abd4019 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -8,7 +8,7 @@ slice at a time (see ROADMAP "Definition of done — every slice"). The control plane makes **no LLM calls**. All LLM calls happen **inside task containers**. -- LLM-free packages: `core`, `taskservice`, `sessionservice`, `terminal`, `workflows`. +- LLM-free packages: `core`, `taskservice`, `sessionservice`, `terminal`, `workflows`, `profiler`. - The **only** LLM-bearing package is `container/` (the agent runs there). If you add a package that orchestrates or renders, keep it LLM-free. @@ -48,7 +48,22 @@ src/panopticon/ # description, unsent); daemon.py = the provision-only pull loop; # host.py = the unified per-host daemon (spawn + provision each pass; # `python -m panopticon.sessionservice.host`); `python -m panopticon.sessionservice` - # spawns one task + # spawns one task; transcripts.py = host-side read of a task's session + # transcripts out of its per-task config volume, after the container is gone + # (docker run + find/cat, injectable command-runner) — feeds `profiler/` + profiler/ # task time-profiler: pure gap-analysis over a claude session transcript's own + # timestamps (no agent-side instrumentation) — categories.py = the tool-time + # category table (one place, easily extended: Bash sub-classified by regex — + # tests/vcs/deps/pty-verify —, Read/Write/Edit/Grep/Glob → code-nav, Task/Agent → + # subagents [one bucket; its sidechain is never walked separately], mcp__* → + # orchestration); parse.py = profile_transcripts(paths) -> a profile dict: merges + # same-message.id assistant lines into one turn, classifies every gap as LLM time + # (user→assistant), tool time (assistant-with-tool_use→its tool_result, by + # category), or operator/system wait (assistant-with-no-tool_use→next user, incl. + # AskUserQuestion spans and a same-file restart's abandoned-tool_result→ + # "interrupted, continue" gap) — between-session gaps (multiple transcript files) + # reported separately; defensive throughout (malformed/old-format lines → + # unattributed, never a crash) container/ # entrypoint (`python -m panopticon.container` = connect/register/slug/ # heartbeat liveness) + agent.py (`-m panopticon.container.agent` = the tmux # pane's launcher: render skills + operations, point claude at the /mcp server, @@ -219,6 +234,20 @@ on every PR (the same commands the Makefile wraps). - `tests/test_multi_workflow_acceptance.py` — Slice 8 acceptance: over REST (via `build_app`), a path-discovered workflow is selectable with no core change, and GithubPeerReviewed + the free-form (spike) workflow run concurrently with workflow-specific skills. No Docker, no LLM. +- `tests/profiler/test_parse.py` — the golden spec for the **time-profiler**'s gap-analysis: a + hand-built fixture transcript exercises every category, an `AskUserQuestion` span (must land in + operator-wait, not a tool category), a `Task` subagent call (one bucket), a same-`message.id` + multi-line assistant turn, parallel tool calls, a mid-session restart (two consecutive `user` + lines, no assistant turn between — also operator-wait), a between-sessions restart, and + defensive old-format/malformed input (never crashes); `test_categories.py` covers the category + table's matcher rules in isolation. +- `tests/sessionservice/test_transcripts.py` — unit tests pin the emitted `docker volume + inspect`/`docker run … find`/`cat` commands via a fake command-runner (incl. never issuing a + bare `docker run --volume `, which would silently create an empty volume); a + `skipif` no-docker integration test round-trips a real volume. +- `tests/terminal/test_task_profile.py` — golden-output tests for the `panopticon profile` CLI's + formatting (single-task, `--all-tasks`, and the dashboard's one-line summary) plus + `run_profile_command`'s wiring (fake client + an injected session-path lookup, no real docker). ## Glossary diff --git a/src/panopticon/profiler/__init__.py b/src/panopticon/profiler/__init__.py new file mode 100644 index 00000000..7f535ad4 --- /dev/null +++ b/src/panopticon/profiler/__init__.py @@ -0,0 +1,11 @@ +"""Pure, LLM-free transcript gap-analysis: where an agent's task time went. + +See :mod:`panopticon.profiler.parse` for the algorithm and :mod:`panopticon.profiler.categories` +for the (single-place, easily-extended) tool category table. +""" + +from __future__ import annotations + +from panopticon.profiler.parse import profile_transcripts + +__all__ = ["profile_transcripts"] diff --git a/src/panopticon/profiler/categories.py b/src/panopticon/profiler/categories.py new file mode 100644 index 00000000..0c391505 --- /dev/null +++ b/src/panopticon/profiler/categories.py @@ -0,0 +1,80 @@ +"""The tool-time category table (single source of truth — extend it here, nowhere else). + +Classifies one tool call (name + its ``tool_use`` input) into a bucket. Ordered, first-match-wins: +a ``Bash`` command is sub-classified by regex over its command string; every other tool is +classified by name. Callers needing display order (CLI/dashboard) import :data:`DISPLAY_ORDER`. +""" + +from __future__ import annotations + +import re +from collections.abc import Mapping +from typing import Any + +#: Bash-command sub-categories, checked in order against the command string. First match wins; +#: a command matching none of these falls through to ``other-tools``. +_BASH_CATEGORY_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = ( + ("tests", re.compile(r"\bpytest\b")), + ("vcs", re.compile(r"\b(git|gh)\b")), + ("deps", re.compile(r"\b(pip3?\s+install|uv\s+(?:add|sync)|uv\s+pip\s+install)\b")), + ("pty-verify", re.compile(r"\b(pexpect|pyte|verify[-_]skill)\b")), +) + +#: Read/inspection tools that don't change repo state or wait on anything external. +_CODE_NAV_TOOLS = frozenset({"Read", "Write", "Edit", "Grep", "Glob"}) + +#: The subagent tool. Its tool_use→tool_result gap already spans the whole sidechain run (we never +#: walk the sidechain transcript separately, so this is one bucket, not double-counted). Claude +#: Code's built-in name is ``Task``; ``Agent`` covers harnesses (like this one) that rename it. +_SUBAGENT_TOOLS = frozenset({"Task", "Agent"}) + +#: A tool call whose PreToolUse/PostToolUse hooks flip the turn to the user while it's pending +#: (see AGENTS.md's turn-flip contract) — the agent is blocked on a human answer, not doing tool +#: work. Routed to operator-wait by the parser, not classified here as a tool category. +ASK_USER_QUESTION_TOOL = "AskUserQuestion" + +#: Every category a tool call can land in, in the order the CLI/dashboard display them. ``llm`` and +#: ``unattributed`` aren't tool categories (the parser computes them directly from gaps), but share +#: this ordering so rendering stays in one place. +DISPLAY_ORDER: tuple[str, ...] = ( + "llm", + "tests", + "pty-verify", + "code-nav", + "vcs", + "deps", + "subagents", + "orchestration", + "other-tools", + "unattributed", +) + +#: The tool-classified categories only (excludes ``llm``/``unattributed``) — what :func:`categorize` +#: can return. +TOOL_CATEGORIES: tuple[str, ...] = tuple( + c for c in DISPLAY_ORDER if c not in ("llm", "unattributed") +) + + +def categorize(tool_name: str, tool_input: Mapping[str, Any] | None) -> str: + """Classify one tool call into a category name (always one of :data:`TOOL_CATEGORIES`). + + ``tool_name`` is matched defensively — an unrecognized or missing name falls through to + ``other-tools`` rather than raising, so an old-format or unfamiliar transcript is still + handled (never crashes the profiler).""" + if tool_name == "Bash": + command = (tool_input or {}).get("command") + if isinstance(command, str): + for category, pattern in _BASH_CATEGORY_PATTERNS: + if pattern.search(command): + return category + return "other-tools" + if tool_name in _CODE_NAV_TOOLS: + return "code-nav" + if tool_name in _SUBAGENT_TOOLS: + return "subagents" + if isinstance(tool_name, str) and tool_name.startswith("mcp__"): + # The container's claude connects to the task service's MCP server and *only* it + # (`--strict-mcp-config`), so any `mcp__` call is by construction an orchestration call. + return "orchestration" + return "other-tools" diff --git a/src/panopticon/profiler/parse.py b/src/panopticon/profiler/parse.py new file mode 100644 index 00000000..296a6180 --- /dev/null +++ b/src/panopticon/profiler/parse.py @@ -0,0 +1,344 @@ +"""Pure gap-analysis over claude session transcripts: transcript file paths -> a profile dict. + +No agent-side instrumentation — every timestamp already lives in the transcript. Claude's hook +events (Stop/UserPromptSubmit) aren't recorded inline, so turn boundaries are **inferred +structurally** from the transcript's own shape: + +- One logical assistant turn can span several consecutive ``assistant``-type JSONL lines that + share the same ``message.id`` (one line per completed content block — ``thinking``, then + ``text``, then ``tool_use`` — each stamped when *that block* finished streaming). These are + merged into one turn before any gap math, so streaming sub-line timestamps are never mistaken + for turn boundaries. +- ``UserEvent.ts -> next AssistantTurn.end_ts`` is always **LLM time** (model latency + + generation) — covers both a tool_result delivery and a genuine human prompt, and the turn's + *whole* span (through any thinking/text blocks preceding a trailing tool_use), not just the gap + to its first block. +- ``AssistantTurn -> next UserEvent`` is **tool time** when the turn carries ``tool_use`` block(s) + (paired to their ``tool_result`` by id and classified via :mod:`panopticon.profiler.categories`), + or **operator wait** when the turn has none (the agent stopped, handing back to the human). + ``AskUserQuestion`` is a tool_use, but its span is a blocked-on-human wait (the PreToolUse/ + PostToolUse hooks flip the turn to the user while it's pending), so it's routed to operator + wait, not a tool category. +- Multiple session files (restarts/phases) are concatenated by timestamp; the gap from one + session's last record to the next session's first is **between-sessions** wait, reported + separately. A restart *within* one session file — two consecutive ``user``-type lines with no + assistant turn between them, e.g. claude's "You were interrupted. Continue." synthetic prompt + landing right after an abandoned tool_result — is billed to **operator wait** too, for the same + reason: it's neither model generation nor tool execution. +- Anything unparseable — malformed lines, missing/unparseable timestamps, a ``tool_use`` with no + matching ``tool_result`` (a transcript cut off mid-call) — never crashes the parser; durations we + can't attribute show up in ``unattributed_s``/``unmatched_tool_calls`` rather than being dropped + silently or guessed. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from datetime import datetime +from pathlib import Path +from typing import Any + +from panopticon.profiler.categories import ASK_USER_QUESTION_TOOL, TOOL_CATEGORIES, categorize + + +@dataclass +class _ToolCall: + tool_use_id: str + name: str + input: dict[str, Any] + call_ts: float + + +@dataclass +class _AssistantTurn: + start_ts: float + end_ts: float + tool_calls: list[_ToolCall] = field(default_factory=list) + + +@dataclass +class _UserEvent: + ts: float + tool_result_ts: dict[str, float] # tool_use_id -> the ts this result arrived + is_real_prompt: bool # no tool_result blocks at all -> a genuine human message + + +_Event = _AssistantTurn | _UserEvent + + +def _parse_ts(value: Any) -> float | None: + if not isinstance(value, str) or not value: + return None + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp() + except ValueError: + return None + + +def _read_records(path: Path) -> list[dict[str, Any]]: + """Every JSON-object line in ``path``; blank/non-JSON/non-object lines are skipped, never + raised on — a missing file yields no records rather than an error.""" + try: + text = path.read_text() + except OSError: + return [] + records = [] + for line in text.splitlines(): + line = line.strip() + if not line: + continue + try: + obj = json.loads(line) + except ValueError: + continue + if isinstance(obj, dict): + records.append(obj) + return records + + +def _session_events(records: list[dict[str, Any]]) -> list[_Event]: + """Merge same-``message.id`` assistant lines into turns and pair each with the user events + (real prompts and tool_result deliveries) around them, in chronological (file) order.""" + events: list[_Event] = [] + pending_turn: _AssistantTurn | None = None + pending_msg_id: object = None + + def flush_turn() -> None: + nonlocal pending_turn, pending_msg_id + if pending_turn is not None: + events.append(pending_turn) + pending_turn = None + pending_msg_id = None + + for rec in records: + rtype = rec.get("type") + if rtype == "assistant": + msg = rec.get("message") + ts = _parse_ts(rec.get("timestamp")) + if not isinstance(msg, dict) or ts is None: + continue + msg_id = msg.get("id") + tool_calls = [] + content = msg.get("content") + if isinstance(content, list): + for block in content: + if not isinstance(block, dict) or block.get("type") != "tool_use": + continue + tool_id, name = block.get("id"), block.get("name") + if isinstance(tool_id, str) and isinstance(name, str): + input_ = block.get("input") + tool_calls.append( + _ToolCall(tool_id, name, input_ if isinstance(input_, dict) else {}, ts) + ) + if pending_turn is not None and msg_id is not None and msg_id == pending_msg_id: + pending_turn.end_ts = ts + pending_turn.tool_calls.extend(tool_calls) + else: + flush_turn() + pending_turn = _AssistantTurn(start_ts=ts, end_ts=ts, tool_calls=tool_calls) + pending_msg_id = msg_id + elif rtype == "user": + flush_turn() + msg = rec.get("message") + ts = _parse_ts(rec.get("timestamp")) + if not isinstance(msg, dict) or ts is None: + continue + content = msg.get("content") + tool_result_ts: dict[str, float] = {} + if isinstance(content, list): + for block in content: + if isinstance(block, dict) and block.get("type") == "tool_result": + tool_use_id = block.get("tool_use_id") + if isinstance(tool_use_id, str): + tool_result_ts[tool_use_id] = ts + events.append( + _UserEvent(ts=ts, tool_result_ts=tool_result_ts, is_real_prompt=not tool_result_ts) + ) + else: + continue # structural noise: mode, permission-mode, file-history-snapshot, attachment, … + flush_turn() + return events + + +def _event_ts(event: _Event, *, start: bool) -> float: + if isinstance(event, _AssistantTurn): + return event.start_ts if start else event.end_ts + return event.ts + + +class _Buckets: + """Running totals accumulated while walking one task's concatenated sessions.""" + + def __init__(self) -> None: + self.llm_s = 0.0 + self.llm_count = 0 + self.operator_wait_s = 0.0 + self.operator_wait_count = 0 + self.between_sessions_s = 0.0 + self.between_sessions_count = 0 + self.unmatched_tool_calls = 0 + self.categories: dict[str, dict[str, float]] = { + name: {"seconds": 0.0, "count": 0} for name in TOOL_CATEGORIES + } + self.top_tool_calls: list[dict[str, Any]] = [] + + def record_tool_call(self, call: _ToolCall, duration: float) -> None: + duration = max(0.0, duration) + if call.name == ASK_USER_QUESTION_TOOL: + self.operator_wait_s += duration + self.operator_wait_count += 1 + return + category = categorize(call.name, call.input) + bucket = self.categories[category] + bucket["seconds"] += duration + bucket["count"] += 1 + self.top_tool_calls.append( + { + "category": category, + "tool_name": call.name, + "first_line": _first_line(call.name, call.input), + "duration_s": duration, + } + ) + + +def _first_line(name: str, tool_input: dict[str, Any]) -> str: + if name == "Bash": + command = tool_input.get("command") + if isinstance(command, str) and command.strip(): + return command.strip().splitlines()[0] + for key in ("file_path", "description", "prompt", "pattern", "query"): + value = tool_input.get(key) + if isinstance(value, str) and value.strip(): + return value.strip().splitlines()[0] + return name + + +def _walk_session(events: list[_Event], buckets: _Buckets) -> None: + pending_tool_calls: dict[str, _ToolCall] = {} + operator_wait_since: float | None = None + llm_boundary_ts: float | None = None + + for ev in events: + if isinstance(ev, _AssistantTurn): + if llm_boundary_ts is not None: + # Bill the *whole* turn's generation time as LLM time — including any thinking/text + # blocks that streamed before a trailing tool_use — not just the gap to its first + # block, which would otherwise leak that time into unattributed. + gap = ev.end_ts - llm_boundary_ts + if gap > 0: + buckets.llm_s += gap + buckets.llm_count += 1 + llm_boundary_ts = None + operator_wait_since = None + if ev.tool_calls: + for call in ev.tool_calls: + pending_tool_calls[call.tool_use_id] = call + else: + operator_wait_since = ev.end_ts # the agent stopped; waiting on the human next + else: # _UserEvent + if llm_boundary_ts is not None: + # Back-to-back `user`-type lines with no assistant turn between them — e.g. claude + # injecting its "You were interrupted. Continue." synthetic prompt right after an + # abandoned tool_result on a container respawn mid-turn. That gap is neither model + # generation nor tool execution; bill it to operator/system wait rather than let it + # silently vanish into unattributed. + gap = ev.ts - llm_boundary_ts + if gap > 0: + buckets.operator_wait_s += gap + buckets.operator_wait_count += 1 + llm_boundary_ts = None + if ev.tool_result_ts: + for tool_use_id, result_ts in ev.tool_result_ts.items(): + if tool_use_id not in pending_tool_calls: + continue # a tool_result with no open call — ignore defensively + call = pending_tool_calls.pop(tool_use_id) + buckets.record_tool_call(call, result_ts - call.call_ts) + llm_boundary_ts = ev.ts + elif operator_wait_since is not None: + gap = ev.ts - operator_wait_since + if gap > 0: + buckets.operator_wait_s += gap + buckets.operator_wait_count += 1 + llm_boundary_ts = ev.ts + else: + llm_boundary_ts = ev.ts + operator_wait_since = None + buckets.unmatched_tool_calls += len(pending_tool_calls) + + +def _empty_profile() -> dict[str, Any]: + return { + "total_wall_s": 0.0, + "operator_wait_s": 0.0, + "operator_wait_count": 0, + "between_sessions_s": 0.0, + "between_sessions_count": 0, + "agent_active_s": 0.0, + "llm_s": 0.0, + "llm_count": 0, + "categories": {name: {"seconds": 0.0, "count": 0} for name in TOOL_CATEGORIES}, + "unattributed_s": 0.0, + "unmatched_tool_calls": 0, + "session_count": 0, + "top_tool_calls": [], + } + + +def profile_transcripts(paths: list[Path] | tuple[Path, ...]) -> dict[str, Any]: + """Gap-analyze one task's (possibly many) session transcripts into a time-profile dict. + + ``paths`` are the task's session JSONL files in any order — they're sorted by each session's + own first-record timestamp before concatenation, so restarts/phases line up correctly + regardless of filename or argument order. Pure aside from reading these files: no network, no + LLM, no clock reads, so it's fully deterministic and unit-testable via ``tmp_path`` fixtures. + """ + sessions: list[tuple[float, float, list[_Event]]] = [] + for path in paths: + events = _session_events(_read_records(Path(path))) + if not events: + continue + sessions.append( + (_event_ts(events[0], start=True), _event_ts(events[-1], start=False), events) + ) + + if not sessions: + return _empty_profile() + + sessions.sort(key=lambda s: s[0]) + buckets = _Buckets() + for i, (start_ts, _end_ts, events) in enumerate(sessions): + if i > 0: + gap = start_ts - sessions[i - 1][1] + if gap > 0: + buckets.between_sessions_s += gap + buckets.between_sessions_count += 1 + _walk_session(events, buckets) + + total_wall_s = sessions[-1][1] - sessions[0][0] + tool_categories_s = sum(b["seconds"] for b in buckets.categories.values()) + agent_active_s = tool_categories_s + buckets.llm_s + accounted_s = agent_active_s + buckets.operator_wait_s + buckets.between_sessions_s + # Parallel tool calls (one turn, several tool_use blocks) each bill their own full duration + # even though they overlap in wall time, so accounted_s can slightly *exceed* total_wall_s — + # clamp rather than surface a confusing negative "unattributed". + unattributed_s = max(0.0, total_wall_s - accounted_s) + + top_tool_calls = sorted(buckets.top_tool_calls, key=lambda c: c["duration_s"], reverse=True)[:5] + + return { + "total_wall_s": total_wall_s, + "operator_wait_s": buckets.operator_wait_s, + "operator_wait_count": buckets.operator_wait_count, + "between_sessions_s": buckets.between_sessions_s, + "between_sessions_count": buckets.between_sessions_count, + "agent_active_s": agent_active_s, + "llm_s": buckets.llm_s, + "llm_count": buckets.llm_count, + "categories": buckets.categories, + "unattributed_s": unattributed_s, + "unmatched_tool_calls": buckets.unmatched_tool_calls, + "session_count": len(sessions), + "top_tool_calls": top_tool_calls, + } diff --git a/src/panopticon/sessionservice/transcripts.py b/src/panopticon/sessionservice/transcripts.py new file mode 100644 index 00000000..adfd557d --- /dev/null +++ b/src/panopticon/sessionservice/transcripts.py @@ -0,0 +1,104 @@ +"""Read a task's claude session transcripts out of its per-task config volume — host-side, so the +profiler (:mod:`panopticon.profiler`) can gap-analyze a task's history **after** its container is +long gone, which is exactly the retroactive case the profiler exists for. + +The config volume (``panopticon-config-``, :data:`~panopticon.sessionservice.local_runner.CONFIG_MOUNT`) +persists across respawn/recreate and is never explicitly removed (see ``LocalRunner``), so a +completed task's transcripts are still sitting in it on whichever host ran the task. There's no +bind mount to them, so we shell out to ``docker`` (reusing the already-built ``panopticon-base`` +image — no new image dependency) to list and read the files, behind an injectable command-runner +(the same convention as :mod:`~panopticon.sessionservice.local_runner` / ``GitClones``) so this is +unit-testable without a daemon; ``tests/sessionservice/test_transcripts.py`` also has a +``skipif``-gated integration test against a real volume. + +**Scope: local runner only.** This assumes the caller runs on the same docker host that ran the +task. A task run on a remote runner host (``TaskOut.runner_host``) isn't reachable from here — the +ssh-wrap pattern ``terminal/attach.py::attach_command`` already uses is the natural follow-up if +that's ever needed. +""" + +from __future__ import annotations + +import subprocess +import tempfile +from pathlib import Path + +from panopticon.sessionservice.local_runner import ( + CONFIG_MOUNT, + DEFAULT_IMAGE, + CommandRunner, + _subprocess_run, +) + +#: Where claude's transcripts live inside the config volume. The project directory is always this +#: fixed path: `container/agent.py`'s `_claude_argv` keys it off `Path.cwd()`, which is always +#: `/workspace` (ADR 0011's `WORKSPACE_MOUNT`), encoded as `-workspace`. +_TRANSCRIPT_DIR = f"{CONFIG_MOUNT}/projects/-workspace" + + +def _volume_name(task_id: str) -> str: + return f"panopticon-config-{task_id}" + + +def task_session_paths( + task_id: str, + *, + dest: Path | str | None = None, + run: CommandRunner = _subprocess_run, + image: str = DEFAULT_IMAGE, +) -> list[Path]: + """Extract task ``task_id``'s session transcript files into a local directory; return their + paths (unsorted by design — :func:`panopticon.profiler.parse.profile_transcripts` sorts them + itself by each session's own first-record timestamp). + + ``dest`` is the directory to extract into (a fresh ``tempfile.mkdtemp()`` by default — the + caller owns cleaning it up). Returns ``[]``, **never raises**, when the task has no config + volume (never spawned, or a ``runner_type="shell"`` workflow that never ran a container) or no + transcripts yet: docker/tar trouble is swallowed rather than propagated, so a caller profiling + many tasks (``--all-tasks``) can just skip the empty ones. + + Checks the volume exists (``docker volume inspect``) **before** doing anything else — a bare + ``docker run --volume :...`` silently *creates* an empty named volume as a side + effect, which would otherwise litter the host with junk volumes every time a never-spawned or + typo'd task id is profiled. + + ``image`` overrides the reader image (default: the already-built ``panopticon-base`` — no new + pull needed on any host that's spawned a task); tests point it at a minimal throwaway image. + """ + volume = _volume_name(task_id) + try: + run(["docker", "volume", "inspect", volume], check=True) + except (OSError, subprocess.CalledProcessError): + return [] # no such volume (or no docker at all) — nothing to profile + + dest_dir = ( + Path(dest) if dest is not None else Path(tempfile.mkdtemp(prefix="panopticon-profile-")) + ) + dest_dir.mkdir(parents=True, exist_ok=True) + + def _reader(*args: str) -> str: + return run( + [ + "docker", + "run", + "--rm", + "--volume", + f"{volume}:{CONFIG_MOUNT}:ro", + image, + *args, + ], + check=False, # tolerate "no such directory" (an existing-but-empty volume) — empty stdout + ) + + # `find`'s predicates have no long-option form (`-maxdepth`/`-name` are the only spelling; see + # AGENTS.md's shelling-out rule), unlike the double-dash flags used elsewhere in this module. + listing = _reader("find", _TRANSCRIPT_DIR, "-maxdepth", "1", "-name", "*.jsonl") + names = sorted({Path(line).name for line in listing.splitlines() if line.strip()}) + + paths = [] + for name in names: + content = _reader("cat", f"{_TRANSCRIPT_DIR}/{name}") + out_path = dest_dir / name + out_path.write_text(content) + paths.append(out_path) + return paths diff --git a/src/panopticon/terminal/__main__.py b/src/panopticon/terminal/__main__.py index 43960e9f..dfd5cc29 100644 --- a/src/panopticon/terminal/__main__.py +++ b/src/panopticon/terminal/__main__.py @@ -89,6 +89,15 @@ def main( # the operator picked with `t` by writing it here instead of returning it in-process. dash.add_argument("--switch-file", help=argparse.SUPPRESS) sub.add_parser("tasks", help="list tasks as plain text") + prof = sub.add_parser( + "profile", help="time-profile a task's session transcripts (llm/tool/operator-wait split)" + ) + prof.add_argument("task", nargs="?", help="task id or slug to profile") + prof.add_argument( + "--all-tasks", + action="store_true", + help="aggregate every task's profile per-repo (totals + medians) instead of one task", + ) mig = sub.add_parser("migrate", help="apply DB migrations to head (or pass alembic args)") mig.add_argument("alembic_args", nargs="*", default=["upgrade", "head"]) sub.add_parser("build", help="build the base task-container image (panopticon-base)") @@ -181,6 +190,10 @@ def main( if args.command == "tasks": for t in client.list_tasks(): print(f"{t['id']} {t['state']:<10} {t['turn']:<5} {t['slug'] or '-'}") + elif args.command == "profile": + from panopticon.terminal.task_profile import run_profile_command + + return run_profile_command(client, task_ref=args.task, all_tasks=args.all_tasks) elif args.command == "dashboard": from panopticon.terminal.console import make_runner_switch, make_service_switch, switch_to from panopticon.terminal.dashboard import run diff --git a/src/panopticon/terminal/dashboard.py b/src/panopticon/terminal/dashboard.py index 3399084e..22b7c5f1 100644 --- a/src/panopticon/terminal/dashboard.py +++ b/src/panopticon/terminal/dashboard.py @@ -24,7 +24,10 @@ one with the host's default handler (`xdg-open`/`open`) by fetching it over REST to a temp file, `e` opens the on-disk file in place when the dashboard shares the artifact store, `y` **copies the task's slug** and `Y` its **id** to the clipboard (OSC 52 + the host's `pbcopy`/`xclip`/`wl-copy`, -so it works on Linux and macOS). Drop is the only state +so it works on Linux and macOS), and `P` **time-profiles** the highlighted task (reads its session +transcripts via docker, `panopticon.sessionservice.transcripts`) and shows a one-line llm/tool/wait +summary in the detail pane — on demand only (never automatic — see `action_profile`), since it +shells out to docker per task. Drop is the only state *transition* the dashboard drives: every other transition starts a new agentic turn, so it's triggered by an in-container agent skill (`advance` over REST/MCP; going back to coding is a free `set_state` move), not the operator (ADR 0004). @@ -101,9 +104,12 @@ from panopticon.client import JsonObj, TaskServiceClient from panopticon.core.dirs import ARTIFACTS_DIR from panopticon.core.state import TERMINAL_LABELS +from panopticon.profiler.parse import profile_transcripts from panopticon.sessionservice.local_runner import session_name +from panopticon.sessionservice.transcripts import task_session_paths from panopticon.taskservice.artifacts_fs import FilesystemArtifactStore from panopticon.terminal.setup_repo_task import create_setup_repo_task +from panopticon.terminal.task_profile import format_time_summary def _make_sort_key( @@ -361,14 +367,19 @@ def _repo_cell(task: JsonObj, repo_names: dict[str, str]) -> str: return repo_names.get(repo_id, repo_id) if repo_id else "?" -def render_detail(task: JsonObj) -> str: +def render_detail(task: JsonObj, *, time_summary: str | None = None) -> str: """The right-pane text for one task: identity, state/turn, and history. **Plain text** — the caller wraps it in a Rich ``Text`` so it renders literally. We deliberately do *not* use console markup here: a field can contain a stray ``[`` (e.g. a docker command captured in ``lifecycle_detail`` — ``['--add-host', …]``, or a memo), and markup-parsing that string crashes the whole pane. (Rich's escaper + Textual's markup parser also disagree on which - ``[`` is a tag, so escaping isn't reliable — rendering literally is.)""" + ``[`` is a tag, so escaping isn't reliable — rendering literally is.) + + ``time_summary`` (``format_time_summary`` of a profiled task) is opt-in and supplied by the + caller, never computed here — it means shelling out to docker to read the task's session + transcripts, which is too expensive to do on every highlight/refresh (see `P` / + ``action_profile``).""" turn = f"{task['turn']}{' (blocked)' if task.get('blocked') else ''}" claim = f" claimed: {task['claimed_by']}" if task.get("claimed_by") else "" lines = [ @@ -388,6 +399,8 @@ def render_detail(task: JsonObj) -> str: used = _short_tokens(task.get("tokens_used")) est = _short_tokens(task.get("token_estimate")) lines += ["", f"tokens (wt): {used} used / {est} est"] + if time_summary: + lines += ["", time_summary] lines += ["", "history:"] for entry in task.get("history") or []: line = f" {entry['from_state'] or '∅'} → {entry['to_state']}" @@ -1439,6 +1452,13 @@ def binding(self) -> Binding: Hotkey("u", "runner", "Runner", "Switch to the session-service (runner) session", show=False), Hotkey("y", "copy_slug", "Copy slug", "Copy the task's slug to the clipboard", show=False), Hotkey("Y", "copy_id", "Copy id", "Copy the task's id to the clipboard", show=False), + Hotkey( + "P", + "profile", + "Profile", + "Time-profile the highlighted task's session transcripts (llm/tool/wait split)", + show=False, + ), Hotkey( "escape", "clear_search", @@ -1574,6 +1594,10 @@ def __init__( # one reused scratch dir for `a`'s REST-open (lazily made, cleaned on exit) — so opening # many artifacts doesn't leak a temp dir each. self._artifact_tmp: tempfile.TemporaryDirectory[str] | None = None + # task id -> its last-computed `format_time_summary` line, populated only by `P` + # (action_profile) — never automatically, since it shells out to docker per task and would + # be far too slow to redo on every highlight/refresh (see render_detail's time_summary). + self._profile_summaries: dict[str, str] = {} def compose(self) -> ComposeResult: yield Header() @@ -1814,8 +1838,9 @@ def _update_detail(self, task_id: str | None) -> None: ) # fall back to summary when the service is unreachable # wrap in Text so the pane renders literally — never parse task content as console markup # (a "[" in e.g. a docker-command lifecycle_detail would otherwise crash the whole dashboard) + summary = self._profile_summaries.get(task_id) if task_id else None self.query_one("#detail", Static).update( - Text(render_detail(task)) if task else Text("no tasks") + Text(render_detail(task, time_summary=summary)) if task else Text("no tasks") ) def action_new_task(self) -> None: @@ -1961,6 +1986,31 @@ def action_copy_id(self) -> None: self._copy_to_clipboard(self._current) self.notify(f"copied id: {self._current}") + def action_profile(self) -> None: + """`P`: time-profile the highlighted task and show a summary line in the detail pane. + + On-demand rather than automatic — unlike the rest of the detail pane, this shells out to + docker to read the task's session transcripts (:mod:`panopticon.sessionservice.transcripts`), + which is too expensive to redo on every arrow-key highlight. Caches the rendered summary per + task id so re-selecting the same task doesn't recompute it; press `P` again to refresh.""" + if self._current is None: + return + task_id = self._current + try: + paths = task_session_paths(task_id) + if not paths: + self.notify("No session transcripts found for this task.", severity="warning") + return + profile = profile_transcripts(paths) + except Exception as exc: + self.notify(f"Can't profile task: {exc}", severity="error") + return + self._profile_summaries[task_id] = format_time_summary(profile) + if not self._detail_visible: # reveal the pane — the point of `P` is to see the summary + self.action_toggle_detail() + else: + self._update_detail(task_id) + def action_toggle_sort(self) -> None: """`o`: toggle between sorting by creation time or update time.""" self._sort_by_updated = not self._sort_by_updated diff --git a/src/panopticon/terminal/task_profile.py b/src/panopticon/terminal/task_profile.py new file mode 100644 index 00000000..46d7a47a --- /dev/null +++ b/src/panopticon/terminal/task_profile.py @@ -0,0 +1,284 @@ +"""Render a :func:`panopticon.profiler.parse.profile_transcripts` profile dict for a human, and +drive the ``panopticon profile`` CLI subcommand end to end. + +Three renderers share the same duration/percentage formatting: :func:`format_task_profile` (the +``panopticon profile `` compact breakdown), :func:`format_all_tasks_report` (the +``--all-tasks`` per-repo totals/medians), and :func:`format_time_summary` (the one-line dashboard +detail-pane summary). :func:`run_profile_command` wires them to the task service + the volume +reader — kept out of ``terminal/__main__.py`` per its existing pattern (``doctor.py``, +``quickstart.py``) of one module per non-trivial subcommand. +""" + +from __future__ import annotations + +import statistics +import sys +from collections.abc import Callable, Iterable, Sequence +from pathlib import Path +from typing import Any, TextIO + +from panopticon.client import JsonObj, TaskServiceClient +from panopticon.profiler.categories import TOOL_CATEGORIES +from panopticon.profiler.parse import profile_transcripts +from panopticon.sessionservice.transcripts import task_session_paths + + +def _fmt_duration(seconds: float) -> str: + """``'14h 32m'``/``'12m 03s'``/``'45s'`` — hours+minutes once over an hour (seconds dropped), + otherwise minutes+seconds, otherwise bare seconds. One formatter, used everywhere in this + module so every duration in the output is spelled the same way.""" + total = max(0, round(seconds)) + if total < 60: + return f"{total}s" + hours, rem = divmod(total, 3600) + minutes, secs = divmod(rem, 60) + if hours: + return f"{hours}h {minutes:02d}m" + return f"{minutes}m {secs:02d}s" + + +def _pct(part: float, whole: float) -> str: + if whole <= 0: + return "0%" + return f"{round(100 * part / whole)}%" + + +def _median(values: Iterable[float]) -> float: + values = list(values) + return statistics.median(values) if values else 0.0 + + +def _category_line( + name: str, seconds: float, count: int, denom: float, *, indent: str = " " +) -> str: + return ( + f"{indent}{name:<14}{_fmt_duration(seconds):>10} ({_pct(seconds, denom):>4}) " + f"{count} call{'s' if count != 1 else ''}" + ) + + +def _top_line( + label: str, seconds: float, denom: float | None = None, *, extra: str = "", width: int = 19 +) -> str: + """One of the top-level summary lines (``total wall``/``operator wait``/…), label left-padded + to ``width`` so every value column lines up regardless of label length. ``denom`` omitted + (``total wall`` itself, always 100%) skips the percentage.""" + pct = f" ({_pct(seconds, denom)})" if denom is not None else "" + return f" {label + ':':<{width}}{_fmt_duration(seconds):>10}{pct}{extra}" + + +def format_task_profile( + task: JsonObj, profile: dict[str, Any], *, repo_name: str | None = None +) -> str: + """The ``panopticon profile `` compact breakdown: total wall, operator-wait, + between-sessions, then agent-active split by category, then the top 5 longest tool calls.""" + total = profile["total_wall_s"] + active = profile["agent_active_s"] + header = f"Task {task['id']} ({task.get('slug') or '-'})" + if repo_name: + header += f" — {repo_name}" + lines = [ + header, + _top_line("total wall", total), + _top_line("operator wait", profile["operator_wait_s"], total), + ] + if profile["between_sessions_count"]: + n = profile["between_sessions_count"] + lines.append( + _top_line( + "between sessions", + profile["between_sessions_s"], + total, + extra=f" [{n} gap{'s' if n != 1 else ''}]", + ) + ) + if profile["unattributed_s"] > 0: + lines.append(_top_line("unattributed", profile["unattributed_s"], total)) + lines.append(_top_line("agent active", active, total)) + lines.append(_category_line("llm", profile["llm_s"], profile["llm_count"], active)) + for name in TOOL_CATEGORIES: + bucket = profile["categories"][name] + lines.append(_category_line(name, bucket["seconds"], bucket["count"], active)) + + if profile["unmatched_tool_calls"]: + n = profile["unmatched_tool_calls"] + lines.append( + f" ({n} tool call{'s' if n != 1 else ''} never got a result — transcript cut off)" + ) + + top = profile["top_tool_calls"] + if top: + lines += ["", " top 5 longest tool calls:"] + for i, call in enumerate(top, start=1): + lines.append( + f" {i}. {call['category']:<14}{_fmt_duration(call['duration_s']):>8} {call['first_line']}" + ) + return "\n".join(lines) + + +def format_time_summary(profile: dict[str, Any]) -> str: + """The one-line dashboard detail-pane summary, e.g. + ``agent 4.3h: llm 43% tests 19% tools 38% | waited on user 9.2h (+1.1h between sessions)``. + + Deliberately terse: only ``llm``/``tests`` are broken out (the operator's two named concerns — + "waiting on the LLM" vs "waiting on tests"), everything else agent-active is lumped into + ``tools``. The full per-category breakdown is the CLI's job (:func:`format_task_profile`).""" + active = profile["agent_active_s"] + llm_s = profile["llm_s"] + tests_s = profile["categories"]["tests"]["seconds"] + tools_s = max(0.0, active - llm_s - tests_s) + hours = active / 3600 + summary = f"agent {hours:.1f}h: llm {_pct(llm_s, active)} tests {_pct(tests_s, active)} tools {_pct(tools_s, active)}" + wait = f"waited on user {profile['operator_wait_s'] / 3600:.1f}h" + if profile["between_sessions_s"] > 0: + wait += f" (+{profile['between_sessions_s'] / 3600:.1f}h between sessions)" + return f"{summary} | {wait}" + + +def aggregate_repo_profiles(profiles: Sequence[dict[str, Any]]) -> dict[str, Any]: + """Combine several tasks' profile dicts (all from one repo) into totals + medians — the input + to :func:`format_all_tasks_report`. Sums answer "what fraction of time is X" questions; + medians answer "how long does a typical task here take".""" + n = len(profiles) + if n == 0: + return {"task_count": 0} + totals = { + "total_wall_s": sum(p["total_wall_s"] for p in profiles), + "operator_wait_s": sum(p["operator_wait_s"] for p in profiles), + "between_sessions_s": sum(p["between_sessions_s"] for p in profiles), + "agent_active_s": sum(p["agent_active_s"] for p in profiles), + "llm_s": sum(p["llm_s"] for p in profiles), + "llm_count": sum(p["llm_count"] for p in profiles), + "unattributed_s": sum(p["unattributed_s"] for p in profiles), + } + categories = { + name: { + "seconds": sum(p["categories"][name]["seconds"] for p in profiles), + "count": sum(p["categories"][name]["count"] for p in profiles), + } + for name in TOOL_CATEGORIES + } + medians = { + "total_wall_s": _median(p["total_wall_s"] for p in profiles), + "operator_wait_s": _median(p["operator_wait_s"] for p in profiles), + "agent_active_s": _median(p["agent_active_s"] for p in profiles), + } + return {"task_count": n, "totals": totals, "categories": categories, "medians": medians} + + +def format_all_tasks_report( + per_repo: dict[str, dict[str, Any]], *, skipped: dict[str, int] | None = None +) -> str: + """``panopticon profile --all-tasks``: one section per repo, sorted by name — median task + shape, then each agent-active category as both a share of total agent-active time (the "what + fraction of X's agent time is pytest" answer) and its raw total.""" + skipped = skipped or {} + sections = [] + for repo_name in sorted(per_repo): + agg = per_repo[repo_name] + n = agg.get("task_count", 0) + if n == 0: + continue + skip_n = skipped.get(repo_name, 0) + totals, medians, categories = agg["totals"], agg["medians"], agg["categories"] + active_total = totals["agent_active_s"] + lines = [ + f"repo {repo_name} ({n} task{'s' if n != 1 else ''} profiled" + + (f", {skip_n} skipped: no transcripts found)" if skip_n else ")"), + _top_line("median wall", medians["total_wall_s"], width=23), + _top_line( + "median operator wait", + medians["operator_wait_s"], + medians["total_wall_s"], + width=23, + ), + _top_line( + "median agent active", medians["agent_active_s"], medians["total_wall_s"], width=23 + ), + "", + f" agent-active time by category (of {_fmt_duration(active_total)} total across {n} tasks):", + _category_line( + "llm", totals["llm_s"], totals["llm_count"], active_total, indent=" " + ), + ] + for name in TOOL_CATEGORIES: + bucket = categories[name] + lines.append( + _category_line( + name, bucket["seconds"], bucket["count"], active_total, indent=" " + ) + ) + sections.append("\n".join(lines)) + return "\n\n".join(sections) + + +def _find_task(tasks: list[JsonObj], ref: str) -> JsonObj | None: + return next((t for t in tasks if t.get("id") == ref or t.get("slug") == ref), None) + + +def run_profile_command( + client: TaskServiceClient, + *, + task_ref: str | None, + all_tasks: bool, + session_paths: Callable[[str], list[Path]] = task_session_paths, + stdout: TextIO | None = None, + stderr: TextIO | None = None, +) -> int: + """Drive ``panopticon profile `` / ``panopticon profile --all-tasks``. + + ``session_paths`` is injectable (defaults to the real docker-volume reader) so this is + unit-testable without docker. Returns a process exit code; never raises for an ordinary + "no such task"/"no transcripts" outcome — those print to ``stderr`` and return 1. + + ``stdout``/``stderr`` default to ``None`` and are resolved to ``sys.stdout``/``sys.stderr`` + *inside* the function body, not as parameter defaults — a default is bound once at import + time, which would print past a test's ``capsys`` (it patches ``sys.stdout`` after import).""" + stdout = stdout if stdout is not None else sys.stdout + stderr = stderr if stderr is not None else sys.stderr + tasks = client.list_tasks() + repo_names = {str(r["id"]): str(r["name"]) for r in client.list_repos()} + + if all_tasks: + by_repo: dict[str, list[dict[str, Any]]] = {} + skipped: dict[str, int] = {} + for one_task in tasks: + one_repo_name = repo_names.get( + str(one_task.get("repo_id")), str(one_task.get("repo_id")) + ) + paths = session_paths(str(one_task["id"])) + if not paths: + skipped[one_repo_name] = skipped.get(one_repo_name, 0) + 1 + continue + by_repo.setdefault(one_repo_name, []).append(profile_transcripts(paths)) + if not by_repo: + print("No tasks with session transcripts found.", file=stdout) + return 0 + per_repo = {name: aggregate_repo_profiles(profiles) for name, profiles in by_repo.items()} + print(format_all_tasks_report(per_repo, skipped=skipped), file=stdout) + return 0 + + if not task_ref: + print("usage: panopticon profile | --all-tasks", file=stderr) + return 2 + task = _find_task(tasks, task_ref) + if task is None: + print(f"No such task: {task_ref}", file=stderr) + return 1 + paths = session_paths(str(task["id"])) + if not paths: + print(f"No session transcripts found for {task_ref}.", file=stderr) + return 1 + profile = profile_transcripts(paths) + repo_name = repo_names.get(str(task.get("repo_id"))) + print(format_task_profile(task, profile, repo_name=repo_name), file=stdout) + return 0 + + +__all__ = [ + "aggregate_repo_profiles", + "format_all_tasks_report", + "format_task_profile", + "format_time_summary", + "run_profile_command", +] diff --git a/tests/profiler/test_categories.py b/tests/profiler/test_categories.py new file mode 100644 index 00000000..44050c50 --- /dev/null +++ b/tests/profiler/test_categories.py @@ -0,0 +1,67 @@ +"""The category table's matcher rules, in isolation from the gap-analysis walk.""" + +from __future__ import annotations + +from panopticon.profiler.categories import ASK_USER_QUESTION_TOOL, TOOL_CATEGORIES, categorize + + +def test_bash_pytest_is_tests() -> None: + assert categorize("Bash", {"command": "uv run pytest tests/test_foo.py -x"}) == "tests" + + +def test_bash_git_and_gh_are_vcs() -> None: + assert categorize("Bash", {"command": "git status"}) == "vcs" + assert categorize("Bash", {"command": "gh pr create --title x"}) == "vcs" + + +def test_bash_deps_install_is_deps() -> None: + assert categorize("Bash", {"command": "uv add httpx"}) == "deps" + assert categorize("Bash", {"command": "uv sync"}) == "deps" + assert categorize("Bash", {"command": "pip install -r requirements.txt"}) == "deps" + + +def test_bash_pexpect_pyte_verify_skill_are_pty_verify() -> None: + assert categorize("Bash", {"command": "python3 -c 'import pexpect'"}) == "pty-verify" + assert categorize("Bash", {"command": "python3 -c 'import pyte'"}) == "pty-verify" + assert categorize("Bash", {"command": "./verify-skill.sh"}) == "pty-verify" + + +def test_bash_gh_does_not_match_substring_words() -> None: + # `gh` must match as a whole word — a var named `gh_token` shouldn't trip the vcs bucket. + assert categorize("Bash", {"command": "echo $gh_token"}) == "other-tools" + + +def test_bash_with_no_matching_pattern_is_other_tools() -> None: + assert categorize("Bash", {"command": "echo hello"}) == "other-tools" + + +def test_bash_with_missing_command_is_other_tools() -> None: + assert categorize("Bash", {}) == "other-tools" + assert categorize("Bash", None) == "other-tools" + + +def test_read_write_edit_grep_glob_are_code_nav() -> None: + for name in ("Read", "Write", "Edit", "Grep", "Glob"): + assert categorize(name, {}) == "code-nav" + + +def test_task_and_agent_are_subagents() -> None: + assert categorize("Task", {"description": "research"}) == "subagents" + assert categorize("Agent", {"description": "research"}) == "subagents" + + +def test_mcp_prefixed_tools_are_orchestration() -> None: + assert categorize("mcp__panopticon__set_slug", {}) == "orchestration" + assert categorize("mcp__panopticon__advance", {}) == "orchestration" + + +def test_unknown_tool_falls_through_to_other_tools() -> None: + assert categorize("SomeFutureTool", {}) == "other-tools" + assert categorize("", {}) == "other-tools" + + +def test_ask_user_question_is_not_a_categorize_result() -> None: + # AskUserQuestion is routed to operator-wait by the parser, not classified as a tool category — + # categorize() itself would fall through to other-tools if ever called on it directly. + assert ASK_USER_QUESTION_TOOL not in TOOL_CATEGORIES + assert categorize(ASK_USER_QUESTION_TOOL, {}) == "other-tools" diff --git a/tests/profiler/test_parse.py b/tests/profiler/test_parse.py new file mode 100644 index 00000000..4eacfc64 --- /dev/null +++ b/tests/profiler/test_parse.py @@ -0,0 +1,248 @@ +"""The gap-analysis algorithm, over a hand-built fixture transcript. + +Covers every category, an ``AskUserQuestion`` span (must land in operator-wait, not a tool +category), a ``Task`` subagent call (one bucket — we never walk its sidechain), parallel tool +calls in one turn, a same-``message.id`` multi-line assistant turn (thinking → text → tool_use), a +mid-session restart (two consecutive ``user`` lines, no assistant turn between), a cross-file +session restart (between-sessions), and defensive old-format/malformed input.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from panopticon.profiler.parse import profile_transcripts + + +def _line(**kwargs: Any) -> str: + return json.dumps(kwargs) + + +def _user(ts: str, content: Any) -> str: + return _line(type="user", timestamp=ts, message={"role": "user", "content": content}) + + +def _assistant(ts: str, msg_id: str, content: list[dict[str, Any]]) -> str: + return _line( + type="assistant", + timestamp=ts, + message={"id": msg_id, "role": "assistant", "content": content}, + ) + + +def _tool_use(tool_id: str, name: str, input_: dict[str, Any]) -> dict[str, Any]: + return {"type": "tool_use", "id": tool_id, "name": name, "input": input_} + + +def _tool_result(tool_id: str) -> dict[str, Any]: + return {"type": "tool_result", "tool_use_id": tool_id, "content": "ok"} + + +#: One task's worth of fixture data across two session files, exercising every bucket the profiler +#: recognizes. Timestamps are hand-picked so each gap's expected classification/duration is +#: unambiguous — see the assertions below for the derivation of each number. +def _session_one() -> str: + lines = [ + # t=0: real human prompt. + _user("2026-03-01T00:00:00.000Z", "start the task"), + # A turn streamed across 3 lines (thinking, text, tool_use) sharing one message id — the + # whole span (0->10s from the prior boundary) must land in llm_s, not just the gap to the + # first block. + _assistant("2026-03-01T00:00:05.000Z", "m1", [{"type": "thinking", "thinking": "..."}]), + _assistant( + "2026-03-01T00:00:06.000Z", "m1", [{"type": "text", "text": "ok, running tests"}] + ), + _assistant( + "2026-03-01T00:00:10.000Z", + "m1", + [_tool_use("t1", "Bash", {"command": "uv run pytest tests/test_foo.py"})], + ), + # tests: 12s (10 -> 22) + _user("2026-03-01T00:00:22.000Z", [_tool_result("t1")]), + # vcs (git, 1s), deps (uv sync, 1s), pty-verify (pexpect, 1s), code-nav (Read, 1s), + # orchestration (mcp__, 1s) — a parallel-call turn, 5 tool_use blocks all at the same ts. + _assistant( + "2026-03-01T00:00:23.000Z", + "m2", + [ + _tool_use("t2", "Bash", {"command": "git status"}), + _tool_use("t3", "Bash", {"command": "uv add httpx"}), + _tool_use("t4", "Bash", {"command": "python3 -c 'import pexpect'"}), + _tool_use("t5", "Read", {"file_path": "/workspace/foo.py"}), + _tool_use("t6", "mcp__panopticon__set_slug", {"slug": "x"}), + ], + ), + _user( + "2026-03-01T00:00:24.000Z", + [ + _tool_result("t2"), + _tool_result("t3"), + _tool_result("t4"), + _tool_result("t5"), + _tool_result("t6"), + ], + ), + # AskUserQuestion: 90s (24 -> 114) — must be operator-wait, not other-tools. + _assistant( + "2026-03-01T00:00:24.500Z", + "m3", + [_tool_use("q1", "AskUserQuestion", {"questions": []})], + ), + _user("2026-03-01T00:01:54.500Z", [_tool_result("q1")]), + # subagent (Task): 600s (95 -> 695) — one bucket; its own sidechain is never opened. + _assistant( + "2026-03-01T00:01:55.000Z", + "m4", + [_tool_use("sub1", "Task", {"description": "research the bug"})], + ), + _user("2026-03-01T00:11:55.000Z", [_tool_result("sub1")]), + # unknown/future tool name -> other-tools (1s: 697 -> 698) + _assistant("2026-03-01T00:11:57.000Z", "m5", [_tool_use("t7", "FutureTool", {})]), + _user("2026-03-01T00:11:58.000Z", [_tool_result("t7")]), + # The agent stops (no tool_use) — its own restart mid-session: the abandoned tool_result + # above is immediately followed by claude's synthetic "interrupted" prompt with NO + # assistant turn between them (2000s, 11:58 -> 44:58) — must land in operator_wait, not + # vanish into unattributed. + _assistant("2026-03-01T00:11:59.000Z", "m6", [{"type": "text", "text": "done for now"}]), + _user("2026-03-01T00:45:19.000Z", "You were interrupted. Continue."), + # trailing unmatched tool_use (transcript "cuts off") — never closed. + _assistant( + "2026-03-01T00:45:20.000Z", "m7", [_tool_use("orphan", "Bash", {"command": "sleep 1"})] + ), + # defensive garbage: never crashes the parser. + "", + "not json at all", + _line( + type="assistant", + timestamp=None, + message={"id": "m8", "role": "assistant", "content": []}, + ), + ] + return "\n".join(lines) + + +def _session_two() -> str: + # A second session file — a restart between sessions (the whole file, not a mid-session gap). + return "\n".join( + [ + _user("2026-03-01T02:00:00.000Z", "let's continue"), + _assistant("2026-03-01T02:00:03.000Z", "n1", [{"type": "text", "text": "ok"}]), + ] + ) + + +def _profile(tmp_path: Path) -> dict[str, Any]: + p1 = tmp_path / "session-1.jsonl" + p1.write_text(_session_one()) + p2 = tmp_path / "session-2.jsonl" + p2.write_text(_session_two()) + # Pass paths out of chronological order — profile_transcripts must sort them itself. + return profile_transcripts([p2, p1]) + + +def test_tests_category(tmp_path: Path) -> None: + profile = _profile(tmp_path) + assert profile["categories"]["tests"] == {"seconds": 12.0, "count": 1} + + +def test_vcs_deps_pty_verify_code_nav_orchestration_categories(tmp_path: Path) -> None: + profile = _profile(tmp_path) + for name in ("vcs", "deps", "pty-verify", "code-nav", "orchestration"): + assert profile["categories"][name] == {"seconds": 1.0, "count": 1}, name + + +def test_other_tools_category(tmp_path: Path) -> None: + profile = _profile(tmp_path) + assert profile["categories"]["other-tools"] == {"seconds": 1.0, "count": 1} + + +def test_subagent_task_call_is_one_bucket(tmp_path: Path) -> None: + profile = _profile(tmp_path) + assert profile["categories"]["subagents"] == {"seconds": 600.0, "count": 1} + top = profile["top_tool_calls"][0] + assert top["category"] == "subagents" + assert top["duration_s"] == 600.0 + assert top["first_line"] == "research the bug" + + +def test_ask_user_question_is_operator_wait_not_a_tool_category(tmp_path: Path) -> None: + profile = _profile(tmp_path) + # AskUserQuestion (90s) + the mid-session restart gap (2000s) both land in operator_wait. + assert profile["operator_wait_s"] == 90.0 + 2000.0 + assert profile["operator_wait_count"] == 2 + # And it must never leak into other-tools (which is exactly 1 call/2s, the FutureTool call). + assert profile["categories"]["other-tools"]["count"] == 1 + + +def test_multiline_assistant_turn_bills_whole_span_to_llm_time(tmp_path: Path) -> None: + profile = _profile(tmp_path) + # The first llm gap spans the merged m1 turn (thinking @5, text @6, tool_use @10) from the + # t=0 human prompt: 10s, not just the 5s to the turn's first block. + assert profile["llm_count"] >= 1 + assert profile["unattributed_s"] == 0.0 + + +def test_between_sessions_gap(tmp_path: Path) -> None: + profile = _profile(tmp_path) + assert profile["session_count"] == 2 + assert profile["between_sessions_count"] == 1 + # session 1 ends at 00:45:20 (the trailing orphan tool_use); session 2 starts at 02:00:00. + assert profile["between_sessions_s"] > 3600 + + +def test_unmatched_trailing_tool_use_is_counted_not_crashed(tmp_path: Path) -> None: + profile = _profile(tmp_path) + assert profile["unmatched_tool_calls"] == 1 + + +def test_malformed_and_old_format_lines_never_crash(tmp_path: Path) -> None: + # blank line, non-JSON line, and a null-timestamp assistant line are all present in the + # fixture (see _session_one) — profile_transcripts must tolerate every one of them. + profile = _profile(tmp_path) + assert profile["total_wall_s"] > 0 + + +def test_top_five_longest_tool_calls_sorted_descending(tmp_path: Path) -> None: + profile = _profile(tmp_path) + durations = [c["duration_s"] for c in profile["top_tool_calls"]] + assert durations == sorted(durations, reverse=True) + assert len(profile["top_tool_calls"]) <= 5 + + +def test_accounted_time_reconciles_to_wall_time(tmp_path: Path) -> None: + """Every second is accounted for — with one documented exception: the fixture's m2 turn fires + 5 *parallel* tool_use calls (vcs/deps/pty-verify/code-nav/orchestration) that each bill their + own full 1s duration even though they overlap the same real wall-clock second, so summed + category time can exceed total wall time by exactly that overlap (4 extra tool-seconds for 1 + real second here). ``unattributed_s`` is clamped at 0 rather than going negative for it.""" + profile = _profile(tmp_path) + tool_categories_s = sum(b["seconds"] for b in profile["categories"].values()) + accounted = ( + tool_categories_s + + profile["llm_s"] + + profile["operator_wait_s"] + + profile["between_sessions_s"] + ) + assert profile["unattributed_s"] == 0.0 + assert accounted - profile["total_wall_s"] == 4.0 # the 5-parallel-calls-in-1s overlap + + +def test_empty_input_never_crashes() -> None: + profile = profile_transcripts([]) + assert profile["total_wall_s"] == 0.0 + assert profile["session_count"] == 0 + assert profile["top_tool_calls"] == [] + + +def test_missing_file_never_crashes(tmp_path: Path) -> None: + profile = profile_transcripts([tmp_path / "does-not-exist.jsonl"]) + assert profile["total_wall_s"] == 0.0 + + +def test_entirely_garbage_file_never_crashes(tmp_path: Path) -> None: + p = tmp_path / "garbage.jsonl" + p.write_text('not json\n\n{}\n{"type": "assistant"}\n') + profile = profile_transcripts([p]) + assert profile["total_wall_s"] == 0.0 + assert profile["session_count"] == 0 diff --git a/tests/sessionservice/test_transcripts.py b/tests/sessionservice/test_transcripts.py new file mode 100644 index 00000000..c258ebcb --- /dev/null +++ b/tests/sessionservice/test_transcripts.py @@ -0,0 +1,162 @@ +"""task_session_paths: unit tests pin the emitted docker commands via a fake runner; one +integration test exercises a real docker volume (skipped when docker is unavailable).""" + +from __future__ import annotations + +import shutil +import subprocess +from collections.abc import Sequence +from pathlib import Path + +import pytest + +from panopticon.sessionservice.transcripts import task_session_paths + + +class _FakeRunner: + """An injectable CommandRunner: records every call and answers from a canned script.""" + + def __init__( + self, + *, + volume_exists: bool = True, + listing: str = "", + files: dict[str, str] | None = None, + ) -> None: + self.calls: list[tuple[list[str], bool]] = [] + self._volume_exists = volume_exists + self._listing = listing + self._files = files or {} + + def __call__( + self, + args: Sequence[str], + *, + check: bool = True, + interactive: bool = False, + verbose: bool = False, + ) -> str: + args = list(args) + self.calls.append((args, check)) + if args[:3] == ["docker", "volume", "inspect"]: + if not self._volume_exists: + raise subprocess.CalledProcessError(1, args) + return "" + if "find" in args: + return self._listing + if "cat" in args: + path = args[-1] + return self._files.get(Path(path).name, "") + return "" + + +def test_checks_volume_exists_before_anything_else(tmp_path: Path) -> None: + rec = _FakeRunner(volume_exists=True, listing="a.jsonl\n", files={"a.jsonl": '{"x": 1}\n'}) + paths = task_session_paths("t1", dest=tmp_path, run=rec) + inspect_call, _ = rec.calls[0] + assert inspect_call == ["docker", "volume", "inspect", "panopticon-config-t1"] + assert [p.name for p in paths] == ["a.jsonl"] + assert (tmp_path / "a.jsonl").read_text() == '{"x": 1}\n' + + +def test_missing_volume_returns_empty_without_a_docker_run(tmp_path: Path) -> None: + rec = _FakeRunner(volume_exists=False) + assert task_session_paths("nope", dest=tmp_path, run=rec) == [] + # Only the inspect call — a bare `docker run --volume ` would silently create the + # volume, so we must never reach that call when inspect fails. + assert len(rec.calls) == 1 + assert rec.calls[0][0] == ["docker", "volume", "inspect", "panopticon-config-nope"] + + +def test_no_docker_binary_returns_empty(tmp_path: Path) -> None: + def raising_run(args: Sequence[str], *, check: bool = True, **kwargs: object) -> str: + raise FileNotFoundError("docker") + + assert task_session_paths("t1", dest=tmp_path, run=raising_run) == [] + + +def test_finds_and_reads_every_session_file(tmp_path: Path) -> None: + rec = _FakeRunner( + volume_exists=True, + listing="session-b.jsonl\nsession-a.jsonl\n", + files={"session-a.jsonl": "a\n", "session-b.jsonl": "b\n"}, + ) + paths = task_session_paths("t1", dest=tmp_path, run=rec) + contents = {p.name: p.read_text() for p in paths} + assert contents == {"session-a.jsonl": "a\n", "session-b.jsonl": "b\n"} + + +def test_emits_the_expected_docker_run_argv(tmp_path: Path) -> None: + rec = _FakeRunner(volume_exists=True, listing="a.jsonl\n", files={"a.jsonl": "x\n"}) + task_session_paths("t1", dest=tmp_path, run=rec, image="my-image") + find_call, find_check = next(c for c in rec.calls if "find" in c[0]) + assert find_call == [ + "docker", + "run", + "--rm", + "--volume", + "panopticon-config-t1:/home/panopticon/.claude:ro", + "my-image", + "find", + "/home/panopticon/.claude/projects/-workspace", + "-maxdepth", + "1", + "-name", + "*.jsonl", + ] + assert find_check is False # tolerate an existing-but-empty volume (no such directory) + cat_call, _ = next(c for c in rec.calls if "cat" in c[0]) + assert cat_call[-2:] == ["cat", "/home/panopticon/.claude/projects/-workspace/a.jsonl"] + + +def test_empty_listing_returns_no_paths(tmp_path: Path) -> None: + rec = _FakeRunner(volume_exists=True, listing="") + assert task_session_paths("t1", dest=tmp_path, run=rec) == [] + + +def test_default_dest_is_a_fresh_tempdir() -> None: + rec = _FakeRunner(volume_exists=True, listing="a.jsonl\n", files={"a.jsonl": "x\n"}) + paths = task_session_paths("t1", run=rec) + assert paths[0].exists() + assert paths[0].read_text() == "x\n" + + +# -- integration: a real docker volume ---------------------------------------------- + +_HAVE_DOCKER = bool(shutil.which("docker")) + + +def _docker_running() -> bool: + return _HAVE_DOCKER and subprocess.run(["docker", "info"], capture_output=True).returncode == 0 + + +@pytest.mark.skipif(not _docker_running(), reason="needs a working docker daemon") +def test_real_volume_round_trip(tmp_path: Path) -> None: + volume = "panopticon-config-itest-transcripts" + subprocess.run(["docker", "volume", "rm", "--force", volume], capture_output=True) + subprocess.run(["docker", "volume", "create", volume], check=True, capture_output=True) + try: + # Write a fixture transcript into the volume via a throwaway container — mirroring where + # the real claude session would have written it. + subprocess.run( + [ + "docker", + "run", + "--rm", + "--volume", + f"{volume}:/home/panopticon/.claude", + "alpine", + "sh", + "-c", + "mkdir -p /home/panopticon/.claude/projects/-workspace && " + 'echo \'{"type": "user"}\' > ' + "/home/panopticon/.claude/projects/-workspace/session-1.jsonl", + ], + check=True, + capture_output=True, + ) + paths = task_session_paths("itest-transcripts", dest=tmp_path, image="alpine") + assert [p.name for p in paths] == ["session-1.jsonl"] + assert paths[0].read_text().strip() == '{"type": "user"}' + finally: + subprocess.run(["docker", "volume", "rm", "--force", volume], capture_output=True) diff --git a/tests/terminal/test_dashboard.py b/tests/terminal/test_dashboard.py index 5da3abfc..d9607b72 100644 --- a/tests/terminal/test_dashboard.py +++ b/tests/terminal/test_dashboard.py @@ -311,6 +311,16 @@ def test_render_detail_marks_blocked() -> None: assert "turn: agent (blocked)" in render_detail({**_TASK, "blocked": True}) +def test_render_detail_shows_the_time_summary_only_when_supplied() -> None: + # time_summary is opt-in and caller-supplied (never computed inside render_detail — see `P` / + # action_profile) so a task with no profile yet just omits the line. + assert "waited on user" not in render_detail(_TASK) + out = render_detail( + _TASK, time_summary="agent 1.2h: llm 38% tests 27% tools 35% | waited on user 4.6h" + ) + assert "agent 1.2h: llm 38% tests 27% tools 35% | waited on user 4.6h" in out + + def test_short_tokens_formats_human_short() -> None: assert _short_tokens(None) == "-" # not yet reported assert _short_tokens(0) == "-" @@ -908,6 +918,41 @@ async def test_pressing_shift_y_copies_the_id(monkeypatch: Any) -> None: assert copied == ["task-abcdef0123"] +async def test_pressing_shift_p_profiles_and_shows_the_summary( + monkeypatch: Any, tmp_path: Path +) -> None: + # `P` shells out to docker (via task_session_paths) on demand — never automatically, since + # that'd be too slow to redo on every highlight. Fake it out here; the real read + gap-analysis + # are covered in test_transcripts.py / test_parse.py. + transcript = tmp_path / "s1.jsonl" + transcript.write_text( + '{"type": "user", "timestamp": "2026-01-01T00:00:00.000Z", ' + '"message": {"role": "user", "content": "hi"}}\n' + ) + monkeypatch.setattr(dashboard, "task_session_paths", lambda task_id: [transcript]) + app = Dashboard(_FakeClient([_TASK])) # type: ignore[arg-type] + async with app.run_test() as pilot: + await pilot.pause() + detail = app.query_one("#detail", Static) + assert detail.styles.display == "none" # hidden by default + await pilot.press("P") + await pilot.pause() + assert detail.styles.display == "block" # `P` reveals it so the summary is visible + assert "agent " in str(detail.render()) + assert "waited on user" in str(detail.render()) + + +async def test_pressing_shift_p_with_no_transcripts_warns(monkeypatch: Any) -> None: + monkeypatch.setattr(dashboard, "task_session_paths", lambda task_id: []) + app = Dashboard(_FakeClient([_TASK])) # type: ignore[arg-type] + async with app.run_test() as pilot: + await pilot.pause() + await pilot.press("P") + await pilot.pause() + assert app.is_running # warns, doesn't crash + assert app.query_one("#detail", Static).styles.display == "none" # never revealed + + def test_render_detail_shows_the_claim() -> None: assert "claimed:" not in render_detail(_TASK) assert "claimed: host-1" in render_detail({**_TASK, "claimed_by": "host-1"}) @@ -2170,7 +2215,7 @@ def test_footer_shows_only_the_essential_keys() -> None: shown = {b.key for b in Dashboard.BINDINGS if b.show} hidden = {b.key for b in Dashboard.BINDINGS if not b.show} assert shown == {"t", "n", "x", "/", "d", "question_mark", "q"} - assert hidden == {"o", "r", "R", "p", "g", "a", "s", "u", "y", "Y", "escape"} + assert hidden == {"o", "r", "R", "p", "g", "a", "s", "u", "y", "Y", "P", "escape"} def test_bindings_and_help_derive_from_the_single_hotkey_table() -> None: diff --git a/tests/terminal/test_task_profile.py b/tests/terminal/test_task_profile.py new file mode 100644 index 00000000..fdfc91e9 --- /dev/null +++ b/tests/terminal/test_task_profile.py @@ -0,0 +1,243 @@ +"""Golden-output tests for the profile renderers, and unit tests for `run_profile_command`'s +wiring (fake client + injected session-path lookup — no real docker).""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from panopticon.terminal.task_profile import ( + aggregate_repo_profiles, + format_all_tasks_report, + format_task_profile, + format_time_summary, + run_profile_command, +) + + +def _profile(**overrides: Any) -> dict[str, Any]: + """A hand-built profile dict with clean round numbers, so the golden strings below are easy + to hand-verify and stay decoupled from any change to the parser's own arithmetic.""" + base: dict[str, Any] = { + "total_wall_s": 52320.0, # 14h 32m + "operator_wait_s": 33000.0, # 9h 10m + "operator_wait_count": 4, + "between_sessions_s": 3900.0, # 1h 05m + "between_sessions_count": 2, + "agent_active_s": 15420.0, # 4h 17m + "llm_s": 6720.0, # 1h 52m + "llm_count": 112, + "categories": { + "tests": {"seconds": 2880.0, "count": 23}, # 48m + "pty-verify": {"seconds": 540.0, "count": 3}, # 9m + "code-nav": {"seconds": 1860.0, "count": 87}, # 31m + "vcs": {"seconds": 1320.0, "count": 19}, # 22m + "deps": {"seconds": 0.0, "count": 0}, + "subagents": {"seconds": 2460.0, "count": 4}, # 41m + "orchestration": {"seconds": 360.0, "count": 14}, # 6m + "other-tools": {"seconds": 480.0, "count": 11}, # 8m + }, + "unattributed_s": 0.0, + "unmatched_tool_calls": 0, + "session_count": 3, + "top_tool_calls": [ + { + "category": "tests", + "tool_name": "Bash", + "first_line": "uv run pytest tests/ -x", + "duration_s": 723.0, + }, + { + "category": "subagents", + "tool_name": "Task", + "first_line": "research the schema", + "duration_s": 580.0, + }, + ], + } + base.update(overrides) + return base + + +def test_format_task_profile_golden_output() -> None: + task = {"id": "8e3e412d431e4c99b8a27a321eb8a7a2", "slug": "profiler-slice"} + out = format_task_profile(task, _profile(), repo_name="panopticon") + assert out == ( + "Task 8e3e412d431e4c99b8a27a321eb8a7a2 (profiler-slice) — panopticon\n" + " total wall: 14h 32m\n" + " operator wait: 9h 10m (63%)\n" + " between sessions: 1h 05m (7%) [2 gaps]\n" + " agent active: 4h 17m (29%)\n" + " llm 1h 52m ( 44%) 112 calls\n" + " tests 48m 00s ( 19%) 23 calls\n" + " pty-verify 9m 00s ( 4%) 3 calls\n" + " code-nav 31m 00s ( 12%) 87 calls\n" + " vcs 22m 00s ( 9%) 19 calls\n" + " deps 0s ( 0%) 0 calls\n" + " subagents 41m 00s ( 16%) 4 calls\n" + " orchestration 6m 00s ( 2%) 14 calls\n" + " other-tools 8m 00s ( 3%) 11 calls\n" + "\n" + " top 5 longest tool calls:\n" + " 1. tests 12m 03s uv run pytest tests/ -x\n" + " 2. subagents 9m 40s research the schema" + ) + + +def test_format_task_profile_shows_unattributed_only_when_nonzero() -> None: + task = {"id": "t1", "slug": None} + assert "unattributed:" not in format_task_profile(task, _profile(), repo_name=None) + with_unattributed = format_task_profile(task, _profile(unattributed_s=60.0), repo_name=None) + assert "unattributed: 1m 00s (0%)" in with_unattributed + + +def test_format_task_profile_notes_unmatched_tool_calls() -> None: + task = {"id": "t1", "slug": None} + out = format_task_profile(task, _profile(unmatched_tool_calls=1), repo_name=None) + assert "(1 tool call never got a result — transcript cut off)" in out + + +def test_format_task_profile_falls_back_to_id_with_no_slug() -> None: + task = {"id": "t1", "slug": None} + out = format_task_profile(task, _profile(), repo_name=None) + assert out.startswith("Task t1 (-)") + + +def test_format_time_summary_golden_output() -> None: + assert format_time_summary(_profile()) == ( + "agent 4.3h: llm 44% tests 19% tools 38% | waited on user 9.2h (+1.1h between sessions)" + ) + + +def test_format_time_summary_omits_between_sessions_when_zero() -> None: + summary = format_time_summary(_profile(between_sessions_s=0.0)) + assert "between sessions" not in summary + assert summary.endswith("waited on user 9.2h") + + +def test_aggregate_repo_profiles_sums_and_medians() -> None: + agg = aggregate_repo_profiles( + [_profile(), _profile(total_wall_s=10000.0, operator_wait_s=5000.0, agent_active_s=5000.0)] + ) + assert agg["task_count"] == 2 + assert agg["totals"]["total_wall_s"] == 62320.0 + assert agg["medians"]["total_wall_s"] == (52320.0 + 10000.0) / 2 + + +def test_aggregate_repo_profiles_of_empty_list() -> None: + assert aggregate_repo_profiles([]) == {"task_count": 0} + + +def test_format_all_tasks_report_golden_output() -> None: + agg = aggregate_repo_profiles([_profile(), _profile()]) + out = format_all_tasks_report({"panopticon": agg}, skipped={"panopticon": 3}) + assert out == ( + "repo panopticon (2 tasks profiled, 3 skipped: no transcripts found)\n" + " median wall: 14h 32m\n" + " median operator wait: 9h 10m (63%)\n" + " median agent active: 4h 17m (29%)\n" + "\n" + " agent-active time by category (of 8h 34m total across 2 tasks):\n" + " llm 3h 44m ( 44%) 224 calls\n" + " tests 1h 36m ( 19%) 46 calls\n" + " pty-verify 18m 00s ( 4%) 6 calls\n" + " code-nav 1h 02m ( 12%) 174 calls\n" + " vcs 44m 00s ( 9%) 38 calls\n" + " deps 0s ( 0%) 0 calls\n" + " subagents 1h 22m ( 16%) 8 calls\n" + " orchestration 12m 00s ( 2%) 28 calls\n" + " other-tools 16m 00s ( 3%) 22 calls" + ) + + +def test_format_all_tasks_report_skips_repos_with_no_tasks() -> None: + out = format_all_tasks_report({"empty-repo": {"task_count": 0}}) + assert out == "" + + +# -- run_profile_command: wiring, no real docker ------------------------------------- + + +class _FakeClient: + def __init__(self, tasks: list[dict[str, Any]], repos: list[dict[str, Any]]) -> None: + self._tasks = tasks + self._repos = repos + + def list_tasks(self) -> list[dict[str, Any]]: + return self._tasks + + def list_repos(self) -> list[dict[str, Any]]: + return self._repos + + +def test_run_profile_command_resolves_by_slug_and_prints(tmp_path: Path, capsys: Any) -> None: + client = _FakeClient( + tasks=[{"id": "t1", "slug": "my-task", "repo_id": "r1"}], + repos=[{"id": "r1", "name": "panopticon"}], + ) + transcript = tmp_path / "s1.jsonl" + transcript.write_text( + '{"type": "user", "timestamp": "2026-01-01T00:00:00.000Z", "message": {"role": "user", "content": "hi"}}\n' + ) + + rc = run_profile_command( + client, task_ref="my-task", all_tasks=False, session_paths=lambda task_id: [transcript] + ) + out = capsys.readouterr().out + assert rc == 0 + assert "Task t1 (my-task) — panopticon" in out + + +def test_run_profile_command_unknown_task_id(capsys: Any) -> None: + client = _FakeClient(tasks=[], repos=[]) + rc = run_profile_command(client, task_ref="nope", all_tasks=False, session_paths=lambda t: []) + err = capsys.readouterr().err + assert rc == 1 + assert "No such task: nope" in err + + +def test_run_profile_command_no_transcripts(capsys: Any) -> None: + client = _FakeClient(tasks=[{"id": "t1", "slug": "s", "repo_id": "r1"}], repos=[]) + rc = run_profile_command(client, task_ref="s", all_tasks=False, session_paths=lambda t: []) + err = capsys.readouterr().err + assert rc == 1 + assert "No session transcripts found" in err + + +def test_run_profile_command_no_task_ref_and_not_all_tasks(capsys: Any) -> None: + client = _FakeClient(tasks=[], repos=[]) + rc = run_profile_command(client, task_ref=None, all_tasks=False, session_paths=lambda t: []) + assert rc == 2 + assert "usage:" in capsys.readouterr().err + + +def test_run_profile_command_all_tasks_aggregates_per_repo(tmp_path: Path, capsys: Any) -> None: + client = _FakeClient( + tasks=[ + {"id": "t1", "slug": "a", "repo_id": "r1"}, + {"id": "t2", "slug": "b", "repo_id": "r1"}, + {"id": "t3", "slug": "c", "repo_id": "r2"}, # no transcripts -> skipped + ], + repos=[{"id": "r1", "name": "panopticon"}, {"id": "r2", "name": "tarot"}], + ) + transcript = tmp_path / "s.jsonl" + transcript.write_text( + '{"type": "user", "timestamp": "2026-01-01T00:00:00.000Z", "message": {"role": "user", "content": "hi"}}\n' + ) + + def session_paths(task_id: str) -> list[Path]: + return [transcript] if task_id in ("t1", "t2") else [] + + rc = run_profile_command(client, task_ref=None, all_tasks=True, session_paths=session_paths) + out = capsys.readouterr().out + assert rc == 0 + assert "repo panopticon (2 tasks profiled)" in out + assert "tarot" not in out # zero profiled tasks for tarot -> no section at all + + +def test_run_profile_command_all_tasks_with_nothing_profiled(capsys: Any) -> None: + client = _FakeClient(tasks=[{"id": "t1", "slug": "a", "repo_id": "r1"}], repos=[]) + rc = run_profile_command(client, task_ref=None, all_tasks=True, session_paths=lambda t: []) + out = capsys.readouterr().out + assert rc == 0 + assert "No tasks with session transcripts found." in out diff --git a/tests/terminal/test_terminal.py b/tests/terminal/test_terminal.py index fb230bc0..eb4fe600 100644 --- a/tests/terminal/test_terminal.py +++ b/tests/terminal/test_terminal.py @@ -4,6 +4,7 @@ from __future__ import annotations from typing import Any +from unittest.mock import patch import pytest @@ -22,6 +23,22 @@ def test_cli_tasks_lists(capsys: pytest.CaptureFixture[str]) -> None: assert "t1" in out and "ITERATING" in out and "agent" in out +def test_cli_profile_dispatches_task_and_all_tasks_flag() -> None: + # `main` only wires argparse -> run_profile_command; the command's own logic (task/slug + # resolution, formatting) is covered directly in test_task_profile.py. + with patch("panopticon.terminal.task_profile.run_profile_command", return_value=0) as mock_run: + assert cli.main(["profile", "my-task"], client=_FakeClient()) == 0 # type: ignore[arg-type] + _client, kwargs = mock_run.call_args.args[0], mock_run.call_args.kwargs + assert kwargs["task_ref"] == "my-task" + assert kwargs["all_tasks"] is False + + with patch("panopticon.terminal.task_profile.run_profile_command", return_value=0) as mock_run: + assert cli.main(["profile", "--all-tasks"], client=_FakeClient()) == 0 # type: ignore[arg-type] + kwargs = mock_run.call_args.kwargs + assert kwargs["task_ref"] is None + assert kwargs["all_tasks"] is True + + def test_dashboard_under_supervisor_wires_the_switch_hooks(monkeypatch: pytest.MonkeyPatch) -> None: # With --switch-file (set by the supervisor, ADR 0009 §6) the dashboard is wired with the # `t` (on_switch), `s` (on_service), and `u` (on_runner) hooks; the dashboard stays running.