feat: token usage transcript parsing (#57) (#91)
* docs: add TechCrunch RSS integration PRD Proposes TechCrunch RSS as SquadScope's first non-GitHub data source, enabling cross-source trend correlation to distinguish press-driven hype from organic developer momentum. Key points: - Implements Decision #7 crawler plugin architecture - Near-zero incremental cost (<$0.01/week) - Phased approach: RSS crawl → correlation engine → editorial integration - Editorial stance: press data is a correlation signal, not content to republish Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * chore(squad): log session — PR #54 merged, TechCrunch PRD opened - Merged 9 decision inbox files into decisions.md (now 27475 bytes) - Updated Leela history: PR #54 cost estimation merged - Updated Farnsworth history: PR #55 TechCrunch integration opened - Added 8 new decisions: Analyze job integration, analysis spec, generate/deploy, run counter, reskill retrospective, cost estimation, topic channels, TechCrunch RSS Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * docs: record Leela's review of PR #55 (rejected — empty branch) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * chore(squad): log review session — PR #55 rejected, Bender revised Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * chore: decompose PRDs into milestoned issues, move to processed - Created milestones v0.5 through v0.9 - Decomposed 3 PRDs into 34 GitHub issues with squad labels - Moved processed PRDs to docs/processed/ - Adopted milestone-based workflow per user directive Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * chore(squad): log milestone decomposition session Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * feat: add pre-flight token cost estimation to analyze workflow Closes #56 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * feat: add transcript/API response parsing to token tracker Closes #57 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Juan Manuel Servera committed
May 19, 2026 at 15:23 UTC
f4550e7e99b05f1b1c38fccca3cf66c252376ba4
5 files changed
+513
-5
.github/workflows/crawl-and-publish.yml
+31
-2
@@ -214,12 +214,28 @@ jobs:
214
echo "current_datetime=$CURRENT_DATETIME"
215
} >> "$GITHUB_OUTPUT"
216
217
+ - name: Pre-flight cost check
218
+ id: preflight-cost
219
+ run: |
220
+ WEEK_FILE="${{ steps.analysis-context.outputs.week_file }}"
221
+ CONTEXT_FILES=("$WEEK_FILE")
222
+ if [ -f ".squad/skills/wisdom.md" ]; then
223
+ CONTEXT_FILES+=(".squad/skills/wisdom.md")
224
+ fi
225
+ for f in .squad/skills/*.md; do
226
+ [ -f "$f" ] && CONTEXT_FILES+=("$f")
227
+ done
228
+ python3 scripts/preflight_cost_check.py \
229
+ --context-files "${CONTEXT_FILES[@]}" \
230
+ --model claude-sonnet-4
231
+
232
- name: Install Copilot CLI
233
id: install-copilot
234
continue-on-error: true
235
run: npm install -g @github/copilot
236
237
- name: Run analysis
238
+ if: steps.preflight-cost.outcome == 'success'
239
id: run-analysis
240
env:
241
COPILOT_GITHUB_TOKEN: ${{ secrets.COPILOT_GH_TOKEN }}
@@ -242,6 +258,7 @@ jobs:
258
--allow-tool=write \
259
--allow-tool=glob \
260
--allow-tool=grep \
261
+ --share=data/metrics/copilot-transcript.md \
262
> "$OUTPUT_FILE"; then
263
ANALYSIS_SOURCE="copilot-cli"
264
ANALYSIS_MODEL="claude-sonnet-4"
@@ -254,6 +271,12 @@ jobs:
271
ANALYSIS_SOURCE="github-models"
272
ANALYSIS_MODEL="${GITHUB_MODELS_MODEL:-openai/gpt-4.1}"
273
fi
274
+
275
+ TRANSCRIPT_ARGS=""
276
+ if [ -f "data/metrics/copilot-transcript.md" ]; then
277
+ TRANSCRIPT_ARGS="--transcript data/metrics/copilot-transcript.md"
278
+ fi
279
+
280
python3 scripts/track_token_usage.py \
281
--stage analysis \
282
--source "$ANALYSIS_SOURCE" \
@@ -261,7 +284,8 @@ jobs:
284
--current-datetime "$CURRENT_DATETIME" \
285
--week "$WEEK" \
286
--prompt-file "$PROMPT_FILE" \
264
- --output-file "$OUTPUT_FILE"
287
+ --output-file "$OUTPUT_FILE" \
288
+ $TRANSCRIPT_ARGS
289
rm -f "$PROMPT_FILE"
290
echo "analysis_source=$ANALYSIS_SOURCE" >> "$GITHUB_OUTPUT"
291
@@ -601,6 +625,10 @@ jobs:
625
mkdir -p .squad/skills .squad/reskill data/metrics
626
627
if python3 scripts/reskill.py --current-datetime "$CURRENT_DATETIME" --output "$RESKILL_OUTPUT" --prompt-output "$RESKILL_PROMPT"; then
628
+ API_RESPONSE_ARGS=""
629
+ if [ -f "data/metrics/reskill-api-response.json" ]; then
630
+ API_RESPONSE_ARGS="--api-response data/metrics/reskill-api-response.json"
631
+ fi
632
python3 scripts/track_token_usage.py \
633
--stage reskill \
634
--source github-models \
@@ -608,7 +636,8 @@ jobs:
636
--current-datetime "$CURRENT_DATETIME" \
637
--week "$WEEK" \
638
--prompt-file "$RESKILL_PROMPT" \
611
- --output-file "$RESKILL_OUTPUT"
639
+ --output-file "$RESKILL_OUTPUT" \
640
+ $API_RESPONSE_ARGS
641
rm -f "$RESKILL_PROMPT"
642
echo "🔄 Reskill report generated for run #$COUNTER"
643
else
scripts/preflight_cost_check.py
new
+91
@@ -0,0 +1,91 @@
1
+#!/usr/bin/env python3
2
+"""Pre-flight cost estimation for the analyze workflow.
3
+
4
+Estimates total input tokens from assembled context files, calculates
5
+expected cost, and aborts (exit 1) if the estimate exceeds the hard cap.
6
+"""
7
+from __future__ import annotations
8
+
9
+import argparse
10
+import sys
11
+from pathlib import Path
12
+
13
+from scripts.track_token_usage import (
14
+ MODEL_RATES,
15
+ estimate_cost_usd,
16
+ estimate_tokens_from_path,
17
+)
18
+
19
+DEFAULT_OUTPUT_TOKENS = 2000
20
+HARD_CAP_USD = 1.00
21
+
22
+
23
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
24
+ parser = argparse.ArgumentParser(
25
+ description="Pre-flight token cost estimation. Aborts if estimate exceeds hard cap."
26
+ )
27
+ parser.add_argument(
28
+ "--context-files",
29
+ nargs="+",
30
+ type=Path,
31
+ required=True,
32
+ help="Paths to context files that will be sent as input (raw JSON, prompt, wisdom, etc.).",
33
+ )
34
+ parser.add_argument(
35
+ "--model",
36
+ default="claude-sonnet-4",
37
+ help="Model name for cost rate lookup.",
38
+ )
39
+ parser.add_argument(
40
+ "--output-tokens",
41
+ type=int,
42
+ default=DEFAULT_OUTPUT_TOKENS,
43
+ help=f"Estimated output tokens (default: {DEFAULT_OUTPUT_TOKENS}).",
44
+ )
45
+ parser.add_argument(
46
+ "--hard-cap",
47
+ type=float,
48
+ default=HARD_CAP_USD,
49
+ help=f"Maximum allowed estimated cost in USD (default: {HARD_CAP_USD}).",
50
+ )
51
+ return parser.parse_args(argv)
52
+
53
+
54
+def estimate_input_tokens(context_files: list[Path]) -> int:
55
+ """Sum estimated tokens across all context files."""
56
+ return sum(estimate_tokens_from_path(p) for p in context_files)
57
+
58
+
59
+def main(argv: list[str] | None = None) -> int:
60
+ args = parse_args(argv)
61
+
62
+ input_tokens = estimate_input_tokens(args.context_files)
63
+ output_tokens = args.output_tokens
64
+ total_tokens = input_tokens + output_tokens
65
+ cost = estimate_cost_usd(args.model, input_tokens, output_tokens)
66
+
67
+ if cost is None:
68
+ print(
69
+ f"::warning::Unknown model '{args.model}' — cannot estimate cost. "
70
+ f"Known models: {', '.join(sorted(MODEL_RATES.keys()))}",
71
+ file=sys.stderr,
72
+ )
73
+ return 1
74
+
75
+ print(
76
+ f"::notice::Pre-flight estimate: {input_tokens} input + {output_tokens} output "
77
+ f"= {total_tokens} tokens → ${cost:.4f} (cap: ${args.hard_cap:.2f}, model: {args.model})"
78
+ )
79
+
80
+ if cost > args.hard_cap:
81
+ print(
82
+ f"::error::Estimated cost ${cost:.4f} exceeds hard cap ${args.hard_cap:.2f}. Aborting.",
83
+ file=sys.stderr,
84
+ )
85
+ return 1
86
+
87
+ return 0
88
+
89
+
90
+if __name__ == "__main__":
91
+ raise SystemExit(main())
scripts/track_token_usage.py
+106
-3
@@ -4,6 +4,7 @@ from __future__ import annotations
4
import argparse
5
import json
6
import math
7
+import re
8
from datetime import UTC, datetime
9
from pathlib import Path
10
@@ -31,6 +32,8 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
32
parser.add_argument("--output-file", type=Path, help="Output file used to estimate output tokens.")
33
parser.add_argument("--input-tokens", type=int, help="Explicit input token count.")
34
parser.add_argument("--output-tokens", type=int, help="Explicit output token count.")
35
+ parser.add_argument("--transcript", type=Path, help="Copilot CLI --share transcript file for parsing token usage.")
36
+ parser.add_argument("--api-response", type=Path, help="GitHub Models API response JSON for extracting usage data.")
37
parser.add_argument("--usage-file", type=Path, default=DEFAULT_USAGE_FILE, help="JSONL path for usage ledger.")
38
return parser.parse_args(argv)
39
@@ -61,6 +64,76 @@ def estimate_tokens_from_path(path: Path | None) -> int:
64
return estimate_tokens_from_text(path.read_text(encoding="utf-8"))
65
66
67
+def parse_copilot_transcript(path: Path) -> tuple[int, int] | None:
68
+ """Parse a Copilot CLI --share transcript for token usage metadata.
69
+
70
+ Searches for patterns like:
71
+ - "Input tokens: 1234" / "Output tokens: 567"
72
+ - "prompt_tokens: 1234" / "completion_tokens: 567"
73
+ - "Tokens used: 1234 input, 567 output"
74
+ Returns (input_tokens, output_tokens) or None if not found.
75
+ """
76
+ if not path.exists():
77
+ return None
78
+ try:
79
+ text = path.read_text(encoding="utf-8")
80
+ except (OSError, UnicodeDecodeError):
81
+ return None
82
+
83
+ input_tokens: int | None = None
84
+ output_tokens: int | None = None
85
+
86
+ # Pattern: "Input tokens: N" and "Output tokens: N"
87
+ m_input = re.search(r"[Ii]nput[\s_]tokens[\s:]+(\d+)", text)
88
+ m_output = re.search(r"[Oo]utput[\s_]tokens[\s:]+(\d+)", text)
89
+ if m_input and m_output:
90
+ return int(m_input.group(1)), int(m_output.group(1))
91
+
92
+ # Pattern: "prompt_tokens: N" and "completion_tokens: N"
93
+ m_prompt = re.search(r"prompt_tokens[\"'\s:]+(\d+)", text)
94
+ m_completion = re.search(r"completion_tokens[\"'\s:]+(\d+)", text)
95
+ if m_prompt and m_completion:
96
+ return int(m_prompt.group(1)), int(m_completion.group(1))
97
+
98
+ # Pattern: "Tokens used: N input, N output"
99
+ m_combined = re.search(r"[Tt]okens\s+used[\s:]+(\d+)\s+input[,;\s]+(\d+)\s+output", text)
100
+ if m_combined:
101
+ return int(m_combined.group(1)), int(m_combined.group(2))
102
+
103
+ # Pattern: "Usage: N/N tokens (input/output)"
104
+ m_usage = re.search(r"[Uu]sage[\s:]+(\d+)\s*/\s*(\d+)\s*tokens", text)
105
+ if m_usage:
106
+ return int(m_usage.group(1)), int(m_usage.group(2))
107
+
108
+ return None
109
+
110
+
111
+def parse_api_response(path: Path) -> tuple[int, int] | None:
112
+ """Parse a GitHub Models API response JSON for usage data.
113
+
114
+ Expects OpenAI-compatible format with usage.prompt_tokens and
115
+ usage.completion_tokens fields.
116
+ Returns (input_tokens, output_tokens) or None if not found.
117
+ """
118
+ if not path.exists():
119
+ return None
120
+ try:
121
+ data = json.loads(path.read_text(encoding="utf-8"))
122
+ except (OSError, UnicodeDecodeError, json.JSONDecodeError):
123
+ return None
124
+
125
+ usage = data.get("usage") if isinstance(data, dict) else None
126
+ if not isinstance(usage, dict):
127
+ return None
128
+
129
+ prompt_tokens = usage.get("prompt_tokens")
130
+ completion_tokens = usage.get("completion_tokens")
131
+ if isinstance(prompt_tokens, int) and isinstance(completion_tokens, int):
132
+ return prompt_tokens, completion_tokens
133
+
134
+ return None
135
+
136
+
137
def estimate_cost_usd(model: str, input_tokens: int, output_tokens: int) -> float | None:
138
rates = MODEL_RATES.get(model)
139
if not rates:
@@ -71,9 +144,39 @@ def estimate_cost_usd(model: str, input_tokens: int, output_tokens: int) -> floa
144
145
def build_record(args: argparse.Namespace) -> dict[str, object]:
146
parsed_datetime = parse_datetime(args.current_datetime).astimezone(UTC)
74
- input_tokens = args.input_tokens if args.input_tokens is not None else estimate_tokens_from_path(args.prompt_file)
75
- output_tokens = args.output_tokens if args.output_tokens is not None else estimate_tokens_from_path(args.output_file)
147
week = args.week or week_slug(parsed_datetime)
148
+
149
+ # Priority: 1) explicit flags, 2) transcript/api-response, 3) file-size estimate
150
+ estimated = True
151
+ input_tokens: int | None = None
152
+ output_tokens: int | None = None
153
+
154
+ # Highest priority: explicit --input-tokens / --output-tokens
155
+ if args.input_tokens is not None and args.output_tokens is not None:
156
+ input_tokens = args.input_tokens
157
+ output_tokens = args.output_tokens
158
+ estimated = False
159
+
160
+ # Second priority: parsed from transcript or API response
161
+ if input_tokens is None or output_tokens is None:
162
+ parsed = None
163
+ transcript_path = getattr(args, "transcript", None)
164
+ api_response_path = getattr(args, "api_response", None)
165
+ if transcript_path is not None:
166
+ parsed = parse_copilot_transcript(transcript_path)
167
+ if parsed is None and api_response_path is not None:
168
+ parsed = parse_api_response(api_response_path)
169
+ if parsed is not None:
170
+ input_tokens = parsed[0]
171
+ output_tokens = parsed[1]
172
+ estimated = False
173
+
174
+ # Lowest priority: file-size estimation
175
+ if input_tokens is None:
176
+ input_tokens = estimate_tokens_from_path(args.prompt_file)
177
+ if output_tokens is None:
178
+ output_tokens = estimate_tokens_from_path(args.output_file)
179
+
180
cost = estimate_cost_usd(args.model, input_tokens, output_tokens)
181
return {
182
"timestamp": parsed_datetime.isoformat().replace("+00:00", "Z"),
@@ -86,7 +189,7 @@ def build_record(args: argparse.Namespace) -> dict[str, object]:
189
"output_tokens": output_tokens,
190
"total_tokens": input_tokens + output_tokens,
191
"cost_usd": cost,
89
- "estimated": args.input_tokens is None or args.output_tokens is None,
192
+ "estimated": estimated,
193
}
194
195
tests/test_preflight_cost_check.py
new
+65
@@ -0,0 +1,65 @@
1
+import tempfile
2
+import unittest
3
+from pathlib import Path
4
+
5
+import scripts.preflight_cost_check as preflight
6
+
7
+
8
+class PreflightCostCheckTests(unittest.TestCase):
9
+ def _make_file(self, base: Path, name: str, size: int) -> Path:
10
+ p = base / name
11
+ p.write_text("x" * size, encoding="utf-8")
12
+ return p
13
+
14
+ def test_passes_under_cap(self) -> None:
15
+ tests_root = Path(__file__).resolve().parent
16
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
17
+ base = Path(tmpdir)
18
+ f1 = self._make_file(base, "small.json", 400)
19
+ rc = preflight.main(["--context-files", str(f1), "--model", "claude-sonnet-4"])
20
+ self.assertEqual(rc, 0)
21
+
22
+ def test_fails_over_cap(self) -> None:
23
+ tests_root = Path(__file__).resolve().parent
24
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
25
+ base = Path(tmpdir)
26
+ # 2M chars → 500k tokens; at $3/M input that's $1.50 — over cap
27
+ f1 = self._make_file(base, "huge.json", 2_000_000)
28
+ rc = preflight.main(["--context-files", str(f1), "--model", "claude-sonnet-4"])
29
+ self.assertEqual(rc, 1)
30
+
31
+ def test_custom_cap(self) -> None:
32
+ tests_root = Path(__file__).resolve().parent
33
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
34
+ base = Path(tmpdir)
35
+ f1 = self._make_file(base, "medium.json", 40_000)
36
+ # 40k chars → 10k tokens; at $3/M input = $0.03; cap $0.01 should fail
37
+ rc = preflight.main(
38
+ ["--context-files", str(f1), "--model", "claude-sonnet-4", "--hard-cap", "0.01"]
39
+ )
40
+ self.assertEqual(rc, 1)
41
+
42
+ def test_unknown_model_fails(self) -> None:
43
+ tests_root = Path(__file__).resolve().parent
44
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
45
+ base = Path(tmpdir)
46
+ f1 = self._make_file(base, "a.json", 100)
47
+ rc = preflight.main(["--context-files", str(f1), "--model", "unknown-model"])
48
+ self.assertEqual(rc, 1)
49
+
50
+ def test_multiple_files_summed(self) -> None:
51
+ tests_root = Path(__file__).resolve().parent
52
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
53
+ base = Path(tmpdir)
54
+ f1 = self._make_file(base, "a.json", 200)
55
+ f2 = self._make_file(base, "b.json", 200)
56
+ tokens = preflight.estimate_input_tokens([f1, f2])
57
+ self.assertEqual(tokens, 100) # 400 chars / 4 = 100 tokens
58
+
59
+ def test_missing_file_counts_zero(self) -> None:
60
+ tokens = preflight.estimate_input_tokens([Path("/nonexistent/path.json")])
61
+ self.assertEqual(tokens, 0)
62
+
63
+
64
+if __name__ == "__main__":
65
+ unittest.main()
tests/test_track_token_usage.py
+220
@@ -87,5 +87,225 @@ class TrackTokenUsageTests(unittest.TestCase):
87
self.assertFalse(record["estimated"])
88
89
90
+class ParseCopilotTranscriptTests(unittest.TestCase):
91
+ def test_parses_input_output_tokens_pattern(self) -> None:
92
+ tests_root = Path(__file__).resolve().parent
93
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
94
+ transcript = Path(tmpdir) / "transcript.md"
95
+ transcript.write_text(
96
+ "# Copilot Session\n\nSome content here.\n\n"
97
+ "---\nInput tokens: 1500\nOutput tokens: 800\n",
98
+ encoding="utf-8",
99
+ )
100
+ result = track_token_usage.parse_copilot_transcript(transcript)
101
+ self.assertEqual(result, (1500, 800))
102
+
103
+ def test_parses_prompt_completion_tokens_pattern(self) -> None:
104
+ tests_root = Path(__file__).resolve().parent
105
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
106
+ transcript = Path(tmpdir) / "transcript.md"
107
+ transcript.write_text(
108
+ "```json\n{\"prompt_tokens\": 2000, \"completion_tokens\": 950}\n```\n",
109
+ encoding="utf-8",
110
+ )
111
+ result = track_token_usage.parse_copilot_transcript(transcript)
112
+ self.assertEqual(result, (2000, 950))
113
+
114
+ def test_parses_tokens_used_combined_pattern(self) -> None:
115
+ tests_root = Path(__file__).resolve().parent
116
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
117
+ transcript = Path(tmpdir) / "transcript.md"
118
+ transcript.write_text(
119
+ "## Summary\nTokens used: 3000 input, 1200 output\n",
120
+ encoding="utf-8",
121
+ )
122
+ result = track_token_usage.parse_copilot_transcript(transcript)
123
+ self.assertEqual(result, (3000, 1200))
124
+
125
+ def test_parses_usage_slash_pattern(self) -> None:
126
+ tests_root = Path(__file__).resolve().parent
127
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
128
+ transcript = Path(tmpdir) / "transcript.md"
129
+ transcript.write_text(
130
+ "Usage: 500/200 tokens (input/output)\n",
131
+ encoding="utf-8",
132
+ )
133
+ result = track_token_usage.parse_copilot_transcript(transcript)
134
+ self.assertEqual(result, (500, 200))
135
+
136
+ def test_returns_none_when_no_pattern_found(self) -> None:
137
+ tests_root = Path(__file__).resolve().parent
138
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
139
+ transcript = Path(tmpdir) / "transcript.md"
140
+ transcript.write_text("# Just a normal transcript\nNo usage info here.\n", encoding="utf-8")
141
+ result = track_token_usage.parse_copilot_transcript(transcript)
142
+ self.assertIsNone(result)
143
+
144
+ def test_returns_none_for_missing_file(self) -> None:
145
+ result = track_token_usage.parse_copilot_transcript(Path("/nonexistent/path.md"))
146
+ self.assertIsNone(result)
147
+
148
+
149
+class ParseApiResponseTests(unittest.TestCase):
150
+ def test_parses_openai_compatible_usage(self) -> None:
151
+ tests_root = Path(__file__).resolve().parent
152
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
153
+ response_file = Path(tmpdir) / "response.json"
154
+ response_file.write_text(
155
+ json.dumps({
156
+ "id": "chatcmpl-abc123",
157
+ "choices": [{"message": {"content": "Hello"}}],
158
+ "usage": {"prompt_tokens": 450, "completion_tokens": 120, "total_tokens": 570},
159
+ }),
160
+ encoding="utf-8",
161
+ )
162
+ result = track_token_usage.parse_api_response(response_file)
163
+ self.assertEqual(result, (450, 120))
164
+
165
+ def test_returns_none_for_missing_usage(self) -> None:
166
+ tests_root = Path(__file__).resolve().parent
167
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
168
+ response_file = Path(tmpdir) / "response.json"
169
+ response_file.write_text(json.dumps({"choices": []}), encoding="utf-8")
170
+ result = track_token_usage.parse_api_response(response_file)
171
+ self.assertIsNone(result)
172
+
173
+ def test_returns_none_for_invalid_json(self) -> None:
174
+ tests_root = Path(__file__).resolve().parent
175
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
176
+ response_file = Path(tmpdir) / "response.json"
177
+ response_file.write_text("not json at all", encoding="utf-8")
178
+ result = track_token_usage.parse_api_response(response_file)
179
+ self.assertIsNone(result)
180
+
181
+ def test_returns_none_for_missing_file(self) -> None:
182
+ result = track_token_usage.parse_api_response(Path("/nonexistent/response.json"))
183
+ self.assertIsNone(result)
184
+
185
+
186
+class TokenSourcePriorityTests(unittest.TestCase):
187
+ """Test the priority ordering: explicit > transcript/api > file-size estimate."""
188
+
189
+ def test_transcript_overrides_file_estimate(self) -> None:
190
+ tests_root = Path(__file__).resolve().parent
191
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
192
+ base = Path(tmpdir)
193
+ usage_file = base / "token-usage.jsonl"
194
+ prompt_path = base / "prompt.txt"
195
+ output_path = base / "output.md"
196
+ transcript = base / "transcript.md"
197
+ prompt_path.write_text("x" * 400, encoding="utf-8")
198
+ output_path.write_text("y" * 200, encoding="utf-8")
199
+ transcript.write_text("Input tokens: 5000\nOutput tokens: 2500\n", encoding="utf-8")
200
+
201
+ exit_code = track_token_usage.main(
202
+ [
203
+ "--stage", "analysis",
204
+ "--source", "copilot-cli",
205
+ "--model", "claude-sonnet-4",
206
+ "--current-datetime", "2026-05-19T08:00:00Z",
207
+ "--prompt-file", str(prompt_path),
208
+ "--output-file", str(output_path),
209
+ "--transcript", str(transcript),
210
+ "--usage-file", str(usage_file),
211
+ ]
212
+ )
213
+
214
+ self.assertEqual(exit_code, 0)
215
+ record = json.loads(usage_file.read_text(encoding="utf-8").strip())
216
+ self.assertEqual(record["input_tokens"], 5000)
217
+ self.assertEqual(record["output_tokens"], 2500)
218
+ self.assertFalse(record["estimated"])
219
+
220
+ def test_api_response_overrides_file_estimate(self) -> None:
221
+ tests_root = Path(__file__).resolve().parent
222
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
223
+ base = Path(tmpdir)
224
+ usage_file = base / "token-usage.jsonl"
225
+ prompt_path = base / "prompt.txt"
226
+ api_response = base / "response.json"
227
+ prompt_path.write_text("x" * 400, encoding="utf-8")
228
+ api_response.write_text(
229
+ json.dumps({"usage": {"prompt_tokens": 800, "completion_tokens": 300, "total_tokens": 1100}}),
230
+ encoding="utf-8",
231
+ )
232
+
233
+ exit_code = track_token_usage.main(
234
+ [
235
+ "--stage", "reskill",
236
+ "--source", "github-models",
237
+ "--model", "gpt-4.1",
238
+ "--current-datetime", "2026-05-19T08:00:00Z",
239
+ "--prompt-file", str(prompt_path),
240
+ "--api-response", str(api_response),
241
+ "--usage-file", str(usage_file),
242
+ ]
243
+ )
244
+
245
+ self.assertEqual(exit_code, 0)
246
+ record = json.loads(usage_file.read_text(encoding="utf-8").strip())
247
+ self.assertEqual(record["input_tokens"], 800)
248
+ self.assertEqual(record["output_tokens"], 300)
249
+ self.assertFalse(record["estimated"])
250
+
251
+ def test_explicit_tokens_override_transcript(self) -> None:
252
+ tests_root = Path(__file__).resolve().parent
253
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
254
+ base = Path(tmpdir)
255
+ usage_file = base / "token-usage.jsonl"
256
+ transcript = base / "transcript.md"
257
+ transcript.write_text("Input tokens: 5000\nOutput tokens: 2500\n", encoding="utf-8")
258
+
259
+ exit_code = track_token_usage.main(
260
+ [
261
+ "--stage", "analysis",
262
+ "--source", "copilot-cli",
263
+ "--model", "claude-sonnet-4",
264
+ "--current-datetime", "2026-05-19T08:00:00Z",
265
+ "--input-tokens", "9999",
266
+ "--output-tokens", "4444",
267
+ "--transcript", str(transcript),
268
+ "--usage-file", str(usage_file),
269
+ ]
270
+ )
271
+
272
+ self.assertEqual(exit_code, 0)
273
+ record = json.loads(usage_file.read_text(encoding="utf-8").strip())
274
+ self.assertEqual(record["input_tokens"], 9999)
275
+ self.assertEqual(record["output_tokens"], 4444)
276
+ self.assertFalse(record["estimated"])
277
+
278
+ def test_fallback_to_estimate_when_transcript_has_no_usage(self) -> None:
279
+ tests_root = Path(__file__).resolve().parent
280
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
281
+ base = Path(tmpdir)
282
+ usage_file = base / "token-usage.jsonl"
283
+ prompt_path = base / "prompt.txt"
284
+ output_path = base / "output.md"
285
+ transcript = base / "transcript.md"
286
+ prompt_path.write_text("x" * 40, encoding="utf-8")
287
+ output_path.write_text("y" * 20, encoding="utf-8")
288
+ transcript.write_text("# No usage info here\n", encoding="utf-8")
289
+
290
+ exit_code = track_token_usage.main(
291
+ [
292
+ "--stage", "analysis",
293
+ "--source", "copilot-cli",
294
+ "--model", "claude-sonnet-4",
295
+ "--current-datetime", "2026-05-19T08:00:00Z",
296
+ "--prompt-file", str(prompt_path),
297
+ "--output-file", str(output_path),
298
+ "--transcript", str(transcript),
299
+ "--usage-file", str(usage_file),
300
+ ]
301
+ )
302
+
303
+ self.assertEqual(exit_code, 0)
304
+ record = json.loads(usage_file.read_text(encoding="utf-8").strip())
305
+ self.assertEqual(record["input_tokens"], 10)
306
+ self.assertEqual(record["output_tokens"], 5)
307
+ self.assertTrue(record["estimated"])
308
+
309
+
310
if __name__ == "__main__":
311
unittest.main()