diff --git a/skills/pr-cost/INSTALL.md b/skills/pr-cost/INSTALL.md new file mode 100644 index 0000000..3f42aa7 --- /dev/null +++ b/skills/pr-cost/INSTALL.md @@ -0,0 +1,68 @@ +# Install PR cost hooks + +Annotates a newly created GitHub PR with estimated AI session cost. Default is +**ledger only** (`~/.local/share/pr-cost/ledger.jsonl`). GitHub comments stay +off unless `PR_COST_HOOK_LIVE=1`. + +Collector: + +```bash +/opt/homebrew/bin/python3 skills/pr-cost/scripts/pr_cost_collect.py from-hook --harness +``` + +## Cursor (installed on this machine) + +User-global: + +- `~/.cursor/hooks.json` — `afterShellExecution` matcher `\bgh\s+pr\s+create\b` +- `~/.cursor/hooks/pr-cost-from-hook.sh` — fail-open wrapper → collector `--harness cursor` + +Reload: Cursor watches `hooks.json`. If it does not fire, restart Cursor and +check the Hooks output channel. + +Versioned copy: `adapters/cursor/`. + +## Claude Code (installed on this machine) + +- `~/.claude/settings.json` `hooks.PostToolUse` matcher `Bash(gh pr create:*)` +- `~/.claude/hooks/pr-cost-from-hook.sh` → `adapters/claude/v1/pr_cost_from_hook.py` + +Existing worklog `PreCompact` / `SessionEnd` hooks must stay. Do not set +`attribution.pr`. + +## Codex (opt-in, not on PATH) + +Codex has no native PR-create hook. Prepend only in shells where Codex runs `gh`: + +```bash +export PATH="$HOME/Documents/oss/dotfiles/skills/pr-cost/adapters/codex/bin:$PATH" +``` + +See `adapters/codex/README.md`. Do not shadow `/opt/homebrew/bin/gh` globally. + +## Enable live PR comments (off by default) + +```bash +export PR_COST_HOOK_LIVE=1 +``` + +The collector posts an idempotent `gh pr comment`. Duplicate `pr_url` + +`session_id` rows are skipped. + +To dogfood a **Claude** cost comment on an open PR, paste +`handovers/claude-comment-pr-cost.md` into a new Claude Code session. That +prompt sums `message.usage` from the session JSONL and runs `annotate` with +`PR_COST_HOOK_LIVE=1`. + +Codex sessions expose running totals as `event_msg.type = token_count` in +`~/.codex/sessions/**/rollout-*.jsonl`. Read the last one with +`scripts/codex_session_usage.py`. + +## Verify without a live PR + +```bash +/opt/homebrew/bin/python3 -m unittest discover -s skills/pr-cost/tests -q +``` + +That suite includes a `from-hook` fixture that writes a temp ledger and fails +if `gh` is invoked while `PR_COST_HOOK_LIVE` is unset. diff --git a/skills/pr-cost/SKILL.md b/skills/pr-cost/SKILL.md new file mode 100644 index 0000000..1a4d338 --- /dev/null +++ b/skills/pr-cost/SKILL.md @@ -0,0 +1,115 @@ +--- +name: pr-cost +description: Collect a typed AI cost payload for a newly created GitHub PR, persist it to a local ledger, and optionally post an idempotent PR comment when live writes are explicitly enabled. +--- + +# pr-cost + +Use this skill from harness-specific hook adapters after a successful `gh pr create`. +It is dry by default: it always prefers the local ledger, and it only writes a +GitHub PR comment when `PR_COST_HOOK_LIVE=1`. + +Install and verify: [INSTALL.md](INSTALL.md). Adapters live in `adapters/{cursor,claude,codex}/`. + +To comment Claude cost on an already-open PR, paste +[handovers/claude-comment-pr-cost.md](handovers/claude-comment-pr-cost.md) into a +fresh Claude Code session. + +## Files + +- Collector: `scripts/pr_cost_collect.py` +- Tests: `tests/test_pr_cost_collect.py` +- Fixtures: `tests/fixtures/` + +## Contract + +The collector emits one JSON object with this required shape: + +```json +{ + "schema_version": "pr-cost/v1", + "harness": "claude | cursor | codex", + "confidence": "metered | estimated | unavailable", + "usd": 1.23, + "tokens_in": 1200, + "tokens_out": 3400, + "model": "claude-sonnet-4-20250514", + "session_id": "session-123", + "window_start": "2026-08-20T19:00:00+00:00", + "window_end": "2026-08-20T19:05:00+00:00", + "pr_url": "https://github.com/owner/repo/pull/123", + "generated_at": "2026-08-20T19:05:01+00:00", + "notes": "optional" +} +``` + +`usd`, `tokens_in`, `tokens_out`, `model`, `session_id`, `pr_url`, and `notes` +may be `null` when the harness cannot supply them. The keys still remain +present so downstream adapters receive a stable typed contract. + +## Privacy rules + +- Never copy prompts, responses, file contents, or shell output beyond the PR URL. +- Never store API keys, tokens, auth headers, or repo-local secrets. +- Prefer safe metadata only: harness, model, token counts, session identifier, + bounded timestamps, PR URL, and a short note about confidence. +- Cursor and Codex adapters should treat unavailable data as `null`, not as a + reason to scrape unrelated local state. + +## Harness guidance + +- `cursor`: hook payload can detect `gh pr create`, but it does not expose + token or USD usage. Default confidence is `unavailable`. +- `claude`: `PostToolUse` can observe `gh pr create`. If an adapter already has + token or pricing inputs, pass them as CLI flags so the collector can emit an + `estimated` payload. Otherwise it will fall back to `unavailable`. +- `codex`: there is no native PR creation hook. Use a wrapper that feeds a + matching hook JSON shape to `from-hook`, or call `emit` / `annotate` + directly with explicit payload fields. + +## Environment + +- `PR_COST_LEDGER`: optional ledger override. Defaults to + `~/.local/share/pr-cost/ledger.jsonl`. +- `PR_COST_HOOK_LIVE=1`: enables live `gh pr comment` writes. Unset keeps the + collector dry and ledger-only. + +## Commands + +Validate and print a payload: + +```bash +/opt/homebrew/bin/python3 scripts/pr_cost_collect.py emit \ + --harness claude \ + --confidence estimated \ + --usd 1.23 \ + --tokens-in 1200 \ + --tokens-out 3400 \ + --model claude-sonnet-4-20250514 \ + --session-id session-123 \ + --window-start 2026-08-20T19:00:00+00:00 \ + --window-end 2026-08-20T19:05:00+00:00 \ + --pr-url https://github.com/owner/repo/pull/123 +``` + +Append to the ledger and optionally comment on the PR: + +```bash +/opt/homebrew/bin/python3 scripts/pr_cost_collect.py annotate \ + --fixture tests/fixtures/emit_valid.json +``` + +Run from a hook adapter by piping the native hook JSON to stdin: + +```bash +printf '%s\n' '{"command":"gh pr create ...","exit_code":0,"stdout":"https://github.com/owner/repo/pull/123"}' \ + | /opt/homebrew/bin/python3 scripts/pr_cost_collect.py from-hook \ + --harness cursor +``` + +`from-hook` is fail-open by design: + +- It ignores `gh pr view`, `gh pr comment`, and unrelated commands. +- It exits `0` on parse failures so the harness never blocks PR creation. +- It skips duplicate annotations when the ledger already contains the same + `pr_url` and `session_id`. diff --git a/skills/pr-cost/adapters/claude/v1/pr_cost_from_hook.py b/skills/pr-cost/adapters/claude/v1/pr_cost_from_hook.py new file mode 100755 index 0000000..c96bf31 --- /dev/null +++ b/skills/pr-cost/adapters/claude/v1/pr_cost_from_hook.py @@ -0,0 +1,125 @@ +#!/opt/homebrew/bin/python3 +"""Normalize Claude PostToolUse payloads for the shared PR cost collector.""" + +from __future__ import annotations + +import json +import subprocess +import sys +from typing import Any + + +PYTHON = "/opt/homebrew/bin/python3" +COLLECTOR = "/Users/fredtran/Documents/oss/dotfiles/skills/pr-cost/scripts/pr_cost_collect.py" + + +def _clean_string(value: Any) -> str | None: + if not isinstance(value, str): + return None + stripped = value.strip() + return stripped or None + + +def _clean_int(value: Any) -> int | None: + if isinstance(value, bool) or not isinstance(value, int): + return None + return value + + +def _clean_number(value: Any) -> float | int | None: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + return value + + +def _exit_code(tool_response: dict[str, Any]) -> int | None: + explicit_exit = _clean_int(tool_response.get("exit_code")) + if explicit_exit is not None: + return explicit_exit + if tool_response.get("interrupted") is True: + return 130 + return None + + +def normalize_payload(payload: dict[str, Any]) -> dict[str, Any]: + tool_input = payload.get("tool_input") + if not isinstance(tool_input, dict): + tool_input = {} + + tool_response = payload.get("tool_response") + if not isinstance(tool_response, dict): + tool_response = {} + + stdout = tool_response.get("stdout") + normalized = { + "command": _clean_string(tool_input.get("command")), + "stdout": stdout if isinstance(stdout, str) else "", + "exit_code": _exit_code(tool_response), + } + + stderr = tool_response.get("stderr") + if isinstance(stderr, str): + normalized["stderr"] = stderr + + return normalized + + +def collector_command(payload: dict[str, Any]) -> list[str]: + command = [PYTHON, COLLECTOR, "from-hook", "--harness", "claude"] + + session_id = _clean_string(payload.get("session_id")) + if session_id is not None: + command.extend(["--session-id", session_id]) + + model = _clean_string(payload.get("model")) + if model is not None: + command.extend(["--model", model]) + + generated_at = _clean_string(payload.get("timestamp")) + if generated_at is not None: + command.extend(["--generated-at", generated_at]) + + usd = _clean_number(payload.get("usd")) + if usd is not None: + command.extend(["--usd", str(usd)]) + + tokens_in = _clean_int(payload.get("tokens_in")) + if tokens_in is not None: + command.extend(["--tokens-in", str(tokens_in)]) + + tokens_out = _clean_int(payload.get("tokens_out")) + if tokens_out is not None: + command.extend(["--tokens-out", str(tokens_out)]) + + return command + + +def main() -> int: + raw_input = sys.stdin.read() + if not raw_input.strip(): + return 0 + + try: + payload = json.loads(raw_input) + except json.JSONDecodeError: + return 0 + + if not isinstance(payload, dict): + return 0 + + try: + subprocess.run( + collector_command(payload), + input=json.dumps(normalize_payload(payload)), + text=True, + capture_output=True, + check=False, + ) + except Exception: + return 0 + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/skills/pr-cost/adapters/codex/README.md b/skills/pr-cost/adapters/codex/README.md new file mode 100644 index 0000000..bcedb74 --- /dev/null +++ b/skills/pr-cost/adapters/codex/README.md @@ -0,0 +1,47 @@ +# Codex PR-cost adapter + +Codex does not expose a native PR-creation hook, so this adapter uses the +least-bad fallback from the survey: an opt-in `gh` wrapper that watches for +successful `gh pr create` commands and forwards a hook-shaped JSON payload to: + +```bash +/opt/homebrew/bin/python3 /Users/fredtran/Documents/oss/dotfiles/skills/pr-cost/scripts/pr_cost_collect.py from-hook --harness codex +``` + +The wrapper is versioned under this skill instead of inventing new +`~/.codex/config.toml` keys or replacing Codex's existing `notify` behavior. + +## Install + +Prepend this adapter directory to `PATH` for the shell where Codex runs `gh`: + +```bash +export PATH="/Users/fredtran/Documents/oss/dotfiles/skills/pr-cost/adapters/codex/bin:$PATH" +``` + +If the real GitHub CLI is not `/opt/homebrew/bin/gh`, point the wrapper at it +explicitly: + +```bash +export PR_COST_REAL_GH="/absolute/path/to/gh" +``` + +This task does not install the wrapper globally, does not edit `~/.codex`, and +does not set `PR_COST_HOOK_LIVE`. + +## Behavior + +- Calls the real `gh` binary and preserves its exit code, stdout, and stderr. +- Only invokes the collector for `gh pr create`. +- Uses the collector's existing fail-open `from-hook` path, so annotation + failures never block PR creation. +- Leaves PR comments disabled unless someone explicitly exports + `PR_COST_HOOK_LIVE=1` outside this adapter. + +## Uninstall + +Remove the adapter directory from `PATH`, or unset the override if you set one: + +```bash +unset PR_COST_REAL_GH +``` diff --git a/skills/pr-cost/adapters/codex/bin/gh b/skills/pr-cost/adapters/codex/bin/gh new file mode 100755 index 0000000..3edc1f7 --- /dev/null +++ b/skills/pr-cost/adapters/codex/bin/gh @@ -0,0 +1,127 @@ +#!/opt/homebrew/bin/python3 +"""Opt-in gh wrapper that annotates successful Codex PR creates.""" + +from __future__ import annotations + +import json +import os +import pathlib +import shlex +import subprocess +import sys +from collections.abc import Iterable + +PYTHON_BIN = "/opt/homebrew/bin/python3" +COLLECTOR = pathlib.Path( + "/Users/fredtran/Documents/oss/dotfiles/skills/pr-cost/scripts/pr_cost_collect.py" +) +DEFAULT_GH_CANDIDATES = ( + pathlib.Path("/opt/homebrew/bin/gh"), + pathlib.Path("/usr/local/bin/gh"), + pathlib.Path("/usr/bin/gh"), +) + + +def iter_gh_candidates(self_path: pathlib.Path) -> Iterable[pathlib.Path]: + override = os.environ.get("PR_COST_REAL_GH") + if override: + yield pathlib.Path(override).expanduser() + return + + seen: set[pathlib.Path] = set() + for candidate in DEFAULT_GH_CANDIDATES: + try: + resolved = candidate.resolve() + except FileNotFoundError: + continue + if resolved == self_path or resolved in seen: + continue + seen.add(resolved) + yield candidate + + for path_entry in os.get_exec_path(): + candidate = pathlib.Path(path_entry) / "gh" + try: + resolved = candidate.resolve() + except FileNotFoundError: + continue + if resolved == self_path or resolved in seen: + continue + seen.add(resolved) + yield candidate + + +def resolve_real_gh(self_path: pathlib.Path) -> pathlib.Path: + for candidate in iter_gh_candidates(self_path): + if candidate.is_file() and os.access(candidate, os.X_OK): + return candidate + raise FileNotFoundError( + "unable to locate the real gh binary; set PR_COST_REAL_GH to an executable path" + ) + + +def is_pr_create(arguments: list[str]) -> bool: + for index, token in enumerate(arguments): + if token != "pr": + continue + return index + 1 < len(arguments) and arguments[index + 1] == "create" + return False + + +def maybe_collect(arguments: list[str], *, exit_code: int, stdout: str) -> None: + if not is_pr_create(arguments): + return + hook_payload = { + "command": shlex_join(["gh", *arguments]), + "exit_code": exit_code, + "stdout": stdout, + } + try: + subprocess.run( + [PYTHON_BIN, str(COLLECTOR), "from-hook", "--harness", "codex"], + input=json.dumps(hook_payload), + capture_output=True, + check=False, + text=True, + ) + except Exception: + return + + +def shlex_join(parts: list[str]) -> str: + # subprocess.run() receives argv, but the collector expects the hook-style + # command string that its existing shlex-based parser already understands. + # Using the same quoting rules here keeps the wrapper aligned with that API. + return " ".join(shlex.quote(part) for part in parts) + + +def main() -> int: + self_path = pathlib.Path(__file__).resolve() + try: + real_gh = resolve_real_gh(self_path) + except FileNotFoundError as exc: + print(f"pr-cost gh wrapper: {exc}", file=sys.stderr) + return 127 + + try: + result = subprocess.run( + [str(real_gh), *sys.argv[1:]], + capture_output=True, + check=False, + text=True, + ) + except OSError as exc: + print(f"pr-cost gh wrapper: failed to execute {real_gh}: {exc}", file=sys.stderr) + return 127 + + if result.stdout: + sys.stdout.write(result.stdout) + if result.stderr: + sys.stderr.write(result.stderr) + + maybe_collect(sys.argv[1:], exit_code=result.returncode, stdout=result.stdout) + return result.returncode + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/skills/pr-cost/adapters/cursor/README.md b/skills/pr-cost/adapters/cursor/README.md new file mode 100644 index 0000000..6b1ea56 --- /dev/null +++ b/skills/pr-cost/adapters/cursor/README.md @@ -0,0 +1,42 @@ +# Cursor PR cost hook adapter + +This adapter installs a user-global Cursor hook that watches successful +`gh pr create` shell commands and forwards the hook JSON to the shared +`pr_cost_collect.py` collector. + +## Files + +- `hooks.json`: sample user-level Cursor hook config +- `pr-cost-from-hook.sh`: thin wrapper that reads hook stdin JSON and pipes it + to the shared collector with `--harness cursor` + +## Install + +Cursor user hooks run from `~/.cursor/`, so the live config should be: + +- `~/.cursor/hooks.json` +- `~/.cursor/hooks/pr-cost-from-hook.sh` + +Install by copying these files into `~/.cursor/`, or point your existing +`~/.cursor/hooks.json` entry at the shared wrapper path if you prefer to keep +the dotfiles copy as the source of truth. + +The recommended `afterShellExecution` hook entry is: + +```json +{ + "version": 1, + "hooks": { + "afterShellExecution": [ + { + "command": "./hooks/pr-cost-from-hook.sh", + "matcher": "\\bgh\\s+pr\\s+create\\b", + "failClosed": false + } + ] + } +} +``` + +The wrapper intentionally fails open, emits only `{}` for Cursor's hook +response, and does not set `PR_COST_HOOK_LIVE`. diff --git a/skills/pr-cost/adapters/cursor/hooks.json b/skills/pr-cost/adapters/cursor/hooks.json new file mode 100644 index 0000000..d510b5a --- /dev/null +++ b/skills/pr-cost/adapters/cursor/hooks.json @@ -0,0 +1,12 @@ +{ + "version": 1, + "hooks": { + "afterShellExecution": [ + { + "command": "./hooks/pr-cost-from-hook.sh", + "matcher": "\\bgh\\s+pr\\s+create\\b", + "failClosed": false + } + ] + } +} diff --git a/skills/pr-cost/adapters/cursor/pr-cost-from-hook.sh b/skills/pr-cost/adapters/cursor/pr-cost-from-hook.sh new file mode 100755 index 0000000..084b717 --- /dev/null +++ b/skills/pr-cost/adapters/cursor/pr-cost-from-hook.sh @@ -0,0 +1,10 @@ +#!/bin/sh + +stdin_json=$(cat) + +printf '%s' "$stdin_json" | /opt/homebrew/bin/python3 \ + /Users/fredtran/Documents/oss/dotfiles/skills/pr-cost/scripts/pr_cost_collect.py \ + from-hook \ + --harness cursor >/dev/null 2>&1 || true + +printf '{}\n' diff --git a/skills/pr-cost/handovers/claude-comment-pr-cost.md b/skills/pr-cost/handovers/claude-comment-pr-cost.md new file mode 100644 index 0000000..38c43d8 --- /dev/null +++ b/skills/pr-cost/handovers/claude-comment-pr-cost.md @@ -0,0 +1,52 @@ +You're picking up a task cold. Assume no prior session memory. + +Project root: /Users/fredtran/Documents/oss +Relevant repos: + - /Users/fredtran/Documents/oss/dotfiles + +Read these first: + - /Users/fredtran/Documents/oss/dotfiles/skills/pr-cost/SKILL.md + - /Users/fredtran/Documents/oss/dotfiles/skills/pr-cost/INSTALL.md + +Task: + Comment the **Claude Code** AI cost for this PR onto the PR itself. Do not + change product code. Do not rewrite the PR body. + + 1. Resolve the PR URL (this dogfood PR): + `https://github.com/cheshirecode/dotfiles/pull/31` + or `gh pr view 31 --repo cheshirecode/dotfiles --json url -q .url` + 2. Find this Claude session JSONL. Prefer `$CLAUDE_PROJECT_DIR` transcripts + under `~/.claude/projects/` matching cwd `dotfiles` or the current + session id. If several files match, use the one whose `sessionId` matches + this conversation. + 3. Run: + `/opt/homebrew/bin/python3 /Users/fredtran/Documents/oss/dotfiles/skills/pr-cost/scripts/claude_session_usage.py --jsonl ` + 4. Live-annotate (this is an explicit write to GitHub): + +```bash +export VIRTUAL_ENV= +export PATH="/opt/homebrew/bin:/usr/bin:/bin:$PATH" +export PR_COST_HOOK_LIVE=1 +PR_URL="$(gh pr view 31 --repo cheshirecode/dotfiles --json url -q .url)" +USAGE="$(/opt/homebrew/bin/python3 /Users/fredtran/Documents/oss/dotfiles/skills/pr-cost/scripts/claude_session_usage.py --jsonl "$JSONL")" +/opt/homebrew/bin/python3 /Users/fredtran/Documents/oss/dotfiles/skills/pr-cost/scripts/pr_cost_collect.py annotate \ + --harness claude \ + --confidence estimated \ + --usd "$(python3 -c 'import json,sys; print(json.load(sys.stdin)["usd_estimated"])' <<<"$USAGE")" \ + --tokens-in "$(python3 -c 'import json,sys; print(json.load(sys.stdin)["tokens_in"])' <<<"$USAGE")" \ + --tokens-out "$(python3 -c 'import json,sys; print(json.load(sys.stdin)["tokens_out"])' <<<"$USAGE")" \ + --model "$(python3 -c 'import json,sys; print(json.load(sys.stdin)["model"] or "claude")' <<<"$USAGE")" \ + --session-id "$(python3 -c 'import json,sys; print(json.load(sys.stdin)["session_id"] or "unknown")' <<<"$USAGE")" \ + --window-start "$(python3 -c 'import json,sys; print(json.load(sys.stdin)["window_start"])' <<<"$USAGE")" \ + --window-end "$(python3 -c 'import json,sys; print(json.load(sys.stdin)["window_end"])' <<<"$USAGE")" \ + --pr-url "$PR_URL" \ + --notes "Claude Code session usage summed from unique assistant message.usage on the JSONL. Cache read/write included in tokens_in. USD uses default Opus-class rates in claude_session_usage.py." +``` + + Privacy: do not paste prompts, diffs, or file contents into the PR comment. + The collector already wraps a JSON payload. If annotate reports + `"status": "duplicate"`, stop — the Claude cost is already on the PR. + +Deliverable: + One PR comment from harness=claude. Print the annotate JSON and the comment + URL. Do not push or open extra PRs. diff --git a/skills/pr-cost/scripts/claude_session_usage.py b/skills/pr-cost/scripts/claude_session_usage.py new file mode 100755 index 0000000..235b425 --- /dev/null +++ b/skills/pr-cost/scripts/claude_session_usage.py @@ -0,0 +1,119 @@ +#!/usr/bin/env python3 +"""Sum unique assistant-message usage from a Claude Code session JSONL.""" + +from __future__ import annotations + +import argparse +import json +import pathlib +import sys +from typing import Any + + +def session_usage(path: pathlib.Path) -> dict[str, Any]: + messages: dict[str, dict[str, Any]] = {} + session_id = None + model = None + window_start = None + window_end = None + with path.open(encoding="utf-8") as handle: + for line in handle: + line = line.strip() + if not line: + continue + try: + event = json.loads(line) + except json.JSONDecodeError: + continue + if not isinstance(event, dict): + continue + session_id = event.get("sessionId") or session_id + timestamp = event.get("timestamp") + if isinstance(timestamp, str): + window_start = window_start or timestamp + window_end = timestamp + message = event.get("message") + if not isinstance(message, dict) or message.get("role") != "assistant": + continue + message_id = message.get("id") + usage = message.get("usage") + if not isinstance(message_id, str) or not isinstance(usage, dict): + continue + messages[message_id] = usage + if isinstance(message.get("model"), str): + model = message["model"] + + tokens_in = 0 + tokens_out = 0 + cache_read = 0 + cache_write = 0 + for usage in messages.values(): + tokens_in += int(usage.get("input_tokens") or 0) + tokens_out += int(usage.get("output_tokens") or 0) + cache_read += int(usage.get("cache_read_input_tokens") or 0) + cache_write += int(usage.get("cache_creation_input_tokens") or 0) + + return { + "session_id": session_id, + "model": model, + "tokens_in": tokens_in + cache_read + cache_write, + "tokens_out": tokens_out, + "uncached_input_tokens": tokens_in, + "cache_read_input_tokens": cache_read, + "cache_creation_input_tokens": cache_write, + "unique_assistant_messages": len(messages), + "window_start": window_start, + "window_end": window_end, + "path": str(path), + } + + +def estimate_usd( + *, + uncached: int, + cache_read: int, + cache_write: int, + tokens_out: int, + input_rate: float, + output_rate: float, + cache_read_rate: float, + cache_write_rate: float, +) -> float: + return round( + ( + uncached * input_rate + + cache_read * cache_read_rate + + cache_write * cache_write_rate + + tokens_out * output_rate + ) + / 1_000_000, + 4, + ) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--jsonl", type=pathlib.Path, required=True) + parser.add_argument("--input-usd-per-mtok", type=float, default=5.0) + parser.add_argument("--output-usd-per-mtok", type=float, default=25.0) + parser.add_argument("--cache-read-usd-per-mtok", type=float, default=0.5) + parser.add_argument("--cache-write-usd-per-mtok", type=float, default=6.25) + args = parser.parse_args() + usage = session_usage(args.jsonl) + usage["usd_estimated"] = estimate_usd( + uncached=int(usage["uncached_input_tokens"]), + cache_read=int(usage["cache_read_input_tokens"]), + cache_write=int(usage["cache_creation_input_tokens"]), + tokens_out=int(usage["tokens_out"]), + input_rate=args.input_usd_per_mtok, + output_rate=args.output_usd_per_mtok, + cache_read_rate=args.cache_read_usd_per_mtok, + cache_write_rate=args.cache_write_usd_per_mtok, + ) + json.dump(usage, sys.stdout, indent=2, sort_keys=True) + sys.stdout.write("\n") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/skills/pr-cost/scripts/codex_session_usage.py b/skills/pr-cost/scripts/codex_session_usage.py new file mode 100755 index 0000000..cc4260b --- /dev/null +++ b/skills/pr-cost/scripts/codex_session_usage.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python3 +"""Read the last Codex token_count event from a session JSONL.""" + +from __future__ import annotations + +import argparse +import json +import pathlib +import sys +from typing import Any + + +def last_token_count(path: pathlib.Path) -> dict[str, Any]: + session_id = None + cwd = None + window_start = None + model = None + last_usage: dict[str, Any] | None = None + last_timestamp = None + with path.open(encoding="utf-8") as handle: + for line in handle: + line = line.strip() + if not line: + continue + try: + event = json.loads(line) + except json.JSONDecodeError: + continue + payload = event.get("payload") if isinstance(event, dict) else None + if not isinstance(payload, dict): + continue + event_type = event.get("type") + if event_type == "session_meta": + session_id = payload.get("session_id") or session_id + cwd = payload.get("cwd") or cwd + window_start = payload.get("timestamp") or event.get("timestamp") or window_start + model = payload.get("model") or payload.get("model_provider") or model + if event_type == "event_msg" and payload.get("type") == "token_count": + info = payload.get("info") if isinstance(payload.get("info"), dict) else {} + usage = info.get("total_token_usage") if isinstance(info.get("total_token_usage"), dict) else {} + if usage: + last_usage = usage + last_timestamp = event.get("timestamp") + if last_usage is None: + raise SystemExit(f"no token_count events in {path}") + return { + "session_id": session_id, + "cwd": cwd, + "model": model, + "tokens_in": last_usage.get("input_tokens"), + "tokens_out": last_usage.get("output_tokens"), + "cached_input_tokens": last_usage.get("cached_input_tokens"), + "window_start": window_start, + "window_end": last_timestamp, + "path": str(path), + } + + +def latest_session(root: pathlib.Path) -> pathlib.Path: + files = sorted(root.rglob("rollout-*.jsonl")) + if not files: + raise SystemExit(f"no rollout-*.jsonl under {root}") + return files[-1] + + +def estimate_usd(tokens_in: int | None, tokens_out: int | None, input_rate: float, output_rate: float) -> float | None: + if tokens_in is None or tokens_out is None: + return None + return round((tokens_in * input_rate + tokens_out * output_rate) / 1_000_000, 4) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--path", type=pathlib.Path, help="session JSONL; default = latest under --root") + parser.add_argument( + "--root", + type=pathlib.Path, + default=pathlib.Path.home() / ".codex" / "sessions", + help="directory to search for rollout-*.jsonl", + ) + parser.add_argument("--input-usd-per-mtok", type=float, default=5.0) + parser.add_argument("--output-usd-per-mtok", type=float, default=30.0) + args = parser.parse_args() + path = args.path or latest_session(args.root) + usage = last_token_count(path) + usage["usd_estimated"] = estimate_usd( + usage.get("tokens_in"), + usage.get("tokens_out"), + args.input_usd_per_mtok, + args.output_usd_per_mtok, + ) + json.dump(usage, sys.stdout, indent=2, sort_keys=True) + sys.stdout.write("\n") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/skills/pr-cost/scripts/pr_cost_collect.py b/skills/pr-cost/scripts/pr_cost_collect.py new file mode 100644 index 0000000..edc5a9f --- /dev/null +++ b/skills/pr-cost/scripts/pr_cost_collect.py @@ -0,0 +1,436 @@ +#!/usr/bin/env python3 +"""Collect and annotate per-PR AI cost payloads.""" + +from __future__ import annotations + +import argparse +import json +import os +import pathlib +import re +import shlex +import subprocess +import sys +import tempfile +from datetime import datetime, timezone +from typing import Any + + +SCHEMA_VERSION = "pr-cost/v1" +HARNESSES = {"claude", "cursor", "codex"} +CONFIDENCE_LEVELS = {"metered", "estimated", "unavailable"} +DEFAULT_LEDGER = "~/.local/share/pr-cost/ledger.jsonl" +PR_URL_PATTERN = re.compile(r"https://github\.com/[^/\s]+/[^/\s]+/pull/\d+") + + +class PrCostError(ValueError): + """Raised when the collector receives invalid input.""" + + +def utc_now() -> str: + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +def is_number(value: Any) -> bool: + return isinstance(value, (int, float)) and not isinstance(value, bool) + + +def is_iso_timestamp(value: str) -> bool: + try: + datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return False + return True + + +def validate_nullable_string(payload: dict[str, Any], field: str) -> None: + value = payload.get(field) + if value is None: + return + if not isinstance(value, str) or not value.strip(): + raise PrCostError(f"{field} must be a non-empty string or null") + + +def validate_nullable_integer(payload: dict[str, Any], field: str) -> None: + value = payload.get(field) + if value is None: + return + if not isinstance(value, int) or isinstance(value, bool): + raise PrCostError(f"{field} must be an integer or null") + + +def validate_payload(payload: dict[str, Any]) -> dict[str, Any]: + required_fields = ( + "schema_version", + "harness", + "confidence", + "usd", + "tokens_in", + "tokens_out", + "model", + "session_id", + "window_start", + "window_end", + "pr_url", + "generated_at", + ) + for field in required_fields: + if field not in payload: + raise PrCostError(f"missing required field: {field}") + + if payload["schema_version"] != SCHEMA_VERSION: + raise PrCostError(f"schema_version must be {SCHEMA_VERSION}") + if payload["harness"] not in HARNESSES: + raise PrCostError("harness must be one of claude, cursor, codex") + if payload["confidence"] not in CONFIDENCE_LEVELS: + raise PrCostError("confidence must be metered, estimated, or unavailable") + + usd = payload["usd"] + if usd is not None and not is_number(usd): + raise PrCostError("usd must be a number or null") + validate_nullable_integer(payload, "tokens_in") + validate_nullable_integer(payload, "tokens_out") + validate_nullable_string(payload, "model") + validate_nullable_string(payload, "session_id") + validate_nullable_string(payload, "pr_url") + notes = payload.get("notes") + if notes is not None and (not isinstance(notes, str) or not notes.strip()): + raise PrCostError("notes must be a non-empty string or null") + + for field in ("window_start", "window_end", "generated_at"): + value = payload.get(field) + if not isinstance(value, str) or not value.strip() or not is_iso_timestamp(value): + raise PrCostError(f"{field} must be a valid ISO 8601 timestamp") + + pr_url = payload["pr_url"] + if pr_url is not None and not PR_URL_PATTERN.fullmatch(pr_url): + raise PrCostError("pr_url must be a GitHub pull request URL or null") + return payload + + +def read_json_file(path: pathlib.Path) -> dict[str, Any]: + try: + raw = json.loads(path.read_text()) + except FileNotFoundError as exc: + raise PrCostError(f"fixture does not exist: {path}") from exc + except json.JSONDecodeError as exc: + raise PrCostError(f"fixture is not valid JSON: {path}: {exc}") from exc + if not isinstance(raw, dict): + raise PrCostError(f"fixture must contain a JSON object: {path}") + return raw + + +def read_json_stdin() -> dict[str, Any]: + raw_stdin = sys.stdin.read().strip() + if not raw_stdin: + raise PrCostError("stdin JSON is required") + try: + value = json.loads(raw_stdin) + except json.JSONDecodeError as exc: + raise PrCostError(f"stdin is not valid JSON: {exc}") from exc + if not isinstance(value, dict): + raise PrCostError("stdin JSON must be an object") + return value + + +def default_confidence(harness: str, usd: float | int | None, tokens_known: bool) -> str: + if usd is not None: + return "estimated" + if harness == "claude" and tokens_known: + return "estimated" + return "unavailable" + + +def default_notes(harness: str, confidence: str) -> str | None: + if confidence != "unavailable": + return None + if harness == "cursor": + return "Cursor hook payloads do not expose cost or token usage." + if harness == "codex": + return "Codex has no native PR creation hook or local cost payload." + return "Hook payload did not include enough data to estimate session cost." + + +def payload_from_args( + args: argparse.Namespace, + *, + default_pr_url: str | None = None, +) -> dict[str, Any]: + payload: dict[str, Any] = {} + if getattr(args, "fixture", None): + payload.update(read_json_file(args.fixture)) + + generated_at = args.generated_at or payload.get("generated_at") or utc_now() + tokens_known = args.tokens_in is not None or args.tokens_out is not None + harness = args.harness or payload.get("harness") + if harness not in HARNESSES: + raise PrCostError("harness is required and must be one of claude, cursor, codex") + confidence = ( + args.confidence + or payload.get("confidence") + or default_confidence(harness, args.usd if args.usd is not None else payload.get("usd"), tokens_known) + ) + + payload.update( + { + "schema_version": SCHEMA_VERSION, + "harness": harness, + "confidence": confidence, + "usd": args.usd if args.usd is not None else payload.get("usd"), + "tokens_in": args.tokens_in if args.tokens_in is not None else payload.get("tokens_in"), + "tokens_out": args.tokens_out if args.tokens_out is not None else payload.get("tokens_out"), + "model": args.model if args.model is not None else payload.get("model"), + "session_id": args.session_id if args.session_id is not None else payload.get("session_id"), + "window_start": args.window_start or payload.get("window_start") or generated_at, + "window_end": args.window_end or payload.get("window_end") or generated_at, + "pr_url": args.pr_url if args.pr_url is not None else payload.get("pr_url", default_pr_url), + "generated_at": generated_at, + } + ) + + if args.notes is not None: + payload["notes"] = args.notes + elif "notes" not in payload: + payload["notes"] = default_notes(harness, confidence) + + return validate_payload(payload) + + +def ledger_path(argument: pathlib.Path | None) -> pathlib.Path: + configured = argument or pathlib.Path(os.environ.get("PR_COST_LEDGER", DEFAULT_LEDGER)).expanduser() + return configured.expanduser() + + +def load_ledger(path: pathlib.Path) -> list[dict[str, Any]]: + if not path.exists(): + return [] + rows: list[dict[str, Any]] = [] + for line_number, raw_line in enumerate(path.read_text().splitlines(), start=1): + if not raw_line.strip(): + continue + try: + parsed = json.loads(raw_line) + except json.JSONDecodeError as exc: + raise PrCostError(f"ledger line {line_number} is not valid JSON: {path}") from exc + if not isinstance(parsed, dict): + raise PrCostError(f"ledger line {line_number} must be a JSON object: {path}") + rows.append(parsed) + return rows + + +def same_annotation(existing: dict[str, Any], payload: dict[str, Any]) -> bool: + return ( + existing.get("pr_url") == payload.get("pr_url") + and existing.get("session_id") == payload.get("session_id") + ) + + +def append_ledger(path: pathlib.Path, payload: dict[str, Any]) -> bool: + rows = load_ledger(path) + if any(same_annotation(row, payload) for row in rows): + return False + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(payload, sort_keys=True)) + handle.write("\n") + return True + + +def comment_body(payload: dict[str, Any]) -> str: + session_marker = payload.get("session_id") or "unknown" + return ( + f"\n" + "AI cost payload for the session that created this PR:\n\n" + "```json\n" + f"{json.dumps(payload, indent=2, sort_keys=True)}\n" + "```" + ) + + +def maybe_comment_pr(payload: dict[str, Any], live: bool) -> bool: + if not live or not payload.get("pr_url"): + return False + subprocess.run( + [ + "gh", + "pr", + "comment", + payload["pr_url"], + "--body", + comment_body(payload), + ], + check=True, + capture_output=True, + text=True, + ) + return True + + +def command_emit(args: argparse.Namespace) -> dict[str, Any]: + return payload_from_args(args) + + +def command_annotate(args: argparse.Namespace) -> dict[str, Any]: + payload = payload_from_args(args) + target_ledger = ledger_path(args.ledger) + wrote_ledger = append_ledger(target_ledger, payload) + commented = False + if wrote_ledger: + commented = maybe_comment_pr(payload, live=os.environ.get("PR_COST_HOOK_LIVE") == "1") + return { + "status": "annotated" if wrote_ledger else "duplicate", + "ledger": str(target_ledger), + "commented": commented, + "payload": payload, + } + + +def extract_command(payload: dict[str, Any]) -> tuple[str | None, str | None, int | None]: + if isinstance(payload.get("command"), str): + return payload["command"], payload.get("stdout"), payload.get("exit_code") + tool_name = payload.get("tool_name") + tool_input = payload.get("tool_input") + tool_response = payload.get("tool_response") + if tool_name == "Shell" and isinstance(tool_input, dict) and isinstance(tool_response, dict): + return ( + tool_input.get("command"), + tool_response.get("stdout"), + tool_response.get("exit_code"), + ) + return None, None, None + + +def detect_harness(args: argparse.Namespace, hook_payload: dict[str, Any]) -> str: + if args.harness: + return args.harness + if "tool_name" in hook_payload: + return "claude" + if "command" in hook_payload: + return "cursor" + return "codex" + + +def is_pr_create_command(command: str) -> bool: + try: + tokens = shlex.split(command) + except ValueError: + return False + if "gh" not in tokens or "pr" not in tokens: + return False + for index, token in enumerate(tokens): + if token != "gh": + continue + remainder = tokens[index + 1 :] + if "pr" not in remainder: + continue + pr_index = remainder.index("pr") + if pr_index + 1 >= len(remainder): + continue + return remainder[pr_index + 1] == "create" + return False + + +def extract_pr_url(stdout: str | None) -> str | None: + if not stdout: + return None + match = PR_URL_PATTERN.search(stdout) + return match.group(0) if match else None + + +def fail_open(message: str, *, extra: dict[str, Any] | None = None) -> int: + response = {"status": "ignored", "reason": message} + if extra: + response.update(extra) + print(json.dumps(response, sort_keys=True)) + return 0 + + +def command_from_hook(args: argparse.Namespace) -> int: + try: + hook_payload = read_json_stdin() + command, stdout, exit_code = extract_command(hook_payload) + if not command or not is_pr_create_command(command): + return fail_open("not-pr-create") + if exit_code not in (0, None): + return fail_open("command-failed") + pr_url = extract_pr_url(stdout) + if pr_url is None: + return fail_open("missing-pr-url") + + args.harness = detect_harness(args, hook_payload) + args.pr_url = pr_url + payload = payload_from_args(args, default_pr_url=pr_url) + target_ledger = ledger_path(args.ledger) + wrote_ledger = append_ledger(target_ledger, payload) + commented = False + if wrote_ledger: + commented = maybe_comment_pr(payload, live=os.environ.get("PR_COST_HOOK_LIVE") == "1") + print( + json.dumps( + { + "status": "annotated" if wrote_ledger else "duplicate", + "ledger": str(target_ledger), + "commented": commented, + "payload": payload, + }, + sort_keys=True, + ) + ) + return 0 + except Exception as exc: # pragma: no cover - fail-open path is behaviorally required + return fail_open("error", extra={"detail": str(exc)}) + + +def add_payload_arguments(parser: argparse.ArgumentParser) -> None: + parser.add_argument("--fixture", type=pathlib.Path, help="read a payload fixture JSON object") + parser.add_argument("--harness", choices=sorted(HARNESSES)) + parser.add_argument("--confidence", choices=sorted(CONFIDENCE_LEVELS)) + parser.add_argument("--usd", type=float) + parser.add_argument("--tokens-in", type=int, dest="tokens_in") + parser.add_argument("--tokens-out", type=int, dest="tokens_out") + parser.add_argument("--model") + parser.add_argument("--session-id", dest="session_id") + parser.add_argument("--window-start") + parser.add_argument("--window-end") + parser.add_argument("--pr-url") + parser.add_argument("--generated-at") + parser.add_argument("--notes") + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + + emit = subparsers.add_parser("emit", help="print a validated payload") + add_payload_arguments(emit) + emit.set_defaults(handler=command_emit) + + annotate = subparsers.add_parser("annotate", help="append payload to local ledger and optionally comment") + add_payload_arguments(annotate) + annotate.add_argument("--ledger", type=pathlib.Path) + annotate.set_defaults(handler=command_annotate) + + from_hook = subparsers.add_parser("from-hook", help="parse hook stdin and annotate matching PR creates") + add_payload_arguments(from_hook) + from_hook.add_argument("--ledger", type=pathlib.Path) + from_hook.set_defaults(handler=command_from_hook) + + return parser + + +def main() -> int: + parser = build_parser() + args = parser.parse_args() + try: + result = args.handler(args) + if args.command == "from-hook": + return result + print(json.dumps(result, indent=2, sort_keys=True)) + return 0 + except PrCostError as exc: + print(f"pr-cost: {exc}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/skills/pr-cost/tests/fixtures/emit_valid.json b/skills/pr-cost/tests/fixtures/emit_valid.json new file mode 100644 index 0000000..cfbee40 --- /dev/null +++ b/skills/pr-cost/tests/fixtures/emit_valid.json @@ -0,0 +1,15 @@ +{ + "schema_version": "pr-cost/v1", + "harness": "claude", + "confidence": "estimated", + "usd": 1.23, + "tokens_in": 1200, + "tokens_out": 3400, + "model": "claude-sonnet-4-20250514", + "session_id": "session-123", + "window_start": "2026-08-20T19:00:00+00:00", + "window_end": "2026-08-20T19:05:00+00:00", + "pr_url": "https://github.com/cheshirecode/dotfiles/pull/123", + "generated_at": "2026-08-20T19:05:01+00:00", + "notes": "Fixture payload for validation." +} diff --git a/skills/pr-cost/tests/fixtures/hook_claude_pr_create.json b/skills/pr-cost/tests/fixtures/hook_claude_pr_create.json new file mode 100644 index 0000000..0acef88 --- /dev/null +++ b/skills/pr-cost/tests/fixtures/hook_claude_pr_create.json @@ -0,0 +1,12 @@ +{ + "session_id": "claude-session-fixture", + "model": "claude-opus-4-8", + "tool_input": { + "command": "gh pr create --title \"Fixture\" --body \"n\"" + }, + "tool_response": { + "exit_code": 0, + "stdout": "https://github.com/cheshirecode/dotfiles/pull/124\n", + "stderr": "" + } +} diff --git a/skills/pr-cost/tests/fixtures/hook_cursor_pr_create.json b/skills/pr-cost/tests/fixtures/hook_cursor_pr_create.json new file mode 100644 index 0000000..99396f6 --- /dev/null +++ b/skills/pr-cost/tests/fixtures/hook_cursor_pr_create.json @@ -0,0 +1,6 @@ +{ + "command": "gh pr create --title \"Add collector\" --body \"Implements the contract\"", + "exit_code": 0, + "stdout": "Creating pull request for fred/pr-cost-hook into main in cheshirecode/dotfiles\nhttps://github.com/cheshirecode/dotfiles/pull/123\n", + "stderr": "" +} diff --git a/skills/pr-cost/tests/fixtures/hook_cursor_pr_view.json b/skills/pr-cost/tests/fixtures/hook_cursor_pr_view.json new file mode 100644 index 0000000..2f204ad --- /dev/null +++ b/skills/pr-cost/tests/fixtures/hook_cursor_pr_view.json @@ -0,0 +1,6 @@ +{ + "command": "gh pr view 123 --json url", + "exit_code": 0, + "stdout": "https://github.com/cheshirecode/dotfiles/pull/123\n", + "stderr": "" +} diff --git a/skills/pr-cost/tests/test_adapters_smoke.py b/skills/pr-cost/tests/test_adapters_smoke.py new file mode 100644 index 0000000..0a90818 --- /dev/null +++ b/skills/pr-cost/tests/test_adapters_smoke.py @@ -0,0 +1,111 @@ +#!/usr/bin/env python3 +"""Smoke the three harness adapters without a live GitHub write.""" + +from __future__ import annotations + +import json +import os +import pathlib +import stat +import subprocess +import sys +import tempfile +import unittest + + +SKILL_DIR = pathlib.Path(__file__).parents[1] +FIXTURES = SKILL_DIR / "tests" / "fixtures" +CURSOR_WRAPPER = SKILL_DIR / "adapters" / "cursor" / "pr-cost-from-hook.sh" +CLAUDE_ADAPTER = SKILL_DIR / "adapters" / "claude" / "v1" / "pr_cost_from_hook.py" +CODEX_GH = SKILL_DIR / "adapters" / "codex" / "bin" / "gh" + + +class AdapterSmokeTest(unittest.TestCase): + def setUp(self) -> None: + self.temporary_directory = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary_directory.cleanup) + self.temporary_path = pathlib.Path(self.temporary_directory.name) + self.ledger = self.temporary_path / "ledger.jsonl" + + def env_with_ledger(self) -> dict[str, str]: + env = os.environ.copy() + env["PR_COST_LEDGER"] = str(self.ledger) + env.pop("PR_COST_HOOK_LIVE", None) + return env + + def ledger_rows(self) -> list[dict[str, object]]: + if not self.ledger.exists(): + return [] + return [ + json.loads(line) + for line in self.ledger.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + + def test_cursor_wrapper_writes_ledger(self) -> None: + result = subprocess.run( + [str(CURSOR_WRAPPER)], + input=(FIXTURES / "hook_cursor_pr_create.json").read_text(encoding="utf-8"), + capture_output=True, + text=True, + env=self.env_with_ledger(), + check=False, + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(json.loads(result.stdout), {}) + rows = self.ledger_rows() + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["harness"], "cursor") + self.assertEqual(rows[0]["pr_url"], "https://github.com/cheshirecode/dotfiles/pull/123") + + def test_claude_adapter_writes_ledger(self) -> None: + result = subprocess.run( + [sys.executable, str(CLAUDE_ADAPTER)], + input=(FIXTURES / "hook_claude_pr_create.json").read_text(encoding="utf-8"), + capture_output=True, + text=True, + env=self.env_with_ledger(), + check=False, + ) + self.assertEqual(result.returncode, 0, result.stderr) + rows = self.ledger_rows() + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["harness"], "claude") + self.assertEqual(rows[0]["pr_url"], "https://github.com/cheshirecode/dotfiles/pull/124") + + def test_codex_gh_wrapper_writes_ledger_without_live_comment(self) -> None: + stub_directory = self.temporary_path / "bin" + stub_directory.mkdir() + marker = self.temporary_path / "gh-comment-called.txt" + real_gh = stub_directory / "gh" + real_gh.write_text( + "#!/bin/sh\n" + "if [ \"$1\" = pr ] && [ \"$2\" = comment ]; then\n" + f" echo commented > {marker}\n" + " exit 99\n" + "fi\n" + "echo 'https://github.com/cheshirecode/dotfiles/pull/125'\n" + "exit 0\n", + encoding="utf-8", + ) + real_gh.chmod(real_gh.stat().st_mode | stat.S_IXUSR) + env = self.env_with_ledger() + env["PR_COST_REAL_GH"] = str(real_gh) + result = subprocess.run( + [str(CODEX_GH), "pr", "create", "--title", "Fixture"], + capture_output=True, + text=True, + env=env, + check=False, + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("https://github.com/cheshirecode/dotfiles/pull/125", result.stdout) + rows = self.ledger_rows() + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["harness"], "codex") + self.assertEqual(rows[0]["pr_url"], "https://github.com/cheshirecode/dotfiles/pull/125") + self.assertFalse(marker.exists(), "gh pr comment must not run when live is unset") + + +if __name__ == "__main__": + unittest.main() diff --git a/skills/pr-cost/tests/test_pr_cost_collect.py b/skills/pr-cost/tests/test_pr_cost_collect.py new file mode 100644 index 0000000..f960944 --- /dev/null +++ b/skills/pr-cost/tests/test_pr_cost_collect.py @@ -0,0 +1,130 @@ +#!/usr/bin/env python3 +"""Tests for the pr-cost collector CLI.""" + +from __future__ import annotations + +import json +import os +import pathlib +import stat +import subprocess +import sys +import tempfile +import unittest + + +SKILL_DIR = pathlib.Path(__file__).parents[1] +SCRIPT = SKILL_DIR / "scripts" / "pr_cost_collect.py" +FIXTURES = SKILL_DIR / "tests" / "fixtures" + + +class PrCostCollectTest(unittest.TestCase): + def setUp(self) -> None: + self.temporary_directory = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary_directory.cleanup) + self.temporary_path = pathlib.Path(self.temporary_directory.name) + self.ledger = self.temporary_path / "ledger.jsonl" + + def run_cli( + self, + *arguments: str, + stdin_text: str | None = None, + env: dict[str, str] | None = None, + expected_returncode: int = 0, + ) -> subprocess.CompletedProcess[str]: + result = subprocess.run( + [sys.executable, str(SCRIPT), *arguments], + input=stdin_text, + capture_output=True, + text=True, + env=env, + check=False, + ) + self.assertEqual( + result.returncode, + expected_returncode, + msg=f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}", + ) + return result + + def make_failing_gh_stub(self) -> tuple[pathlib.Path, pathlib.Path]: + stub_directory = self.temporary_path / "bin" + stub_directory.mkdir() + marker = self.temporary_path / "gh-called.txt" + script = stub_directory / "gh" + script.write_text( + "#!/bin/sh\n" + f"echo called > {marker}\n" + "exit 99\n", + encoding="utf-8", + ) + script.chmod(script.stat().st_mode | stat.S_IXUSR) + return stub_directory, marker + + def test_emit_valid_fixture(self) -> None: + result = self.run_cli( + "emit", + "--fixture", + str(FIXTURES / "emit_valid.json"), + ) + payload = json.loads(result.stdout) + self.assertEqual(payload["schema_version"], "pr-cost/v1") + self.assertEqual(payload["harness"], "claude") + self.assertEqual(payload["confidence"], "estimated") + self.assertEqual(payload["usd"], 1.23) + + def test_from_hook_rejects_gh_pr_view(self) -> None: + result = self.run_cli( + "from-hook", + "--harness", + "cursor", + "--ledger", + str(self.ledger), + stdin_text=(FIXTURES / "hook_cursor_pr_view.json").read_text(), + ) + response = json.loads(result.stdout) + self.assertEqual(response["status"], "ignored") + self.assertEqual(response["reason"], "not-pr-create") + self.assertFalse(self.ledger.exists()) + + def test_from_hook_writes_ledger_without_calling_gh_when_live_unset(self) -> None: + stub_directory, marker = self.make_failing_gh_stub() + env = os.environ.copy() + env["PATH"] = f"{stub_directory}:{env.get('PATH', '')}" + result = self.run_cli( + "from-hook", + "--harness", + "cursor", + "--ledger", + str(self.ledger), + stdin_text=(FIXTURES / "hook_cursor_pr_create.json").read_text(), + env=env, + ) + response = json.loads(result.stdout) + self.assertEqual(response["status"], "annotated") + self.assertFalse(response["commented"]) + self.assertTrue(self.ledger.exists()) + rows = [json.loads(line) for line in self.ledger.read_text().splitlines() if line.strip()] + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["harness"], "cursor") + self.assertEqual(rows[0]["pr_url"], "https://github.com/cheshirecode/dotfiles/pull/123") + self.assertFalse(marker.exists(), "gh should not run when PR_COST_HOOK_LIVE is unset") + + def test_annotate_is_idempotent_for_same_pr_and_session(self) -> None: + base_arguments = ( + "annotate", + "--fixture", + str(FIXTURES / "emit_valid.json"), + "--ledger", + str(self.ledger), + ) + first = json.loads(self.run_cli(*base_arguments).stdout) + second = json.loads(self.run_cli(*base_arguments).stdout) + self.assertEqual(first["status"], "annotated") + self.assertEqual(second["status"], "duplicate") + rows = [json.loads(line) for line in self.ledger.read_text().splitlines() if line.strip()] + self.assertEqual(len(rows), 1) + + +if __name__ == "__main__": + unittest.main()