test_brief.py

analysis/test_brief.py

1. #!/usr/bin/env python3
2. """
3. Tests for brief.py. Offline: no network, no credentials, no model.
5. python3 -m unittest discover -s analysis -v
7. Every case below states the failure it prevents rather than restating the
8. rule. The failures that matter here are silent ones: a briefing that renders
9. empty, or cut off, with nothing saying why.
11. Following the same convention as the TypeScript suite, these assert properties
12. (it raised, the message carries the stop_reason) rather than exact wording: a
13. test that matches on message text can pin a wrong diagnosis in place.
14. """
16. from __future__ import annotations
18. import io
19. import json
20. import sys
21. import tempfile
22. import unittest
23. from pathlib import Path
24. from unittest import mock
26. sys.path.insert(0, str(Path(__file__).resolve().parent))
28. import brief # noqa: E402
31. def response(body: dict) -> mock.MagicMock:
32. """A stand-in for the urlopen context manager."""
33. res = mock.MagicMock()
34. res.read.return_value = json.dumps(body).encode()
35. res.__enter__.return_value = res
36. res.__exit__.return_value = False
37. return res
40. def anthropic(**over) -> dict:
41. base = {
42. "content": [{"type": "text", "text": "A briefing."}],
43. "stop_reason": "end_turn",
44. "usage": {"input_tokens": 100, "output_tokens": 20},
45. }
46. base.update(over)
47. return base
50. def call(body: dict, max_tokens: int = 4000):
51. with mock.patch("urllib.request.urlopen", return_value=response(body)):
52. return brief.call_model(
53. "https://api.anthropic.com/v1", "claude-opus-5", "sys", "usr",
54. "key", 30, "anthropic", max_tokens,
55. )
58. PAYLOAD = {
59. "schemaVersion": "1.0.0",
60. "people": [{"name": "Nora Lindqvist"}, {"name": "Sam Whitaker"}],
61. }
64. class NoTextBlockRaises(unittest.TestCase):
65. """
66. Joining the text blocks of a response that has none returns "", which would
67. be written to briefings.json and rendered as a heading with nothing under
68. it: a silent failure in the one project built to eliminate them.
69. """
71. def test_raises_rather_than_returning_empty(self):
72. with self.assertRaises(brief.EmptyResponseError):
73. call(anthropic(content=[]))
75. def test_raises_when_only_non_text_blocks_came_back(self):
76. with self.assertRaises(brief.EmptyResponseError):
77. call(anthropic(content=[{"type": "thinking", "thinking": "..."}]))
79. def test_raises_on_whitespace_only_text(self):
80. with self.assertRaises(brief.EmptyResponseError):
81. call(anthropic(content=[{"type": "text", "text": " \n "}]))
83. def test_message_names_the_model_and_the_stop_reason(self):
84. # The handoff asked for both by name: without them the error says a
85. # briefing failed but not what the API reported about it.
86. with self.assertRaises(brief.EmptyResponseError) as caught:
87. call(anthropic(content=[], stop_reason="max_tokens"))
88. msg = str(caught.exception)
89. self.assertIn("claude-opus-5", msg)
90. self.assertIn("max_tokens", msg)
92. def test_message_names_the_block_types_that_did_come_back(self):
93. with self.assertRaises(brief.EmptyResponseError) as caught:
94. call(anthropic(content=[{"type": "thinking", "thinking": "..."}]))
95. self.assertIn("thinking", str(caught.exception))
97. def test_does_not_assert_an_unobserved_cause(self):
98. # Report what was observed, not a guessed cause (rule 1). Whether
99. # reasoning consumed the budget is inferred, not observed, so it may be
100. # offered as a candidate but never stated as the cause.
101. with self.assertRaises(brief.EmptyResponseError) as caught:
102. call(anthropic(content=[], stop_reason="max_tokens"))
103. msg = str(caught.exception).lower()
104. self.assertIn("candidates", msg)
105. for claim in ("most likely", "must be", "because the model"):
106. self.assertNotIn(claim, msg)
109. class TruncationIsReportedNotDiscarded(unittest.TestCase):
110. """
111. A brief cut mid-word at the output limit must not be written to the page
112. with no indication it is incomplete.
114. Truncated-and-labelled beats absent. The text is kept and
115. the caller is told, rather than the brief being thrown away.
116. """
118. def test_keeps_the_text(self):
119. text, _, _ = call(anthropic(
120. content=[{"type": "text", "text": "Half a sen"}], stop_reason="max_tokens"))
121. self.assertEqual(text, "Half a sen")
123. def test_surfaces_the_stop_reason_so_the_caller_can_label_it(self):
124. _, _, stop = call(anthropic(stop_reason="max_tokens"))
125. self.assertEqual(stop, "max_tokens")
126. self.assertIn(stop, brief.TRUNCATED_STOPS)
128. def test_a_complete_brief_is_not_flagged(self):
129. _, _, stop = call(anthropic(stop_reason="end_turn"))
130. self.assertNotIn(stop, brief.TRUNCATED_STOPS)
133. class UsageIsReadFromTheResponse(unittest.TestCase):
134. def test_exact_token_counts_not_estimates(self):
135. _, usage, _ = call(anthropic(usage={"input_tokens": 1984, "output_tokens": 430}))
136. self.assertEqual(usage, {"input": 1984, "output": 430})
139. class OpenAiBranchFailsTheSameWay(unittest.TestCase):
140. """
141. The OpenAI-compatible path indexed ["choices"][0] directly, so a malformed
142. response raised KeyError -- which says nothing about what went wrong. It gets
143. the same contract as the Anthropic path: raise, and say what came back.
144. """
146. def openai(self, body: dict):
147. with mock.patch("urllib.request.urlopen", return_value=response(body)):
148. return brief.call_model("http://127.0.0.1:1234/v1", "qwen3", "s", "u",
149. None, 30, "openai", 4000)
151. def test_no_choices_raises_something_informative(self):
152. with self.assertRaises(brief.EmptyResponseError):
153. self.openai({"error": "model not loaded"})
155. def test_empty_message_raises(self):
156. with self.assertRaises(brief.EmptyResponseError):
157. self.openai({"choices": [{"message": {"content": ""}}]})
159. def test_length_finish_reason_counts_as_truncated(self):
160. _, _, stop = self.openai({
161. "choices": [{"message": {"content": "Half a sen"}, "finish_reason": "length"}],
162. "usage": {"prompt_tokens": 10, "completion_tokens": 5},
163. })
164. self.assertIn(stop, brief.TRUNCATED_STOPS)
167. class AllIncludesTheCompanyBrief(unittest.TestCase):
168. """
169. surfaces.ts gates the entire "Written briefing" card on the `company` key,
170. with the leadership view nested inside it. An --all run without `company`
171. would generate every brief while the board showed "Not generated for this
172. sync", leadership included.
173. """
175. def test_all_generates_a_company_brief(self):
176. keys = [j.key for j in brief.build_jobs(PAYLOAD, all_=True, person=None, view="standard")]
177. self.assertIn("company", keys)
179. def test_all_generates_exec_and_every_person(self):
180. keys = [j.key for j in brief.build_jobs(PAYLOAD, all_=True, person=None, view="standard")]
181. self.assertIn("exec", keys)
182. for p in PAYLOAD["people"]:
183. self.assertIn(p["name"], keys)
185. def test_every_key_the_page_reads_is_covered(self):
186. # The three shapes surfaces.ts looks up: `company` for the main card,
187. # `exec` for the disclosure, and a person's roster name for "My day".
188. keys = set(j.key for j in brief.build_jobs(PAYLOAD, all_=True, person=None, view="standard"))
189. self.assertTrue({"company", "exec", "Nora Lindqvist"} <= keys)
191. def test_no_duplicate_keys(self):
192. keys = [j.key for j in brief.build_jobs(PAYLOAD, all_=True, person=None, view="standard")]
193. self.assertEqual(len(keys), len(set(keys)))
195. def test_a_single_person_run_writes_only_that_person(self):
196. jobs = brief.build_jobs(PAYLOAD, all_=False, person="Sam Whitaker", view="standard")
197. self.assertEqual([j.key for j in jobs], ["Sam Whitaker"])
199. def test_exec_view_alone_keys_as_exec(self):
200. jobs = brief.build_jobs(PAYLOAD, all_=False, person=None, view="exec")
201. self.assertEqual([j.key for j in jobs], ["exec"])
204. class BudgetsAreNotOneNumberForEverything(unittest.TestCase):
205. """
206. A single max_tokens of 800 covered a ~2k-token exec prompt and a ~6.7k-token
207. company prompt alike, and both ran out. Each kind of brief carries its own.
208. """
210. def test_every_job_resolves_to_a_known_budget(self):
211. for job in brief.build_jobs(PAYLOAD, all_=True, person=None, view="standard"):
212. self.assertIn(job.budget, brief.MAX_TOKENS)
214. def test_people_share_one_budget_rather_than_one_each(self):
215. jobs = brief.build_jobs(PAYLOAD, all_=True, person=None, view="standard")
216. people = [j for j in jobs if not j.key.startswith("company") and j.key != "exec"]
217. self.assertTrue(people)
218. for j in people:
219. self.assertEqual(j.budget, "person")
221. def test_the_all_hands_brief_is_priced_as_a_company_brief(self):
222. # Same view over a slightly smaller payload. Priced as a person's brief it
223. # would truncate: company measured 1715 output tokens against person 1029.
224. jobs = brief.build_jobs(PAYLOAD, all_=True, person=None, view="standard")
225. allhands = next(j for j in jobs if j.key == "company-allhands")
226. self.assertEqual(allhands.budget, "company")
228. def test_the_old_default_is_gone(self):
229. # An 800-token ceiling truncates live briefings.
230. for kind, budget in brief.MAX_TOKENS.items():
231. self.assertGreater(budget, 800, kind)
233. def test_every_budget_clears_what_was_measured_with_room_to_spare(self):
234. # Live output_tokens from the run the budgets were set from.
235. # A budget that merely equalled these would truncate on a busier day.
236. measured = {"exec": 872, "company": 1715, "person": 1029}
237. for kind, observed in measured.items():
238. self.assertGreater(brief.MAX_TOKENS[kind], observed * 2, kind)
240. def test_company_gets_the_largest_budget(self):
241. # Exec, the most constrained prompt, might be expected to need the most.
242. # Measured, company needs the most by a wide margin. This pins the
243. # measurement, not the guess.
244. self.assertEqual(max(brief.MAX_TOKENS, key=brief.MAX_TOKENS.get), "company")
247. class NearCeilingWarnsBeforeItBreaks(unittest.TestCase):
248. """
249. stop_reason only reports truncation once a brief has already gone out
250. half-finished. This is the warning one run earlier.
251. """
253. def test_the_threshold_leaves_usable_headroom(self):
254. self.assertTrue(0.5 < brief.NEAR_CEILING < 1.0)
256. def test_the_measured_worst_case_does_not_trip_it(self):
257. # company at 1715 against a 4000 budget is 43% -- comfortable. If a
258. # budget change made today's real data warn, the budget is wrong.
259. for kind, observed in {"exec": 872, "company": 1715, "person": 1029}.items():
260. self.assertLess(observed, brief.MAX_TOKENS[kind] * brief.NEAR_CEILING, kind)
263. MIN_PAYLOAD = {
264. "schemaVersion": "1.0.0",
265. "generatedAt": "2026-08-25T18:38:31.943Z",
266. "day": "2026-08-25",
267. "timezone": "America/Chicago",
268. "mode": "live",
269. "sources": [{"name": "plinth", "ok": True, "error": None}],
270. "rules": ["A rule that must reach the system prompt verbatim."],
271. "execRules": ["Capacity, not performance."],
272. "company": {"ourMove": 3},
273. "exec": {"holding": 3},
274. "dataHealth": {"unmeasurable": []},
275. "people": [{"name": "Nora Lindqvist"}, {"name": "Sam Whitaker"}],
276. }
279. class RunEndToEnd(unittest.TestCase):
280. """
281. Drives main() with a stubbed transport. Covers the parts that only show up
282. once the loop, the merge and the exit code are working together.
283. """
285. def setUp(self):
286. self.dir = tempfile.TemporaryDirectory()
287. self.addCleanup(self.dir.cleanup)
288. self.payload = Path(self.dir.name) / "briefing-payload.json"
289. self.payload.write_text(json.dumps(MIN_PAYLOAD), encoding="utf-8")
290. self.out = Path(self.dir.name) / "briefings.json"
292. def run_main(self, bodies: list[dict], argv: list[str]) -> int:
293. full = ["brief.py", "--payload", str(self.payload), "--out", str(self.out),
294. "--base-url", "https://api.anthropic.com/v1", "--api-key", "k",
295. "--model", "claude-opus-5", *argv]
296. with mock.patch.object(sys, "argv", full), \
297. mock.patch("urllib.request.urlopen",
298. side_effect=[response(b) for b in bodies]), \
299. mock.patch.object(sys, "stdout", io.StringIO()), \
300. mock.patch.object(sys, "stderr", io.StringIO()):
301. return brief.main()
303. def written(self) -> dict:
304. return json.loads(self.out.read_text(encoding="utf-8"))
306. def test_all_writes_the_company_key_the_page_reads(self):
307. code = self.run_main([anthropic()] * 5, ["--all"])
308. self.assertEqual(code, 0)
309. self.assertIn("company", self.written()["briefings"])
311. def test_all_writes_exec_company_and_every_person(self):
312. self.run_main([anthropic()] * 5, ["--all"])
313. got = set(self.written()["briefings"])
314. self.assertEqual(got, {"exec", "company", "company-allhands", "Nora Lindqvist", "Sam Whitaker"})
316. def test_one_empty_response_does_not_lose_the_other_briefs(self):
317. # Making call_model raise turned a single bad response into a run that
318. # wrote nothing at all. The rest of the briefs and a named failure beats none.
319. bodies = [anthropic(content=[]), anthropic(), anthropic(), anthropic(), anthropic()]
320. code = self.run_main(bodies, ["--all"])
321. self.assertEqual(code, 1, "a failed brief must still fail the run")
322. got = self.written()["briefings"]
323. self.assertNotIn("exec", got, "the failed brief must be absent, not blank")
324. self.assertEqual(set(got), {"company", "company-allhands", "Nora Lindqvist", "Sam Whitaker"})
326. def test_a_failed_brief_is_absent_rather_than_empty_string(self):
327. # Absent renders as "not generated"; an empty string renders as a blank
328. # briefing, which is the failure this whole change is about.
329. self.run_main([anthropic(content=[])] + [anthropic()] * 4, ["--all"])
330. for key, text in self.written()["briefings"].items():
331. self.assertTrue(text.strip(), key)
333. def test_truncated_briefs_are_listed_for_the_page(self):
334. bodies = [anthropic(stop_reason="max_tokens")] + [anthropic()] * 4
335. self.run_main(bodies, ["--all"])
336. self.assertEqual(self.written()["truncated"], ["exec"])
338. def test_a_clean_run_lists_nothing_as_truncated(self):
339. self.run_main([anthropic()] * 5, ["--all"])
340. self.assertEqual(self.written()["truncated"], [])
342. def test_rerunning_one_person_clears_only_their_stale_flag(self):
343. # Union-ing the old and new lists would label a brief truncated forever,
344. # so a successful re-run has to drop the key it rewrote and no other.
345. self.run_main([anthropic(stop_reason="max_tokens")] * 5, ["--all"])
346. self.assertEqual(self.written()["truncated"],
347. ["Nora Lindqvist", "Sam Whitaker", "company", "company-allhands", "exec"])
348. self.run_main([anthropic()], ["--person", "Nora Lindqvist"])
349. self.assertEqual(self.written()["truncated"],
350. ["Sam Whitaker", "company", "company-allhands", "exec"])
352. def test_a_single_person_run_does_not_wipe_the_morning(self):
353. self.run_main([anthropic()] * 5, ["--all"])
354. self.run_main([anthropic()], ["--person", "Nora Lindqvist"])
355. self.assertEqual(set(self.written()["briefings"]),
356. {"exec", "company", "company-allhands", "Nora Lindqvist", "Sam Whitaker"})
358. def test_ties_the_briefing_to_the_sync_it_was_written_from(self):
359. # main.ts and briefingAge() compare this against snap.syncedAt to decide
360. # whether the prose and the numbers describe the same moment.
361. self.run_main([anthropic()] * 5, ["--all"])
362. self.assertEqual(self.written()["payloadGeneratedAt"], MIN_PAYLOAD["generatedAt"])
365. class RulesReachTheSystemPrompt(unittest.TestCase):
366. """
367. The payload ships its own `rules` array and brief.py's own docstring says
368. they go in verbatim -- they are the traps that produce fluent, wrong
369. sentences. A prompt built without them looks identical and is not.
370. """
372. def test_standard_view_carries_the_payload_rules(self):
373. got = brief.system_prompt(MIN_PAYLOAD, "standard")
374. self.assertIn(MIN_PAYLOAD["rules"][0], got)
376. def test_exec_view_carries_both_rule_sets(self):
377. got = brief.system_prompt(MIN_PAYLOAD, "exec")
378. self.assertIn(MIN_PAYLOAD["execRules"][0], got)
379. self.assertIn(MIN_PAYLOAD["rules"][0], got)
381. def test_exec_rules_come_first(self):
382. # Stated reason in the source: a model that runs out of attention should
383. # lose the style guidance, not the limits.
384. got = brief.system_prompt(MIN_PAYLOAD, "exec")
385. self.assertLess(got.index(MIN_PAYLOAD["execRules"][0]),
386. got.index(MIN_PAYLOAD["rules"][0]))
389. if __name__ == "__main__":
390. unittest.main()
393. class RestrictedTodosNeverReachThePrompt(unittest.TestCase):
394. """
395. Leadership Team to-dos are readable only by the leadership team. The company briefing names
396. to-dos by title, and prose cannot be redacted once a model has written it:
397. there is no reliable way to strip a title from a paragraph that reasons about
398. it, and asking the model not to mention it is a request, not a control.
400. So the all-hands company brief is generated from a payload with those rows
401. removed. These tests assert the removal, at the payload level, because that is
402. the only place the guarantee is enforceable.
403. """
405. def payload(self):
406. # Shaped like the real payload, which is the whole point.
407. #
408. # company.lateTodos is a count; the list lives under queues. A fixture
409. # that put the list under company.lateTodos would pass against a shape
410. # briefing-payload.ts never emits, while the filter removed nothing in
411. # production. A fixture that does not match the contract tests the
412. # fixture.
413. return {
414. **MIN_PAYLOAD,
415. "company": {"ourMove": 3, "lateTodos": 2},
416. "queues": {"lateTodos": [
417. {"title": "SECRET leadership item", "restricted": True},
418. {"title": "Open all-hands item", "restricted": False},
419. ]},
420. "people": [
421. {"name": "Nora Lindqvist", "todos": [
422. {"title": "SECRET leadership item", "restricted": True},
423. {"title": "Open all-hands item", "restricted": False},
424. ]},
425. {"name": "Sam Whitaker", "todos": [
426. {"title": "Another SECRET one", "restricted": True},
427. ]},
428. ],
429. }
431. def test_restricted_rows_are_gone(self):
432. out = brief.without_restricted(self.payload())
433. titles = [t["title"] for p in out["people"] for t in p["todos"]]
434. self.assertEqual(titles, ["Open all-hands item"])
436. def test_the_serialized_prompt_does_not_contain_the_titles(self):
437. # The real guarantee: whatever ends up in the user turn must not carry
438. # them. A filter that missed a nested copy would still pass the test above.
439. out = brief.without_restricted(self.payload())
440. prompt = brief.user_prompt(out, None, "standard")
441. self.assertNotIn("SECRET leadership item", prompt)
442. self.assertNotIn("Another SECRET one", prompt)
443. self.assertIn("Open all-hands item", prompt)
445. def test_the_queue_list_is_filtered_too(self):
446. # queues.lateTodos is the company-level list of to-do TITLES. It was
447. # never filtered, so every Leadership title reached the all-hands prompt
448. # through it even once the rows were flagged.
449. out = brief.without_restricted(self.payload())
450. self.assertEqual([t["title"] for t in out["queues"]["lateTodos"]],
451. ["Open all-hands item"])
453. def test_a_count_that_looks_like_a_list_is_left_alone(self):
454. # company.lateTodos is an int. Filtering must not touch it, and must not
455. # crash on it either -- reading it as a list is what pointed the filter at
456. # nothing for a week.
457. out = brief.without_restricted(self.payload())
458. self.assertEqual(out["company"]["lateTodos"], 2)
460. def test_every_list_of_titles_in_the_payload_is_covered(self):
461. # The assertion that would have caught the original bug without knowing
462. # where the list lived: after filtering, NO to-do dict anywhere in the
463. # payload may still carry restricted=True.
464. out = brief.without_restricted(self.payload())
466. def walk(node):
467. if isinstance(node, dict):
468. if node.get("restricted") is True:
469. yield node
470. for v in node.values():
471. yield from walk(v)
472. elif isinstance(node, list):
473. for v in node:
474. yield from walk(v)
476. self.assertEqual(list(walk(out)), [])
478. def test_the_original_payload_is_untouched(self):
479. # The full company brief is generated from the same object in the same
480. # run. Mutating it in place would silently redact that one too.
481. src = self.payload()
482. brief.without_restricted(src)
483. self.assertEqual(len(src["people"][0]["todos"]), 2)
485. def test_it_reports_how_many_it_removed(self):
486. out = brief.without_restricted(self.payload())
487. self.assertEqual(out["_restrictedTodosRemoved"], 3)
489. def test_a_todo_with_no_restricted_flag_is_kept(self):
490. # `restricted` is computed in TypeScript. An older payload without the
491. # field must not silently drop every row -- that would empty the brief and
492. # look like a quiet week.
493. p = {**MIN_PAYLOAD, "people": [{"name": "X", "todos": [{"title": "No flag"}]}]}
494. out = brief.without_restricted(p)
495. self.assertEqual([t["title"] for t in out["people"][0]["todos"]], ["No flag"])
498. class TheAllHandsBriefIsItsOwnGeneration(unittest.TestCase):
499. def test_all_produces_both_company_briefs(self):
500. keys = [j.key for j in brief.build_jobs(PAYLOAD, all_=True, person=None, view="standard")]
501. self.assertIn("company", keys)
502. self.assertIn("company-allhands", keys)
504. def test_they_are_distinct_keys(self):
505. # surfaces.ts picks one by permission. If they collided, everyone would
506. # read whichever was written last.
507. keys = [j.key for j in brief.build_jobs(PAYLOAD, all_=True, person=None, view="standard")]
508. self.assertEqual(len(keys), len(set(keys)))