#!/usr/bin/env python3 """Merge the analyst output into evidences/workspace_evidence_report.json. Mechanical companion to diagnosis-analyst: the analyst authors evidences/agent-runs/diagnosis-analyst.json (its single source of truth), and this script owns the report merge — preserving all collector-generated raw fields, replacing only the `analysis` section, repairing the epoch-seconds time-range contract when needed, and validating the result. This replaces the read-report → whole-file-rewrite → validate → retry loop inside the agent. Run from the workspace root (the directory containing evidences/). """ from __future__ import annotations import argparse import json import re import sys from datetime import datetime from pathlib import Path ANALYST_PATH = Path("evidences") / "agent-runs" / "diagnosis-analyst.json" REPORT_PATH = Path("evidences") / "workspace_evidence_report.json" SUMMARY_PATH = Path("evidences") / "evidence_summary_for_agent.json" ROUTING_PATH = Path("evidences") / "agent-runs" / "diagnosis-routing.json" HANDOFF_PATH = Path("evidences") / "agent-runs" / "evidence-agent.json" # Hand-written core: everything else is normalized in by _normalize_analyst. CORE_ANALYST_FIELDS = ( "agent", "root_cause", "key_evidence", "signal_tags", "fault_classification", "final_response", ) REQUIRED_ANALYST_FIELDS = ( "agent", "status", "root_cause", "confidence", "key_evidence", "rejected_causes", "needs_more_evidence", "missing_evidence", "quality_checks", "signal_tags", "fault_classification", "selected_diagnosis_skills", "critical_evidence_hints", "final_response", ) GUARDRAIL_MARKER = "KB Step B complete: Pitfall catalog loaded" LEGACY_TO_STANDARD_CLASS = { "sql_plan_or_query": "sql_plan_or_query", "lock_or_transaction_contention": "transaction_or_lock_contention", "tikv_read_hotspot": "tikv_hotspot_or_coprocessor", "tidb_cpu_gc_or_memory": "memory_or_oom_pressure", "component_restart_or_panic": "component_failure_or_bug", "resource_control_or_capacity": "resource_control_or_capacity", "storage_or_infra_io": "storage_or_io_bottleneck", "service_unavailability_or_worker_failure": "service_unavailability_or_worker_failure", "unknown_insufficient_evidence": "unknown_insufficient_evidence", } ROUTE_SKILL_MARKERS = { "sql_plan_or_query": ("sql-plan-or-query/subskill.md", "tidb-query-tuning-autox"), "transaction_or_lock_contention": ( "transaction-or-lock-contention/subskill.md", "transaction-oncall-handbook", ), "tikv_hotspot_or_coprocessor": ("tikv-hotspot-or-coprocessor/subskill.md",), "storage_or_io_bottleneck": ("storage-or-io-bottleneck/subskill.md",), "memory_or_oom_pressure": ("memory-or-oom-pressure/subskill.md",), "gc_or_allocator_pressure": ("gc-or-allocator-pressure/subskill.md",), "resource_control_or_capacity": ("resource-control-or-capacity/subskill.md",), "pd_tso_or_scheduling": ("pd-tso-or-scheduling/subskill.md",), "cdc_or_replication_lag": ( "cdc-or-replication-lag/subskill.md", "ticdc-health-inspection", ), "network_or_connectivity": ("network-or-connectivity/subskill.md",), "component_failure_or_bug": ("component-failure-or-bug/subskill.md",), } def _check_guardrail_marker(analyst: dict) -> bool: """The misdiagnosis guardrail catalog must be loaded before attribution. The marker is required in quality_checks or final_response; refusing the artifact without it keeps the KB Step B gate honest and eval-visible. """ haystack = json.dumps( [analyst.get("quality_checks"), analyst.get("final_response")], ensure_ascii=False, ) if GUARDRAIL_MARKER in haystack: return True print( "error: guardrail marker missing. Read " "tidb-knowledge-base/knowledge/incorrect-conclusions.json (KB Step B), " f"then record '{GUARDRAIL_MARKER}' in quality_checks and final_response.", file=sys.stderr, ) return False def _final_standard_class(analyst: dict) -> str: classification = analyst.get("fault_classification") if not isinstance(classification, dict): return "" precise = str(classification.get("standard_root_cause_class") or "").strip() if precise: return precise return LEGACY_TO_STANDARD_CLASS.get( str(classification.get("primary_class") or "").strip(), "" ) def _check_route_override_contract(analyst: dict) -> bool: """Require the final family and mandatory specialist after a Copilot reroute.""" routing = _load_json(ROUTING_PATH) selection = routing.get("selection") if isinstance(routing, dict) else None selected_routes = selection.get("routes") if isinstance(selection, dict) else None if not isinstance(selected_routes, list) or not selected_routes: return True selected_routes = [str(route) for route in selected_routes if str(route)] final_route = _final_standard_class(analyst) if not final_route or final_route in selected_routes: return True if final_route not in ROUTE_SKILL_MARKERS: print( "error: rerouted diagnosis requires a standard_root_cause_class from " "the canonical taxonomy", file=sys.stderr, ) return False override = analyst.get("classification_override") if not isinstance(override, dict): print( "error: final classification differs from the Copilot selection; " "classification_override is required", file=sys.stderr, ) return False from_routes = override.get("from_routes") if not isinstance(from_routes, list) or set(map(str, from_routes)) != set(selected_routes): print( "error: classification_override.from_routes must match the selected Copilot routes", file=sys.stderr, ) return False if str(override.get("to_route") or "") != final_route: print( "error: classification_override.to_route must match the final standard_root_cause_class", file=sys.stderr, ) return False if not str(override.get("reason") or "").strip(): print("error: classification_override.reason is required", file=sys.stderr) return False skill_text = "\n".join( str(skill).lower() for skill in analyst.get("selected_diagnosis_skills") or [] ) missing = [ marker for marker in ROUTE_SKILL_MARKERS.get(final_route, ()) if marker not in skill_text ] if missing: print( "error: rerouted diagnosis must load the final classification family and " f"mandatory specialist; missing: {', '.join(missing)}", file=sys.stderr, ) return False return True SIGNAL_TAG_DEFAULTS = ( ("baseline", "incident_window_only"), ("severity", "medium"), ("confidence", "medium"), ) SOURCE_BY_REF_HINT = ( ("resource_control", "resource_control"), ("component_health", "component_health"), ("topsql", "topsql"), ("queries", "queries"), ("logs", "logs"), ("events", "events"), ("metrics", "metrics"), ) def _slug(text: str) -> str: slug = re.sub(r"[^a-z0-9]+", "_", str(text or "").lower()).strip("_") return slug[:60] or "signal" def _source_from_ref(evidence_ref: str) -> str: ref = str(evidence_ref or "").lower() for hint, source in SOURCE_BY_REF_HINT: if hint in ref: return source return "metrics" def _normalize_analyst(analyst: dict) -> bool: """Mechanically complete fields the agent may omit for generation economy. The on-disk contract stays full; the agent only hand-writes what carries information. Returns True when anything was filled in. """ changed = False for tag in analyst.get("signal_tags") or []: if not isinstance(tag, dict): continue if not tag.get("id"): tag["id"] = _slug(f"{tag.get('metric_or_signal', '')}_{tag.get('direction', '')}") changed = True if not tag.get("source"): tag["source"] = _source_from_ref(tag.get("evidence_ref", "")) changed = True for key, default in SIGNAL_TAG_DEFAULTS: if not tag.get(key): tag[key] = default changed = True classification = analyst.get("fault_classification") if isinstance(classification, dict): candidates = classification.get("candidates") or [] structured = [c for c in candidates if isinstance(c, dict)] if not classification.get("rejected_classes"): rejected = [str(c.get("class", "")) for c in structured if c.get("role") == "rejected" and c.get("class")] if rejected: classification["rejected_classes"] = rejected changed = True for key, default in (("rejected_classes", []), ("candidates", []), ("classification_confidence", "medium"), ("missing_evidence_for_higher_confidence", [])): if key not in classification: classification[key] = default changed = True if not analyst.get("rejected_causes"): derived = [ f"{c.get('class')}: {'; '.join(str(x) for x in (c.get('evidence_against') or [])) or 'no supporting evidence in window'}" for c in structured if c.get("role") == "rejected" and c.get("class") ] if derived: analyst["rejected_causes"] = derived changed = True if not analyst.get("confidence"): analyst["confidence"] = str(classification.get("classification_confidence") or "medium") changed = True for key, default in (("status", "ok"), ("needs_more_evidence", False), ("missing_evidence", []), ("key_evidence", []), ("rejected_causes", []), ("quality_checks", []), ("confidence", "medium")): if key not in analyst: analyst[key] = default changed = True if not analyst.get("critical_evidence_hints"): handoff = _load_json(HANDOFF_PATH) or {} hints = handoff.get("critical_evidence_hints") or [] analyst["critical_evidence_hints"] = hints changed = True if _restore_hint_time_scopes(analyst): changed = True if not analyst.get("selected_diagnosis_skills"): # Default shared contract only; never infer family/specialist reads. analyst["selected_diagnosis_skills"] = ["tidb-knowledge-base", "diagnosis-overview"] changed = True return changed def _write_analyst_file(analyst: dict) -> None: ANALYST_PATH.parent.mkdir(parents=True, exist_ok=True) ANALYST_PATH.write_text( json.dumps(analyst, ensure_ascii=False, indent=2, default=str) + "\n", encoding="utf-8", ) def _read_stdin_analyst() -> dict | None: """Read, validate, normalize, and persist the analyst JSON from stdin.""" try: data = json.loads(sys.stdin.read()) except json.JSONDecodeError as exc: print(f"error: stdin is not valid JSON: {exc}", file=sys.stderr) return None if not isinstance(data, dict): print("error: stdin JSON must be an object", file=sys.stderr) return None missing = [key for key in CORE_ANALYST_FIELDS if key not in data] if missing: print(f"error: analyst JSON missing required fields: {', '.join(missing)}", file=sys.stderr) return None if not _check_guardrail_marker(data): return None _normalize_analyst(data) if not _check_route_override_contract(data): return None if not _check_canonical_window_evidence(data): return None missing = [key for key in REQUIRED_ANALYST_FIELDS if key not in data] if missing: # normalization guarantees the full contract; guard regressions print(f"error: normalization left fields missing: {', '.join(missing)}", file=sys.stderr) return None _write_analyst_file(data) print(f"diagnosis-analyst.json written: agent={data.get('agent')} status={data.get('status')}") return data def _load_json(path: Path) -> dict | None: try: data = json.loads(path.read_text(encoding="utf-8")) return data if isinstance(data, dict) else None except (OSError, json.JSONDecodeError): return None def _derive_evidence_chain(analyst: dict) -> list[str]: chain = [] for tag in analyst.get("signal_tags") or []: if not isinstance(tag, dict): continue parts = [ str(tag.get("component", "")), str(tag.get("metric_or_signal", "")), str(tag.get("direction", "")), ] line = " ".join(p for p in parts if p) ref = tag.get("evidence_ref") baseline = tag.get("baseline") if baseline and baseline != "unknown": line += f" vs {baseline}" if ref: line += f" ({ref})" if line.strip(): chain.append(line.strip()) return chain def _context_only_hint(hint: object) -> bool: return ( isinstance(hint, dict) and bool(hint.get("time_scope")) and hint["time_scope"] != "canonical_window" ) def _hint_id(hint: object) -> str: if isinstance(hint, dict): return str(hint.get("id") or hint.get("hint") or "") return hint if isinstance(hint, str) else "" def _verified_followup_hint(hint: dict, handoff: dict, analyst: dict) -> bool: """Qualify a source hint by its raw sample, not by an agent's scope label.""" if hint.get("time_scope") != "canonical_window": return False scope = analyst.get("scope") or {} source_scope = handoff.get("scope") or {} if not scope.get("cluster_id") or source_scope.get("cluster_id") != scope["cluster_id"]: return False ref = hint.get("evidence_ref") if not isinstance(ref, str) or "#/" not in ref: return False filename, pointer = ref.split("#", 1) path = Path(filename).resolve() root = Path("evidences").resolve() if not path.is_relative_to(root) or path.is_relative_to(root / "agent-runs"): return False if path in {SUMMARY_PATH.resolve(), REPORT_PATH.resolve()}: return False try: sample = json.loads(path.read_text(encoding="utf-8")) if isinstance(sample, dict) and sample.get("cluster_id") not in (None, scope["cluster_id"]): return False for part in pointer[1:].split("/"): part = part.replace("~1", "/").replace("~0", "~") if isinstance(sample, list): if not part.isdigit(): return False sample = sample[int(part)] else: sample = sample[part] if not isinstance(sample, dict) or not (sample.get("line") or sample.get("message")): return False # A bucket timestamp does not date its representative log sample. if sample.get("aggregation_window") or sample.get("count_semantics"): return False timestamp = sample.get("timestamp_ns") if timestamp is not None: timestamp_ns = int(timestamp) else: observed = str(sample.get("observed_time") or sample.get("timestamp") or "") dt = datetime.fromisoformat(observed.replace("Z", "+00:00")) if dt.tzinfo is None: return False timestamp_ns = int(dt.timestamp() * 1_000_000_000) for item in (scope, source_scope): window = item.get("time_range") or {} start, end = _window_epoch_bounds(window) if start is None or end is None or not int(start) * 1_000_000_000 <= timestamp_ns <= int(end) * 1_000_000_000: return False except (OSError, ValueError, TypeError, KeyError, IndexError, OverflowError): return False return True def _restore_hint_time_scopes(analyst: dict) -> bool: # Analyst status/note are judgments, not replacements for source time scope. # Check the collector too: the evidence handoff may have omitted metadata. restricted = {} handoff = _load_json(HANDOFF_PATH) or {} for source in (_load_json(SUMMARY_PATH) or {}, handoff): for hint in source.get("critical_evidence_hints") or []: if _hint_id(hint) and _context_only_hint(hint): restricted.setdefault(_hint_id(hint), []).append(hint) confirmed = { _hint_id(hint): hint for hint in handoff.get("critical_evidence_hints") or [] if isinstance(hint, dict) and _hint_id(hint) in restricted and _verified_followup_hint(hint, handoff, analyst) } changed = False hints = analyst.get("critical_evidence_hints") or [] for index, hint in enumerate(hints): sources = restricted.get(_hint_id(hint)) if not sources: continue followup = confirmed.get(_hint_id(hint)) ref = hint.get("evidence_ref") if isinstance(hint, dict) else None if followup and (not ref or ref == followup["evidence_ref"]) and not _context_only_hint(hint): restored = dict(followup) if isinstance(hint, dict): for field in ("status", "note"): if field in hint: restored[field] = hint[field] if restored != hint: hints[index] = restored changed = True continue source = next((item for item in sources if ref and item.get("evidence_ref") == ref), sources[-1]) restored = {**source, **(hint if isinstance(hint, dict) else {})} restored["time_scope"] = source["time_scope"] if restored != hint: hints[index] = restored changed = True # Explicit supporting refs may mention a hint omitted from the analyst list. present_ids = {_hint_id(hint) for hint in hints} context = list(analyst.get("outside_window_context") or []) for hint_id, sources in restricted.items(): for source in sources: if (hint_id not in present_ids or hint_id in confirmed) and source not in context: context.append(source) changed = True if context != (analyst.get("outside_window_context") or []): analyst["outside_window_context"] = context return changed def _derive_supporting(analyst: dict) -> list[str]: active = analyst.get("critical_evidence_hints") or [] canonical_ids = { _hint_id(hint) for hint in active if isinstance(hint, dict) and hint.get("time_scope") == "canonical_window" and hint.get("evidence_ref") } context_ids = { _hint_id(hint) for hint in ( list(analyst.get("critical_evidence_hints") or []) + list(analyst.get("outside_window_context") or []) ) if _hint_id(hint) and _context_only_hint(hint) } restricted_refs = { hint["evidence_ref"] for hint in list(active) + list(analyst.get("outside_window_context") or []) if _context_only_hint(hint) and hint.get("evidence_ref") } restricted_refs -= { hint["evidence_ref"] for hint in active if isinstance(hint, dict) and hint.get("time_scope") == "canonical_window" and hint.get("evidence_ref") } context_ids -= canonical_ids context_refs = {f"hint: {hint_id}" for hint_id in context_ids} if analyst.get("supporting_evidence"): return [ item for item in analyst["supporting_evidence"] if not _context_only_hint(item) and not (isinstance(item, dict) and _hint_id(item) in context_ids) and not (isinstance(item, dict) and item.get("evidence_ref") in restricted_refs) and not (isinstance(item, str) and item in restricted_refs) and not (isinstance(item, str) and item in context_refs) ] supporting = [] for hint in analyst.get("critical_evidence_hints") or []: if isinstance(hint, dict): if _context_only_hint(hint): continue hint_id = hint.get("id") or hint.get("hint") status = str(hint.get("status", "")).lower() if hint_id and status in ("", "matched", "present"): supporting.append(f"hint: {hint_id}") elif isinstance(hint, str): supporting.append(f"hint: {hint}") return supporting def _derive_context(analyst: dict) -> list: context = list(analyst.get("outside_window_context") or []) for hint in analyst.get("critical_evidence_hints") or []: if _context_only_hint(hint) and hint not in context: # Preserve the full hint, including unknown/overlapping time scope. context.append(hint) return context def _stamp_canonical_scope(report: dict, analyst: dict) -> None: """Stamp the canonical investigation window from the analyst scope. metadata.time_range is the investigation scope; targeted data commands may have seeded it with a narrower query slice. Preserve an already wider canonical window, but allow the analyst scope to expand a missing or narrower report window. """ scope = analyst.get("scope") if not isinstance(scope, dict): return time_range = scope.get("time_range") if isinstance(time_range, dict): start = time_range.get("start") or time_range.get("from") end = time_range.get("end") or time_range.get("to") if start and end: metadata = report.setdefault("metadata", {}) existing = metadata.get("time_range") if not _incoming_window_is_strictly_narrower(existing, start=start, end=end): metadata["time_range"] = {"start": str(start), "end": str(end)} cluster_id = str(scope.get("cluster_id") or "") if cluster_id: metadata = report.setdefault("metadata", {}) if not metadata.get("cluster_id"): metadata["cluster_id"] = cluster_id cluster = metadata.get("cluster") if not (isinstance(cluster, dict) and cluster.get("cluster_id")): metadata["cluster"] = {"cluster_id": cluster_id} def _epoch_seconds_str(value: object) -> str | None: """Return the value as an epoch-seconds string, converting ISO if needed.""" text = str(value or "").strip() if not text: return None if text.replace(".", "", 1).isdigit(): number = float(text) if number > 1e12: # milliseconds number /= 1000 return str(int(number)) try: return str(int(datetime.fromisoformat(text.replace("Z", "+00:00")).timestamp())) except ValueError: return None def _window_epoch_bounds(window: dict) -> tuple[str | None, str | None]: """Use the same boundary aliases for report and supplemental validation.""" return ( _epoch_seconds_str(window.get("start") or window.get("from")), _epoch_seconds_str(window.get("end") or window.get("to")), ) def _incoming_window_is_strictly_narrower( existing: object, *, start: object, end: object, ) -> bool: if not isinstance(existing, dict): return False existing_start = _epoch_seconds_str(existing.get("start") or existing.get("from")) existing_end = _epoch_seconds_str(existing.get("end") or existing.get("to")) incoming_start = _epoch_seconds_str(start) incoming_end = _epoch_seconds_str(end) if not all((existing_start, existing_end, incoming_start, incoming_end)): return False existing_start_i = int(existing_start) existing_end_i = int(existing_end) incoming_start_i = int(incoming_start) incoming_end_i = int(incoming_end) if existing_start_i > existing_end_i or incoming_start_i > incoming_end_i: return False return ( existing_start_i <= incoming_start_i and incoming_end_i <= existing_end_i and (incoming_start_i > existing_start_i or incoming_end_i < existing_end_i) ) def _repair_time_range(metadata: dict) -> None: time_range = metadata.get("time_range") if not isinstance(time_range, dict): return for key, iso_key in (("start", "start_iso"), ("end", "end_iso")): raw = str(time_range.get(key, "") or "") epoch = _epoch_seconds_str(raw) if epoch is None or epoch == raw: continue if raw and not raw.replace(".", "", 1).isdigit() and not time_range.get(iso_key): time_range[iso_key] = raw time_range[key] = epoch ISO_TIMESTAMP_RE = re.compile( r"\b20\d{2}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})\b" ) def _load_bearing_evidence_items(analyst: dict) -> list[tuple[str, object]]: items: list[tuple[str, object]] = [] for index, item in enumerate(analyst.get("key_evidence") or []): items.append((f"key_evidence[{index}]", item)) for index, item in enumerate(analyst.get("signal_tags") or []): items.append((f"signal_tags[{index}]", item)) classification = analyst.get("fault_classification") candidates = classification.get("candidates") if isinstance(classification, dict) else [] for candidate_index, candidate in enumerate(candidates or []): if not isinstance(candidate, dict) or candidate.get("role") != "primary": continue for evidence_index, item in enumerate(candidate.get("evidence_for") or []): items.append( ( f"fault_classification.candidates[{candidate_index}]." f"evidence_for[{evidence_index}]", item, ) ) return items def _check_canonical_window_evidence(analyst: dict) -> bool: """Reject explicitly dated load-bearing evidence outside the canonical window.""" scope = analyst.get("scope") time_range = scope.get("time_range") if isinstance(scope, dict) else None if not isinstance(time_range, dict): print("error: analyst scope.time_range is required", file=sys.stderr) return False start, end = _window_epoch_bounds(time_range) if start is None or end is None or int(start) > int(end): print("error: analyst scope.time_range must be a valid canonical window", file=sys.stderr) return False start_i, end_i = int(start), int(end) for field, item in _load_bearing_evidence_items(analyst): if isinstance(item, dict) and item.get("time_scope") == "outside_window_context": print( f"error: {field} is load-bearing and cannot use outside_window_context; " "move it to outside_window_context", file=sys.stderr, ) return False text = item if isinstance(item, str) else json.dumps(item, ensure_ascii=False) for timestamp in ISO_TIMESTAMP_RE.findall(text): epoch = _epoch_seconds_str(timestamp) if epoch is not None and not start_i <= int(epoch) <= end_i: print( f"error: {field} cites {timestamp} outside the canonical window; " "move it to outside_window_context or replace it with in-window evidence", file=sys.stderr, ) return False return True def main() -> int: parser = argparse.ArgumentParser(description="Merge the analyst output into the workspace evidence report") parser.add_argument( "--from-stdin", action="store_true", help="Read the analyst JSON from stdin, write evidences/agent-runs/diagnosis-analyst.json, then merge the report.", ) args = parser.parse_args() if args.from_stdin: analyst = _read_stdin_analyst() if analyst is None: return 1 return _merge_report(analyst) analyst = _load_json(ANALYST_PATH) if analyst is None: print(f"error: {ANALYST_PATH} missing or invalid; write it first", file=sys.stderr) return 1 return _merge_report(analyst) def _merge_report(analyst: dict) -> int: if not _check_guardrail_marker(analyst): return 1 normalized = _normalize_analyst(analyst) if not _check_route_override_contract(analyst): return 1 if not _check_canonical_window_evidence(analyst): return 1 if normalized: _write_analyst_file(analyst) report = _load_json(REPORT_PATH) created = report is None if created: summary = _load_json(SUMMARY_PATH) or {} scope = summary.get("scope") or analyst.get("scope") or {} report = { "type": "workspace_evidence_report", "metadata": { "cluster_id": str(scope.get("cluster_id", "")), "cluster": {"cluster_id": str(scope.get("cluster_id", ""))}, "time_range": dict(scope.get("time_range") or {}), }, "evidence": {"metrics": [], "queries": [], "logs": [], "events": []}, "links": {}, } report["analysis"] = { "root_cause": analyst.get("root_cause", ""), "confidence": analyst.get("confidence", "medium"), "evidence_chain": analyst.get("evidence_chain") or _derive_evidence_chain(analyst), "key_evidence": analyst.get("key_evidence") or [], "supporting_evidence": _derive_supporting(analyst), "rejected_causes": analyst.get("rejected_causes") or [], "quality_checks": analyst.get("quality_checks") or [], "recommendations": analyst.get("recommendations") or [], "outside_window_context": _derive_context(analyst), } _stamp_canonical_scope(report, analyst) if isinstance(report.get("metadata"), dict): _repair_time_range(report["metadata"]) text = json.dumps(report, ensure_ascii=False, indent=2, default=str) json.loads(text) # validate before touching the file REPORT_PATH.write_text(text + "\n", encoding="utf-8") json.loads(REPORT_PATH.read_text(encoding="utf-8")) # round-trip validation print( "workspace_evidence_report.json updated:", f"created={str(created).lower()}", f"chain={len(report['analysis']['evidence_chain'])}", f"key={len(report['analysis']['key_evidence'])}", f"rejected={len(report['analysis']['rejected_causes'])}", f"recommendations={len(report['analysis']['recommendations'])}", ) return 0 if __name__ == "__main__": raise SystemExit(main())