#!/usr/bin/env python3 """ build_corpus.py: merge extracted quote files into data/corpus.json.

Reads every data/quotes/*.jsonl (one quote per line, written during setup or a refresh), checks each row against the codebook in SETUP.md, de-duplicates by quote_key, and writes data/corpus.json with a fresh generated stamp.

Reports by set math, not row count: rows read, distinct keys, duplicate keys, conflicting duplicates (same key, different text), and rejected rows with reasons. Exits 1 if anything was rejected or conflicted, so a bad batch never slips in.

./build_corpus.py # build ./build_corpus.py --check # validate only, write nothing """ import argparse, collections, datetime, glob, json, os, re, sys

HERE = os.path.dirname(os.path.abspath(file)) ROOT = os.path.normpath(os.path.join(HERE, "..")) QUOTES = os.path.join(ROOT, "data", "quotes") OUT = os.path.join(ROOT, "data", "corpus.json") PROFILE = os.path.join(HERE, "profile.json")

STANCES = {"problem", "desire", "failed-solution", "objection", "proof", "decision-trigger"} STAGES = {"Problem Aware", "Solution/Category Aware", "Product/Brand Aware", "Most Aware"} SOURCES = {"discovery-call", "sales-call", "client-call", "interview"} CONFIDENCE = {"high", "med", "low"} KEY = re.compile(r"^[a-z-]+:[^#]+(#\d+)?$") REQUIRED = ("quote_key", "quote", "stance", "stage", "source_type", "confidence")

def check_row(q, personas): errs = [] for f in REQUIRED: if not q.get(f): errs.append("missing %s" % f) if q.get("quote_key") and not KEY.match(q["quote_key"]): errs.append("quote_key %r must look like source:id#n" % q["quote_key"]) if q.get("stance") and q["stance"] not in STANCES: errs.append("stance %r not in %s" % (q["stance"], sorted(STANCES))) if q.get("stage") and q["stage"] not in STAGES: errs.append("stage %r not in %s" % (q["stage"], sorted(STAGES))) if q.get("source_type") and q["source_type"] not in SOURCES: errs.append("source_type %r not in %s" % (q["source_type"], sorted(SOURCES))) if q.get("confidence") and q["confidence"] not in CONFIDENCE: errs.append("confidence %r not in %s" % (q["confidence"], sorted(CONFIDENCE))) if q.get("persona") and personas and q["persona"] not in personas: errs.append("persona %r is not one of the personas in references/profile.json" % q["persona"]) if q.get("date"): try: datetime.date.fromisoformat(q["date"]) except ValueError: errs.append("date %r must be YYYY-MM-DD" % q["date"]) return errs

def main(): ap = argparse.ArgumentParser(description=doc, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--check", action="store_true", help="validate only; do not write data/corpus.json") a = ap.parse_args()

personas = set(json.load(open(PROFILE)).get("personas") or []) if os.path.exists(PROFILE) else set()
files = sorted(glob.glob(os.path.join(QUOTES, "*.jsonl")))
if not files:
    sys.exit("no quote files in %s: extract quotes first (SETUP.md step 3)" % QUOTES)

rows, rejected = 0, []
by_key, dup_same, dup_conflict = {}, 0, []
for path in files:
    for i, line in enumerate(open(path), 1):
        if not line.strip():
            continue
        rows += 1
        where = "%s:%d" % (os.path.basename(path), i)
        try:
            q = json.loads(line)
        except json.JSONDecodeError as e:
            rejected.append((where, ["not valid JSON: %s" % e]))
            continue
        errs = check_row(q, personas)
        if errs:
            rejected.append((where, errs))
            continue
        k = q["quote_key"]
        if k in by_key:
            if by_key[k]["quote"] == q["quote"]:
                dup_same += 1
            else:
                dup_conflict.append((where, k))
            continue
        by_key[k] = q

print("files read:          %d" % len(files))
print("rows read:           %d" % rows)
print("distinct quote_keys: %d" % len(by_key))
print("exact duplicates:    %d (dropped, harmless)" % dup_same)
print("conflicting dupes:   %d" % len(dup_conflict))
print("rejected rows:       %d" % len(rejected))
for where, k in dup_conflict[:20]:
    print("  ! %s reuses quote_key %s with different text" % (where, k))
for where, errs in rejected[:20]:
    print("  x %s: %s" % (where, "; ".join(errs)))
if len(rejected) + len(dup_conflict) > 20:
    print("  ... and more")

quotes = sorted(by_key.values(), key=lambda q: q["quote_key"])
counts = collections.Counter(q["source_type"] for q in quotes)
print("by source_type:      %s" % dict(counts))

if rejected or dup_conflict:
    print("\nNOT WRITTEN: fix the rows above and rerun.")
    sys.exit(1)
if a.check:
    print("\nCHECK PASSED (nothing written)")
    return

corpus = {"generated": datetime.datetime.now().isoformat(timespec="seconds"),
          "total": len(quotes), "quotes": quotes}
json.dump(corpus, open(OUT, "w"), indent=1, ensure_ascii=False)
if os.path.exists(PROFILE):
    prof = json.load(open(PROFILE))
    prof["last_corpus_build"] = corpus["generated"]
    json.dump(prof, open(PROFILE, "w"), indent=2, ensure_ascii=False)
print("\nWROTE %s: %d quotes, generated %s" % (os.path.relpath(OUT, ROOT), len(quotes), corpus["generated"]))

if name == "main": main()

#!/usr/bin/env python3 """ panel.py: evidence retrieval for the ICP panel.

Slices data/corpus.json down to one seat's evidence pack, widening the filter along a fixed ladder when the narrow slice is too thin, and reporting exactly which precision level the answer is actually standing on.

Stdlib only: no pip install needed.

./panel.py seats ./panel.py evidence --seat cold_skeptic --persona "Mid-size firm" ./panel.py evidence --seat objector --pattern "Slow onboarding" --n 8

--config points at a different seats.json (default: seats.json beside this script). """ import argparse, datetime, hashlib, json, os, sys, collections

HERE = os.path.dirname(os.path.abspath(file)) CFG = os.environ.get("PANEL_CONFIG") or os.path.join(HERE, "seats.json") NARROW_DIMS = ("pattern", "persona", "stage", "topic")

CONF_RANK = {"high": 0, "med": 1, "low": 2}

def load(): cfg = json.load(open(CFG)) # cfg["corpus"] is relative to the panel root (the folder holding config/) corpus_path = os.path.normpath(os.path.join(os.path.dirname(os.path.abspath(CFG)), "..", cfg["corpus"])) if not os.path.exists(corpus_path): sys.exit("no corpus at %s: run setup first (see SETUP.md), then references/build_corpus.py" % corpus_path) corpus = json.load(open(corpus_path)) return cfg, corpus

def apply_base(quotes, base): out = quotes if "source_type_in" in base: out = [q for q in out if q.get("source_type") in base["source_type_in"]] if "stance_in" in base: out = [q for q in out if q.get("stance") in base["stance_in"]] if "stage_in" in base: out = [q for q in out if q.get("stage") in base["stage_in"]] if "persona_in" in base: out = [q for q in out if q.get("persona") in base["persona_in"]] return out

def apply_narrow(quotes, **filters): out = quotes for dim, val in filters.items(): if val: out = [q for q in out if (q.get(dim) or "") == val] return out

def rid(q): p = q.get("provenance") or {} return p.get("recording_id") or p.get("doc_id") or q.get("source_ref") or "?"

def sample(quotes, n): """Prefer high-confidence, then spread across distinct calls so n quotes from one call never masquerade as n independent data points.""" by_call = collections.OrderedDict() for q in sorted(quotes, key=lambda x: CONF_RANK.get(x.get("confidence"), 3)): by_call.setdefault(rid(q), []).append(q) # Visit calls in a stable hash order, not corpus order: corpus order would draw only # from the alphabetically-first sources. by_call = collections.OrderedDict( sorted(by_call.items(), key=lambda kv: hashlib.sha1(str(kv[0]).encode()).hexdigest())) picked, exhausted = [], False while len(picked) < n and not exhausted: exhausted = True for call in list(by_call): if by_call[call]: picked.append(by_call[call].pop(0)) exhausted = False if len(picked) >= n: break return picked

def age_profile(quotes, stale_days=540): """Surface how old the evidence is. Never used to discard: pruning would gut a thin corpus, and 'is this pain still showing up' is a question you can only ask if you kept the old quotes.""" ds = sorted(q["date"] for q in quotes if q.get("date")) out = {"dated": len(ds), "undated": len(quotes) - len(ds)} if not ds: return out today = datetime.date.today() ages = [(today - datetime.date.fromisoformat(d)).days for d in ds] ages.sort() out["oldest"] = ds[0] out["newest"] = ds[-1] out["median_age_days"] = ages[len(ages) // 2] recent = sum(1 for a in ages if a <= 365) out["within_12mo"] = recent out["share_within_12mo"] = round(recent / len(ages), 2) if ages[len(ages) // 2] > stale_days: out["flag"] = ("STALE: median evidence is %d days old. Say so in the verdict: this " "describes what buyers said, not necessarily what they say now." % ages[len(ages) // 2]) return out

def band(n, bands): if n >= bands["sufficient"]: return "sufficient" if n >= bands["thin"]: return "thin" return "insufficient"

def retrieve(seat_key, n=12, **filters): cfg, corpus = load() if seat_key not in cfg["seats"]: sys.exit("unknown seat %r: choose from %s" % (seat_key, ", ".join(cfg["seats"]))) seat = cfg["seats"][seat_key] bands = cfg["density_bands"] pool = apply_base(corpus["quotes"], seat["base_filter"])

requested = {d: filters.get(d) for d in NARROW_DIMS}
active = dict(requested)
dropped = []

# Widen one dimension at a time until the slice can carry a verdict.
# Ladder from config, then any other requested dimension the config didn't rank.
ladder = list(cfg["widening_ladder"]) + [d for d in NARROW_DIMS if d not in cfg["widening_ladder"]]
for dim in [None] + ladder:
    if dim is not None:
        if not active.get(dim):
            continue
        active[dim] = None
        dropped.append(dim)
    hits = apply_narrow(pool, **active)
    if band(len(hits), bands) == "sufficient":
        break

hits = apply_narrow(pool, **active)
surviving = [k for k, v in active.items() if v] or ["seat base filter only"]
picked = sample(hits, n)

return {
    "seat": seat_key,
    "seat_role": seat["role"],
    "may_not_answer": seat["may_not_answer"],
    "corpus_version": {"generated": corpus.get("generated"), "total": corpus.get("total")},
    "retrieval": {
        "requested_filter": {k: v for k, v in requested.items() if v},
        "precision_level": "+".join(surviving),
        "dropped": dropped,
        "n": len(hits),
        "distinct_calls": len({rid(q) for q in hits}),
        "density_band": band(len(hits), bands),
        "source_mix": dict(collections.Counter(q.get("source_type") for q in hits)),
        "age_mix": age_profile(hits),
    },
    "evidence": [
        {
            "quote_key": q["quote_key"],
            "quote": q["quote"],
            "stance": q.get("stance"),
            "stage": q.get("stage"),
            "persona": q.get("persona") or None,
            "pattern": q.get("pattern"),
            "topic": q.get("topic"),
            "source_type": q.get("source_type"),
            "confidence": q.get("confidence"),
            "public_ok": q.get("public_ok"),
            "date": q.get("date"),
        }
        for q in picked
    ],
}

def main(): global CFG ap = argparse.ArgumentParser(description=doc, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--config", help="seats.json of the panel to query (default: seats.json beside this script)") sub = ap.add_subparsers(dest="cmd", required=True) sub.add_parser("seats", help="list seats and their sizes") e = sub.add_parser("evidence", help="build an evidence pack for one seat") e.add_argument("--seat", required=True) e.add_argument("--persona") e.add_argument("--stage") e.add_argument("--pattern") e.add_argument("--topic") e.add_argument("--n", type=int, default=12) a = ap.parse_args() if a.config: CFG = os.path.abspath(a.config)

if a.cmd == "seats":
    cfg, corpus = load()
    bands = cfg["density_bands"]
    print("corpus: %s quotes (generated %s)\n" % (corpus.get("total"), corpus.get("generated")))
    for k, s in cfg["seats"].items():
        pool = apply_base(corpus["quotes"], s["base_filter"])
        calls = len({rid(q) for q in pool})
        flag = "" if band(len(pool), bands) == "sufficient" else \
            "   <-- too thin to judge anything; add more %s" % ", ".join(s["base_filter"].get("stance_in", ["quotes"]))
        print("%-14s n=%-5s calls=%-4s %s%s" % (k, len(pool), calls, s["role"], flag))
    return

filters = {d: getattr(a, d) for d in NARROW_DIMS}
print(json.dumps(retrieve(a.seat, a.n, **filters), indent=2))

if name == "main": main()

{ "_doc": "Your answers to the setup questions in references/setup.md. Setup fills this in; the skill reads it on every run. setup_complete stays false until the corpus is built and references/test_gate.py passes.", "setup_complete": false, "company": "", "offer": "", "buyer": "", "personas": [], "transcript_sources": [], "patterns": [], "excluded_docs": [], "public_quote_rule": "Quotes are for internal judgment only. Never publish one, and never attribute one to a named person or company.", "last_corpus_build": null }

Routing, seats and reporting

Open this on every review, after setup is done. It decides which seats run, what each seat may judge, and the order the findings are reported in.

Routing

Work the play out from what the user hands over. Do not make them pick a seat. Say in one line which seats ran and why.

objector runs in every play except Brief. It is the seat the user would never think to ask for and the one that carries most of the value. Infer the funnel stage from what the asset actually does, not from what it is called. Never blend seats into one verdict: the disagreement between them is the output.

Play notes

• Brief. The returned questions are the outline: the piece answers them in the order they are felt. Better than a keyword outline, because it is what the reader wants answered, not what a tool says gets searched.
• Gate. Lead with stopped_reading_at and strongest_objection; fix those two. If the user waves the objection off without a reason, say plainly that the run produced nothing.
• Choose. Feed the seat the differing unit in isolation: two openers, not two whole posts. Then pre-register: if a real outcome is coming (an A/B test, a send, a title change), append the prediction to predictions.jsonl with truth_winner: null BEFORE the result exists. When the result lands, fill it in and run python3 references/score.py. Never skip it; it is the only way to learn whether the panel is right.
• Diagnose. The panel explains why an observed pattern happened. It never invents the pattern. Get that from real behaviour (analytics, reply rates, sales data).

predictions.jsonl lives in the skill folder, one JSON object per line:

{"id": "2026-11-03-subject-lines", "date": "2026-11-03", "question": "...", "variants": ["A text", "B text"], "panel_pick": "A", "truth_winner": null, "metric": "open rate", "notes": ""}

The seats

Run python3 references/panel.py seats for live sizes. A seat flagged too thin cannot judge anything until more quotes are added. Say so rather than running it.

Seats are not characters. Never give one a name, a backstory, a mood, or a biography. The moment it becomes a person it improvises to fill gaps, which is the failure this skill exists to prevent.

Persona

Personas are the groups in references/profile.json. Map from what the asset says about its reader, not from who the user wishes were reading. If unsure, run unfiltered; the widening ladder drops persona and says so, which beats a wrong persona.

Output files

All paths are relative to the skill folder.

Reporting order, inverted on purpose

1. density_band / dropped: thin means weak signal; widened means not persona-specific.
2. age_mix.flag: a STALE slice describes what buyers said, not what they say now.
3. strongest_objection: the most useful field in the document.
4. unresolved_questions: second most useful.
5. verdict.answer: last, and most suspect. It tends to agree with whoever wrote the draft, because the draft came from the same instincts.

distinct_calls is the honest denominator: 12 quotes from 3 calls is not 12 data points. Watch source_mix for client-side creep. If client calls keep getting added, a pre-sale seat filling with them means the corpus is drifting toward your own language.

When not to run it

• When the user has already decided and wants backing. This is not a second-opinion service.
• On every draft. Reserve it for assets that get spent on, or the dissent becomes noise and gets ignored on the one that mattered.
• On a stage the corpus barely covers, without saying so.

What a passing verdict still does not prove

The gate proves the evidence exists and was retrieved. It cannot prove the inference is sound: a verdict can cite four real quotes and reason wrongly from them and still pass. Only score.py on real outcomes catches that, and it is underpowered until roughly 30 scored predictions. Until then, say out loud that these are well-sourced opinions, not calibrated predictions.

#!/usr/bin/env python3 """ score.py: score the panel's pre-registered predictions against what really happened.

The citation gate proves a verdict's evidence exists. It cannot prove the reasoning is right. This does: every time the panel picks between variants BEFORE the real result exists (an A/B test, two subject lines, a title rewrite), log it in predictions.jsonl. When the result comes in, fill truth_winner. This script runs an exact one-sided binomial test against a coin flip.

The headline number is not accuracy: it is power. A good panel lands around 65-70%, which is undetectable until roughly 30-40 scored predictions. Below that this prints UNDERPOWERED rather than letting a lucky 4/5 read as success.

predictions.jsonl, one JSON object per line: {"id": "2026-11-03-subject-lines", "date": "2026-11-03", "question": "...", "variants": ["A text", "B text"], "panel_pick": "A", "truth_winner": null, "metric": "open rate", "notes": ""}

./score.py """ import json, math, os

HERE = os.path.dirname(os.path.abspath(file)) LEDGER = os.path.join(HERE, "..", "predictions.jsonl")

def binom_p_ge(c, n, p=0.5): return sum(math.comb(n, k) * p ** k * (1 - p) ** (n - k) for k in range(c, n + 1)) if n else 1.0

def min_correct_for_sig(n, alpha=0.05): for c in range(n + 1): if binom_p_ge(c, n) <= alpha: return c return None

def main(): rows = [json.loads(l) for l in open(LEDGER) if l.strip()] if os.path.exists(LEDGER) else [] scored = [r for r in rows if r.get("truth_winner") not in (None, "")] pending = len(rows) - len(scored) n = len(scored) correct = sum(1 for r in scored if r.get("panel_pick") == r.get("truth_winner")) print("predictions logged: %d scored: %d waiting on a result: %d" % (len(rows), n, pending)) if not n: print("Nothing to score yet. Log a prediction before every real A/B result.") return p = binom_p_ge(correct, n) need = min_correct_for_sig(n) print("panel correct: %d/%d (%.0f%%) p vs coin flip = %.3f" % (correct, n, 100.0 * correct / n, p)) if n < 30: bar = "%d/%d" % (need, n) if need is not None else "more predictions: no score" print("UNDERPOWERED: at n=%d you need %s to beat chance. Treat verdicts as " "well-sourced opinion, not calibrated prediction." % (n, bar)) elif p <= 0.05: print("CALIBRATED: the panel beats a coin flip on real outcomes.") else: print("NOT BEATING CHANCE at n=%d. Read the misses before trusting another verdict." % n)

if name == "main": main()

{ "_doc": "Seat definitions for the ICP panel. Each seat is a fixed slice of data/corpus.json. Seats are NOT characters; they are scoped query surfaces. Never add a backstory, name, mood, or biography to a seat. Setup replaces {{COMPANY}} with your company name; the filters work as shipped.", "corpus": "data/corpus.json", "excluded_sources": { "your own messaging": "Any positioning doc, message house, website copy or sales script you wrote. Grounding a verdict in it grades your copy against your copy. Never load it into an evidence pack. Setup records the file names in references/profile.json under excluded_docs.", "draft under review": "The copy being judged is never added to the evidence base." }, "density_bands": { "sufficient": 12, "thin": 5, "_note": "n >= 12 allows a full verdict. 5-11 allows a verdict that MUST be labeled weak signal. n < 5 forbids a verdict; return no-evidence." }, "widening_ladder": [ "pattern", "topic", "persona", "stage" ], "_widening_note": "Retrieve at the narrowest requested filter first. If below the sufficient band, drop one dimension at a time in this order and re-retrieve. Never widen past the seat's base_filter; that is the seat's identity. Always report the precision level actually used.", "seats": { "cold_skeptic": { "role": "Judges whether a piece of content lands on someone who has never heard of {{COMPANY}}.", "base_filter": { "source_type_in": ["discovery-call", "sales-call", "interview"], "stance_in": ["problem", "failed-solution"], "stage_in": ["Problem Aware"] }, "_interview_note": "Interviews (podcast guests, research calls) are ICP people who are not in a buying conversation, which is exactly this seat's job. Their problem language is legitimate here. It is NOT legitimate in the objector or decider seats.", "may_answer": [ "Does this opening land", "Is this the problem they actually have", "Is this framing foreign to them", "What unresolved questions does this reader carry" ], "may_not_answer": [ "Anything about buying from {{COMPANY}}", "Pricing reactions", "Post-sale experience" ] }, "objector": { "role": "Surfaces what makes a reader stop, bounce, or not reply.", "base_filter": { "source_type_in": ["discovery-call", "sales-call"], "stance_in": ["objection", "failed-solution"] }, "may_answer": [ "What in this makes them stop reading", "What objection does this raise and not answer", "Where does this overreach" ], "may_not_answer": [ "Whether they would ultimately buy", "Conversion rate estimates" ] }, "decider": { "role": "Judges close-stage copy: proposals, late sequence emails, the offer or pricing page.", "base_filter": { "source_type_in": ["discovery-call", "sales-call"], "stage_in": ["Most Aware"], "stance_in": ["objection", "decision-trigger"] }, "_source_note": "Restricted to buying-context calls. An existing client's late-stage objection is about the next phase of work, not about deciding to buy.", "may_answer": [ "Does this resolve the last objection", "Is the ask clear", "What has to be true before they move" ], "may_not_answer": [ "Top-of-funnel resonance", "Whether a cold reader understands the category" ] }, "client": { "role": "Post-sale reality only. Retention, expansion, what working with {{COMPANY}} feels like.", "base_filter": { "source_type_in": ["client-call"] }, "_scope_warning": "These people have already absorbed your framing and mostly discuss the work itself. This seat is near-worthless for judging acquisition copy and must never be used for it.", "may_answer": [ "Would an existing client find this useful", "Does this match what working with us actually feels like", "Expansion and retention language" ], "may_not_answer": [ "Whether cold copy resonates", "Whether an article lands", "Anything about the buying decision" ] } } }

icp-review setup

Run this once, with the user, before the first review. It takes one sitting: about 10 minutes of questions, then however long it takes to read the transcripts. Paths are relative to this skill's base directory.

The goal: a corpus of real, verbatim things buyers said, each tagged so the four seats can slice it. The panel is only as honest as this corpus. Never pad it with invented or paraphrased quotes to make a seat look fuller. A thin seat that says "I can't judge this" is working correctly.

Step 1: Ask the setup questions

Ask these one or two at a time, in plain language. Write each answer into references/profile.json.

Then replace every {{COMPANY}} in references/seats.json with the company name.

Minimum to be useful: about 20 sales or discovery calls. Fewer works, but most seats will come back "thin" or "insufficient" and say so. If the user has fewer than 10 sales calls, tell them plainly before starting that the panel will mostly decline to judge until more calls are added.

Step 2: Get the transcripts

Use whatever the user named in question 4: a connected meeting-notes tool (search the available tools for its own verbs, like list_meetings or get_meeting_transcript, before saying it is not connected), a folder of files, or pasted text. Work in batches of 10 to 20 calls.

Skip a call entirely if the buyer barely speaks or it is not with your kind of buyer.

Step 3: Extract and tag quotes

Read each transcript and pull out the buyer's words, never the user's own words, never a teammate's. Take quotes that show what the buyer struggles with, wants, tried, doubts, or needs to decide. Copy them verbatim: trim filler words and false starts, but never reword. A good call yields 5 to 15 quotes; a weak one, zero.

Create data/quotes/ in the skill folder if it is missing (mkdir -p data/quotes), then write one JSON object per line to data/quotes/<source>-<YYYY-MM-DD>.jsonl (for example data/quotes/gong-2026-11-03.jsonl). One file per batch; never edit an older batch file.

{"quote_key": "gong:8812734#3", "quote": "We tried two vendors before and both disappeared after month two.", "stance": "failed-solution", "stage": "Problem Aware", "persona": "Owner-led, under 20 staff", "pattern": "Vendors vanish", "topic": "past vendors", "source_type": "discovery-call", "confidence": "high", "date": "2026-09-14", "public_ok": false, "provenance": {"recording_id": "8812734", "call_title": "Discovery call"}}

Codebook

Tagging rules that matter most:

• The same person can be Problem Aware early in a call and Most Aware at the end. Tag the moment.
• An objection from a current client is about the work, not about buying. It is client-call, never sales-call.
• When unsure between two stances, pick the one that would change what the copy should say, and set confidence to med.

Step 4: Build and check

python3 references/build_corpus.py
python3 references/panel.py seats
python3 references/test_gate.py

• build_corpus.py rejects any row that breaks the codebook and writes nothing until every row passes. Fix the rows in the batch file and rerun.
• panel.py seats shows how big each seat is. Under 12 means that seat can't give a full verdict yet. Tell the user which seats are thin and what kind of calls would fill them (for decider: late-stage sales calls; for client: client calls).
• test_gate.py must print ALL 11 TESTS PASS. It needs at least 6 quotes in the objector seat.

Step 5: Finish

Set setup_complete to true in references/profile.json. Tell the user, in two or three sentences: how many quotes from how many calls, which seats are ready, which are thin, and that they can now say /icp-review and hand over a draft.

#!/usr/bin/env python3 """ test_gate.py: regression test for the citation gate.

Run after any change to validate_verdict.py, seats.json, or verdict-schema.json. Builds a live evidence pack, constructs a valid verdict, then mutates it into each known failure mode and asserts the gate catches it. Two cases assert the gate does NOT fire: a validator that rejects everything is as useless as one that rejects nothing.

./test_gate.py -> "ALL 11 TESTS PASS" and exit 0 """ import copy, json, os, subprocess, sys, tempfile

HERE = os.path.dirname(os.path.abspath(file)) PANEL = os.path.join(HERE, "panel.py") VALIDATE = os.path.join(HERE, "validate_verdict.py")

def main(): # Optional: ./test_gate.py --config <other panel seats.json> --seat <a seat in it> cfg = os.path.abspath(sys.argv[sys.argv.index("--config") + 1]) if "--config" in sys.argv
else os.path.join(HERE, "seats.json") seat = sys.argv[sys.argv.index("--seat") + 1] if "--seat" in sys.argv else "objector" CORPUS = os.path.normpath(os.path.join(os.path.dirname(cfg), "..", json.load(open(cfg))["corpus"])) cflag = ["--config", cfg] tmp = tempfile.mkdtemp(prefix="panel-gate-") pack_path = os.path.join(tmp, "pack.json") with open(pack_path, "w") as fh: fh.write(subprocess.run([sys.executable, PANEL] + cflag + ["evidence", "--seat", seat, "--n", "6"], capture_output=True, text=True, check=True).stdout) pack = json.load(open(pack_path)) k = [e["quote_key"] for e in pack["evidence"]] outside = next(q["quote_key"] for q in json.load(open(CORPUS))["quotes"] if q["quote_key"] not in k)

good = {
    "seat": seat,
    "question": "Does the pricing page answer the cost objection?",
    "corpus_version": pack["corpus_version"],
    "retrieval": pack["retrieval"],
    "verdict": {"answer": "Names the price, never addresses the switching-cost fear.",
                "confidence": "supported", "citations": [k[0], k[1]]},
    "strongest_objection": {"text": "Any new tool reads as extra work for a stretched team.", "citations": [k[2]]},
    "stopped_reading_at": "The pricing tier table.",
}

cases = [("valid verdict", good, 0)]

def mut(name, fn, expected):
    v = copy.deepcopy(good)
    fn(v)
    cases.append((name, v, expected))

def reband(v, **kw):
    v["retrieval"] = dict(v["retrieval"], **kw)

mut("fabricated citation", lambda v: v["verdict"]["citations"].append("calls:does-not-exist#42"), 1)
mut("real key outside the pack", lambda v: v["verdict"]["citations"].append(outside), 1)
mut("missing forced dissent", lambda v: v.pop("strongest_objection"), 1)
mut("dissent with no citations", lambda v: v["strongest_objection"].update({"citations": []}), 1)
mut("insufficient density, answered anyway", lambda v: reband(v, density_band="insufficient", n=3), 1)
mut("thin density claiming 'supported'", lambda v: reband(v, density_band="thin", n=7), 1)
mut("thin density, correctly hedged",
    lambda v: (reband(v, density_band="thin", n=7), v["verdict"].__setitem__("confidence", "weak signal")), 0)
mut("insufficient density, correctly refuses",
    lambda v: (reband(v, density_band="insufficient", n=3), v.__setitem__("verdict", None)), 0)
mut("stale corpus version", lambda v: v.__setitem__("corpus_version", {"generated": "2026-01-01", "total": 900}), 1)
mut("unknown seat", lambda v: v.__setitem__("seat", "the_persona"), 1)

fails = 0
for name, v, expected in cases:
    p = os.path.join(tmp, "t.json")
    json.dump(v, open(p, "w"))
    r = subprocess.run([sys.executable, VALIDATE, p, "--evidence", pack_path] + cflag, capture_output=True, text=True)
    ok = r.returncode == expected
    fails += 0 if ok else 1
    print(("  OK  " if ok else " FAIL ") + "%-40s exit=%s (want %s)" % (name, r.returncode, expected))

if fails:
    print("\n%d TEST(S) FAILED" % fails)
    sys.exit(1)
print("\nALL %d TESTS PASS" % len(cases))

if name == "main": main()

#!/usr/bin/env python3 """ validate_verdict.py: the hard gate.

A prompt asking the model to cite its sources is a suggestion. This is a check. Every quote_key in a verdict must resolve to a real row in data/corpus.json, and (when an evidence pack is supplied) must have actually been retrieved for the question. Density and verdict strength must agree. Forced dissent must be present.

Exits 0 on pass, 1 on failure, printing every violation.

./validate_verdict.py verdict.json ./validate_verdict.py verdict.json --evidence pack.json

WHAT THIS DOES NOT DO: it proves the cited evidence exists and was retrieved. It cannot prove the inference drawn from that evidence is sound: a verdict can cite four real quotes and reason wrongly from them and still pass. Detecting that is the backtest's job, not this gate's. """ import argparse, json, os, sys

HERE = os.path.dirname(os.path.abspath(file)) SEATS = CORPUS_PATH = VALID_SEATS = None

def configure(cfg_path=None): """Bind the gate to one panel's seats + corpus.""" global SEATS, CORPUS_PATH, VALID_SEATS cfg_path = os.path.abspath(cfg_path or os.environ.get("PANEL_CONFIG") or os.path.join(HERE, "seats.json")) SEATS = json.load(open(cfg_path)) CORPUS_PATH = os.path.normpath(os.path.join(os.path.dirname(cfg_path), "..", SEATS["corpus"])) VALID_SEATS = set(SEATS["seats"])

configure() BANDS = {"sufficient", "thin", "insufficient"}

def corpus_keys(): c = json.load(open(CORPUS_PATH)) return {q["quote_key"] for q in c["quotes"]}, c

def check(v, evidence=None): errs = [] keys, corpus = corpus_keys()

def need(obj, field, where, allow_empty=False):
    """allow_empty: the key must be PRESENT but may be empty: an unfiltered
    query legitimately has requested_filter == {}."""
    if field not in obj:
        errs.append("missing required field: %s.%s" % (where, field))
        return False
    if not allow_empty and obj.get(field) in (None, "", [], {}):
        errs.append("empty required field: %s.%s" % (where, field))
        return False
    return True

# --- structure -----------------------------------------------------------
for f in ("seat", "question", "corpus_version", "retrieval", "strongest_objection"):
    need(v, f, "verdict")
if "verdict" not in v:
    errs.append("missing required field: verdict.verdict (use null for no-evidence)")

if v.get("seat") not in VALID_SEATS:
    errs.append("unknown seat %r (expected one of %s)" % (v.get("seat"), ", ".join(sorted(VALID_SEATS))))

cv = v.get("corpus_version") or {}
if cv.get("total") != corpus.get("total") or cv.get("generated") != corpus.get("generated"):
    errs.append("corpus_version %s does not match the corpus on disk %s: the verdict was "
                "produced against a different corpus state"
                % (json.dumps(cv), json.dumps({"generated": corpus.get("generated"), "total": corpus.get("total")})))

r = v.get("retrieval") or {}
need(r, "requested_filter", "retrieval", allow_empty=True)
need(r, "precision_level", "retrieval")
need(r, "n", "retrieval", allow_empty=True)   # n == 0 is meaningful
need(r, "density_band", "retrieval")
band = r.get("density_band")
if band not in BANDS:
    errs.append("retrieval.density_band %r must be one of %s" % (band, sorted(BANDS)))

# --- density must agree with verdict strength ----------------------------
verdict = v.get("verdict")
if band == "insufficient" and verdict is not None:
    errs.append("density_band is 'insufficient' (n=%s) but a verdict was returned: "
                "this seat has no standing to answer" % r.get("n"))
if band == "thin" and isinstance(verdict, dict) and verdict.get("confidence") != "weak signal":
    errs.append("density_band is 'thin' (n=%s) so verdict.confidence must be 'weak signal', got %r"
                % (r.get("n"), verdict.get("confidence")))
if isinstance(verdict, dict):
    for f in ("answer", "confidence", "citations"):
        need(verdict, f, "verdict.verdict")

# --- forced dissent ------------------------------------------------------
so = v.get("strongest_objection") or {}
if not so.get("text"):
    errs.append("strongest_objection.text is required on EVERY verdict, including favourable "
                "ones: if the evidence contains no objection, say so and cite the nearest friction")
if not so.get("citations"):
    errs.append("strongest_objection.citations must contain at least one quote_key")

# --- scope ---------------------------------------------------------------
seat_cfg = SEATS["seats"].get(v.get("seat"))
if seat_cfg and verdict is not None and "unresolved_questions" not in v and seat_cfg.get("_scope_warning"):
    pass  # advisory only

# --- citations resolve ---------------------------------------------------
cited = []
if isinstance(verdict, dict):
    cited += list(verdict.get("citations") or [])
cited += list(so.get("citations") or [])
for uq in v.get("unresolved_questions") or []:
    if not uq.get("citations"):
        errs.append("unresolved_questions entry %r carries no citations" % uq.get("question", "")[:50])
    cited += list(uq.get("citations") or [])

unknown = [k for k in cited if k not in keys]
for k in sorted(set(unknown)):
    errs.append("FABRICATED CITATION: quote_key %r does not exist in the corpus" % k)

if evidence is not None:
    retrieved = {e["quote_key"] for e in evidence.get("evidence", [])}
    outside = [k for k in cited if k in keys and k not in retrieved]
    for k in sorted(set(outside)):
        errs.append("citation %r is a real quote but was NOT in the retrieved evidence pack "
                    "- it was sourced from outside the seat's slice" % k)

return errs, len(set(cited))

def main(): ap = argparse.ArgumentParser(description=doc, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("verdict") ap.add_argument("--evidence", help="the evidence pack the verdict was built from") ap.add_argument("--config", help="seats.json of the panel (default: seats.json beside this script)") a = ap.parse_args() if a.config: configure(a.config)

v = json.load(open(a.verdict))
ev = json.load(open(a.evidence)) if a.evidence else None
errs, n_cited = check(v, ev)

if errs:
    print("REJECTED: %d violation(s):\n" % len(errs))
    for e in errs:
        print("  x %s" % e)
    sys.exit(1)
print("PASS: %d citations, all resolve%s" % (n_cited, " and all within the evidence pack" if ev else ""))

if name == "main": main()

{ "$schema": "http://json-schema.org/draft-07/schema#", "title": "ICP Panel Verdict", "_doc": "Every panel answer must serialize to this shape and pass references/validate_verdict.py. The citation requirement is enforced mechanically: a verdict whose quote_keys do not resolve to real rows in data/corpus.json is rejected by the validator, not merely discouraged by a prompt. NOTE: this gate proves the evidence EXISTS and was RETRIEVED. It does not prove the inference drawn from it is sound. That is what the backtest is for.", "type": "object", "required": [ "seat", "question", "corpus_version", "retrieval", "verdict", "strongest_objection" ], "additionalProperties": false, "properties": { "seat": { "type": "string", "_doc": "Must be a seat key from references/seats.json; validate_verdict.py checks it." }, "question": { "type": "string", "minLength": 1 }, "corpus_version": { "type": "object", "required": [ "generated", "total" ], "properties": { "generated": { "type": "string" }, "total": { "type": "integer" } }, "_doc": "Copied verbatim from data/corpus.json. Pins the answer to a corpus state so a changed verdict can be traced to a changed corpus." }, "retrieval": { "type": "object", "required": [ "requested_filter", "precision_level", "n", "density_band" ], "properties": { "requested_filter": { "type": "object" }, "precision_level": { "type": "string", "_doc": "Which dimensions survived widening, e.g. 'persona+stage' or 'stage only (persona dropped, n=3)'." }, "dropped": { "type": "array", "items": { "type": "string" } }, "n": { "type": "integer", "minimum": 0 }, "density_band": { "enum": [ "sufficient", "thin", "insufficient" ] }, "source_mix": { "type": "object", "_doc": "Counts by source_type in the evidence pack. Guards against current-client language quietly dominating a pre-sale seat as client calls accumulate." }, "age_mix": { "type": "object", "_doc": "Dated vs undated evidence, oldest/newest, median age, share within 12 months, and a STALE flag above a 540-day median. Surfaced, never used to silently discard." } } }, "verdict": { "oneOf": [ { "type": "null", "_doc": "REQUIRED to be null when density_band is 'insufficient'. The seat has no standing to answer." }, { "type": "object", "required": [ "answer", "confidence", "citations" ], "properties": { "answer": { "type": "string", "minLength": 1 }, "confidence": { "enum": [ "weak signal", "supported" ] }, "citations": { "type": "array", "minItems": 1, "items": { "type": "string", "pattern": "^[a-z-]+:[^#]+(#\d+)?$" }, "_doc": "quote_keys. Must resolve against data/corpus.json AND must have been present in the retrieved evidence pack." } } } ] }, "strongest_objection": { "type": "object", "required": [ "text", "citations" ], "_doc": "FORCED DISSENT. Mandatory on every verdict, including favourable ones. Asking an LLM 'does this resonate?' has a polite answer; this field has no polite answer. If the evidence genuinely contains no objection, say so explicitly in text and cite the nearest friction quote. Do not leave it empty.", "properties": { "text": { "type": "string", "minLength": 1 }, "citations": { "type": "array", "minItems": 1, "items": { "type": "string" } } } }, "stopped_reading_at": { "type": [ "string", "null" ], "_doc": "Required for any judgment on a written asset (article, email, page). The exact line or moment this seat would disengage, or null with a stated reason if it would read to the end." }, "unresolved_questions": { "type": "array", "_doc": "The navigation use case, scoped honestly. Informational needs this seat carries INTO the page, ranked by evidence density, never a simulated click path. Each item carries its own citations.", "items": { "type": "object", "required": [ "question", "citations" ], "properties": { "question": { "type": "string" }, "citations": { "type": "array", "minItems": 1, "items": { "type": "string" } }, "n_supporting": { "type": "integer" } } } } } }