Deep live pipeline: reference-depth reports generated LIVE (fact-store + per-report fan-out)

.gitignore

@@ -33,3 +33,8 @@ v1/htmlcov/
 .specify/
 specs/
 CLAUDE.md
+
+# ── framework spike (LangGraph evaluation; NOT product code) ──────
+# Recorded the plain-code-vs-LangGraph decision in specs/003; the spike files exercised the
+# comparison and are kept local only — they are not part of the shipped engine.
+v1/spike/

v1/README.md

@@ -64,13 +64,21 @@ uv run python scripts/doctor.py          # verify provider connectivity
 ```bash
 uv run python run.py --domain o2c                  # live run (generates everything live)
 uv run python run.py --domain o2c --auto-resolve   # non-interactive
-uv run python run.py --domain o2c --use-fixture    # use the pre-built O2C fixture (demo-safe)
+uv run python run.py --domain o2c --use-fixture    # pre-built O2C fixture — the full reference-depth suite
 uv run python run.py --domain o2c --golden         # replay a saved run offline
 uv run python run.py --domain o2c --refresh        # diff against the previous run (new/resolved/changed)
 uv run python run.py --domain o2c --no-verify      # skip the adversarial verification pass
 uv run python run.py --domain p2p --auto-resolve   # any other domain — generated live
 ```
 
+> **Reference-depth O2C suite.** `--use-fixture` renders the hand-grounded O2C fixture — the full
+> reference-grade suite (per-report cover + own TOC, the channel-mix / lead-time / credit-band /
+> collections / EDI-connection / top-account tables, the five pain-point detail tables, the evidence
+> register, success-metrics, risk register and traceability matrix). Every figure traces to the raw
+> CSVs / source documents and passes the grounding gate. The live path emits the core suite today;
+> mining the corpus into these structured sections from the live agent is a planned follow-up, so
+> richer depth flows automatically for any domain.
+
 Findings are variable in number and ranked by impact; each is adversarially **verified** (a
 challenged finding is flagged for review, not dropped); the agent can **conformance-check** a
 documented rule against the data; every report number **links to its source**; and `--refresh`

v1/discovery/agent_loop.py

@@ -1,7 +1,9 @@
 """The agent tool-use loop — the agent genuinely discovers the findings.
 
-The agent is given generic data/text tools and asked to investigate an O2C document set and
-emit exactly 3 findings. It is NOT told what the findings are. It calls describe/group_by/
+The agent is given generic data/text tools and asked to investigate a document set and emit the
+material findings the evidence supports (typically 3-5; the report DEPTH comes from the fact-store +
+per-report fan-out downstream, not from the finding count). It is NOT told what the findings are. It
+calls describe/group_by/
 join_diff/filter_count/aggregate/find_mentions, reasons over the results, and terminates by
 calling emit_findings. Every quantitative claim is grounded against a real tool result.
 
@@ -163,8 +165,10 @@ def run_discovery(llm: LLMClient, csv_ids: list[str], doc_ids: list[str],
     system = build_system_prompt(csv_ids, doc_ids, narrative_text, domain_label)
     schemas = tools.schemas() + [EMIT_TOOL]
     messages: list[dict] = [{"role": "user", "content":
-                             f"Investigate this {domain_label} landscape and emit exactly 3 "
-                             "findings. Begin by orienting on each data file per the protocol."}]
+                             f"Investigate this {domain_label} landscape and emit the material, "
+                             "evidence-backed findings (typically 3-5), each a distinct cross-source "
+                             "issue ranked by business impact. Begin by orienting on each data file "
+                             "per the protocol, then emit_findings once."}]
     seen, findings = set(), None
     for _ in range(MAX_TURNS):
         turn = llm.messages_with_tools(system=system, messages=messages, tools=schemas, model=model)
@@ -173,7 +177,10 @@ def run_discovery(llm: LLMClient, csv_ids: list[str], doc_ids: list[str],
         if on_activity:
             for tu in tool_uses:
                 on_activity(narrate(tu))
-        if turn.stop_reason != "tool_use" or not tool_uses:
+        # The API requires a tool_result for EVERY tool_use, regardless of stop_reason — a turn that
+        # ends with tool_use blocks (even on a non-"tool_use" stop, e.g. max_tokens mid-call) must
+        # still be answered with results, or the next request 400s on an unpaired tool_use.
+        if not tool_uses:
             messages.append({"role": "user", "content":
                              "You must finish by calling emit_findings exactly once. Do that now."})
             continue

v1/discovery/factstore.py

@@ -0,0 +1,231 @@
+"""Build a grounded FactStore (the KG-lite) from a discovery run + the registered sources.
+
+This is the Block-1+2 output the per-report synthesis fan-out expands from — it replaces the flat
+"~3 findings" waist with a structured, sourced collection of measured numbers, verbatim document
+quotes, typed entities (accounts / systems / connections with field-level attributes), and
+relations. Everything carries its source(s) + confidence tier.
+
+GENERIC by construction: facts are derived from whatever findings + CSV columns + narrative text a
+domain provides. No domain constants — a thinner domain simply yields fewer facts. Deterministic:
+ordering is stable and row harvesting is capped + sorted, so a golden replay is byte-stable.
+"""
+from __future__ import annotations
+
+import re
+
+from . import docnames, tools
+from .models import DocQuote, EntityFact, FactStore, QuantFact, Relation, StrategyProfile
+
+# Raw tool-output field names / engine jargon a model may copy verbatim into a quote. These are
+# internal to the tool layer and must never reach a client-facing brief or report. We strip the
+# token (keeping the human numbers/words around it); a quote that is *only* such jargon is dropped.
+_TOOL_JARGON = re.compile(
+    r"\b(?:n_mismatch|sum_delta|from_tool|group_by|join_diff|filter_count|find_mentions)\b"
+    r"\s*[:=]?\s*",
+    re.I,
+)
+# residue left after a key is stripped: braces/quotes, then orphaned separators (": :" / ", :" /
+# leading-or-trailing ";"). Applied repeatedly so chained residue ("{ : : 267") fully collapses.
+_QUOTE_BRACES = re.compile(r"[{}\"]")
+_QUOTE_RESIDUE = re.compile(r"\s*[:,]\s*(?=[:,])|^\s*[:,;]\s*|\s*[:,;]\s*$")
+
+# how many entity rows to harvest per CSV (deterministic cap — the reports surface the top accounts,
+# not every row; full data lives in the source pages / provenance).
+_ENTITY_CAP = 12
+# relation keywords a finding/handoff may carry (generic, not domain-specific)
+_REL_KINDS = ("handoff", "conflict", "owns", "runs on", "triggers", "depends")
+
+
+def build_fact_store(raw_payload: dict, reg: dict) -> FactStore:
+    """Assemble the grounded fact-store from the run's findings + the registered sources."""
+    fs = FactStore()
+    _harvest_quants_and_quotes(raw_payload, fs)
+    _harvest_entities(reg, fs)
+    _harvest_relations(raw_payload, fs)
+    return fs
+
+
+# ── measured numbers + verbatim quotes (from the findings the tools already grounded) ───────────
+def _harvest_quants_and_quotes(raw_payload: dict, fs: FactStore) -> None:
+    seen_q: set[tuple] = set()
+    seen_quote: set[tuple] = set()
+    for f in raw_payload.get("findings", []):
+        tier = _tier(f)
+        srcs = sorted({docnames.stem(s.get("doc_id", "")) for s in f.get("sources", [])
+                       if s.get("doc_id")})
+        for cv in f.get("computed_values", []):
+            label, val = str(cv.get("label", "")).strip(), cv.get("value")
+            num = _num(val)
+            if num is None or not label:
+                continue
+            key = (label.lower(), round(num, 4))
+            if key in seen_q:
+                continue
+            seen_q.add(key)
+            fs.quant.append(QuantFact(label=label, value=num, unit=_unit_of(label),
+                                      sources=srcs, tier=tier))
+        for nv in f.get("narrative_values", []):
+            quote = _clean_quote(str(nv.get("quote", "")))
+            doc = docnames.stem(nv.get("doc_id", ""))
+            if not quote or not doc:
+                continue
+            key = (doc, quote[:60].lower())
+            if key in seen_quote:
+                continue
+            seen_quote.add(key)
+            fs.quotes.append(DocQuote(text=quote, doc_id=doc,
+                                      locator=str(nv.get("label", "")), tier=tier))
+        for s in f.get("sources", []):
+            quote = _clean_quote(str(s.get("quote", "")))
+            doc = docnames.stem(s.get("doc_id", ""))
+            if not quote or not doc:
+                continue
+            key = (doc, quote[:60].lower())
+            if key in seen_quote:
+                continue
+            seen_quote.add(key)
+            fs.quotes.append(DocQuote(text=quote, doc_id=doc,
+                                      locator=str(s.get("locator", "")), tier=tier))
+
+
+# ── typed entities (one per CSV row, generic name + attributes derived from the columns) ─────────
+def _harvest_entities(reg: dict, fs: FactStore) -> None:
+    for csv_id in sorted(reg.get("csv_ids", [])):
+        path = tools.FILE_REGISTRY.get(csv_id)
+        if path is None:
+            continue
+        try:
+            cols, rows = tools._read_rows(path)
+        except (OSError, ValueError):
+            continue
+        if not cols or not rows:
+            continue
+        kind = _entity_kind(csv_id)
+        name_col = _name_column(cols)
+        # keep a deterministic, capped slice (rows are already in file order)
+        for r in rows[:_ENTITY_CAP]:
+            name = str(r.get(name_col, "")).strip() if name_col else ""
+            if not name:
+                continue
+            attrs = {c: str(r.get(c, "")).strip() for c in cols
+                     if c != name_col and str(r.get(c, "")).strip()}
+            fs.entities.append(EntityFact(kind=kind, name=name, attributes=attrs,
+                                          sources=[csv_id], tier="verified"))
+
+
+# ── relations (from handoff/conflict-flavoured findings) ─────────────────────────────────────────
+def _harvest_relations(raw_payload: dict, fs: FactStore) -> None:
+    seen: set[tuple] = set()
+    for f in raw_payload.get("findings", []):
+        blob = (str(f.get("title", "")) + " " + str(f.get("description", ""))).lower()
+        kind = next((k.replace(" ", "_") for k in _REL_KINDS if k in blob), "")
+        if not kind:
+            continue
+        srcs = sorted({docnames.stem(s.get("doc_id", "")) for s in f.get("sources", [])
+                       if s.get("doc_id")})
+        rel = Relation(src=str(f.get("id", "")), kind=kind, dst=str(f.get("title", ""))[:60],
+                       sources=srcs)
+        key = (rel.src, rel.kind)
+        if key in seen:
+            continue
+        seen.add(key)
+        fs.relations.append(rel)
+
+
+# ── StrategyProfile from the domain manifest (neutral default when none declared) ────────────────
+def strategy_from_manifest(manifest: dict | None) -> StrategyProfile:
+    s = (manifest or {}).get("strategy_profile") or {}
+    return StrategyProfile(
+        direction_type=str(s.get("direction_type", "")),
+        horizon=str(s.get("horizon", "")),
+        strategic_constraints=str(s.get("strategic_constraints", "")),
+        stakeholder_priorities=list(s.get("stakeholder_priorities", []) or []),
+        out_of_scope=str(s.get("out_of_scope", "")),
+        success_definition=str(s.get("success_definition", "")))
+
+
+# ── helpers ──────────────────────────────────────────────────────────────────────────────────────
+def _clean_quote(quote: str) -> str:
+    """Strip raw tool-output field names / engine jargon a model may have copied into a quote, so
+    internal tokens never reach a client-facing brief. Returns "" when, after stripping, nothing of
+    substance remains (the quote was essentially a raw tool dump) — caller then drops it.
+
+    Deterministic and domain-agnostic: it removes only the fixed internal token set, leaving the
+    human numbers and words intact (e.g. "n_mismatch 267; sum_delta 30675000" → "267; 30675000",
+    which carries no meaning on its own and so is dropped; a real prose quote is left untouched)."""
+    q = quote.strip()
+    if not q or not _TOOL_JARGON.search(q):
+        return q
+    cleaned = _QUOTE_BRACES.sub("", _TOOL_JARGON.sub("", q))
+    prev = None
+    while prev != cleaned:               # collapse orphaned separators until stable
+        prev = cleaned
+        cleaned = _QUOTE_RESIDUE.sub("", cleaned)
+    cleaned = cleaned.strip(" ;,:")
+    # If what remains is only digits / punctuation / separators, the quote carried no prose meaning
+    # of its own — it was a raw tool-output echo. Drop it rather than surface a bare number string.
+    if not re.search(r"[A-Za-z]", cleaned):
+        return ""
+    return cleaned
+
+
+def _num(v) -> float | None:
+    try:
+        return float(v)
+    except (TypeError, ValueError):
+        return None
+
+
+def _tier(f: dict) -> str:
+    c = str(f.get("confidence", "")).lower()
+    if c in ("verified", "amber", "gap"):
+        return c
+    # an adversarially-challenged finding is at most amber
+    if f.get("verification", {}).get("supported") is False:
+        return "amber"
+    return "verified"
+
+
+def _unit_of(label: str) -> str:
+    lo = label.lower()
+    # a COUNT-of-things label wins over the field name it counts ("Accounts with mismatched
+    # credit_limit_eur" is a count of accounts, not a EUR amount).
+    if lo.startswith("account") or "accounts with" in lo or "number of account" in lo:
+        return "accounts"
+    if "pct" in lo or "percent" in lo or "%" in label or " rate" in lo or "share" in lo:
+        return "percent"
+    if "escalation" in lo or "incident" in lo or "case" in lo:
+        return "escalations"
+    if "eur" in lo or "€" in label or "overstatement" in lo or "limit" in lo or "value" in lo:
+        return "eur"
+    if "account" in lo:
+        return "accounts"
+    return "count"
+
+
+def _entity_kind(csv_id: str) -> str:
+    """Derive a generic entity kind from the filename (no domain constants). Check the more specific
+    signals (escalation/incident log, connection register) before the broad 'customer/master' one,
+    so a 'customer-service-escalation-log' is an incident, not an account."""
+    lo = csv_id.lower()
+    if "escalation" in lo or "incident" in lo or "ticket" in lo or "log" in lo:
+        return "incident"
+    if "connection" in lo or "integration" in lo or "edi" in lo:
+        return "connection"
+    if "order" in lo or "flow" in lo or "transaction" in lo:
+        return "transaction"
+    if "customer" in lo or "account" in lo or "master" in lo or "crm" in lo:
+        return "account"
+    return "record"
+
+
+def _name_column(cols: list[str]) -> str:
+    """Pick the most name-like column generically: prefer a 'name', else an 'id', else the first."""
+    lowered = [(c, c.lower()) for c in cols]
+    for c, lo in lowered:
+        if lo.endswith("name") or lo == "name" or "customer_name" in lo:
+            return c
+    for c, lo in lowered:
+        if lo.endswith("_id") or lo == "id":
+            return c
+    return cols[0] if cols else ""