diff --git a/.gitignore b/.gitignore index 3f9b8bf..f78d89f 100644 --- a/.gitignore +++ b/.gitignore @@ -17,6 +17,7 @@ v1/.venv/ # ── v1 runtime cache (regenerated; noisy) ───────────────────────── v1/.cache/ +v1/.cache.tmp/ # ── coverage artifacts (regenerated) ────────────────────────────── .coverage diff --git a/explorer/.gitignore b/explorer/.gitignore index d076302..c6502d0 100644 --- a/explorer/.gitignore +++ b/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 diff --git a/v1/discovery/fanout.py b/v1/discovery/fanout.py index 5cf61e8..31ea42b 100644 --- a/v1/discovery/fanout.py +++ b/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: diff --git a/v1/discovery/fanout_specs.py b/v1/discovery/fanout_specs.py index ae080f3..1d73f84 100644 --- a/v1/discovery/fanout_specs.py +++ b/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 diff --git a/v1/discovery/llm.py b/v1/discovery/llm.py index 86cb663..7b7422e 100644 --- a/v1/discovery/llm.py +++ b/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,17 +168,26 @@ 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" @@ -163,18 +195,32 @@ def _provider_error(provider: str, e: Exception) -> 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), diff --git a/v1/discovery/models.py b/v1/discovery/models.py index 722a3da..756169f 100644 --- a/v1/discovery/models.py +++ b/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(), diff --git a/v1/discovery/reportsuite/build.py b/v1/discovery/reportsuite/build.py index 1cea206..daa1fac 100644 --- a/v1/discovery/reportsuite/build.py +++ b/v1/discovery/reportsuite/build.py @@ -44,12 +44,13 @@ def build_synthesis(raw_payload: dict, *, domain: str = "o2c", live=False, llm=N grounded fact-store; it falls back to the legacy path when absent.""" if live and fanout and reg is not None: from .. import fanout_specs - merged, planning, fs, strat = fanout_specs.run_report_fanout( + merged, planning, fs, strat, omitted = fanout_specs.run_report_fanout( llm, raw_payload, reg, doc_keys=doc_keys, model=model) content = _from_payload(merged) content.fact_store = fs content.strategy = strat content.planning_assumptions = planning + content.omitted_sections = omitted # surface only the NON-EMPTY strategy fields alongside r05's posture (don't blank anything). # Defensive: a live section occasionally emits strategy_profile as a bare string instead of # an object — spreading that would raise "'str' object is not a mapping" and abort the whole diff --git a/v1/run.py b/v1/run.py index 73e8515..d20de6f 100644 --- a/v1/run.py +++ b/v1/run.py @@ -198,6 +198,21 @@ def main(argv=None) -> int: print(" Re-run with a working model. NOT rendering wrong-domain content.") return 1 + # ---- omitted-section signal (adaptive to transient/rate-limit failures) ---- + # The parallel fan-out omits a section rather than aborting the suite when a transient provider + # failure (rate limit / overload) outlives the SDK's retries. Surface that loudly so a short + # deliverable is never silent — and refuse to bake a golden snapshot that is missing sections. + omitted = getattr(result.synthesis, "omitted_sections", []) or [] + if omitted: + print(f"! WARNING: {len(omitted)} section(s) could not be generated this run " + f"(transient/grounding): {', '.join(omitted)}.") + print(" The suite is rendered without them. Re-run to fill the gaps " + "(often a rate limit — DISCOVERY_MAX_WORKERS=1 avoids bursting the API).") + if args.save_golden: + print("! refusing --save-golden: a golden snapshot must be complete, not missing " + "sections. Re-run until all sections generate, then --save-golden.") + return 1 + # ---- render ----------------------------------------------------------- OUT.mkdir(exist_ok=True) suite_dir = OUT / args.domain # the 6-report client suite (the deliverable) diff --git a/v1/tests/test_fanout.py b/v1/tests/test_fanout.py index 3b7762b..633371a 100644 --- a/v1/tests/test_fanout.py +++ b/v1/tests/test_fanout.py @@ -186,7 +186,7 @@ def test_fanout_assembles_reports_opportunities_and_planning(): "planning_assumptions": [{"statement": "owner: CS Lead", "kind": "owner"}]}, ] llm = FakeLLM(script) - merged, planning = run_synthesis_fanout( + merged, planning, _ = run_synthesis_fanout( llm, fs, m.StrategyProfile(), doc_keys={"flow"}, report_specs=_specs(), opp_seeds=[{"id": "OPP1", "title": "Exception handling", "topic": "edi"}]) @@ -204,7 +204,7 @@ def test_fanout_skips_report_with_no_spec_and_omits_failed_opp(): "business_impact": {"quantified": [{"value": 5, "unit": "x", "text": "5"}]}}, {"id": "OPP9", "title": "t", "overview": "o", "business_impact": {"quantified": [{"value": 5, "unit": "x", "text": "5"}]}}]) - merged, planning = run_synthesis_fanout( + merged, planning, _ = run_synthesis_fanout( llm, fs, m.StrategyProfile(), doc_keys={"flow"}, report_specs={"04-opportunity-portfolio": {}}, opp_seeds=[{"id": "OPP9", "title": "t", "topic": ""}]) @@ -232,8 +232,8 @@ def flaky(*a, **k): "instruction": "y"}, } llm = FakeLLM([{"sequencing_rationale": "OPP1 first"}]) # only r03 will reach the LLM - merged, planning = run_synthesis_fanout(llm, fs, m.StrategyProfile(), doc_keys={"flow"}, - report_specs=specs) + merged, planning, _ = run_synthesis_fanout(llm, fs, m.StrategyProfile(), doc_keys={"flow"}, + report_specs=specs) assert "pain_points" not in merged # r02 omitted (raised), suite survived assert merged.get("sequencing_rationale") == "OPP1 first" # r03 still assembled @@ -247,9 +247,9 @@ def flaky(*a, **k): raise ValueError("bad opp shape") return real(*a, **k) monkeypatch.setattr(fanout, "synth_section", flaky) - merged, _ = run_synthesis_fanout(FakeLLM([]), fs, m.StrategyProfile(), doc_keys={"flow"}, - report_specs={"04-opportunity-portfolio": {}}, - opp_seeds=[{"id": "OPP1", "title": "t", "topic": ""}]) + merged, _, _ = run_synthesis_fanout(FakeLLM([]), fs, m.StrategyProfile(), doc_keys={"flow"}, + report_specs={"04-opportunity-portfolio": {}}, + opp_seeds=[{"id": "OPP1", "title": "t", "topic": ""}]) assert "opportunities" not in merged # the raising opp omitted, no crash @@ -265,8 +265,129 @@ def test_fanout_determinism_same_inputs_same_output(): fs = _fs() specs = {"03-recommendation": {"tool": "emit_r03", "schema": {"type": "object", "properties": {}}, "instruction": "rec"}} - out1, _ = run_synthesis_fanout(FakeLLM([{"sequencing_rationale": "x"}]), fs, - m.StrategyProfile(), doc_keys={"flow"}, report_specs=specs) - out2, _ = run_synthesis_fanout(FakeLLM([{"sequencing_rationale": "x"}]), fs, - m.StrategyProfile(), doc_keys={"flow"}, report_specs=specs) + out1, _, _ = run_synthesis_fanout(FakeLLM([{"sequencing_rationale": "x"}]), fs, + m.StrategyProfile(), doc_keys={"flow"}, report_specs=specs) + out2, _, _ = run_synthesis_fanout(FakeLLM([{"sequencing_rationale": "x"}]), fs, + m.StrategyProfile(), doc_keys={"flow"}, report_specs=specs) assert out1 == out2 + + +# ── parallelization: order-independence, omission signal, worker count ──────────────────────────── +import threading # noqa: E402 + + +class _ConcurrentLLM: + """Thread-safe fake: returns a fixed emit keyed by the offered tool name (no mutable per-call + state), and records how many calls were IN FLIGHT simultaneously so we can prove the pool really + ran sections concurrently (not silently serialized).""" + + def __init__(self, by_tool, barrier_n=0): + self._by_tool = by_tool + self._lock = threading.Lock() + self.max_inflight = 0 + self._inflight = 0 + # an optional barrier forces N calls to overlap, proving real concurrency + self._barrier = threading.Barrier(barrier_n) if barrier_n else None + + def messages_with_tools(self, *, system, messages, tools, model=None, max_tokens=4096): + with self._lock: + self._inflight += 1 + self.max_inflight = max(self.max_inflight, self._inflight) + if self._barrier: + try: + self._barrier.wait(timeout=5) + except threading.BrokenBarrierError: + pass + try: + name = tools[0]["name"] + return ToolTurn(content=[{"type": "tool_use", "id": "e", "name": name, + "input": self._by_tool.get(name, {})}], stop_reason="tool_use") + finally: + with self._lock: + self._inflight -= 1 + + +_MULTI_SPECS = { + "00-executive-summary": {"tool": "emit_r00", "schema": {"type": "object", "properties": {}}, + "instruction": "exec"}, + "03-recommendation": {"tool": "emit_r03", "schema": {"type": "object", "properties": {}}, + "instruction": "rec"}, + "05-roadmap": {"tool": "emit_r05", "schema": {"type": "object", "properties": {}}, + "instruction": "road"}, +} +_MULTI_EMITS = { + "emit_r00": {"target_state": "t"}, + "emit_r03": {"sequencing_rationale": "seq"}, + "emit_r05": {"strategic_readiness": "ready"}, + "emit_opportunity": {"id": "OPPx", "title": "t", "overview": "o"}, +} + + +def test_max_workers_env_parsing(monkeypatch): + monkeypatch.setenv("DISCOVERY_MAX_WORKERS", "3"); assert fanout._max_workers() == 3 + monkeypatch.setenv("DISCOVERY_MAX_WORKERS", "0"); assert fanout._max_workers() == 1 # floored + monkeypatch.setenv("DISCOVERY_MAX_WORKERS", "nonsense"); assert fanout._max_workers() == 4 # fallback + monkeypatch.delenv("DISCOVERY_MAX_WORKERS", raising=False); assert fanout._max_workers() == 4 + + +def test_fanout_output_identical_at_workers_1_and_4(monkeypatch): + """The whole point: byte-identical merged/planning regardless of worker count (submission-order + consume). Same fixed emits, run serial then parallel — outputs must be equal.""" + def run(n): + monkeypatch.setenv("DISCOVERY_MAX_WORKERS", str(n)) + return run_synthesis_fanout(_ConcurrentLLM(_MULTI_EMITS), _fs(), m.StrategyProfile(), + doc_keys={"flow"}, report_specs=_MULTI_SPECS, + opp_seeds=[{"id": "OPPx", "title": "t", "topic": ""}]) + m1, p1, o1 = run(1) + m4, p4, o4 = run(4) + assert m1 == m4 and o1 == o4 == [] + assert [str(x.statement) for x in p1] == [str(x.statement) for x in p4] + + +def test_fanout_actually_runs_concurrently(): + """Prove the pool overlaps calls (not silently serialized): 3 sections + a 3-way barrier; if any + call ran alone the barrier would time out, so reaching max_inflight==3 proves real concurrency.""" + llm = _ConcurrentLLM(_MULTI_EMITS, barrier_n=3) + run_synthesis_fanout(llm, _fs(), m.StrategyProfile(), doc_keys={"flow"}, + report_specs=_MULTI_SPECS) + assert llm.max_inflight == 3 + + +def test_fanout_reports_omitted_sections(): + """A section that fails grounding twice is omitted AND reported in the third return value, so + run.py can warn / refuse --save-golden.""" + fs = _fs() + # opp grounds-fails (a measured number not in allow) on both attempts -> omitted + bad_opp = {"id": "OPP9", "title": "t", "overview": "o", + "business_impact": {"quantified": [{"value": 999, "unit": "x", "text": "999"}]}} + llm = _ConcurrentLLM({**_MULTI_EMITS, "emit_opportunity": bad_opp}) + _, _, omitted = run_synthesis_fanout(llm, fs, m.StrategyProfile(), doc_keys={"flow"}, + report_specs={"04-opportunity-portfolio": {}}, + opp_seeds=[{"id": "OPP9", "title": "t", "topic": ""}]) + assert omitted == ["OPP9"] # labelled by the seed id, surfaced for the operator + + +def test_fanout_placeholder_report_not_flagged_omitted(): + """A report spec with an EMPTY instruction is a structural placeholder (the 04 portfolio's + content comes from opp seeds). Its None return must NOT be reported as a missing section — + otherwise the operator gets a false-positive warning and --save-golden is wrongly blocked.""" + fs = _fs() + # report 04 has an empty instruction AND its emit grounds-empty -> returns None, but it's a + # placeholder so it must NOT appear in `omitted`. A substantive report (03) succeeds normally. + specs = { + "03-recommendation": {"tool": "emit_r03", "schema": {"type": "object", "properties": {}}, + "instruction": "rec"}, + "04-opportunity-portfolio": {"tool": "emit_portfolio", + "schema": {"type": "object", "properties": {}}, + "instruction": ""}, # <- placeholder + } + # emit_portfolio grounds-fails (a measured number not in `allow`) on both attempts -> the worker + # returns None for it. Because its instruction is empty (placeholder), that None must NOT be + # tracked. The opp seed still produces the real portfolio content. + bad_portfolio = {"numbers": [{"value": 12345, "unit": "x", "text": "12,345 widgets"}]} + llm = _ConcurrentLLM({**_MULTI_EMITS, "emit_portfolio": bad_portfolio}) + merged, _, omitted = run_synthesis_fanout( + llm, fs, m.StrategyProfile(), doc_keys={"flow"}, report_specs=specs, + opp_seeds=[{"id": "OPPx", "title": "t", "topic": ""}]) + assert "04-opportunity-portfolio" not in omitted # placeholder None is expected, not flagged + assert merged["opportunities"][0]["id"] == "OPPx" # portfolio content came from the opp seed diff --git a/v1/tests/test_fanout_specs.py b/v1/tests/test_fanout_specs.py index 6fb318b..0294d4a 100644 --- a/v1/tests/test_fanout_specs.py +++ b/v1/tests/test_fanout_specs.py @@ -101,7 +101,7 @@ def test_fanout_assembles_reference_depth_synthesis_content(monkeypatch): llm = Fake() reg = {"csv_ids": ["flow"], "doc_ids": [], "manifest": {}} raw = {"_tool_numbers": [1196], "findings": []} - merged, planning, fs, strat = fspec.run_report_fanout( + merged, planning, fs, strat, _ = fspec.run_report_fanout( llm, raw, reg, strategy=m.StrategyProfile(direction_type="consolidate"), doc_keys={"flow"}) c = _from_payload(merged) @@ -161,6 +161,6 @@ def test_fanout_determinism(monkeypatch): monkeypatch.setattr(fspec.factstore, "build_fact_store", lambda raw, reg: _fs()) reg = {"csv_ids": ["flow"], "doc_ids": [], "manifest": {}} raw = {"_tool_numbers": [1196], "findings": []} - a, _, _, _ = fspec.run_report_fanout(Fake(), raw, reg, doc_keys={"flow"}) - b, _, _, _ = fspec.run_report_fanout(Fake(), raw, reg, doc_keys={"flow"}) + a, _, _, _, _ = fspec.run_report_fanout(Fake(), raw, reg, doc_keys={"flow"}) + b, _, _, _, _ = fspec.run_report_fanout(Fake(), raw, reg, doc_keys={"flow"}) assert a == b diff --git a/v1/tests/test_llm_concurrency.py b/v1/tests/test_llm_concurrency.py new file mode 100644 index 0000000..e5e62c7 --- /dev/null +++ b/v1/tests/test_llm_concurrency.py @@ -0,0 +1,116 @@ +"""Concurrency-safety coverage for discovery/llm.py (the synthesis fan-out runs many sections on a +thread pool). llm.py is omitted from the coverage gate (real HTTP client), but these behaviours are +load-bearing for parallel synthesis and were flagged by the design review, so they're tested here. + +All offline — no network. We exercise: (1) _provider_error classification (transient → retryable, +auth/config → hard, so a misconfigured run never silently omits a report); (2) atomic _write_cache +under concurrent same-key writes (no torn file → no corrupt golden artifact); (3) the locked lazy +client init building exactly one client across threads. +""" +from __future__ import annotations + +import json +import sys +import threading +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT)) + +from discovery.llm import LLMClient, LLMError, LLMRetryableError # noqa: E402 + + +# ── _provider_error: transient vs hard ──────────────────────────────────────────────────────────── +class _NamedExc(Exception): + """An exception whose class NAME we control, to simulate the SDK's error classes by name.""" + def __init__(self, name, msg=""): + super().__init__(msg) + self.__class__.__name__ = name + + +def test_provider_error_auth_is_hard_llmerror(): + # missing/invalid credentials must stay a plain LLMError (NOT retryable) so it aborts loudly + e = LLMClient._provider_error("anthropic", _NamedExc("AuthenticationError", "bad api_key")) + assert isinstance(e, LLMError) and not isinstance(e, LLMRetryableError) + assert "credentials" in str(e).lower() + + +def test_provider_error_transient_is_retryable(): + for name in ("RateLimitError", "APIConnectionError", "InternalServerError", "OverloadedError"): + e = LLMClient._provider_error("anthropic", _NamedExc(name, "boom")) + assert isinstance(e, LLMRetryableError), name + # also classified transient by message signal even with a generic class name + assert isinstance(LLMClient._provider_error("anthropic", _NamedExc("X", "Error 429 overloaded")), + LLMRetryableError) + + +def test_provider_error_unknown_is_hard(): + # an unrecognised, non-auth, non-transient failure stays a hard LLMError (don't silently omit) + e = LLMClient._provider_error("anthropic", _NamedExc("WeirdError", "??")) + assert isinstance(e, LLMError) and not isinstance(e, LLMRetryableError) + + +# ── _write_cache: atomic + concurrent same-key ───────────────────────────────────────────────────── +def test_write_cache_atomic_concurrent_same_key(tmp_path): + """Many threads writing the SAME key concurrently must never leave a torn/half-written file — + every read must parse, and the final file must be valid JSON with the expected shape.""" + c = LLMClient(cache_dir=tmp_path / ".cache") + key = "k" * 32 + errors: list[str] = [] + + def writer(i): + try: + for _ in range(20): + c._write_cache(key, "sys", "prompt", f"response-{i}") + # read it straight back — must always be parseable (never a torn file) + json.loads(c._cache_path(key).read_text()) + except Exception as e: # noqa: BLE001 - capture for the assertion + errors.append(repr(e)) + + threads = [threading.Thread(target=writer, args=(i,)) for i in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + assert errors == [] # no torn reads + final = json.loads(c._cache_path(key).read_text()) # final file is valid + well-shaped + assert final["system"] == "sys" and final["response"].startswith("response-") + + +def test_write_cache_uses_scratch_outside_cache_dir(tmp_path): + """The temp file must NOT be written inside .cache/ (else --save-golden's copytree could capture + a crash-orphaned temp). After a write, .cache/ holds exactly the final {key}.json, no *.tmp.""" + c = LLMClient(cache_dir=tmp_path / ".cache") + c._write_cache("a" * 32, "s", "p", "r") + names = [p.name for p in (tmp_path / ".cache").iterdir()] + assert names == [f"{'a' * 32}.json"] # only the final file, no temp leftover + + +# ── locked lazy client init ───────────────────────────────────────────────────────────────────── +def test_client_init_is_locked_and_single(monkeypatch, tmp_path): + """Under concurrent first-use from many threads, exactly ONE Anthropic client is constructed.""" + c = LLMClient(cache_dir=tmp_path / ".cache") + built = {"n": 0} + + class _FakeAnthropic: + def __init__(self, **kw): + built["n"] += 1 + + import types + fake_mod = types.ModuleType("anthropic") + fake_mod.Anthropic = _FakeAnthropic # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "anthropic", fake_mod) + + start = threading.Barrier(8) + + def use(): + start.wait(timeout=5) + c._anthropic_client() + + threads = [threading.Thread(target=use) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + assert built["n"] == 1 # double-checked lock → constructed once + assert c._client is not None