fix: per-item drop of invalid brief facts; relax LLM-produced singleton fields

Generalizes the fail-closed-per-fact principle to Pydantic validation
itself, closing the whole class of bugs behind this incident: a single
malformed item produced by the LLM (mis-cased source, malformed risk,
business line missing a description) used to crash
BriefOutput.model_validate or CompanyProfileSection.model_validate
outright, discarding the entire brief or the entire overview section.

A _drop_invalid_list_items helper (generalizing the existing
_sanitize_quality_signals pattern) validates each list item on its own
in a model_validator(mode="before") and quarantines only the failures,
with a stderr log. Invalid singletons fall back instead of raising:
standout_number / sentiment / market_expectations -> None, mda_summary
-> defaults, status -> PARTIAL. Same treatment on the company profile
side, plus revenue_share_pct normalization (non-numeric or out of
[0,100] -> None) and a direction fallback to "insufficient".

Deliberately excluded: management_commentary and
earnings_quality_signals, which already have dedicated field
sanitizers with their own log-tag tests. EvidenceRecord stays strict —
it is built by the tools, not copied by the model.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

agent/company_profile_schemas.py

@@ -4,9 +4,9 @@ from __future__ import annotations
 import sys
 from typing import Any, Literal, Optional
 
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
-from agent.schemas import SourcedFact
+from agent.schemas import SourcedFact, _drop_invalid_list_items
 
 
 _CANONICAL_TRENDS = {"growing", "stable", "declining", "mixed", "not_disclosed"}
@@ -35,6 +35,14 @@ _CANONICAL_EXPOSURE_TYPES = {
     "geopolitical",
 }
 
+_CANONICAL_ATTENTION_DIRECTIONS = {
+    "rising",
+    "stable",
+    "falling",
+    "new",
+    "insufficient",
+}
+
 _EXPOSURE_TYPE_ALIASES: dict[str, str] = {
     "regulatory": "regulation",
     "legal": "regulation",
@@ -88,6 +96,23 @@ class BusinessLine(BaseModel):
         )
         return "not_disclosed"
 
+    @field_validator("revenue_share_pct", mode="before")
+    @classmethod
+    def _normalize_revenue_share_pct(cls, v: object) -> object:
+        if v is None:
+            return None
+        try:
+            numeric = float(v)
+        except (TypeError, ValueError):
+            numeric = None
+        if numeric is not None and 0 <= numeric <= 100:
+            return numeric
+        print(
+            f"[revenue-share] invalid {cls.__name__} revenue_share_pct '{v}', falling back to None",
+            file=sys.stderr,
+        )
+        return None
+
 
 class GeographicExposure(BaseModel):
     model_config = ConfigDict(extra="ignore")
@@ -127,6 +152,23 @@ class GeographicExposure(BaseModel):
                 seen.add(normalized)
         return normalized_items
 
+    @field_validator("revenue_share_pct", mode="before")
+    @classmethod
+    def _normalize_revenue_share_pct(cls, v: object) -> object:
+        if v is None:
+            return None
+        try:
+            numeric = float(v)
+        except (TypeError, ValueError):
+            numeric = None
+        if numeric is not None and 0 <= numeric <= 100:
+            return numeric
+        print(
+            f"[revenue-share] invalid {cls.__name__} revenue_share_pct '{v}', falling back to None",
+            file=sys.stderr,
+        )
+        return None
+
 
 class StrategicChange(BaseModel):
     model_config = ConfigDict(extra="ignore")
@@ -157,6 +199,19 @@ class AttentionTheme(BaseModel):
     retrieved_news_count: int = Field(default=0, ge=0)
     direction: Literal["rising", "stable", "falling", "new", "insufficient"] = "insufficient"
 
+    @field_validator("direction", mode="before")
+    @classmethod
+    def _normalize_direction(cls, v: object) -> str:
+        if isinstance(v, str):
+            normalized = "_".join(v.strip().lower().replace("-", " ").split())
+            if normalized in _CANONICAL_ATTENTION_DIRECTIONS:
+                return normalized
+        print(
+            f"[attention-direction] unknown direction '{v}', falling back to 'insufficient'",
+            file=sys.stderr,
+        )
+        return "insufficient"
+
 
 class WatchVariable(BaseModel):
     model_config = ConfigDict(extra="ignore")
@@ -180,6 +235,40 @@ class CompanyProfileSection(BaseModel):
     attention_themes: list[AttentionTheme] = Field(default_factory=list)
     watch_variables: list[WatchVariable] = Field(default_factory=list)
 
+    @model_validator(mode="before")
+    @classmethod
+    def _sanitize_llm_fields(cls, data: object) -> object:
+        if not isinstance(data, dict):
+            return data
+
+        sanitized = dict(data)
+        list_fields: dict[str, type[BaseModel]] = {
+            "business_lines": BusinessLine,
+            "geographic_exposures": GeographicExposure,
+            "strategic_changes": StrategicChange,
+            "attention_themes": AttentionTheme,
+            "watch_variables": WatchVariable,
+        }
+        for field_name, model_cls in list_fields.items():
+            if field_name in sanitized:
+                sanitized[field_name] = _drop_invalid_list_items(
+                    sanitized[field_name], model_cls, field_name
+                )
+
+        if "identity" in sanitized:
+            try:
+                sanitized["identity"] = CompanyIdentity.model_validate(
+                    sanitized["identity"]
+                )
+            except Exception as exc:
+                first_line = str(exc).splitlines()[0] if str(exc) else type(exc).__name__
+                print(
+                    f"[brief-sanitizer] replacing invalid identity: {first_line}",
+                    file=sys.stderr,
+                )
+                sanitized["identity"] = {}
+        return sanitized
+
 
 class AnnualTrend(BaseModel):
     model_config = ConfigDict(extra="ignore")

agent/schemas.py

@@ -407,6 +407,27 @@ def _normalize_signal_type(v: object) -> object:
     return "language_drift"
 
 
+def _drop_invalid_list_items(
+    items: object,
+    model_cls: type[BaseModel],
+    field_name: str,
+) -> object:
+    """Validate LLM-produced list items independently and quarantine failures."""
+    if not isinstance(items, list):
+        return items
+    kept = []
+    for item in items:
+        try:
+            kept.append(model_cls.model_validate(item))
+        except Exception as exc:
+            first_line = str(exc).splitlines()[0] if str(exc) else type(exc).__name__
+            print(
+                f"[brief-sanitizer] dropping invalid {field_name} item: {first_line}",
+                file=sys.stderr,
+            )
+    return kept
+
+
 class SourcedFact(BaseModel):
     model_config = ConfigDict(extra="ignore")
 
@@ -463,18 +484,35 @@ class MDASection(BaseModel):
     model_config = ConfigDict(extra="ignore")
 
     drivers: list[SourcedFact] = Field(
+        default_factory=list,
         description="2-4 key revenue or margin drivers cited in the MD&A or transcript."
     )
     headwinds: list[SourcedFact] = Field(
+        default_factory=list,
         description="1-3 headwinds or drags on performance cited in the MD&A or transcript."
     )
     language_shift: str = Field(
+        default="",
         description="1-2 sentences on how management language evolved vs prior periods: more confident, more cautious, more defensive? If cross-period data is unavailable, state so explicitly."
     )
-    key_quote: SourcedFact = Field(
+    key_quote: Optional[SourcedFact] = Field(
+        default=None,
         description="The single most revealing management statement from the filing or transcript this period."
     )
 
+    @model_validator(mode="before")
+    @classmethod
+    def _drop_invalid_fact_items(cls, data: object) -> object:
+        if not isinstance(data, dict):
+            return data
+        sanitized = dict(data)
+        for field_name in ("drivers", "headwinds"):
+            if field_name in sanitized:
+                sanitized[field_name] = _drop_invalid_list_items(
+                    sanitized[field_name], SourcedFact, field_name
+                )
+        return sanitized
+
 
 class CategorizedRisk(BaseModel):
     model_config = ConfigDict(extra="ignore")
@@ -819,7 +857,8 @@ class BriefOutput(BaseModel):
         description="2-5 earnings quality signals across distinct dimensions. Use evidence already retrieved."
     )
 
-    standout_number: SourcedFact = Field(
+    standout_number: Optional[SourcedFact] = Field(
+        default=None,
         description="The single most remarkable quantitative fact this quarter — the number a journalist would lead with."
     )
 
@@ -843,7 +882,10 @@ class BriefOutput(BaseModel):
         description="Deterministic verification coverage populated after synthesis.",
     )
 
-    mda_summary: MDASection = Field(description="Structured MD&A analysis from the 10-K/10-Q.")
+    mda_summary: MDASection = Field(
+        default_factory=MDASection,
+        description="Structured MD&A analysis from the 10-K/10-Q.",
+    )
     risks_categorized: list[CategorizedRisk] = Field(description="3-6 categorized risks from the filing.")
     management_commentary: list[ManagementCommentaryTopic] = Field(
         description="3-5 key management themes drawn from MD&A (preferred) or earnings call transcript."
@@ -868,6 +910,64 @@ class BriefOutput(BaseModel):
         description="Verbatim text deltas computed deterministically across consecutive filing periods. Populated by code, not LLM.",
     )
 
+    @model_validator(mode="before")
+    @classmethod
+    def _sanitize_llm_fields(cls, data: object) -> object:
+        if not isinstance(data, dict):
+            return data
+
+        sanitized = dict(data)
+        list_fields: dict[str, type[BaseModel]] = {
+            "what_changed": SourcedFact,
+            "bull_points": SourcedFact,
+            "bear_points": SourcedFact,
+            "risks_categorized": CategorizedRisk,
+            "guidance_history": GuidancePoint,
+            "analytical_tensions": AnalyticalTension,
+            "between_the_lines": SubtextRead,
+            "trends": TrendPoint,
+            "quarter_deltas": QuarterDelta,
+        }
+        for field_name, model_cls in list_fields.items():
+            if field_name in sanitized:
+                sanitized[field_name] = _drop_invalid_list_items(
+                    sanitized[field_name], model_cls, field_name
+                )
+
+        singleton_fields: tuple[tuple[str, type[BaseModel], object], ...] = (
+            ("standout_number", SourcedFact, None),
+            ("sentiment", SentimentScores, None),
+            ("market_expectations", MarketExpectations, None),
+            ("mda_summary", MDASection, {}),
+        )
+        for field_name, model_cls, fallback in singleton_fields:
+            if field_name not in sanitized:
+                continue
+            value = sanitized[field_name]
+            if value is None and fallback is None:
+                continue
+            try:
+                sanitized[field_name] = model_cls.model_validate(value)
+            except Exception as exc:
+                first_line = str(exc).splitlines()[0] if str(exc) else type(exc).__name__
+                print(
+                    f"[brief-sanitizer] replacing invalid {field_name}: {first_line}",
+                    file=sys.stderr,
+                )
+                sanitized[field_name] = fallback
+        return sanitized
+
+    @field_validator("status", mode="before")
+    @classmethod
+    def _normalize_status(cls, v: object) -> object:
+        if isinstance(v, str) and v in {"COMPLETE", "PARTIAL"}:
+            return v
+        print(
+            f"[brief-status] unknown status '{v}', falling back to 'PARTIAL'",
+            file=sys.stderr,
+        )
+        return "PARTIAL"
+
     @field_validator("earnings_quality_signals", mode="before")
     @classmethod
     def _sanitize_quality_signals(cls, v: object) -> object:

tests/test_company_profile.py

@@ -130,6 +130,64 @@ def test_business_line_trend_normalizes_synonyms_and_unknowns():
     ]
 
 
+def test_profile_out_of_range_revenue_shares_become_none(capsys):
+    section = CompanyProfileSection.model_validate(
+        {
+            "business_lines": [
+                {
+                    "name": "Services",
+                    "description": _fact(),
+                    "revenue_share_pct": 150,
+                }
+            ],
+            "geographic_exposures": [
+                {
+                    "name": "Americas",
+                    "description": _fact(),
+                    "revenue_share_pct": -1,
+                }
+            ],
+        }
+    )
+
+    assert section.business_lines[0].revenue_share_pct is None
+    assert section.geographic_exposures[0].revenue_share_pct is None
+    assert "[revenue-share]" in capsys.readouterr().err
+
+
+def test_profile_unknown_attention_direction_falls_back(capsys):
+    section = CompanyProfileSection.model_validate(
+        {
+            "attention_themes": [
+                {
+                    "theme": "AI",
+                    "why_it_matters": "Material investment cycle.",
+                    "evidence": _fact(),
+                    "direction": "unknown_direction",
+                }
+            ]
+        }
+    )
+
+    assert section.attention_themes[0].direction == "insufficient"
+    assert "[attention-direction]" in capsys.readouterr().err
+
+
+def test_profile_drops_only_invalid_business_line(capsys):
+    section = CompanyProfileSection.model_validate(
+        {
+            "business_lines": [
+                {"name": "Services", "description": _fact()},
+                {"name": "Malformed line without description"},
+                {"name": "Products", "description": _fact()},
+            ]
+        }
+    )
+
+    assert [line.name for line in section.business_lines] == ["Services", "Products"]
+    assert "[brief-sanitizer] dropping invalid business_lines item:" in capsys.readouterr().err
+
+
 def test_synthesis_prompt_enumerates_profile_literals():
     assert (
         "revenue, operations, supply_chain, regulation, geopolitical"

tests/test_schemas.py

@@ -131,6 +131,81 @@ def test_brief_output_tolerates_truncated_content_hash_in_evidence_ref():
     assert brief.risks_categorized[0].evidence_ref.content_hash == truncated_hash
 
 
+def test_brief_output_drops_only_invalid_bull_point(capsys):
+    payload = _minimal_brief().model_dump(mode="json")
+    payload["bull_points"] = [
+        {
+            "text": "Services revenue grew.",
+            "source": "10-Q",
+            "reliability": "HIGH",
+            "evidence_snippet": "Services revenue grew.",
+        },
+        {
+            "text": "Unusable composite source.",
+            "source": "bloomberg",
+            "reliability": "LOW",
+            "evidence_snippet": "Unusable composite source.",
+        },
+        {
+            "text": "Margins expanded.",
+            "source": "transcript",
+            "reliability": "MEDIUM",
+            "evidence_snippet": "Margins expanded.",
+        },
+    ]
+
+    brief = BriefOutput.model_validate(payload)
+
+    assert [item.text for item in brief.bull_points] == [
+        "Services revenue grew.",
+        "Margins expanded.",
+    ]
+    assert "[brief-sanitizer] dropping invalid bull_points item:" in capsys.readouterr().err
+
+
+def test_brief_output_invalid_standout_number_becomes_none():
+    payload = _minimal_brief().model_dump(mode="json")
+    payload["standout_number"] = {
+        "source": "10-Q",
+        "reliability": "HIGH",
+        "evidence_snippet": "Revenue grew.",
+    }
+
+    brief = BriefOutput.model_validate(payload)
+
+    assert brief.standout_number is None
+
+
+def test_brief_output_evidence_ref_ignores_extra_content_key():
+    payload = _minimal_brief().model_dump(mode="json")
+    payload["bull_points"][0]["evidence_ref"] = {
+        "evidence_id": "ev_abc123",
+        "source": "10-Q",
+        "content_hash": "short-hash-is-tolerated",
+        "document_id": "sec:AAPL:0001",
+        "content": "This belongs to an evidence record, not an evidence ref.",
+    }
+
+    brief = BriefOutput.model_validate(payload)
+
+    assert brief.bull_points[0].evidence_ref.evidence_id == "ev_abc123"
+    assert not hasattr(brief.bull_points[0].evidence_ref, "content")
+
+
+def test_brief_output_defaults_llm_singletons_and_status(capsys):
+    payload = _minimal_brief().model_dump(mode="json")
+    payload.pop("standout_number")
+    payload.pop("mda_summary")
+    payload["status"] = {"unexpected": "shape"}
+
+    brief = BriefOutput.model_validate(payload)
+
+    assert brief.standout_number is None
+    assert brief.mda_summary == MDASection()
+    assert brief.status == "PARTIAL"
+    assert "[brief-status]" in capsys.readouterr().err
+
+
 def test_brief_output_accepts_metrics_sources_for_risk_and_guidance():
     brief = _minimal_brief(
         risks_categorized=[dict(