mattrguerra/meridian-ops-boardmainPrivate on GitHub, published here in full

brief.py

Language Python33.9 KB759 lines

analysis/brief.py

#!/usr/bin/env python3
"""
Generate the daily briefing from briefing-payload.json.
python3 analysis/brief.py --dry-run # prompt only, no model call
python3 analysis/brief.py # local endpoint (default)
python3 analysis/brief.py --base-url https://... --model claude-...
python3 analysis/brief.py --person "Marco Ibáñez"
Why this is Python and separate from the TypeScript:
Everything upstream (three API clients, the stage taxonomies, the review-gate
arithmetic, the assignment join) is deterministic and costs zero model tokens.
It runs in TypeScript because that is where it already works and is tested.
This file is the only place a model is involved. It reads one small JSON file
and nothing else. That split is the whole cost strategy: a model reasoning over
raw Bitrix and Plinth payloads reads 200k-500k tokens per run; reading this payload
it reads roughly 3-4k. Same conclusions, about 1% of the budget.
It speaks two wire formats, chosen from --base-url by detect_api(): the OpenAI
chat-completions shape, which works unchanged against LM Studio, mlx_lm.server,
Ollama, vLLM or any compatible hosted API, and Anthropic's Messages API when the
host is api.anthropic.com. Point --base-url wherever you want the tokens spent.
What makes the output safe:
The payload ships its own `rules` array. Those are not documentation; they are
the four traps in Meridian's data that produce fluent, confident, wrong sentences,
the worst of which get read aloud in a meeting. They go into the system prompt
verbatim. Do not summarize them, do not "clean them up", and do not drop them to
save tokens; they are the cheapest part of the prompt and the reason the output
can be trusted.
Small models drop nuance like this more readily than large ones, which is
precisely why the rules sit in the system role rather than buried in the user
turn.
"""
from __future__ import annotations
import argparse
import json
import os
import sys
import urllib.error
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, NamedTuple
REPO_ROOT = Path(__file__).resolve().parent.parent
DEFAULT_PAYLOAD = REPO_ROOT / "briefing-payload.json"
# No .env reader, deliberately.
#
# Credentials come from the process environment only, exported from the shell
# profile. A credentials file in the working tree is a liability: any tool or
# agent working in this repo can open it, and a secret that has been read is a
# secret that has to be rotated.
#
# Consequence to know about: a new terminal must have sourced the profile. If a
# credential is missing, the shell is the thing to check, not the repo.
# Local by default. An OpenAI-compatible server (LM Studio, mlx_lm.server,
# Ollama, vLLM) exposes this path, so the same script serves a Mac Studio on the
# LAN and a hosted API without a code change.
DEFAULT_BASE_URL = "http://127.0.0.1:1234/v1"
DEFAULT_MODEL = "qwen3-30b-a3b-instruct"
# The schema majors this script knows how to read. A payload from a future major
# may have moved a field this prompt relies on, and a briefing built from a
# misread payload is worse than no briefing.
SUPPORTED_MAJORS = {1}
class Job(NamedTuple):
"""One brief to write."""
label: str
person: str | None
view: str
"""What it is stored under in briefings.json: 'exec', 'company', or a name."""
key: str
@property
def budget(self) -> str:
"""Which MAX_TOKENS entry pays for it. Every person shares one budget."""
if self.key == "exec":
return "exec"
# company-allhands is the same view over a slightly smaller payload, so it
# is priced as a company brief rather than as a person's.
if self.key.startswith("company"):
return "company"
return "person"
# Output budget per kind of brief.
#
# A model that reasons before it answers is charged for that reasoning against
# max_tokens, so a budget sized for the answer alone cuts the brief off
# mid-sentence, or leaves no text block at all.
#
# Measured, not guessed: every brief run against live data at max_tokens 4000,
# output_tokens read off the response (this counts reasoning as well as text,
# which is what the budget actually pays for):
#
# company 1715 <- the largest, on an 11.5k-token prompt
# person (largest) 1029
# exec 872
# person (smallest) 380
#
# The company brief needs the most, not the exec one, even though exec is the
# most constrained prompt. Follow the numbers.
#
# The ceilings below sit 2-3x above what was observed, because a prompt grows
# with the team and the order book. This costs nothing to set high: max_tokens is
# a limit, not a reservation, and only tokens actually produced are billed.
#
# Re-measure rather than nudging these if the prompts change.
MAX_TOKENS = {
"exec": 3000,
"company": 4000,
"person": 3000,
}
# Warn when a brief comes this close to its ceiling. Truncation is caught by
# stop_reason after the fact; this catches the run before the one that truncates.
NEAR_CEILING = 0.8
# stop_reason / finish_reason values that mean "there was more to say".
TRUNCATED_STOPS = {"max_tokens", "length"}
def _now_iso() -> str:
return datetime.now(timezone.utc).isoformat(timespec="seconds")
class PayloadError(RuntimeError):
pass
def load_payload(path: Path, allow_fixtures: bool = False) -> dict[str, Any]:
if not path.exists():
raise PayloadError(
f"No payload at {path}.\n"
f"Run `pnpm dashboard` (or `pnpm dashboard --fixtures` for sample data) first."
)
payload = json.loads(path.read_text(encoding="utf-8"))
version = str(payload.get("schemaVersion", "0.0.0"))
major = int(version.split(".")[0])
if major not in SUPPORTED_MAJORS:
raise PayloadError(
f"Payload schema {version} is a major this script has not been updated for "
f"(knows: {sorted(SUPPORTED_MAJORS)}). Fields it relies on may have moved. "
f"Read apps/dashboard/src/briefing-payload.ts and update this file rather "
f"than forcing it. A briefing built from a misread payload is worse than none."
)
if payload.get("mode") == "fixtures" and not allow_fixtures:
raise PayloadError(
f"This payload is SAMPLE DATA captured {payload.get('capturedAt', 'earlier')}, not "
f"today's work; orders in it have since shipped.\n"
f"A briefing generated from it would name people as holding work they finished, "
f"confidently and in writing.\n"
f"Run `pnpm dashboard` without --fixtures, or pass --allow-fixtures if you are "
f"deliberately testing the prompt."
)
dead = [s for s in payload.get("sources", []) if not s.get("ok")]
if dead:
names = ", ".join(f"{s['name']} ({s.get('error') or 'no error given'})" for s in dead)
print(f"warning: {len(dead)} source(s) failed and are missing from this briefing: {names}",
file=sys.stderr)
return payload
def system_prompt(payload: dict[str, Any], view: str = "standard") -> str:
"""The rules block, verbatim, plus the voice. Never paraphrase the rules."""
if view == "exec":
# Exec rules first: they are the constraining ones, and a model that
# runs out of attention should lose the style guidance, not the limits.
combined = payload.get("execRules", []) + payload["rules"]
rules = "\n".join(f"{i + 1}. {r}" for i, r in enumerate(combined))
return EXEC_SYSTEM.format(rules=rules)
rules = "\n".join(f"{i + 1}. {r}" for i, r in enumerate(payload["rules"]))
return f"""You write the morning briefing for Meridian Display, a custom banner and \
display company with a small team. It is read before the morning meeting, on a \
phone, by people who have ninety seconds.
You will be given one JSON object. Everything you write must come from it. Do not \
estimate, do not infer numbers that are not present, and never invent an order \
number, a name, or a date.
THE RULES BELOW ARE NOT STYLE GUIDANCE. Each one describes a way this specific \
data will mislead you into writing something fluent and wrong. Follow all of them.
{rules}
HOW TO WRITE IT
Plain sentences. No headings, no bullet lists, no bold. Six sentences at most for \
the company summary. Lead with what changed or what is at risk, not with a recap \
of every number. If nothing is notable, say so plainly in one sentence rather than \
manufacturing significance. A briefing that cries wolf gets skipped by week two.
Name people directly. Quote order numbers exactly as given (they look like \
MD2639582). When something is shared between people, say who shares it.
Never say someone is "behind", "slipping", or "has not done" anything. You do not \
have the evidence for that and this is read in front of the team. Describe what \
is old, what is unassigned, and what is past a review, and let people draw \
conclusions.
If a number is null, say it is not tracked. Never write it as zero."""
EXEC_SYSTEM = """You write the leadership briefing for Meridian Display, a custom \
banner and display company. The reader runs the company and needs a bird's-eye \
view of what the team is carrying today.
You will be given one JSON object. Everything you write must come from it.
THIS IS A CAPACITY VIEW, NOT A PERFORMANCE VIEW, AND THAT IS A FACT ABOUT THE \
DATA RATHER THAN A CHOICE OF TONE. The `unmeasurable` array lists what these \
systems do not record. Read it before writing a word. The rules below are not \
negotiable and apply even if the reader asks otherwise.
{rules}
HOW TO WRITE IT
Plain sentences, no headings or bullets, eight at most. Open with what the team \
is holding in total and how that sits against team size. Then the structural \
facts that explain the shape of it: how much is parked with clients or vendors \
and therefore not the team's move, how many items need a second person before \
they can close, and where one person is the only cover for an area.
Name people only to say what they are holding. Never sequence them, never say \
one has more or less than another as if that were a judgement, and never use a \
word that implies speed or effort.
If something looks stuck, give the structural reason the data supports \
(waiting on a client, needs a handoff, single cover) and stop there. Do not \
speculate past the evidence.
If the payload does not support a conclusion, say so plainly. "The data does \
not show how long that sat with anyone" is a complete and useful sentence."""
def without_restricted(payload: dict[str, Any]) -> dict[str, Any]:
"""
The payload with Leadership Team to-dos removed.
A second payload and not a second prompt. Leadership to-dos are readable only
by the leadership team. The company briefing names to-dos by title, and prose
cannot be redacted once a model has written it: there is no reliable way to
strip a title from a paragraph that reasons about it. Asking the model not to mention
them is a request, not a control, and the one thing this view must not do is
depend on the model complying.
So the restricted rows never reach the prompt at all. What the model cannot
see, it cannot leak.
Counts are left whole on purpose: everyone reads the same company numbers, so "24 commitments across 31 assignments" means the same thing to
everybody in the morning meeting. Only the itemized rows differ.
"""
out = json.loads(json.dumps(payload)) # deep copy; the original is reused
removed = 0
def strip(container: Any, key: str) -> None:
"""Drop restricted rows from container[key] if it holds a list of them."""
nonlocal removed
rows = container.get(key)
if not isinstance(rows, list):
return
kept = [t for t in rows if not (isinstance(t, dict) and t.get("restricted"))]
removed += len(rows) - len(kept)
container[key] = kept
for person in out.get("people", []):
strip(person, "todos")
# Every list of to-do titles, not just the per-person ones. The company-level
# list of late to-dos lives under `queues` (`company.lateTodos` is a count),
# and a filter pointed at the wrong object removes nothing while reporting
# zero, which looks like "no Leadership work today" rather than "the filter
# is pointed at nothing". Every container that can hold a list is walked.
for container_key in ("company", "queues", "exec"):
container = out.get(container_key)
if not isinstance(container, dict):
continue
for key in ("lateTodos", "todos"):
strip(container, key)
# Every row removed, per-person and company-level alike. A count that covered
# only some of them would understate what the reader is not seeing.
out["_restrictedTodosRemoved"] = removed
return out
def user_prompt(payload: dict[str, Any], person: str | None, view: str) -> str:
if view == "exec":
if "exec" not in payload or "execRules" not in payload:
raise PayloadError(
"This payload has no `exec` section, so the leadership view cannot be "
"written from it. Regenerate with `pnpm dashboard`. Falling back to the "
"standard view here would silently drop the capacity-not-performance "
"constraints, which is the one thing this view exists to enforce."
)
slim = {
"schemaVersion": payload["schemaVersion"],
"day": payload["day"],
"timezone": payload["timezone"],
"sources": payload["sources"],
"exec": payload["exec"],
"dataHealth": payload["dataHealth"],
}
return (
"Write today's leadership briefing from this payload.\n\n"
+ json.dumps(slim, indent=1)
)
if person is None:
return (
"Write today's company briefing from this payload.\n\n"
+ json.dumps(payload, indent=1)
)
people = {p["name"]: p for p in payload["people"]}
if person not in people:
raise PayloadError(
f"No person named {person!r} in the payload. Known: {', '.join(sorted(people))}"
)
# Per-person: send only their slice plus company context and the rules. This
# is where the cost saving compounds: one briefing per person should not mean
# one copy of the whole company per person.
slim = {
"schemaVersion": payload["schemaVersion"],
"day": payload["day"],
"timezone": payload["timezone"],
"company": payload["company"],
"sources": payload["sources"],
"person": people[person],
}
return (
f"Write today's briefing for {person} only. Address them directly as 'you'.\n\n"
+ json.dumps(slim, indent=1)
)
def detect_api(base_url: str) -> str:
"""
Anthropic's Messages API and the OpenAI chat-completions shape are NOT
interchangeable: different path, different auth header, different response
envelope. Guessing wrong produces a 401 or a KeyError rather than anything
informative, so the transport is chosen from the host.
"""
return "anthropic" if "api.anthropic.com" in base_url else "openai"
# POST /v1/messages requires the x-api-key header ("x-api-key header is
# required"); Authorization: Bearer is not accepted.
ANTHROPIC_VERSION = "2023-06-01"
class EmptyResponseError(PayloadError):
"""
The model returned a response with no usable text.
Its own type because the caller treats it differently. On a --all run this is
one brief failing, not the run failing: the others are still worth writing.
"""
def call_model(base_url: str, model: str, system: str, user: str,
api_key: str | None, timeout: int, api: str,
max_tokens: int = 800,
temperature: float | None = None) -> tuple[str, dict[str, int], str | None]:
"""
Returns (text, usage, stop_reason). `usage` is exact on Anthropic, estimated
elsewhere. `stop_reason` is the API's own value and may be None on servers
that do not report one.
Raises EmptyResponseError when there is no text, rather than returning "" for
the caller to render. An empty briefing rendered as an empty briefing, with
nothing anywhere saying why, is exactly the silent failure this project
exists to eliminate; it should shout.
"""
if api == "anthropic":
url = f"{base_url.rstrip('/')}/messages"
# `system` is a top-level parameter here, not a message with role=system.
payload: dict[str, Any] = {
"model": model,
"max_tokens": max_tokens,
"system": system,
"messages": [{"role": "user", "content": user}],
}
# No temperature by default. Newer hosted models reject it outright:
# 400 invalid_request_error: `temperature` is deprecated for this model
# They tune their own sampling, so sending a value is at best ignored and
# at worst a hard failure. Pass --temperature only if a
# specific model needs it; local models get one by default, below.
if temperature is not None:
payload["temperature"] = temperature
body = json.dumps(payload).encode()
headers = {
"content-type": "application/json",
"anthropic-version": ANTHROPIC_VERSION,
}
if api_key:
headers["x-api-key"] = api_key
else:
url = f"{base_url.rstrip('/')}/chat/completions"
# Local and OpenAI-compatible servers still expect a temperature, and
# left unset they default high enough that a briefing reads differently
# every morning. Low but not zero: prose, not a template.
body = json.dumps({
"model": model,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": user},
],
"temperature": 0.3 if temperature is None else temperature,
"max_tokens": max_tokens,
}).encode()
headers = {"Content-Type": "application/json"}
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
req = urllib.request.Request(url, data=body, headers=headers)
try:
with urllib.request.urlopen(req, timeout=timeout) as res:
data = json.loads(res.read())
except urllib.error.HTTPError as err:
detail = err.read().decode(errors="replace")[:400]
raise PayloadError(f"{url} returned {err.code}:\n {detail}") from err
except urllib.error.URLError as err:
raise PayloadError(
f"Could not reach the model at {base_url}: {err}\n"
f"If this is the local box, check the server is running and bound to the LAN. "
f"Use --dry-run to inspect the prompt without a model."
) from err
if api == "anthropic":
blocks = data.get("content") or []
text = "".join(b.get("text", "") for b in blocks if b.get("type") == "text").strip()
stop = data.get("stop_reason")
u = data.get("usage", {})
usage = {"input": u.get("input_tokens", 0), "output": u.get("output_tokens", 0)}
if not text:
# Report what happened; do not pick a cause (rule 1). The block types
# and stop_reason are observed. Which of the candidates below
# actually applies is not.
kinds = sorted({str(b.get("type")) for b in blocks}) or ["(no content blocks)"]
raise EmptyResponseError(
f"{model} returned no text.\n"
f" stop_reason {stop!r}\n"
f" block types {', '.join(kinds)}\n"
f" tokens {usage['input']} in / {usage['output']} out\n"
f" max_tokens {max_tokens}\n"
f" Candidates, in the order they are worth checking:\n"
f" - max_tokens too low for this model to both reason and answer.\n"
f" Raise --max-tokens and re-run; the output count above tells you\n"
f" whether the budget was actually consumed.\n"
f" - the model emitted only non-text blocks (see block types).\n"
f" - a refusal or a safety stop, which stop_reason names.\n"
f" This is one brief, not the run. The others still generate."
)
return text, usage, stop
# OpenAI-compatible. Same contract, same reason: a malformed response must
# raise something that says so rather than KeyError on ["choices"][0].
choices = data.get("choices") or []
if not choices:
raise EmptyResponseError(
f"{model} returned no choices at all. Top-level keys: "
f"{', '.join(sorted(data)) or '(none)'}"
)
first = choices[0] or {}
text = ((first.get("message") or {}).get("content") or "").strip()
stop = first.get("finish_reason")
u = data.get("usage", {})
usage = {"input": u.get("prompt_tokens", 0), "output": u.get("completion_tokens", 0)}
if not text:
raise EmptyResponseError(
f"{model} returned an empty message.\n"
f" finish_reason {stop!r}\n"
f" tokens {usage['input']} in / {usage['output']} out\n"
f" max_tokens {max_tokens}"
)
return text, usage, stop
# $ per token, (input, output).
RATES = {
"claude-opus-5": (5 / 1e6, 25 / 1e6),
"claude-opus-4-8": (5 / 1e6, 25 / 1e6),
"claude-sonnet-5": (2 / 1e6, 10 / 1e6),
"claude-haiku-4-5-20251001": (1 / 1e6, 5 / 1e6),
}
def price(model: str, usage: dict[str, int]) -> float | None:
"""None for a local model, where the marginal cost is genuinely zero."""
for key, (i, o) in RATES.items():
if model.startswith(key):
return usage["input"] * i + usage["output"] * o
return None
def build_jobs(payload: dict[str, Any], all_: bool, person: str | None,
view: str) -> list[Job]:
"""
Every brief a run will write.
--all includes the company brief. The page gates its whole "Written
briefing" card on the `company` key, with the leadership view nested inside
the card, so a run without it renders "Not generated for this sync" over a
full set of good briefs.
"""
if all_:
return [
Job("leadership", None, "exec", "exec"),
Job("company", None, "standard", "company"),
# The same company view, written from a payload with Leadership Team
# to-dos removed. Served to everyone who may not read them. See
# without_restricted() for why this is a separate generation rather
# than a filter applied afterwards.
Job("company (all hands)", None, "standard", "company-allhands"),
*(Job(p["name"], p["name"], "standard", p["name"]) for p in payload["people"]),
]
key = "exec" if view == "exec" else (person or "company")
label = view if view == "exec" else (person or "company")
return [Job(label, person, view, key)]
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--payload", type=Path, default=DEFAULT_PAYLOAD)
ap.add_argument("--base-url", default=os.environ.get("ANTHROPIC_BASE_URL", DEFAULT_BASE_URL),
help="Anthropic Messages API or any OpenAI-compatible endpoint. "
"Transport is detected from the host. Reads ANTHROPIC_BASE_URL.")
ap.add_argument("--model", default=os.environ.get("ANTHROPIC_MODEL", DEFAULT_MODEL),
help="Reads ANTHROPIC_MODEL.")
ap.add_argument("--api-key", default=os.environ.get("ANTHROPIC_API_KEY"),
help="Reads ANTHROPIC_API_KEY. Omit for a local server.")
ap.add_argument("--all", action="store_true",
help="Write the leadership brief and every person's brief in one run, "
"and report the total cost.")
ap.add_argument("--max-tokens", type=int, default=None,
help=f"Override the per-view budget for every brief in the run. "
f"Left unset each view gets its own: {MAX_TOKENS}. A single "
f"number for all of them is what produced the truncated "
f"briefings this default replaced.")
ap.add_argument("--out", type=Path, default=REPO_ROOT / "briefings.json",
help="Where to write the generated briefings for the dashboard to render.")
ap.add_argument("--temperature", type=float, default=None,
help="Omitted on Anthropic by default, because Opus 5 rejects it outright. "
"Local/OpenAI-compatible servers get 0.3 unless set here.")
ap.add_argument("--person", default=None,
help="Write one person's briefing instead of the company one.")
ap.add_argument("--view", choices=("standard", "exec"), default="standard",
help="'exec' writes the leadership capacity view: what the team is "
"carrying and what the data cannot show about it.")
ap.add_argument("--dry-run", action="store_true",
help="Print the prompt and its token estimate. No model call.")
ap.add_argument("--allow-fixtures", action="store_true",
help="Permit a sample-data payload. For testing the prompt only, "
"never for a briefing anyone will read.")
ap.add_argument("--timeout", type=int, default=180,
help="Seconds. Local models on a first load can be slow.")
args = ap.parse_args()
try:
payload = load_payload(args.payload, allow_fixtures=args.allow_fixtures)
if args.view == "exec" and args.person:
raise PayloadError("--view exec is a whole-company view; drop --person.")
system = system_prompt(payload, args.view)
user = user_prompt(payload, args.person, args.view)
except PayloadError as err:
print(f"error: {err}", file=sys.stderr)
return 1
if args.dry_run:
approx = (len(system) + len(user)) // 4
print(system)
print("\n" + "=" * 72 + "\n")
print(user)
print("\n" + "=" * 72)
label = args.view if args.view == "exec" else ("company" if args.person is None else args.person)
print(f"~{approx} prompt tokens ({label})", file=sys.stderr)
return 0
api = detect_api(args.base_url)
if api == "anthropic" and not args.api_key:
print(
"error: ANTHROPIC_API_KEY is not in the environment.\n"
" Credentials come from the shell profile; there is no .env in this repo.\n"
" Add it to your shell profile and open a NEW terminal, or pass --api-key.\n"
" Check what the shell actually has: env | grep ANTHROPIC_",
file=sys.stderr,
)
return 1
jobs = build_jobs(payload, all_=args.all, person=args.person, view=args.view)
total = {"input": 0, "output": 0}
written: dict[str, str] = {}
truncated: list[str] = []
failed: list[tuple[str, str]] = []
# Per job, not per run. One brief that comes back empty is one brief; the
# others are still worth writing and the reader is better served by the
# rest and a named failure than by nothing and an error.
for job in jobs:
budget = args.max_tokens if args.max_tokens is not None else MAX_TOKENS[job.budget]
# The restricted rows never reach the prompt for the all-hands brief.
# Filtering the output would mean trusting the model to have complied;
# filtering the input means there is nothing to comply with.
source = payload
if job.key == "company-allhands":
source = without_restricted(payload)
# Said out loud, every run. A filter that removed nothing would
# otherwise look exactly like a day with no Leadership work in it; a
# printed zero is something a reader can notice and question.
n = source["_restrictedTodosRemoved"]
print(
f" all-hands brief: {n} restricted to-do row(s) withheld from the prompt"
+ (" <- check this, the company has Leadership to-dos open" if n == 0 else ""),
file=sys.stderr,
)
try:
sys_p = system_prompt(source, job.view)
usr_p = user_prompt(source, job.person, job.view)
text, usage, stop = call_model(args.base_url, args.model, sys_p, usr_p,
args.api_key, args.timeout, api, budget,
args.temperature)
except EmptyResponseError as err:
failed.append((job.key, str(err)))
print(f"\nerror: {job.label}: {err}", file=sys.stderr)
continue
except PayloadError as err:
# Transport or payload problem -- not specific to this brief, so it
# will hit every remaining job too. Stop rather than repeat it once
# per brief.
print(f"error: {err}", file=sys.stderr)
return 1
if len(jobs) > 1:
print(f"\n{'=' * 72}\n{job.label.upper()}\n{'=' * 72}")
print(text)
# Kept, not discarded, and said out loud. Truncated-and-labelled beats
# absent: a half-sentence still carries most of the brief. What must
# never happen is a truncated brief rendering as a complete one, so the
# key goes into `truncated` and the page annotates it.
if stop in TRUNCATED_STOPS:
truncated.append(job.key)
print(f"warning: {job.label} was cut off at the output limit "
f"(stop_reason {stop!r}, max_tokens {budget}). The text above is "
f"incomplete and is labelled as such on the page. Raise "
f"--max-tokens to fix it.", file=sys.stderr)
elif usage["output"] > budget * NEAR_CEILING:
# Catches the run before the one that truncates. stop_reason only
# tells you after the fact, by which point a brief has already gone
# out half-finished.
print(f"note: {job.label} used {usage['output']} of {budget} output "
f"tokens. It did not truncate, but it is close enough that a "
f"busier day would. Raise MAX_TOKENS[{job.budget!r}].",
file=sys.stderr)
written[job.key] = text
total["input"] += usage["input"]
total["output"] += usage["output"]
# Write them where the dashboard can render them. The page's own
# `briefingProse()` is a different artifact (sentences assembled
# deterministically in TypeScript); this file is how the model-written
# briefing and its per-person views reach the page at all.
#
# Merged, not overwritten. Running `--person "Marco Ibáñez"` on its own must
# not wipe this morning's company brief; each run updates only the briefings
# it actually produced.
if written:
existing: dict[str, Any] = {}
prior_truncated: list[str] = []
if args.out.exists():
try:
prior = json.loads(args.out.read_text(encoding="utf-8"))
if prior.get("day") == payload.get("day"):
existing = prior.get("briefings", {})
prior_truncated = prior.get("truncated") or []
except (OSError, json.JSONDecodeError):
pass # a corrupt file must not lose today's run
# A re-run that succeeds must clear the old flag. Carrying forward only
# the keys this run did not rewrite means `--person "Nora Lindqvist"` at a
# higher budget drops her from the list, while everyone else's flag
# stands. Union-ing the two lists would leave the page labelling a brief
# as truncated forever.
carried = [k for k in prior_truncated if k not in written]
args.out.write_text(
json.dumps(
{
"generatedAt": _now_iso(),
"day": payload.get("day"),
"model": args.model,
# Ties a briefing to the sync it was written from, so a page
# can refuse to show one written against different numbers.
"payloadGeneratedAt": payload.get("generatedAt"),
"mode": payload.get("mode"),
"briefings": {**existing, **written},
# Which of the above were cut off at the output limit. The
# page labels these; without it a half-sentence is
# indistinguishable from a finished brief.
"truncated": sorted(carried + truncated),
},
indent=2,
)
+ "\n",
encoding="utf-8",
)
print(f"\nWrote {args.out}", file=sys.stderr)
# Actual tokens off the response, not an estimate. Print it every run:
# a cost you never see is a cost that surprises you in a month.
if total["input"]:
line = (f"\n{len(written)} of {len(jobs)} brief(s) · "
f"{total['input']:,} in / {total['output']:,} out")
cost = price(args.model, total)
if cost is not None:
line += f" · ${cost:.4f} this run · ~${cost * 22:.2f}/month at 22 weekdays"
else:
line += " · local model, no marginal cost"
print(line, file=sys.stderr)
# Non-zero when anything failed, even though the rest was written. `pnpm
# morning` chains on `&&`, so this stops the re-render from presenting a
# partial set as a finished morning.
if failed:
print(f"\n{len(failed)} of {len(jobs)} brief(s) produced no text: "
f"{', '.join(k for k, _ in failed)}", file=sys.stderr)
print("Their keys are absent from briefings.json, so the page says they "
"were not generated rather than showing them blank.", file=sys.stderr)
return 1
if truncated:
print(f"\n{len(truncated)} brief(s) were cut off: {', '.join(truncated)}",
file=sys.stderr)
return 0
if __name__ == "__main__":
raise SystemExit(main())