bin/brief.py

#!/usr/bin/env python3 """brief.py - turn a transcript into the smallest thing that still carries the session. WHY THIS EXISTS Compaction re-reads the whole conversation to write a summary. On this machine a working session reached 5,995,006 bytes - about 1.5M tokens of reading to produce a few hundred lines of output. Most of that reading is wasted, and it is wasted the same way every time. Measured on a real 3,787-line transcript: message 64.2% incl. 10.9% opaque thinking SIGNATURES toolUseResult 10.3% a SECOND copy of output already in message uuid/parentUuid 3.6% envelope, one pair per line sessionId/session_id 4.0% the same id repeated 3,787 times assistant text 2.8% <- conclusions user text 0.6% <- the actual instructions The signal is 3.4% of the file. Everything else is machinery, duplication, or reasoning that had already served its purpose by the time it was written down. So this does the extraction ONCE, deterministically, with no model call - and hands compaction the spine instead of the haystack. WHAT IT KEEPS, AND WHY THAT ORDER 1. Every user instruction, verbatim. It is 0.6% of the bytes and 100% of the intent. Never summarised, never trimmed. If a budget is tight, other sections lose first. 2. Errors and how they were resolved. This is where the lessons are. 3. Assistant conclusions - what was decided and what was reported done. 4. Files touched, and commands collapsed by shape. WHAT IT DROPS ON PURPOSE - thinking blocks. Worth being exact about these, because the first measurement here was wrong: in a persisted transcript the reasoning TEXT is not stored at all. 411 thinking blocks in the sample carried 0 bytes of readable text and 655,712 bytes of cryptographic signature. So this is not summarising reasoning away - the reasoning is already gone, and what remains is an opaque blob worth 10.9% of the file. Dropping it loses nothing that could ever be read. - toolUseResult. It is a byte-for-byte second copy of the tool_result already read. - tool call arguments. The command line is kept; the full argument blob is not. - successful tool output. Kept only when it failed, because that is when it matters. TWO RULES THIS FILE FOLLOWS Redact BEFORE truncating. A secret cut in half is still a secret, and a secret that survives because the scrubber only saw the first 200 characters is a leak this project has already shipped once. scrub() runs on the full text, then it is cut. Never cap silently. Anything dropped for budget is counted and printed. A brief that quietly loses half the session while looking complete is worse than no brief. """ from __future__ import annotations import json import os import re import sys from pathlib import Path BIN = Path(__file__).resolve().parent sys.path.insert(0, str(BIN)) try: from redact import scrub except Exception: # pragma: no cover def scrub(t): # type: ignore return (t or "", 0, []) DEFAULT_BUDGET = 24_000 # characters of brief, ~6k tokens # How many of the person's own requests to carry, newest kept. # # Not a silent cap: whatever is dropped is counted and named in the brief itself. # 60 covers a long working session without letting a very large transcript hand back # the whole window it just freed. ASK_KEEP = 60 # A standing constraint is an instruction that stays true after the sentence that # carried it. "make sure it works on windows" is still binding tomorrow; "check the # board" is not. Keeping the first kind and dropping the second is the difference # between context that helps and context that just takes up room. CONSTRAINT_MARKS = ("never ", "always ", "do not ", "don't ", "dont ", "must ", "make sure", "ensure ", "are we clear", "no ", "stop ", "should not", "shouldn't", "cannot ", "can't ", "under no ") # Something left open. These read as unfinished when a session ends. OPEN_MARKS = ("still ", "blocked", "pending", "waiting", "todo", "to do", "next ", "not yet", "unfinished", "remaining", "left to", "awaiting", "needs to") ERROR_MARKS = ("Traceback (most recent", "Error:", "error:", "ERROR", "FAILED", "FAIL ", "exception", "Exception", "fatal", "refused", "Permission denied", "No such file", "not found", "exit code 1", "command not found", "SyntaxError", "cannot ", "Cannot ") # Text Claude Code injects that is not the user talking. It is STRIPPED, not used as # a reason to discard the message. # # The first version treated these as a filter: if a marker appeared anywhere in a user # message, the whole message was dropped. But the tool appends reminders TO real # instructions, so this silently deleted the user's own words along with the # boilerplate - including a standing order about supported platforms. The brief looked # fine; it was just missing things, which is the hardest kind of wrong to notice. # # So: cut out the injected spans, keep whatever the person actually wrote. INJECTED_SPANS = ( ("<system-reminder>", "</system-reminder>"), ("<command-name>", "</command-message>"), ("<local-command-stdout>", "</local-command-stdout>"), ("<local-command-stderr>", "</local-command-stderr>"), # ⚠️ A task-notification ARRIVES AS A USER MESSAGE. The harness delivers it in the # same record type a person's typing lands in, so every one of them was being read # as something the user said - and the fragments that survived truncation were # promoted into STANDING INSTRUCTIONS, the one section that is never trimmed and is # handed to the next session as binding. Measured on a real transcript: entries like # - last-failure skills" finished</summary> <note>A task-notification fires... # were being carried forward as the user's own standing orders. Same failure as # mining a docstring: an instruction that nobody ever gave. ("<task-notification>", "</task-notification>"), ) # These mark a whole block that is genuinely not the user speaking. NOT_THE_USER = ("Caveat: The messages below", "This session is being continued from", "[Request interrupted") def _blocks(line: str): try: d = json.loads(line) except Exception: return None, [] m = d.get("message") or {} role = m.get("role") or d.get("type") or "" c = m.get("content") if isinstance(c, str): return role, [{"type": "text", "text": c}] if isinstance(c, list): return role, [b for b in c if isinstance(b, dict)] return role, [] def _strip_injected(t: str) -> str: """Remove tool-injected spans, keep the human text around them.""" for open_tag, close_tag in INJECTED_SPANS: while True: a = t.find(open_tag) if a < 0: break b = t.find(close_tag, a) t = (t[:a] + t[b + len(close_tag):]) if b >= 0 else t[:a] return t def _clean(s: str, limit: int) -> str: """Scrub first, THEN shorten. Never the other way round.""" s, _, _ = scrub(s or "") s = " ".join(s.split()) return s if len(s) <= limit else s[:limit] + " ..." def _tidy_rule(piece: str) -> str: """Strip markdown scaffolding off a mined rule, or reject it. Mining a previous summary means reading formatted prose, so list bullets, bold markers and stray numbering come along with the sentence. A rules section is only useful if every line in it is actually a rule - "5." is noise, and noise in the one section that never gets trimmed is expensive noise. """ piece = piece.strip().strip("-*`\"' \t") piece = re.sub(r"^\d+[.)]\s*", "", piece) # "5. " / "7) " piece = re.sub(r"\*\*|`|^#+\s*", "", piece).strip() if not (12 < len(piece) <= 200): return "" if not piece[:1].isalpha(): return "" if len(piece.split()) < 4: return "" # A quoted CODE fragment is not an instruction. Summaries quote source as often # as they quote people, and a comment lifted out of a function reads like an # imperative ("NO locked FIELD ON PURPOSE") without being one. if "\n" in piece or "#" in piece or "()" in piece or "```" in piece: return "" return piece # Headings under which a summary is reporting what the PERSON said. Matched loosely # against the heading text, because summary formats vary, but never against body # prose - a heading is a claim about provenance and body text is not. USER_VOICE_HEADINGS = ( "user message", "all user messages", "user request", "primary request", "stated verbatim by the user", "user instruction", "verbatim, in order", ) def _user_voiced(summary: str) -> str: """The parts of a carried summary that are reporting the user's own words. Returns "" when the summary does not mark any such section. That is a real answer, not a failure: it means this document does not tell us which words were the person's, and guessing is how a docstring becomes a standing order. """ keep, out = False, [] for line in summary.splitlines(): h = line.lstrip("#*- \t").rstrip(":*# \t").lower() is_heading = bool(re.match(r"\s*(#{1,6}\s|\*\*|\d+\.\s+\*\*)", line)) if is_heading: keep = any(m in h for m in USER_VOICE_HEADINGS) if keep: out.append(line) return "\n".join(out) def _dedupe_rules(rules: list[str]) -> list[str]: """One rule per idea. The same order given twice is still one order.""" out, seen = [], set() for r in rules: key = re.sub(r"[^a-z0-9]", "", r.lower())[:60] if key and key not in seen: seen.add(key) out.append(r) return out def _record_ask(out: dict, clean: str) -> None: """One path for everything the person said, however it reached the transcript.""" if not clean.strip(): return out["asks"].append(clean) # An instruction can carry a rule AND a task. Split on sentence boundaries so the # binding half survives even when it arrived attached to a one-off request. for piece in re.split(r"(?<=[.!?])\s+|\s+(?:and|but|also)\s+", clean): low = piece.lower().strip() if len(low) < 8 or len(piece) > 220: continue # Fewer than three words is a reaction, not a rule: "Stop here", "no go". # # ⚠️ THE THRESHOLD IS THREE, NOT FOUR, AND THE TEST IS WHY. Four looked tidier # and it silently dropped "no outside servers" - three words, and exactly the # kind of absolute the whole section exists to carry. Losing one real rule is # worse than keeping several bits of noise: noise is visible and a person can # ignore it, whereas a rule that is gone is gone quietly and the next session # breaks it without ever knowing it existed. if len(piece.split()) < 3: continue # Markup remnants mean this came from a harness block that was cut mid-tag, # not from a person. Anything still carrying </...> or <...> is not speech. if "</" in piece or "<note>" in piece or "<summary>" in piece: continue if any(m in low for m in CONSTRAINT_MARKS): out["rules"].append(piece.strip()) def _cmd_shape(cmd: str) -> str: """Collapse a command to its shape so 200 near-identical runs count as one.""" c = " ".join((cmd or "").split()) c = re.sub(r"'[^']*'|\"[^\"]*\"", "'..'", c) c = re.sub(r"/[\w./~-]{12,}", "/..", c) c = re.sub(r"\b[0-9a-f]{8,}\b", "..", c) c = re.sub(r"\b\d{3,}\b", "N", c) return c[:110] def harvest(path: Path) -> dict: """Single streaming pass. Nothing is held that is not going into the brief.""" out = {"asks": [], "did": [], "errors": [], "files": {}, "cmds": {}, "rules": [], "open": [], "carried": "", "src_bytes": 0, "lines": 0, "thinking_bytes": 0, "sig_bytes": 0, "dup_bytes": 0} pending: dict[str, str] = {} # tool_use_id -> short label with open(path, encoding="utf-8", errors="replace") as fh: for line in fh: out["src_bytes"] += len(line) out["lines"] += 1 try: d = json.loads(line) except Exception: continue # counted so the header can show what was skipped, never stored tur = d.get("toolUseResult") if tur is not None: out["dup_bytes"] += len(json.dumps(tur)) # A message typed while the assistant is working is stored as its own # record type, with the text at TOP LEVEL rather than inside "message". # Reading only "message" therefore misses it completely - and for anyone # who works by queueing instructions mid-task, that is not an edge case, # it is most of what they said. In the transcript this was written # against, 110 such records held several standing orders that appeared # nowhere else in the file. if d.get("type") == "queue-operation": if d.get("operation") == "enqueue": q = (d.get("content") or "").strip() # task notifications are the harness talking to itself if q and not q.startswith(("<task-notification>", "<system-reminder>")): q = _strip_injected(q) if q.strip(): _record_ask(out, _clean(q, 600)) continue role, blocks = _blocks(line) for b in blocks: t = b.get("type") if t == "thinking": # Counted apart because they are different things. Text is content # that was dropped; signature is a blob that never held content. # Reporting them together produced a header line that read "0B # thinking" while the file was 10.9% signature - a true number # attached to the wrong name, which is its own kind of wrong. out["thinking_bytes"] += len(b.get("thinking") or "") out["sig_bytes"] += len(b.get("signature") or "") continue # dropped on purpose if t == "text": txt = b.get("text") or "" if role == "user": if not txt.strip(): continue if any(k in txt for k in NOT_THE_USER): # A compaction summary is not the user speaking, but it is # the ONLY surviving record of everything before the # compaction. Dropping it loses the first half of the # session outright - which is how a standing order about # not naming people went missing from an early draft of # this brief. Keep it, clearly labelled as prior context # so it is never mistaken for a fresh instruction. if "This session is being continued from" in txt: out["carried"] = txt continue txt = _strip_injected(txt) if not txt.strip(): continue _record_ask(out, _clean(txt, 600)) elif role == "assistant": s = txt.strip() if len(s) > 40: c = _clean(s, 400) out["did"].append(c) for piece in re.split(r"(?<=[.!?])\s+", c): low = piece.lower() if 12 < len(piece) <= 200 and any(m in low for m in OPEN_MARKS): out["open"].append(piece.strip()) continue if t == "tool_use": name = b.get("name") or "?" inp = b.get("input") or {} label = "" if name == "Bash": label = _cmd_shape(str(inp.get("command", ""))) out["cmds"][label] = out["cmds"].get(label, 0) + 1 elif name in ("Write", "Edit", "NotebookEdit", "Read"): fp = str(inp.get("file_path", "")) if fp: v = out["files"].setdefault(fp, set()) v.add("read" if name == "Read" else "write") label = f"{name} {fp}" else: label = name pending[str(b.get("id"))] = f"{name}: {label}"[:140] continue if t == "tool_result": body = b.get("content") if not isinstance(body, str): body = json.dumps(body) if body is not None else "" failed = bool(b.get("is_error")) or any(m in body for m in ERROR_MARKS) if failed: who = pending.get(str(b.get("tool_use_id")), "") out["errors"].append({"what": who, "out": _clean(body, 300)}) continue # the repeated-prompt collapse: a loop that ran 40 times is one line, not 40 # A rule does not stop being a rule because it survived only inside a previous # summary. Mine the carried-over text for standing instructions too, so binding # orders land in the protected section wherever they happen to have survived - # this is how "never list anyone by name", given before a compaction, stays in # front of the next session instead of being trimmed with the old history. if out["carried"]: # ONLY quoted spans, and ONLY from the parts of the summary that are ABOUT # what the person said. # # Taking every quoted span was the earlier rule, on the reasoning that a # summary quotes the user faithfully. It does - and it also quotes source # code, other agents' messages, and my own docstrings, with identical # punctuation. Measured on a real transcript, that rule promoted # "Scrub first, THEN shorten. Never the other way round." (a docstring) # "NO RESPONDER, NOT EVEN A BOUNDED ONE" (another crew) # into STANDING INSTRUCTIONS - the one section that is never trimmed and is # presented to the next session as orders the user is still bound by. # Attributing a code comment to the Commander is not a formatting slip. It # invents an instruction, and inventing one is worse than losing one. # # A summary marks its own provenance in its headings. Use it: quotes are only # promoted from sections that say they hold the user's words. If no such # section exists, nothing is promoted. Carrying too few rules is a cost the # next session can see and fix; carrying a fabricated one is not. for q in re.findall(r'"([^"]{12,200})"', _user_voiced(out["carried"])): tidy = _tidy_rule(q) if tidy and any(m in tidy.lower() for m in CONSTRAINT_MARKS): out["rules"].append(tidy) out["rules"] = _dedupe_rules(out["rules"]) out["asks"] = _collapse(out["asks"]) # Only the tail of the open-thread list is meaningful: something described as # blocked early on was usually resolved later in the same session, and carrying # it forward as still-open is worse than not mentioning it. out["open"] = _collapse(out["open"][-14:]) return out def _collapse(items: list[str]) -> list[str]: """Fold consecutive-or-repeated identical entries into one with a count.""" seen: dict[str, int] = {} order: list[str] = [] for s in items: key = s[:120] if key not in seen: seen[key] = 0 order.append(s) seen[key] += 1 return [(s if seen[s[:120]] == 1 else f"{s} [x{seen[s[:120]]}]") for s in order] def _carry_extract(text: str, limit: int = 6000) -> str: """Pull the durable part out of a previous compaction summary. Its instruction list is the valuable part: it is the only place the user's earlier words still exist. Prefer that section; fall back to the head of the summary. """ for header in ("## 6. All User Messages", "All User Messages", "## 1. Primary Request", "Primary Request"): i = text.find(header) if i >= 0: return _neutralise_headings(_clean(text[i:i + limit * 2], limit)) return _neutralise_headings(_clean(text[:limit * 2], limit)) def _neutralise_headings(t: str) -> str: """Stop quoted content from impersonating a section heading. _clean collapses whitespace, so a carried-over summary becomes ONE line that can begin with "## 6. All User Messages". Anything that later scans for lines starting with "## " then reads that as a real heading and treats the entire block as a section - which is exactly how the short mode ended up printing the whole brief it was supposed to be an alternative to. Content is content. It does not get to define the document's structure. """ t = re.sub(r"^#+\s*", "", t) return t.replace("## ", "- ").replace("#", "") def render(h: dict, budget: int = DEFAULT_BUDGET) -> str: """Assemble under a budget by PRIORITY, not by position. The first version protected the head of the document and trimmed the tail. That sounds reasonable and is wrong: it made the carried-over history - the only surviving record of everything before a compaction - the first thing thrown away, while a list of 500 command shapes sat safely near the top. So sections declare what they are worth. Intent is never cut. Command lists go first. Whatever goes is named in NOT INCLUDED, because a brief that silently loses half the session while looking complete is worse than no brief at all. """ src = h["src_bytes"] drops = [] if h["sig_bytes"]: drops.append(f"{h['sig_bytes']:,}B opaque thinking signatures (no readable content)") if h["thinking_bytes"]: drops.append(f"{h['thinking_bytes']:,}B reasoning text") if h["dup_bytes"]: drops.append(f"{h['dup_bytes']:,}B duplicate tool output") head = ["# SESSION BRIEF - precomputed by Recall, no model call", f"# source {src:,} bytes / {h['lines']:,} lines", "# dropped: " + ("; ".join(drops) if drops else "nothing"), ""] # (priority, title, lines, note-if-cut). Lower number = protected for longer. sections: list[tuple[int, str, list[str], str]] = [] if h["rules"]: sections.append((0, "## STANDING INSTRUCTIONS (still binding - rules, not history)", [f"- {r}" for r in h["rules"]], "standing instructions")) # ⚠️ THE REQUEST LOG IS HISTORY. THE RULES ARE NOT. Do not merge these again. # # Both used to sit under "intent is never trimmed", which is right for a binding # rule and wrong for a transcript of every request ever made. On a 22 MB session # that produced 527 verbatim requests - 107,485 of the brief's 119,031 bytes - # and the whole thing was injected straight back after the compaction that was # supposed to free the window. Measured on this machine: the context came down # from nearly full to 81% and stopped there, because roughly 30,000 tokens of # request history went back in behind it. The saving the tool exists for was # being handed back at the door. # # So the OLDEST requests go first and the count that went is stated. The most # recent ones are what the next session is actually continuing; a request from # four hours and nine subjects ago is context, not instruction. STANDING # INSTRUCTIONS stay uncapped and unpriorityised - a rule is binding however old # it is, and that section is small because it is deduplicated. asks = h["asks"] ask_lines = [f"- {a}" for a in asks[-ASK_KEEP:]] if len(asks) > ASK_KEEP: ask_lines.insert(0, f"- ({len(asks) - ASK_KEEP} earlier requests not listed - " f"the most recent {ASK_KEEP} are here, oldest first)") sections.append((0, "## WHAT WAS ASKED (verbatim, most recent last)", ask_lines, "instructions")) if h["carried"]: sections.append((1, "## CARRIED OVER FROM AN EARLIER COMPACTION (prior context, not new orders)", [_carry_extract(h["carried"])], "carried-over history")) if h["open"]: sections.append((2, "## LEFT OPEN AT THE END (may already be resolved - verify first)", [f"- {o}" for o in h["open"][-10:]], "open threads")) if h["errors"]: sections.append((3, "## WHERE IT WENT WRONG (and what the output actually said)", [f"- {e['what']} -> {e['out']}" for e in h["errors"][:40]], "errors")) if h["did"]: sections.append((4, "## WHAT WAS DECIDED AND REPORTED DONE (most recent last)", [f"- {x}" for x in h["did"][-60:]], "decisions")) if h["files"]: wrote = sorted(pth for pth, v in h["files"].items() if "write" in v) read_n = sum(1 for v in h["files"].values() if "write" not in v) lines = [f"- W {pth}" for pth in wrote[:60]] if read_n: lines.append(f"- (read only, not listed: {read_n} file(s))") sections.append((5, "## FILES TOUCHED", lines, "files touched")) if h["cmds"]: top = sorted(h["cmds"].items(), key=lambda kv: -kv[1])[:35] sections.append((6, "## COMMANDS, COLLAPSED BY SHAPE (counts are real; the text is normalised, " "not the exact command that ran)", [f"- {n:>3} x {shape}" for shape, n in top], "command shapes")) def size(sec) -> int: return len(sec[1]) + 1 + sum(len(x) + 1 for x in sec[2]) + 1 cut: list[str] = [] total = sum(len(x) + 1 for x in head) + sum(size(x) for x in sections) for prio in (6, 5, 4, 3, 2, 1): if total <= budget: break for sec in [x for x in sections if x[0] == prio]: if total <= budget: break total -= size(sec) sections.remove(sec) cut.append(f"the {sec[3]} section ({len(sec[2])} entries)") out = list(head) for _, title, lines, _n in sections: out.append(title) out.extend(lines) out.append("") body = "\n".join(out) if len(body) > budget: # STANDING INSTRUCTIONS are still never trimmed - a rule the person is bound # by does not get cut to fit a number. But say so loudly and say by how much, # because this is now the only way the brief can exceed its budget, and if it # is exceeding it by a lot that is worth someone looking at. cut.append(f"{len(body) - budget:,} characters over budget, kept anyway " f"(standing instructions are never trimmed)") if cut: body += "\n## NOT INCLUDED\n" + "\n".join(f"- {c}" for c in cut) + "\n" return body def main() -> int: args = sys.argv[1:] budget = DEFAULT_BUDGET if "--budget" in args: i = args.index("--budget") budget = int(args[i + 1]) del args[i:i + 2] as_json = "--json" in args args = [a for a in args if not a.startswith("--")] if args: path = Path(args[0]) else: env = os.environ.get("CLAUDE_TRANSCRIPT_PATH") if not env: print("usage: brief.py <transcript.jsonl> [--budget N] [--json]", file=sys.stderr) return 2 path = Path(env) if not path.exists(): print(f"no such transcript: {path}", file=sys.stderr) return 2 h = harvest(path) text = render(h, budget) if as_json: print(json.dumps({"brief": text, "src_bytes": h["src_bytes"], "out_bytes": len(text), "lines": h["lines"]})) else: print(text) return 0 if __name__ == "__main__": raise SystemExit(main())