test: add SubtextRead + _detect_trend tests

Adds 19 new test cases covering SubtextRead signal_type coercion (all 5
canonical values, 6 aliases, unknown fallback, uppercase fix), BriefOutput
between_the_lines field (empty default + round-trip), and _detect_trend
pure function (rising/falling/flat/short/empty edge cases).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

tests/test_schemas.py

@@ -3,7 +3,9 @@ from pydantic import ValidationError
 from agent.schemas import (
     SourcedFact, BriefOutput, TrendPoint, ManagementCommentaryTopic,
     MDASection, CategorizedRisk, GuidancePoint,
+    EarningsQualitySignal, AnalyticalTension, SectionSentiment, SubtextRead,
 )
+from analysis.textdiff import _detect_trend
 
 
 def test_sourced_fact_valid():
@@ -155,6 +157,37 @@ def test_brief_output_has_management_commentary_field():
     assert "transcript_topics" not in BriefOutput.model_fields
 
 
+def test_brief_output_drops_news_commentary_mixed():
+    """BriefOutput drops news-sourced items but keeps valid ones (filing/transcript)."""
+    brief = _minimal_brief(
+        management_commentary=[
+            {"topic": "AI strategy", "summary": "Management outlined AI plans.", "source": "transcript",
+             "reliability": "MEDIUM", "evidence_snippet": "We are fully committed to AI integration."},
+            {"topic": "Breaking news", "summary": "CNBC reported earnings beat.", "source": "news",
+             "reliability": "LOW", "evidence_snippet": "CNBC reported strong results."},
+            {"topic": "Revenue guidance", "summary": "Guided 3-5% growth.", "source": "10-Q",
+             "reliability": "HIGH", "evidence_snippet": "We expect revenue growth of 3-5%."},
+        ]
+    )
+    assert len(brief.management_commentary) == 2
+    topics = [t.topic for t in brief.management_commentary]
+    assert "AI strategy" in topics
+    assert "Revenue guidance" in topics
+    assert "Breaking news" not in topics
+
+
+def test_brief_output_all_news_commentary_yields_empty_list():
+    """BriefOutput validates even when all management_commentary items are news-sourced (MSFT regression)."""
+    brief = _minimal_brief(
+        management_commentary=[
+            {"topic": f"Item {i}", "summary": "News summary.", "source": "news",
+             "reliability": "LOW", "evidence_snippet": f"News snippet {i}."}
+            for i in range(5)
+        ]
+    )
+    assert brief.management_commentary == []
+
+
 # ── CategorizedRisk category normalization ───────────────────────────────────
 
 def _risk(**kw) -> CategorizedRisk:
@@ -310,3 +343,373 @@ def test_brief_output_tolerates_compound_source_in_analytical_tension():
     brief = _minimal_brief(analytical_tensions=[tension])
     assert brief.analytical_tensions[0].bullish_evidence.source == "10-Q"
     assert brief.analytical_tensions[0].bearish_evidence.source == "10-Q"
+
+
+# ── EarningsQualitySignal.assessment ─────────────────────────────────────────
+
+def _eqs(**kw) -> EarningsQualitySignal:
+    defaults = dict(
+        dimension="guidance_dynamics",
+        assessment="positive",
+        rationale="Guidance was raised.",
+        evidence=SourcedFact(text="x", source="10-Q", reliability="HIGH", evidence_snippet="x"),
+    )
+    return EarningsQualitySignal(**(defaults | kw))
+
+
+def test_eqs_assessment_valid_passthrough():
+    assert _eqs(assessment="concerning").assessment == "concerning"
+    assert _eqs(assessment="positive").assessment == "positive"
+    assert _eqs(assessment="neutral").assessment == "neutral"
+
+
+def test_eqs_assessment_coerces_negative_to_concerning(capsys):
+    s = _eqs(assessment="negative")
+    assert s.assessment == "concerning"
+    assert "[quality-assessment]" in capsys.readouterr().err
+
+
+def test_eqs_assessment_coerces_bearish_to_concerning():
+    assert _eqs(assessment="bearish").assessment == "concerning"
+
+
+def test_eqs_assessment_coerces_bullish_to_positive():
+    assert _eqs(assessment="bullish").assessment == "positive"
+
+
+def test_eqs_assessment_coerces_mixed_to_neutral():
+    assert _eqs(assessment="mixed").assessment == "neutral"
+
+
+def test_eqs_assessment_unknown_falls_back_to_neutral(capsys):
+    s = _eqs(assessment="unclear")
+    assert s.assessment == "neutral"
+    assert "[quality-assessment]" in capsys.readouterr().err
+
+
+# ── EarningsQualitySignal.dimension ──────────────────────────────────────────
+
+def test_eqs_dimension_valid_passthrough():
+    assert _eqs(dimension="consensus_beat_mix").dimension == "consensus_beat_mix"
+    assert _eqs(dimension="capital_allocation").dimension == "capital_allocation"
+
+
+def test_eqs_dimension_coerces_guidance_synonym(capsys):
+    s = _eqs(dimension="guidance")
+    assert s.dimension == "guidance_dynamics"
+    assert "[quality-dimension]" in capsys.readouterr().err
+
+
+def test_eqs_dimension_coerces_beat_mix_synonym():
+    assert _eqs(dimension="beat_mix").dimension == "consensus_beat_mix"
+
+
+def test_eqs_dimension_coerces_narrative_synonym():
+    assert _eqs(dimension="tone").dimension == "narrative_vs_numbers"
+
+
+def test_eqs_dimension_coerces_segment_synonym():
+    assert _eqs(dimension="segment").dimension == "segment_mix"
+
+
+def test_eqs_dimension_coerces_capital_synonym():
+    assert _eqs(dimension="capex").dimension == "capital_allocation"
+
+
+def test_eqs_dimension_unknown_still_raises():
+    """Unknown dimension passes through _normalize_dimension unchanged → Literal rejects it."""
+    with pytest.raises(ValidationError):
+        _eqs(dimension="completely_unknown_dim")
+
+
+# ── BriefOutput._sanitize_quality_signals ────────────────────────────────────
+
+def test_brief_output_drops_unknown_dimension_keeps_valid(capsys):
+    """Unknown-dimension item is dropped at brief level; valid item is kept."""
+    sf = {"text": "x", "source": "10-Q", "reliability": "HIGH", "evidence_snippet": "x"}
+    brief = _minimal_brief(
+        earnings_quality_signals=[
+            {"dimension": "totally_unknown", "assessment": "positive", "rationale": "r", "evidence": sf},
+            {"dimension": "guidance_dynamics", "assessment": "neutral", "rationale": "r", "evidence": sf},
+        ]
+    )
+    assert len(brief.earnings_quality_signals) == 1
+    assert brief.earnings_quality_signals[0].dimension == "guidance_dynamics"
+    assert "[quality-dimension]" in capsys.readouterr().err
+
+
+def test_brief_output_all_unknown_dimension_yields_empty_list():
+    """All items with unknown dimension → empty list, brief still validates."""
+    sf = {"text": "x", "source": "10-Q", "reliability": "HIGH", "evidence_snippet": "x"}
+    brief = _minimal_brief(
+        earnings_quality_signals=[
+            {"dimension": "foo", "assessment": "positive", "rationale": "r", "evidence": sf},
+            {"dimension": "bar", "assessment": "neutral", "rationale": "r", "evidence": sf},
+        ]
+    )
+    assert brief.earnings_quality_signals == []
+
+
+# ── AnalyticalTension.weight ──────────────────────────────────────────────────
+
+def _tension(**kw) -> AnalyticalTension:
+    sf = SourcedFact(text="x", source="10-Q", reliability="HIGH", evidence_snippet="x")
+    defaults = dict(
+        headline="H", bullish_reading="B", bearish_reading="Be",
+        weight="watch", bullish_evidence=sf, bearish_evidence=sf,
+    )
+    return AnalyticalTension(**(defaults | kw))
+
+
+def test_tension_weight_valid_passthrough():
+    assert _tension(weight="material").weight == "material"
+    assert _tension(weight="watch").weight == "watch"
+    assert _tension(weight="minor").weight == "minor"
+
+
+def test_tension_weight_coerces_high_to_material(capsys):
+    t = _tension(weight="high")
+    assert t.weight == "material"
+    assert "[tension-weight]" in capsys.readouterr().err
+
+
+def test_tension_weight_coerces_medium_to_watch():
+    assert _tension(weight="medium").weight == "watch"
+
+
+def test_tension_weight_coerces_low_to_minor():
+    assert _tension(weight="low").weight == "minor"
+
+
+def test_tension_weight_coerces_critical_to_material():
+    assert _tension(weight="critical").weight == "material"
+
+
+def test_tension_weight_unknown_falls_back_to_watch(capsys):
+    t = _tension(weight="extreme")
+    assert t.weight == "watch"
+    assert "[tension-weight]" in capsys.readouterr().err
+
+
+# ── GuidancePoint.verdict ─────────────────────────────────────────────────────
+
+def _gp(**kw) -> GuidancePoint:
+    defaults = dict(period="Q1 2025", text="Revenue guided flat.", source="10-Q")
+    return GuidancePoint(**(defaults | kw))
+
+
+def test_guidance_verdict_valid_passthrough():
+    assert _gp(verdict="beat").verdict == "beat"
+    assert _gp(verdict="in-line").verdict == "in-line"
+    assert _gp(verdict="missed").verdict == "missed"
+    assert _gp(verdict="pending").verdict == "pending"
+
+
+def test_guidance_verdict_none_passthrough():
+    assert _gp(verdict=None).verdict is None
+
+
+def test_guidance_verdict_coerces_miss_to_missed(capsys):
+    g = _gp(verdict="miss")
+    assert g.verdict == "missed"
+    assert "[guidance-verdict]" in capsys.readouterr().err
+
+
+def test_guidance_verdict_coerces_inline_to_in_line():
+    assert _gp(verdict="inline").verdict == "in-line"
+
+
+def test_guidance_verdict_coerces_met_to_in_line():
+    assert _gp(verdict="met").verdict == "in-line"
+
+
+def test_guidance_verdict_coerces_above_to_beat():
+    assert _gp(verdict="above").verdict == "beat"
+
+
+def test_guidance_verdict_unknown_becomes_none(capsys):
+    g = _gp(verdict="partial")
+    assert g.verdict is None
+    assert "[guidance-verdict]" in capsys.readouterr().err
+
+
+# ── SectionSentiment.score ────────────────────────────────────────────────────
+
+def _ss(**kw) -> SectionSentiment:
+    defaults = dict(rationale="Revenue beat.")
+    return SectionSentiment(**(defaults | kw))
+
+
+def test_sentiment_score_valid_passthrough():
+    for s in (-2, -1, 0, 1, 2):
+        assert _ss(score=s).score == s
+
+
+def test_sentiment_score_none_passthrough():
+    assert _ss(score=None).score is None
+
+
+def test_sentiment_score_coerces_string_int(capsys):
+    s = _ss(score="1")
+    assert s.score == 1
+    assert "[sentiment-score]" in capsys.readouterr().err
+
+
+def test_sentiment_score_coerces_positive_string():
+    assert _ss(score="+2").score == 2
+
+
+def test_sentiment_score_coerces_float():
+    assert _ss(score=1.0).score == 1
+
+
+def test_sentiment_score_clamps_out_of_range(capsys):
+    s = _ss(score=3)
+    assert s.score == 2
+    assert "[sentiment-score]" in capsys.readouterr().err
+
+
+def test_sentiment_score_non_numeric_becomes_none(capsys):
+    s = _ss(score="abc")
+    assert s.score is None
+    assert "[sentiment-score]" in capsys.readouterr().err
+
+
+# ── SectionSentiment.label ────────────────────────────────────────────────────
+
+def test_sentiment_label_valid_passthrough():
+    for lbl in ("Strongly Bearish", "Bearish", "Neutral", "Bullish", "Strongly Bullish"):
+        assert _ss(label=lbl).label == lbl
+
+
+def test_sentiment_label_none_passthrough():
+    assert _ss(label=None).label is None
+
+
+def test_sentiment_label_coerces_positive_to_bullish(capsys):
+    s = _ss(label="positive")
+    assert s.label == "Bullish"
+    assert "[sentiment-label]" in capsys.readouterr().err
+
+
+def test_sentiment_label_coerces_negative_to_bearish():
+    assert _ss(label="negative").label == "Bearish"
+
+
+def test_sentiment_label_unknown_becomes_none(capsys):
+    s = _ss(label="meh")
+    assert s.label is None
+    assert "[sentiment-label]" in capsys.readouterr().err
+
+
+# ── SubtextRead ──────────────────────────────────────────────────────────────
+
+def _str(signal_type="language_drift", **kw) -> SubtextRead:
+    sf = SourcedFact(text="x", source="10-Q", reliability="HIGH", evidence_snippet="x")
+    defaults = dict(
+        observation="Management used 'we expect growth' instead of 'we expect strong growth'.",
+        reading="The softening qualifier signals reduced conviction in guidance.",
+        signal_type=signal_type,
+        implication="Watch for a guidance cut next quarter if this trend continues.",
+        evidence=sf,
+    )
+    defaults.update(kw)
+    return SubtextRead(**defaults)
+
+
+def test_subtext_read_valid_signal_types():
+    for st in ("language_drift", "qa_evasion", "omission", "emphasis_shift", "accounting_quality"):
+        assert _str(signal_type=st).signal_type == st
+
+
+def test_subtext_read_coerces_evasion_to_qa_evasion(capsys):
+    s = _str(signal_type="evasion")
+    assert s.signal_type == "qa_evasion"
+    assert "[signal-type]" in capsys.readouterr().err
+
+
+def test_subtext_read_coerces_drift_to_language_drift():
+    assert _str(signal_type="drift").signal_type == "language_drift"
+
+
+def test_subtext_read_coerces_kpi_dropped_to_emphasis_shift():
+    assert _str(signal_type="kpi_dropped").signal_type == "emphasis_shift"
+
+
+def test_subtext_read_coerces_silence_to_omission():
+    assert _str(signal_type="silence").signal_type == "omission"
+
+
+def test_subtext_read_coerces_accounting_quality_alias():
+    assert _str(signal_type="earnings_quality").signal_type == "accounting_quality"
+
+
+def test_subtext_read_unknown_falls_back_to_language_drift(capsys):
+    s = _str(signal_type="completely_unknown_signal")
+    assert s.signal_type == "language_drift"
+    assert "[signal-type]" in capsys.readouterr().err
+
+
+def test_subtext_read_uppercase_canonical_coerced(capsys):
+    s = _str(signal_type="Accounting_Quality")
+    assert s.signal_type == "accounting_quality"
+    # The .lower() fix means the canonical is recognised directly — no stderr expected
+    capsys.readouterr()  # consume any output
+
+
+def test_subtext_read_evidence_is_sourced_fact():
+    s = _str()
+    assert isinstance(s.evidence, SourcedFact)
+
+
+def test_brief_output_between_the_lines_empty_by_default():
+    brief = _minimal_brief()
+    assert brief.between_the_lines == []
+
+
+def test_brief_output_accepts_between_the_lines_list():
+    sf = {"text": "x", "source": "10-Q", "reliability": "HIGH", "evidence_snippet": "x"}
+    item = {
+        "observation": "Analyst asked about China; management pivoted.",
+        "reading": "Evasion signals China pricing under pressure.",
+        "signal_type": "qa_evasion",
+        "implication": "Watch for China revenue disclosure next quarter.",
+        "evidence": sf,
+    }
+    brief = _minimal_brief(between_the_lines=[item])
+    assert len(brief.between_the_lines) == 1
+    assert brief.between_the_lines[0].signal_type == "qa_evasion"
+
+
+# ── _detect_trend ────────────────────────────────────────────────────────────
+
+def test_detect_trend_rising_4():
+    assert _detect_trend([1, 3, 5, 8]) == "rising 4 quarters"
+
+
+def test_detect_trend_falling_4():
+    assert _detect_trend([8, 5, 3, 1]) == "falling 4 quarters"
+
+
+def test_detect_trend_rising_3_at_tail():
+    assert _detect_trend([1, 5, 2, 4, 6]) == "rising 3 quarters"
+
+
+def test_detect_trend_flat_returns_none():
+    assert _detect_trend([1, 2, 2, 4]) is None
+
+
+def test_detect_trend_too_short_returns_none():
+    assert _detect_trend([1, 3]) is None
+
+
+def test_detect_trend_empty_returns_none():
+    assert _detect_trend([]) is None
+
+
+def test_detect_trend_single_value_returns_none():
+    assert _detect_trend([5]) is None
+
+
+def test_detect_trend_falling_3_at_tail():
+    assert _detect_trend([10, 2, 8, 6, 4]) == "falling 3 quarters"