Parallelize live synthesis fan-out (~3x faster), determinism preserved

.gitignore

@@ -17,6 +17,7 @@ v1/.venv/
 
 # ── v1 runtime cache (regenerated; noisy) ─────────────────────────
 v1/.cache/
+v1/.cache.tmp/
 
 # ── coverage artifacts (regenerated) ──────────────────────────────
 .coverage

explorer/.gitignore

@@ -1,5 +1,6 @@
 node_modules/
 dist/
+.vite/
 *.tsbuildinfo
 .DS_Store
 # synced from v1/out by `npm run sync:data` — regenerated, not source

v1/discovery/fanout.py

@@ -19,9 +19,13 @@
 """
 from __future__ import annotations
 
+import os
 import re
+from concurrent.futures import ThreadPoolExecutor
+from typing import Any
 
 from .agent_loop import GroundingError, _close
+from .llm import LLMRetryableError
 from .models import FactStore, PlanningAssumption, StrategyProfile
 from .synthesis import _STRUCTURAL, _SOURCED_TABLE_KEYS, assert_factual
 
@@ -179,6 +183,48 @@ def _facts_brief(fs: FactStore) -> str:
 
 
 # ── orchestrator: build the fact-store, fan out per report (+ per opportunity), assemble ────────
+def _max_workers() -> int:
+    """Concurrency for the synthesis fan-out. Default 4 (≈24K output tokens in flight at 6K/section
+    — safe under a standard Opus OTPM tier even with the attempts=2 grounding-retry burst). Set
+    DISCOVERY_MAX_WORKERS=1 to revert to fully serial (the regression baseline)."""
+    try:
+        return max(1, int(os.environ.get("DISCOVERY_MAX_WORKERS", "4")))
+    except ValueError:
+        return 4
+
+
+# A section worker runs on a pool thread and is fully self-contained: it does its own slice + the
+# exact per-section kwargs INSIDE the guard, so a malformed spec or a transient provider blip omits
+# just that one section (returns None) instead of propagating to the main thread and aborting the
+# suite. Note the catch is narrow: LLMRetryableError (transient) is omitted, but a plain LLMError
+# (auth/config) is NOT caught here — it propagates so a misconfigured run fails loudly, never
+# silently produces an empty report.
+_SECTION_OMIT = (AttributeError, TypeError, KeyError, ValueError, LLMRetryableError)
+
+
+def _safe_report(llm, key, spec, fact_store, allow, strategy, doc_keys, model):
+    try:
+        fs = fact_store.slice_for(*spec["slice"]) if spec.get("slice") else fact_store
+        return synth_section(
+            llm, tool_name=spec["tool"], schema=spec["schema"], fact_store=fs, allow=allow,
+            strategy=strategy if key in _STRATEGIC else None,
+            instruction=spec["instruction"], doc_keys=doc_keys,
+            factual=key in _FACTUAL, max_tokens=spec.get("max_tokens", 6000), model=model)
+    except _SECTION_OMIT:
+        return None
+
+
+def _safe_opp(llm, seed, spec, fact_store, allow, doc_keys, model):
+    try:
+        return synth_section(
+            llm, tool_name="emit_opportunity", schema=spec.get("opp_schema", _MIN_OPP_SCHEMA),
+            fact_store=fact_store.slice_for(*([seed["topic"]] if seed.get("topic") else [])),
+            allow=allow, strategy=None, instruction=_opp_instruction(seed), doc_keys=doc_keys,
+            max_tokens=spec.get("opp_max_tokens", 6000), model=model)
+    except _SECTION_OMIT:
+        return None
+
+
 # Each report owns a slice of SynthesisContent fields; the orchestrator merges the per-report emits.
 # (Phase 1 wires the control flow + gate + planning channel; Phase 2 fills the per-report schemas to
 # reference depth.) The seed names a small number of opportunities to expand individually for r04.
@@ -201,43 +247,48 @@ def run_synthesis_fanout(llm, fact_store: FactStore, strategy: StrategyProfile,
     if allow is None:
         allow = fact_store.numbers_allow()
 
-    for key in REPORT_KEYS:
-        spec = report_specs.get(key)
-        if not spec:
-            continue
-        fs = fact_store.slice_for(*spec.get("slice", [])) if spec.get("slice") else fact_store
-        # a malformed/oddly-shaped emit from ONE report must omit just that report, never abort the
-        # suite (the live model can return an unexpected shape; resilience over all-or-nothing).
-        try:
-            section = synth_section(
-                llm, tool_name=spec["tool"], schema=spec["schema"], fact_store=fs, allow=allow,
-                strategy=strategy if key in _STRATEGIC else None,
-                instruction=spec["instruction"], doc_keys=doc_keys,
-                factual=key in _FACTUAL, max_tokens=spec.get("max_tokens", 6000), model=model)
-            planning += collect_planning(section)
-            _merge(merged, section)
-        except (AttributeError, TypeError, KeyError, ValueError):
-            continue
+    # Fan the independent sections out across a small thread pool (the calls are I/O-bound LLM
+    # round-trips, so threads give real wall-clock parallelism). Reports and per-opportunity
+    # generations are mutually independent here — the only ordering dependency (pain-points pre-pass
+    # → opportunity seeds) is enforced by the CALLER (fanout_specs.run_report_fanout), not here.
+    #
+    # Determinism is preserved by consuming futures in SUBMISSION ORDER (never as_completed):
+    # _merge's first-write-wins / list-extend semantics and the planning-list order then reproduce
+    # exactly what the old sequential loops produced, independent of thread scheduling. Each call's
+    # cache key is content-derived and order-independent, so --golden replay is byte-identical too.
+    report_jobs = [key for key in REPORT_KEYS if report_specs.get(key)]
+    ospec = report_specs.get("04-opportunity-portfolio", {})
+    omitted: list[str] = []
+    futures: list[tuple[str, str, bool, Any]] = []   # (kind, label, track_omission, future)
+    with ThreadPoolExecutor(max_workers=_max_workers()) as pool:
+        for key in report_jobs:                                   # submit reports first…
+            # A report with no instruction is a structural placeholder (the 04 portfolio's content
+            # comes from the per-opportunity seeds, not its report spec) — its None is EXPECTED, so
+            # don't track it as a missing deliverable (avoids a false-positive omission warning).
+            track = bool((report_specs[key].get("instruction") or "").strip())
+            futures.append(("report", key, track, pool.submit(
+                _safe_report, llm, key, report_specs[key], fact_store, allow, strategy, doc_keys, model)))
+        for i, seed in enumerate(opp_seeds):                      # …then opps → one combined wave
+            label = str(seed.get("id") or seed.get("title") or f"opp-{i}")
+            futures.append(("opp", label, True, pool.submit(
+                _safe_opp, llm, seed, ospec, fact_store, allow, doc_keys, model)))
 
-    # per-opportunity deep generation for the centrepiece portfolio (report 04)
     opps = []
-    for seed in opp_seeds:
-        spec = report_specs.get("04-opportunity-portfolio", {})
-        try:
-            opp = synth_section(
-                llm, tool_name="emit_opportunity", schema=spec.get("opp_schema", _MIN_OPP_SCHEMA),
-                fact_store=fact_store.slice_for(*([seed.get("topic")] if seed.get("topic") else [])),
-                allow=allow, strategy=None, instruction=_opp_instruction(seed), doc_keys=doc_keys,
-                max_tokens=spec.get("opp_max_tokens", 6000), model=model)
-        except (AttributeError, TypeError, KeyError, ValueError):
-            opp = None
-        if opp is not None:
-            planning += collect_planning(opp)
-            opps.append(opp)
+    for kind, label, track, fut in futures:        # SUBMISSION ORDER — deterministic merge/planning
+        section = fut.result()                     # never raises: the worker swallowed _SECTION_OMIT
+        if section is None:
+            if track:                              # genuine missing content (placeholders excluded)
+                omitted.append(label)
+            continue
+        planning += collect_planning(section)      # single-threaded, on the main thread
+        if kind == "report":
+            _merge(merged, section)                # single-threaded → deterministic ownership order
+        else:
+            opps.append(section)
     if opps:
         merged.setdefault("opportunities", [])
         merged["opportunities"] = opps + merged.get("opportunities", [])
-    return merged, planning
+    return merged, planning, omitted
 
 
 def _merge(into: dict, section: dict | None) -> None:

v1/discovery/fanout_specs.py

@@ -288,8 +288,10 @@ def opp_seeds_from_pain_points(payload: dict) -> list[dict]:
 def run_report_fanout(llm, raw_payload: dict, reg: dict, strategy: StrategyProfile | None = None,
                       doc_keys=None, model=None):
     """Top-level live deep synthesis: build the grounded fact-store, fan out per report, expand one
-    opportunity per pain point, and return (merged_payload, planning, fact_store, strategy). The
-    caller maps merged_payload via build._from_payload and attaches fact_store/strategy/planning."""
+    opportunity per pain point, and return (merged_payload, planning, fact_store, strategy, omitted).
+    `omitted` lists the section labels that could not be produced (transient/grounding failures) so
+    the caller can warn or, on --save-golden, refuse a short deliverable. The caller maps
+    merged_payload via build._from_payload and attaches fact_store/strategy/planning."""
     from .synthesis import allowed_numbers
     fs = factstore.build_fact_store(raw_payload, reg)
     strat = strategy or factstore.strategy_from_manifest(reg.get("manifest"))
@@ -298,14 +300,17 @@ def run_report_fanout(llm, raw_payload: dict, reg: dict, strategy: StrategyProfi
     # the AUTHORITATIVE grounding allow-list for this run (tool numbers + finding values + derived
     # ratios) — same source the monolith gate uses; the fact-store slice only shapes the prompt.
     allow = allowed_numbers(raw_payload)
-    # first pass for the pain points (report 02) so we can seed one opportunity per pain point
-    pp_only, _ = run_synthesis_fanout(llm, fs, strat, dk, allow=allow,
-                                      report_specs={"02-pain-points": specs["02-pain-points"]})
+    # first pass for the pain points (report 02) so we can seed one opportunity per pain point.
+    # This MUST precede seed derivation — it is the one hard barrier in the fan-out (the full pass
+    # below parallelises everything after it). The pre-pass's own omissions don't matter (02 is
+    # re-run in the full pass and folded with first-write-wins below), so we ignore them here.
+    pp_only, _, _ = run_synthesis_fanout(llm, fs, strat, dk, allow=allow,
+                                         report_specs={"02-pain-points": specs["02-pain-points"]})
     seeds = opp_seeds_from_pain_points(pp_only)
-    merged, planning = run_synthesis_fanout(llm, fs, strat, dk, allow=allow, report_specs=specs,
-                                            opp_seeds=seeds)
+    merged, planning, omitted = run_synthesis_fanout(
+        llm, fs, strat, dk, allow=allow, report_specs=specs, opp_seeds=seeds)
     # fold the first-pass pain points in (report_specs ran them again in the full pass too; merge
     # keeps the first, so they are consistent)
     for k, v in pp_only.items():
         merged.setdefault(k, v)
-    return merged, planning, fs, strat
+    return merged, planning, fs, strat, omitted

v1/discovery/llm.py

@@ -20,17 +20,28 @@
 import hashlib
 import json
 import os
+import threading
 from pathlib import Path
 from typing import Any
 
 DEFAULT_ANTHROPIC_MODEL = "claude-opus-4-8"
 CACHE_DIR = Path(__file__).resolve().parent.parent / ".cache"
+# Scratch dir for atomic cache writes — a SIBLING of .cache/ (same filesystem so os.replace is
+# atomic), and deliberately OUTSIDE .cache/ so a crash-orphaned temp file can never be captured by
+# --save-golden's copytree of .cache/. See _write_cache.
+_CACHE_SCRATCH = CACHE_DIR.parent / ".cache.tmp"
 
 
 class LLMError(RuntimeError):
     pass
 
 
+class LLMRetryableError(LLMError):
+    """A transient provider failure (rate limit / overload / connection) that one section may
+    safely retry-or-omit without it being a hard, run-aborting error. Auth/config failures stay a
+    plain LLMError so they still surface loudly instead of silently omitting a report section."""
+
+
 class LLMClient:
     def __init__(self, *, cache_dir: Path | None = None, offline: bool | None = None) -> None:
         self.cache_dir = cache_dir or CACHE_DIR
@@ -46,6 +57,9 @@ def __init__(self, *, cache_dir: Path | None = None, offline: bool | None = None
         # offline — a fresh run by definition needs the network.
         self.no_cache = os.environ.get("DISCOVERY_NO_CACHE", "0") == "1" and not self.offline
         self._client = None  # lazily created
+        # Guards the lazy self._client init when sections run on multiple threads (the synthesis
+        # fan-out). The constructed SDK client is itself thread-safe; only its construction races.
+        self._client_lock = threading.Lock()
 
     # ---- caching -----------------------------------------------------------
     def _cache_key(self, system: str, prompt: str, model: str) -> str:
@@ -69,13 +83,22 @@ def _read_cache(self, key: str) -> str | None:
         return None
 
     def _write_cache(self, key: str, system: str, prompt: str, response: str) -> None:
-        self._cache_path(key).write_text(
+        # Atomic write: a plain write_text truncates-then-writes, so two threads writing the SAME
+        # key (the grounding-retry loop re-issues identical turns) could tear the file → a corrupt
+        # JSON or a torn golden artifact. Write to a unique temp in a sibling scratch dir on the
+        # same filesystem, then os.replace (atomic same-FS rename) into .cache/. The scratch dir is
+        # OUTSIDE .cache/ so a crash-orphaned temp is never copied into a --save-golden snapshot.
+        _CACHE_SCRATCH.mkdir(parents=True, exist_ok=True)
+        tmp = _CACHE_SCRATCH / f"{key}.{os.getpid()}.{threading.get_ident()}.tmp"
+        tmp.write_text(
             json.dumps(
                 {"system": system, "prompt": prompt, "response": response},
                 indent=2,
                 ensure_ascii=False,
-            )
+            ),
+            encoding="utf-8",
         )
+        os.replace(tmp, self._cache_path(key))
 
     # ---- public API --------------------------------------------------------
     def complete(self, system: str, prompt: str, *, model: str | None = None,
@@ -145,36 +168,59 @@ def _temp_kwargs(model: str) -> dict:
         deprecated = ("claude-opus-4-8",)
         return {} if any(model.startswith(d) for d in deprecated) else {"temperature": 0}
 
+    # SDK exception classes (and signals) that mean "transient — safe to retry or omit one section"
+    # rather than "the run is misconfigured". Kept as names so we don't hard-import the SDK here.
+    _RETRYABLE_NAMES = frozenset({
+        "RateLimitError", "APIConnectionError", "APITimeoutError",
+        "InternalServerError", "APIStatusError", "ServiceUnavailableError", "OverloadedError",
+    })
+
     @staticmethod
     def _provider_error(provider: str, e: Exception) -> LLMError:
-        """Turn any provider-SDK failure into one clean, actionable LLMError (no traceback leak).
+        """Turn any provider-SDK failure into one clean, actionable error (no traceback leak).
 
-        The most common first-run failure is no/invalid credentials: the Anthropic SDK raises a
-        TypeError ("Could not resolve authentication method") when no key is set, or an
-        AuthenticationError on a bad key. Either way, point the operator at the real fix instead of
-        surfacing a stack trace or the misleading 'offline / use golden' message."""
+        Returns an LLMRetryableError for TRANSIENT failures (rate limit / overload / connection) —
+        the synthesis fan-out may omit just that one section for these. Returns a plain LLMError for
+        everything else (auth/config), which stays a HARD, run-aborting error: the most common
+        first-run failure is no/invalid credentials, and that must surface loudly, never be silently
+        swallowed into an empty report."""
         name = type(e).__name__
         msg = str(e) or name
-        auth = "authentication" in msg.lower() or "api_key" in msg.lower() or name in (
+        low = msg.lower()
+        auth = "authentication" in low or "api_key" in low or name in (
             "AuthenticationError", "PermissionDeniedError")
         if auth:
             keyvar = "AZURE_OPENAI_API_KEY" if provider == "azure" else "ANTHROPIC_API_KEY"
             return LLMError(
                 f"{provider} credentials rejected or missing ({name}). Check {keyvar} in v1/.env "
                 "(verify with: uv run python scripts/doctor.py), or run with --golden for the "
                 "offline demo.")
+        # transient? — by SDK class name, or by a 429/529/overloaded signal in the message
+        transient = (name in LLMClient._RETRYABLE_NAMES
+                     or "429" in msg or "529" in msg or "overloaded" in low or "rate limit" in low)
+        if transient:
+            return LLMRetryableError(f"{provider} transient failure ({name}): {msg}")
         return LLMError(f"{provider} call failed ({name}): {msg}")
 
+    def _anthropic_client(self):
+        """Lazily build the shared Anthropic client under a lock (the fan-out calls this from many
+        threads). max_retries=2 lets the SDK absorb a transient 429/529 before we degrade a section;
+        capped low so a rate-limit storm can't wedge every pool slot in backoff for minutes."""
+        if self._client is None:
+            with self._client_lock:
+                if self._client is None:           # double-checked: only construct once
+                    import anthropic
+                    self._client = anthropic.Anthropic(max_retries=2)
+        return self._client
+
     def _call_anthropic_tools(self, system, messages, tools, model, max_tokens):
         try:
-            import anthropic
+            client = self._anthropic_client()
         except ImportError as e:  # pragma: no cover
             raise LLMError("pip install anthropic") from e
-        if self._client is None:
-            self._client = anthropic.Anthropic()
         try:
             # we never stream, so .create() returns a Message (not a Stream); narrow for the checker
-            msg: Any = self._client.messages.create(
+            msg: Any = client.messages.create(
                 model=model, max_tokens=max_tokens, **self._temp_kwargs(model),
                 system=system, tools=tools, tool_choice={"type": "auto"},
                 messages=messages,
@@ -204,13 +250,11 @@ def _call_provider(self, system: str, prompt: str, model: str, max_tokens: int)
 
     def _call_anthropic(self, system: str, prompt: str, model: str, max_tokens: int) -> str:
         try:
-            import anthropic
+            client = self._anthropic_client()  # locked lazy init; reads ANTHROPIC_API_KEY
         except ImportError as e:  # pragma: no cover
             raise LLMError("pip install anthropic") from e
-        if self._client is None:
-            self._client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY
         try:
-            msg: Any = self._client.messages.create(   # non-streaming -> Message; narrow for the checker
+            msg: Any = client.messages.create(   # non-streaming -> Message; narrow for the checker
                 model=model,
                 max_tokens=max_tokens,
                 **self._temp_kwargs(model),

v1/discovery/models.py

@@ -708,6 +708,10 @@ class SynthesisContent:                    # everything reports 00-06 render
     fact_store: "FactStore | None" = None
     strategy: "StrategyProfile | None" = None
     planning_assumptions: list[PlanningAssumption] = field(default_factory=list)
+    # Operator signal only (NOT serialized to the client JSON): section labels the fan-out could not
+    # produce this run (transient/grounding failures). run.py warns on these and refuses --save-golden
+    # if any are present, so a polished snapshot is never silently short a section.
+    omitted_sections: list[str] = field(default_factory=list)
 
     def to_dict(self) -> dict[str, Any]:
         return {"current_state": self.current_state.to_dict(),