feat: Deterministic data gate on the Researcher->Analyzer edge

Applies the AgentAsk edge-contract principle deterministically (no LLM):
- DG: per-category required-metric rules audited on the exact extracted
view the Analyzer consumes; gaps render as an explicit MISSING section
in the reference table instructing 'DATA NOT PROVIDED' - the silent
empty-valuation case can no longer pass unflagged, and the Critic's
existing constraint rule finally has something to enforce
- SC: sanity bounds quarantine impossible magnitudes (decimal/unit slips)
before they can enter the reference table and be cited; bounds reject
the impossible, not the unusual (Boeing's real 9.92 D/E passes)
- Both findings logged to the activity stream; 7 gate tests added

src/nodes/analyzer.py

@@ -970,13 +970,17 @@ def _format_metric_for_reference(key: str, value, temporal_info: dict = None) ->
     return formatted, as_of_date
 
 
-def _generate_metric_reference_table(extracted: dict, is_financial: bool = False) -> tuple:
+def _generate_metric_reference_table(extracted: dict, is_financial: bool = False,
+                                     data_gaps: list = None) -> tuple:
     """
     Generate an immutable metric reference table for LLM grounding.
 
     Args:
         extracted: Extracted metrics dictionary from _extract_key_metrics()
         is_financial: If True, exclude EV/EBITDA
+        data_gaps: Required metrics found missing by the data gate; rendered
+                   as an explicit MISSING section so the LLM must say
+                   "DATA NOT PROVIDED" rather than silently omit or invent
 
     Returns:
         tuple: (table_string, metric_lookup_dict)
@@ -1099,6 +1103,15 @@ def _generate_metric_reference_table(extracted: dict, is_financial: bool = False
         lines.extend(sentiment_lines)
         lines.append("")
 
+    if data_gaps:
+        lines.append("[MISSING - REQUIRED DATA NOT AVAILABLE]")
+        lines.append("  The following required metrics could not be obtained.")
+        lines.append("  If a point would depend on one, write exactly: DATA NOT PROVIDED")
+        lines.append("  Do NOT estimate, infer, or substitute values for these:")
+        for gap in data_gaps:
+            lines.append(f"  - {gap}")
+        lines.append("")
+
     lines.append("=" * 60)
     lines.append("")
 
@@ -1250,7 +1263,8 @@ def _build_revision_prompt(
     company_data: str,
     current_draft: str,
     is_financial: bool,
-    extracted: dict = None
+    extracted: dict = None,
+    data_gaps: list = None
 ) -> str:
     """Build revision prompt with conditional focus areas based on failed criteria.
 
@@ -1267,7 +1281,7 @@ def _build_revision_prompt(
     # Generate metric reference table for revision (same as initial mode)
     reference_table = ""
     if extracted:
-        reference_table, _ = _generate_metric_reference_table(extracted, is_financial)
+        reference_table, _ = _generate_metric_reference_table(extracted, is_financial, data_gaps=data_gaps)
     scores = critique_details.get("scores", {})
 
     # Determine which focus areas to include based on failed criteria
@@ -1406,7 +1420,8 @@ Simply output the improved SWOT as a clean, final deliverable."""
 
 
 def _build_analyzer_prompt(company: str, ticker: str, formatted_data: str,
-                           is_financial: bool, extracted: dict = None) -> tuple:
+                           is_financial: bool, extracted: dict = None,
+                           data_gaps: list = None) -> tuple:
     """Build analyzer prompt with metric reference table for hallucination prevention.
 
     Args:
@@ -1425,7 +1440,8 @@ def _build_analyzer_prompt(company: str, ticker: str, formatted_data: str,
     ref_hash = ""
 
     if extracted:
-        reference_table, metric_lookup = _generate_metric_reference_table(extracted, is_financial)
+        reference_table, metric_lookup = _generate_metric_reference_table(
+            extracted, is_financial, data_gaps=data_gaps)
         ref_hash = _compute_reference_hash(metric_lookup)
 
     if is_financial:
@@ -1509,6 +1525,12 @@ def analyzer_node(state, workflow_id=None, progress_store=None):
 
     # Extract and format metrics for better LLM understanding
     extracted = _extract_key_metrics(raw)
+
+    # Data gate (SC): drop values the researcher flagged as implausible so
+    # they can never enter the reference table or be cited
+    from src.utils.data_gate import scrub_suspect_metrics
+    extracted = scrub_suspect_metrics(extracted, state.get("suspect_metrics") or [])
+
     formatted_data = _format_metrics_for_prompt(extracted, is_financial=is_financial)
 
     # Generate detailed data report (shown before SWOT)
@@ -1533,7 +1555,8 @@ def analyzer_node(state, workflow_id=None, progress_store=None):
             company_data=formatted_data,
             current_draft=state.get("draft_report", ""),
             is_financial=is_financial,
-            extracted=extracted
+            extracted=extracted,
+            data_gaps=state.get("data_gaps")
         )
 
         # Update progress with revision info
@@ -1547,7 +1570,8 @@ def analyzer_node(state, workflow_id=None, progress_store=None):
         _add_activity_log(workflow_id, progress_store, "analyzer",
                           f"Calling LLM to generate SWOT analysis...")
         prompt, metric_lookup, ref_hash = _build_analyzer_prompt(
-            company, ticker, formatted_data, is_financial, extracted
+            company, ticker, formatted_data, is_financial, extracted,
+            data_gaps=state.get("data_gaps")
         )
         # Store metric reference for validation (Layer 1 hallucination prevention)
         state["metric_reference"] = metric_lookup

src/nodes/researcher.py

@@ -153,6 +153,23 @@ def researcher_node(state, workflow_id=None, progress_store=None):
             state["raw_data"] = json.dumps(result, indent=2, default=str)
             state["sources_failed"] = sources_failed
 
+            # Deterministic data gate on the exact view the Analyzer consumes:
+            # metric-level gaps (DG) and impossible magnitudes (SC)
+            try:
+                from src.nodes.analyzer import _extract_key_metrics
+                from src.utils.data_gate import audit_extracted_metrics
+                audit = audit_extracted_metrics(_extract_key_metrics(state["raw_data"]))
+                state["data_gaps"] = audit["gaps"]
+                state["suspect_metrics"] = audit["suspect"]
+                if audit["gaps"]:
+                    add_log("researcher",
+                            f"Data gaps (will be marked DATA NOT PROVIDED): {', '.join(audit['gaps'])}")
+                for cat, key, value in audit["suspect"]:
+                    add_log("researcher",
+                            f"Discarded implausible value: {cat}.{key} = {value:g} (outside sanity bounds)")
+            except Exception as gate_err:
+                print(f"[data-gate] audit skipped: {gate_err}")
+
             print(f"  - Sources available: {result['sources_available']}")
             if sources_failed:
                 print(f"  - Sources failed: {sources_failed}")

src/utils/data_gate.py

@@ -0,0 +1,105 @@
+"""
+Deterministic data gate for the Researcher -> Analyzer edge.
+
+Two checks, run on the same extracted-metrics view the Analyzer consumes:
+
+- DG (data gap): required metrics per category. Gaps are surfaced to the
+  Analyzer as explicit DATA NOT PROVIDED entries instead of silent omission,
+  giving the Critic's constraint-compliance rule something to enforce.
+- SC (signal corruption): impossible magnitudes (unit/decimal slips upstream)
+  are quarantined before they can enter the reference table and be cited.
+
+Bounds are deliberately loose - they reject the impossible, not the unusual.
+"""
+
+# Per-category requirement rules: (metric keys, minimum count present)
+REQUIRED_METRICS = {
+    "fundamentals": (["revenue", "net_margin", "eps"], 2),
+    "valuation": (["pe_trailing", "pe_forward", "pb_ratio", "ps_ratio"], 1),
+    "volatility": (["beta", "vix", "historical_volatility"], 1),
+    "macro": (["gdp_growth", "interest_rate", "inflation", "unemployment"], 2),
+}
+
+# (min, max) inclusive. Values are in the units the extractor produces
+# (margins/rates/vol in percent, ratios as multiples, currency in dollars).
+SANITY_BOUNDS = {
+    "revenue": (0, 1e13),
+    "net_income": (-1e12, 1e12),
+    "free_cash_flow": (-1e12, 1e12),
+    "net_margin": (-200, 100),
+    "gross_margin": (-200, 100),
+    "operating_margin": (-200, 100),
+    "eps": (-10000, 10000),
+    "debt_to_equity": (-100, 100),
+    "revenue_cagr_3yr": (-100, 300),
+    "pe_trailing": (-1000, 1000),
+    "pe_forward": (-1000, 1000),
+    "pb_ratio": (-100, 500),
+    "ps_ratio": (0, 500),
+    "ev_ebitda": (-1000, 1000),
+    "beta": (-5, 10),
+    "vix": (5, 150),
+    "historical_volatility": (0, 500),
+    "gdp_growth": (-30, 30),
+    "interest_rate": (-5, 50),
+    "inflation": (-20, 100),
+    "unemployment": (0, 50),
+}
+
+
+def _numeric_value(metric_val):
+    """Extracted metrics are either numbers or {'value': number, ...} dicts."""
+    if isinstance(metric_val, dict):
+        metric_val = metric_val.get("value")
+    if isinstance(metric_val, (int, float)) and not isinstance(metric_val, bool):
+        return float(metric_val)
+    return None
+
+
+def audit_extracted_metrics(extracted: dict) -> dict:
+    """
+    Audit the extracted metrics view.
+
+    Args:
+        extracted: dict from analyzer._extract_key_metrics -
+                   {"fundamentals": {...}, "valuation": {...}, ...}
+
+    Returns:
+        {
+          "gaps": ["fundamentals: revenue", ...]        # DG findings
+          "suspect": [("volatility", "vix", 1673.0), ...]  # SC findings
+        }
+    """
+    gaps = []
+    suspect = []
+
+    for category, (keys, min_present) in REQUIRED_METRICS.items():
+        data = extracted.get(category) or {}
+        present = [k for k in keys if _numeric_value(data.get(k)) is not None]
+        if len(present) < min_present:
+            for k in keys:
+                if k not in present:
+                    gaps.append(f"{category}: {k}")
+
+    for category in REQUIRED_METRICS:
+        data = extracted.get(category) or {}
+        if not isinstance(data, dict):
+            continue
+        for key, raw in data.items():
+            value = _numeric_value(raw)
+            bounds = SANITY_BOUNDS.get(key)
+            if value is None or bounds is None:
+                continue
+            lo, hi = bounds
+            if not (lo <= value <= hi):
+                suspect.append((category, key, value))
+
+    return {"gaps": gaps, "suspect": suspect}
+
+
+def scrub_suspect_metrics(extracted: dict, suspect: list) -> dict:
+    """Remove quarantined values so they cannot enter the reference table."""
+    for category, key, _value in suspect:
+        if isinstance(extracted.get(category), dict):
+            extracted[category].pop(key, None)
+    return extracted

tests/test_data_gate.py

@@ -0,0 +1,66 @@
+"""Tests for the deterministic Researcher -> Analyzer data gate."""
+
+from src.utils.data_gate import audit_extracted_metrics, scrub_suspect_metrics
+
+
+def _healthy():
+    return {
+        "fundamentals": {"revenue": 89.5e9, "net_margin": 2.5, "eps": 2.49},
+        "valuation": {"pe_trailing": {"value": 24.1}},
+        "volatility": {"beta": 1.15, "vix": 16.73},
+        "macro": {"gdp_growth": 2.1, "interest_rate": 3.63},
+    }
+
+
+def test_healthy_data_passes_clean():
+    audit = audit_extracted_metrics(_healthy())
+    assert audit["gaps"] == []
+    assert audit["suspect"] == []
+
+
+def test_empty_valuation_is_a_gap():
+    # The MSFT run shipped an empty valuation basket with no flag anywhere
+    data = _healthy()
+    data["valuation"] = {}
+    audit = audit_extracted_metrics(data)
+    assert any(g.startswith("valuation:") for g in audit["gaps"])
+
+
+def test_missing_fundamentals_below_minimum():
+    data = _healthy()
+    data["fundamentals"] = {"eps": 2.49}  # only 1 of required 2
+    audit = audit_extracted_metrics(data)
+    assert "fundamentals: revenue" in audit["gaps"]
+    assert "fundamentals: net_margin" in audit["gaps"]
+
+
+def test_impossible_magnitude_is_quarantined():
+    data = _healthy()
+    data["fundamentals"]["net_margin"] = 1218.0  # decimal-slip: 12.18 -> 1218
+    audit = audit_extracted_metrics(data)
+    assert ("fundamentals", "net_margin", 1218.0) in audit["suspect"]
+
+
+def test_unusual_but_possible_values_pass():
+    # Boeing's real D/E of 9.92 and hist vol of 34% must NOT be flagged
+    data = _healthy()
+    data["fundamentals"]["debt_to_equity"] = 9.92
+    data["volatility"]["historical_volatility"] = 34.13
+    audit = audit_extracted_metrics(data)
+    assert audit["suspect"] == []
+
+
+def test_scrub_removes_only_quarantined_values():
+    data = _healthy()
+    data["volatility"]["vix"] = 1673.0
+    audit = audit_extracted_metrics(data)
+    scrubbed = scrub_suspect_metrics(data, audit["suspect"])
+    assert "vix" not in scrubbed["volatility"]
+    assert scrubbed["volatility"]["beta"] == 1.15
+
+
+def test_dict_wrapped_values_audited():
+    data = _healthy()
+    data["fundamentals"]["net_margin"] = {"value": 1218.0, "end_date": "2025-12-31"}
+    audit = audit_extracted_metrics(data)
+    assert ("fundamentals", "net_margin", 1218.0) in audit["suspect"]