"""SPGI relationship-layer remediation (forensic package SPGI_6901d42876e85a17): RelationshipClaim, typed break-even
and story metrics, model-evidence scope, guidance epistemic state, semantic returns and typed blend sources.
Corruption tests A–K each assert the named validator fails. Structure only: no SPGI rating or value is asserted.
"""
import copy
import os
import sys
import pytest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import catalog as C # noqa: E402
import evidence as EV # noqa: E402
import guidance as G # noqa: E402
import pipeline_fixtures as F # noqa: E402
import prose as P # noqa: E402
import relations as RL # noqa: E402
import test_spgi_remediation as T # noqa: E402 (SPGI-shaped data, story and guidance fixtures)
def _story_with_breakeven(bm=0.45, be_rev=31.5, be_margin=0.55):
d = T.spgi_data()
start = d["story_base"]["revenue"]
st = T.spgi_story(base=d["story_base"])
st["assumptions"].update(revenue_N_usd_b=25.2, tam_usd_b=85.0, years=10, margin_target=0.494)
st["breakeven"] = {"price": 403.3, "story_value": 333.11, "story_revenue_N_usd_b": 25.2, "story_margin": bm, "years": 10,
"revenue_needed_N_usd_b": be_rev, "rev_cagr_needed": round((be_rev * 1e9 / start) ** 0.1 - 1.0, 4),
"revenue_needed_reachable": True, "margin_needed": be_margin, "margin_needed_reachable": True}
return d, st
def spgi_run(llm=None, summary=None, panel_visible=False, data=None, story=None, panel_stats=True):
d, st = _story_with_breakeven()
d = data or d
st = story or st
pf_ = F.pred_fund(); pf_.update(revenue_annual=11.8e9, eps_annual=4.3)
over = {"pred_fund": pf_, "story": st}
import policy as pol
if panel_visible:
over["panel_rel"] = dict(pol.panel_reliability(None), state="LIMITED", persona_visibility=True, usable_for_argument=True,
rank_ic=0.05, ic_positive_share=0.62)
over["panel_row"] = {"decile": 8, "xs_rank_pct": 80.0, "pred_excess_12m": 0.05}
elif panel_stats:
# as in the live SPGI run: the rank model is UNRELIABLE and hidden, yet its statistics exist in the state
over["panel_rel"] = dict(pol.panel_reliability(None), state="UNRELIABLE", persona_visibility=False, rank_ic=0.0346,
ic_positive_share=0.688)
return F.run_pipeline(data=d, forecast_over=over, llm=llm, summary=summary)
@pytest.fixture(scope="module")
def good():
run, payload, failed = spgi_run()
assert failed is None, [(v.stage, c.id, c.detail) for v in run.validations for c in v.blocking]
return run, payload
def _rules(run):
cat = C.build_catalog(run.state, include_deferred=True)
return cat, run.rules(cat)
def _audit_ids(run, extra_text_frags):
"""Run the report audit on the run's report plus one authored (llm) sentence."""
import report_model as RM
cat = C.build_catalog(run.state, include_deferred=True)
rep = RM.build_user_report(run.state, cat, {"summary": [], "thesis": [], "risks": [], "seats": {}}, [])
rep.blocks.append(RM.Block("para", "summary", extra_text_frags))
a = RM.audit(rep, run.state, RM.render_markdown(rep), RM.render_html(rep))
return {c.id: c for c in a.checks}
def _llm_frags(run, sentence):
"""Fragments of an authored sentence, as ingestion would build them, bypassing the ingestion checks (a corruption)."""
cat = C.build_catalog(run.state, include_deferred=True)
frags = P.to_fragments(sentence, P.Rules(catalog=cat))
for f in frags:
f.origin = "llm"
return frags
# ── component: RelationshipClaim ──────────────────────────────────────────────────────────────────────
def test_relationships_have_explicit_sides_operator_and_magnitude(good):
run, _ = good
items = run.state.relations.get("state")["items"]
r = items["REL_PRICE_MARKET_COMPOSITE"]
assert (r["lhs_claim_id"], r["rhs_claim_id"]) == ("PRICE", "MARKET_COMPOSITE")
comp, price = run.state.value("pricing.market_composite"), run.state.value("market.price")
assert r["relation_type"] == (RL.BELOW if price < comp else RL.ABOVE)
assert r["magnitude"] == pytest.approx(abs(price / comp - 1)) and r["magnitude_definition"] == RL.RATIO_GAP
assert r["rendered"].startswith("the price (") and "the market composite (" in r["rendered"]
for cid in ("REL_PRICE_INTRINSIC", "REL_PRICE_MECHANICAL", "REL_FORECAST_EPS_CONSENSUS", "REL_TARGET_PRICE",
"REL_BREAK_EVEN_REVENUE_STORY", "REL_BREAK_EVEN_MARGIN_STORY", "REL_BREAK_EVEN_REVENUE_GROWTH", "REL_EXPECTATIONS_STORY"):
assert cid in items, cid
assert RL.check(run.state) == []
def test_break_even_relations_are_typed_and_bound(good):
run, _ = good
items = run.state.relations.get("state")["items"]
lv = items["REL_BREAK_EVEN_REVENUE_STORY"]
assert lv["relation_type"] == RL.GREATER_THAN and lv["magnitude"] == pytest.approx(31.5 / 25.2 - 1) # +25% level gap
gr = items["REL_BREAK_EVEN_REVENUE_GROWTH"]
assert (gr["lhs_claim_id"], gr["rhs_claim_id"], gr["years_claim_id"]) == ("BE_REVENUE", "STORY_START_REVENUE", "STORY_YEARS")
assert gr["magnitude_definition"] == RL.CAGR and gr["magnitude"] != pytest.approx(lv["magnitude"])
mg = items["REL_BREAK_EVEN_MARGIN_STORY"]
assert mg["magnitude_unit"] == "PERCENTAGE_POINTS" and mg["magnitude"] == pytest.approx(0.55 - 0.494)
for path, concept in (("valuation.breakeven_revenue", "BREAK_EVEN_REVENUE"), ("valuation.breakeven_margin", "BREAK_EVEN_OPERATING_MARGIN"),
("valuation.breakeven_revenue_cagr", "BREAK_EVEN_REVENUE_CAGR"), ("valuation.story_revenue_maturity", "REVENUE"),
("valuation.story_tam", "MARKET_SIZE"), ("valuation.story_years", "YEARS")):
assert run.state.get(path).economic_concept == concept
assert run.state.valuation.get("breakeven_growth_binding")["target_revenue_metric_id"] == "breakeven_revenue"
# the revenue relation states the basis it holds fixed when the break-even solve differs from the adopted story
assert "holding the operating margin at 45.0%" in lv["rendered"]
def test_expectations_polarity_is_deterministic(good):
run, _ = good
ex = run.state.relations.get("state")["items"]["REL_EXPECTATIONS_STORY"]
assert ex["relation_type"] == RL.MORE_DEMANDING == "MORE_DEMANDING_COMBINATION" and ex["overall_requirement"] == RL.MORE_DEMANDING
assert ex["components"] == {"revenue_relation": "GREATER_THAN", "margin_relation": "GREATER_THAN",
"price_implied_revenue": 31.5e9, "story_revenue": 25.2e9, "price_implied_margin": 0.55,
"story_margin": 0.494}
def test_persona_payload_separates_fact_relationship_and_evidence_tokens(good):
run, _ = good
cat = C.build_catalog(run.state, include_deferred=False)
txt = C.persona_catalog_text(cat, run.state)
assert "FACT TOKENS" in txt and "RELATIONSHIP TOKENS" in txt
for gone in ("[[PRICE_VS_COMPOSITE]]", "[[PRICE_VS_INTRINSIC]]", "[[FORECAST_EPS_VS_CONSENSUS]]", "[[TARGET_VS_PRICE]]",
"[[BE_REVENUE_CAGR]]", "[[RETURN_12M]]", "[[OWN_13F_HOLDERS_CHANGE]]", "[[ML_RANK_IC]]", "[[ML_RANK_POSITIVE]]"):
assert gone not in txt, gone
assert "[[REL_PRICE_MARKET_COMPOSITE]]" in txt and "[[RETURN_12M_PHRASE]]" in txt
def test_descriptive_separate_sentences_remain_allowed(good):
run, _ = good
cat, rules = _rules(run)
g = next(n for n, m in enumerate(run.state.guidance.company["metrics"], 1)
if m["concept"] == "OPERATING_MARGIN_CHANGE" and m["accounting_basis"] == "ADJUSTED")
assert P.check_sentence(f"Management guides to [[GUIDANCE_{g}]] of adjusted margin expansion.", rules)[0]
assert P.check_sentence("The break-even operating-margin level is [[BE_MARGIN]].", rules)[0]
assert P.check_sentence("Put plainly, [[REL_BREAK_EVEN_REVENUE_STORY]].", rules)[0]
# ── corruption matrix A–K ────────────────────────────────────────────────────────────────────────────
def test_A_reversed_gap_fails_relation_identity(good):
run, _ = good
cat, rules = _rules(run)
ok, why = P.check_sentence("The market composite of [[MARKET_COMPOSITE]] sits [[PRICE_VS_COMPOSITE]] current price.", rules)
assert not ok and why.startswith("RELATION_IDENTITY")
ok2, why2 = P.check_sentence("The market composite [[MARKET_COMPOSITE]] is below the price [[PRICE]].", rules)
assert not ok2 and "RELATION_REQUIRED_FOR_PAIR" in why2 and "REL_PRICE_MARKET_COMPOSITE" in why2
# a corrupted relation that swaps its sides fails the state-level identity check
st = run.state
items = dict(st.relations.get("state")["items"])
bad = dict(items["REL_PRICE_MARKET_COMPOSITE"], lhs_claim_id="MARKET_COMPOSITE", rhs_claim_id="PRICE")
object.__setattr__(st, "_frozen", False)
rs_backup = st.relations.get("state")
try:
st.relations._data["state"] = dict(rs_backup, items=dict(items, REL_PRICE_MARKET_COMPOSITE=bad))
assert any(k == "REL1" for k, _p in RL.check(st))
# and the audit catches the swapped relation fragment in authored prose
ids = _audit_ids(run, _llm_frags(run, "The composite sits [[PRICE_VS_COMPOSITE]] the price."))
assert ids["RELATION_IDENTITY"].status == "FAIL"
finally:
st.relations._data["state"] = rs_backup
def test_B_wrong_relation_subject_fails(good):
run, _ = good
cat, rules = _rules(run)
ok, why = P.check_sentence("The market composite of [[MARKET_COMPOSITE]] is [[PRICE_VS_INTRINSIC]] the story value.", rules)
assert not ok and why.startswith("RELATION_IDENTITY")
# a relation attempted with the composite as its subject and the intrinsic magnitude is rejected by the gate
rel, rej = RL.make(run.state, RL.Spec("REL_PRICE_INTRINSIC", "MARKET_COMPOSITE", "INTRINSIC_VALUE", "valuation", RL.RATIO_GAP,
RL.ABOVE, RL.BELOW, "PRICE_VS_INTRINSIC", lhs_words="the price", rhs_words="the story value"))
corrupt = dict(run.state.relations.get("state")["items"]["REL_PRICE_INTRINSIC"], lhs_claim_id="MARKET_COMPOSITE")
st = run.state
rs_backup = st.relations.get("state")
try:
st.relations._data["state"] = dict(rs_backup, items=dict(rs_backup["items"], REL_PRICE_INTRINSIC=corrupt))
assert any(k == "REL1" and "lhs/rhs" in p for k, p in RL.check(st))
finally:
st.relations._data["state"] = rs_backup
def test_C_change_vs_level_relation_is_never_created(good):
run, _ = good
g = next(n for n, m in enumerate(run.state.guidance.company["metrics"], 1) if m["label"] == "Operating Profit Margin expansion (Adjusted)")
rel, rej = RL.attempt(run.state, f"GUIDANCE_{g}", "BE_MARGIN", "scenario", RL.LEVEL_DIFF)
assert rel is None and "quantity MARGIN_CHANGE ≠ MARGIN_LEVEL" in rej["reason"]
cat, rules = _rules(run)
for s in (f"Guidance of [[GUIDANCE_{g}]] is small relative to the [[BE_MARGIN]] margin needed.",
f"Guidance of [[GUIDANCE_{g}]] is trivial against the [[BE_MARGIN]] break-even margin hurdle.",
f"Margin expansion guidance of [[GUIDANCE_{g}]] and a break-even margin of [[BE_MARGIN]] frame the debate."):
ok, why = P.check_sentence(s, rules)
assert not ok and "RELATION_REQUIRED_FOR_PAIR" in why and "a level and a change" in why
# an incompatible relation smuggled into the state fails RELATION_COMPATIBILITY
st = run.state
rs_backup = st.relations.get("state")
fake = dict(rs_backup["items"]["REL_BREAK_EVEN_MARGIN_STORY"], rhs_claim_id=f"GUIDANCE_{g}")
try:
st.relations._data["state"] = dict(rs_backup, items=dict(rs_backup["items"], REL_BREAK_EVEN_MARGIN_STORY=fake))
assert any(k == "REL1" for k, _p in RL.check(st)) # sides no longer the defined pair
import validation as V
spec = next(s for s in RL._specs(st) if s.claim_id == "REL_BREAK_EVEN_MARGIN_STORY")
from state import compatible
lm, _ = RL._metric(st, spec.lhs)
gm = C._claim_metric(st, f"GUIDANCE_{g}")
assert compatible(lm, gm, "scenario")[0] is False
finally:
st.relations._data["state"] = rs_backup
def test_D_cagr_used_as_level_gap_fails_relation_magnitude(good):
run, _ = good
cat, rules = _rules(run)
ok, why = P.check_sentence("[[BE_REVENUE]] exceeds [[STORY_REVENUE_MATURITY]] by [[BE_REVENUE_CAGR]].", rules)
assert not ok and why.startswith("RELATION_MAGNITUDE")
ids = _audit_ids(run, _llm_frags(run, "[[BE_REVENUE]] exceeds [[STORY_REVENUE_MATURITY]] by [[BE_REVENUE_CAGR]]."))
assert ids["RELATION_MAGNITUDE"].status == "FAIL"
# the correct statement uses the level-gap relation
assert "25.0% above" in run.state.relations.get("state")["items"]["REL_BREAK_EVEN_REVENUE_STORY"]["rendered"]
def test_E_reversed_implied_expectations_fails(good):
run, _ = good
cat, rules = _rules(run)
ok, why = P.check_sentence("The price embeds less optimism than the story.", rules)
assert not ok and why.startswith("EXPECTATIONS_POLARITY")
ok, why = P.check_sentence("[[REL_EXPECTATIONS_STORY]], so the price is conservative relative to the story case.", rules)
assert not ok and "EXPECTATIONS_POLARITY" in why
assert P.check_sentence("In short, [[REL_EXPECTATIONS_STORY]].", rules)[0]
ids = _audit_ids(run, _llm_frags(run, "The break-even math is conservative relative to the story case."))
assert ids["EXPECTATIONS_POLARITY"].status == "FAIL"
# a stored polarity that contradicts its components fails RELATION_IDENTITY
st = run.state
rs_backup = st.relations.get("state")
ex = dict(rs_backup["items"]["REL_EXPECTATIONS_STORY"], relation_type=RL.LESS_DEMANDING)
try:
st.relations._data["state"] = dict(rs_backup, items=dict(rs_backup["items"], REL_EXPECTATIONS_STORY=ex))
assert any(k == "REL1" and "polarity" in p for k, p in RL.check(st))
finally:
st.relations._data["state"] = rs_backup
def test_F_cross_model_evidence_fails_scope():
run, payload, failed = spgi_run(panel_visible=True)
assert failed is None
cat, rules = _rules(run)
assert "ML_RANK_IC" in rules.offered or run.state.value("quant.panel_rank.reliability.rank_ic") is None
s = "[[REL_FORECAST_EPS_CONSENSUS]], but that optimism may be overfit given [[ML_RANK_IC]] rank correlation."
ok, why = P.check_sentence(s, rules)
assert not ok and why.startswith("MODEL_EVIDENCE_SCOPE") and "FUNDAMENTAL_FORECAST" in why
ids = _audit_ids(run, _llm_frags(run, s))
assert ids["MODEL_EVIDENCE_SCOPE"].status == "FAIL"
def test_G_unreliable_model_claim_is_not_offered(good):
run, _ = good
assert run.state.quant.get("panel_rank")["reliability"]["persona_visibility"] is False
cat, rules = _rules(run)
assert "ML_RANK_IC" not in rules.offered and "ML_RANK_POSITIVE" not in rules.offered
ok, why = P.check_sentence("The rank correlation is only [[ML_RANK_IC]].", rules)
assert not ok and why.startswith("MODEL_EVIDENCE_VISIBILITY")
ids = _audit_ids(run, _llm_frags(run, "The rank correlation is only [[ML_RANK_IC]]."))
assert ids["MODEL_EVIDENCE_VISIBILITY"].status == "FAIL"
import brief as B
assert "0.03" not in B.panel_line(run.state) and "rank correlation" not in B._quant_block(run.state)
def test_H_rejected_guidance_cannot_become_absence():
d, st = _story_with_breakeven()
q = "The Company now expects to repurchase more than $7 billion in shares in total in 2026."
rel = dict(T.RELEASE, text=T.TABLE + " " + q)
prop = copy.deepcopy(T.PROPOSED)
prop["metrics"].append({"concept": "SHARE_REPURCHASE", "label": "Share repurchases", "period_label": "2026", "low": 7500,
"high": None, "unit": "USD_MILLION", "operator": "GREATER_THAN", "basis": "GAAP", "quote": q})
d["company_guidance"] = G.build_state(rel, prop)
assert d["company_guidance"]["concept_states"]["SHARE_REPURCHASE"]["state"] == G.FOUND_BUT_REJECTED
run, payload, failed = spgi_run(data=d, story=st)
cat, rules = _rules(run)
for s in ("No buyback guidance was given.", "No buyback or capital return acceleration was signaled post-spin."):
ok, why = P.check_sentence(s, rules)
assert not ok and why.startswith("GUIDANCE_EPISTEMIC_STATE"), s
ids = _audit_ids(run, [P.Fragment("No buyback guidance was given.", origin="llm")])
assert ids["GUIDANCE_EPISTEMIC_STATE"].status == "FAIL"
assert "do not state that guidance on it was or was not given" in G.persona_block(run.state.guidance.company)
def test_I_unit_equivalent_bound_guidance_verifies():
q = "The Company now expects to repurchase more than $7 billion in shares in total in 2026."
m = {"concept": "SHARE_REPURCHASE", "label": "Share repurchases", "period_label": "2026", "low": 7000, "high": None,
"unit": "USD_MILLION", "operator": "GREATER_THAN", "basis": "GAAP", "quote": q}
assert G.verify(m, q) == (True, "")
row = G.build_state({"text": q, "form": "8-K", "filed": "2026-07-28"}, {"metrics": [m]})["metrics"][0]
assert (row["operator"], row["raw_unit"], row["raw_low"], row["low"]) == ("GREATER_THAN", "USD_BILLION", 7.0, 7e9)
assert row["rendered_value"].startswith("more than $7")
assert not G.verify(dict(m, operator="POINT"), q)[0] and not G.verify(dict(m, operator="RANGE", high=8000), q)[0]
def test_J_signed_return_duplication_is_not_constructible(good):
run, _ = good
cat, rules = _rules(run)
for s in ("Shares have fallen [[RETURN_12M]] over twelve months.", "The bear case envisions a [[SCENARIO_BEAR_RETURN]] drawdown."):
ok, why = P.check_sentence(s, rules)
assert not ok and why.startswith("RETURN_DIRECTION_RENDER"), s
r12 = cat["RETURN_12M_PHRASE"].rendered
v = run.state.value("market.return_12m")
assert r12.startswith("down " if v < 0 else "up ") and "-" not in r12 and "+" not in r12
bear = cat["SCENARIO_BEAR_RETURN_PHRASE"].rendered
assert "downside case" in bear and "-" not in bear
ids = _audit_ids(run, _llm_frags(run, "Shares have fallen [[RETURN_12M]] over twelve months."))
assert ids["RETURN_DIRECTION_RENDER"].status == "FAIL"
def test_K_market_pricing_source_cannot_be_labelled_valuation_methods(good):
run, payload = good
src = run.state.blend.get("sources")
mp = next(x for x in src if x["kind"] == "MARKET_PRICING")
assert mp["label"] == "Market pricing" and "pricing.market_composite" in mp["underlying_object_ids"]
assert "Valuation methods" not in payload["markdown"] and "| Market pricing |" in payload["markdown"]
import report_model as RM
cat = C.build_catalog(run.state, include_deferred=True)
rep = RM.build_user_report(run.state, cat, {"summary": [], "thesis": [], "risks": [], "seats": {}}, [])
for b in rep.blocks:
if b.kind == "table" and b.section == "blend":
for row in b.content["rows"]:
if "".join(f.text for f in row[0]) == "Market pricing":
row[0][0] = P.Fragment("Valuation methods")
a = RM.audit(rep, run.state, RM.render_markdown(rep), RM.render_html(rep))
assert next(c for c in a.checks if c.id == "BLEND_SOURCE_KIND").status == "FAIL"
import validation as V
corrupt = [dict(x, label="Valuation methods") if x["kind"] == "MARKET_PRICING" else x for x in src]
st = run.state
try:
st.blend._data["sources"] = corrupt
assert next(c for c in V.validate_analysis(st).checks if c.id == "BLEND_SOURCE_KIND").status == "FAIL"
finally:
st.blend._data["sources"] = src
def test_final_spgi_shaped_report_passes_every_new_validator(good):
run, payload = good
new = ("RELATION_IDENTITY", "RELATION_MAGNITUDE", "RELATION_COMPATIBILITY", "RELATION_REQUIRED_FOR_PAIR", "MODEL_EVIDENCE_SCOPE",
"MODEL_EVIDENCE_VISIBILITY", "GUIDANCE_EPISTEMIC_STATE", "GUIDANCE_CANONICAL_VALUE", "RETURN_DIRECTION_RENDER",
"BLEND_SOURCE_KIND", "EXPECTATIONS_POLARITY")
seen = {}
for v in run.validations + [run.report_audit]:
for c in v.checks:
if c.id in new:
seen.setdefault(c.id, []).append(c.status)
assert set(seen) == set(new), set(new) - set(seen)
assert all(st in ("PASS", "NOT_APPLICABLE") for sts in seen.values() for st in sts), seen
md = payload["markdown"]
assert "What the price implies" in md and "requires a more demanding revenue/margin combination than the adopted story" in md
def test_a_claim_that_carries_its_own_qualifier_cannot_repeat_it():
"""Live SPGI (3fda3e3efd4f1e56): "…is 7.8% above consensus ($19.04) above consensus" and "more than more than $7.00B"."""
d, st = _story_with_breakeven()
q = "The Company now expects to repurchase more than $7 billion in shares in total in 2026."
prop = copy.deepcopy(T.PROPOSED)
prop["metrics"].append({"concept": "SHARE_REPURCHASE", "label": "Share repurchases", "period_label": "2026", "low": 7000,
"high": None, "unit": "USD_MILLION", "operator": "GREATER_THAN", "basis": "GAAP", "quote": q})
d["company_guidance"] = G.build_state(dict(T.RELEASE, text=T.TABLE + " " + q), prop)
run, payload, failed = spgi_run(data=d, story=st)
cat, rules = _rules(run)
n = next(i for i, m in enumerate(run.state.guidance.company["metrics"], 1) if m["concept"] == "SHARE_REPURCHASE")
ok, why = P.check_sentence(f"The company guides share repurchases of more than [[GUIDANCE_{n}]].", rules)
assert not ok and why.startswith("CLAIM_QUALIFIER_DOUBLED")
assert P.check_sentence(f"The company guides share repurchases of [[GUIDANCE_{n}]].", rules)[0]
ok, why = P.check_sentence("Only [[REL_FORECAST_EPS_CONSENSUS]] above consensus.", rules)
assert not ok and why.startswith("CLAIM_QUALIFIER_DOUBLED")
ids = _audit_ids(run, _llm_frags(run, f"The company guides repurchases of more than [[GUIDANCE_{n}]]."))
assert ids["CLAIM_QUALIFIER_DOUBLED"].status == "FAIL"
def test_a_phrase_claim_carries_its_own_direction_and_article(good):
"""Live SPGI (4436fda6840d7db9): "down to 51 fewer institutions" and "a an 8.5% upside case"."""
run, _ = good
cat, rules = _rules(run)
ok, why = P.check_sentence("Holders fell, down to [[OWN_13F_HOLDERS_CHANGE_PHRASE]].", rules)
assert not ok and (why.startswith("CLAIM_QUALIFIER_DOUBLED") or why.startswith("RELATION_CLAUSE_GRAMMAR"))
r = run.rules(cat, own="SCENARIO_NEUTRAL_RETURN_PHRASE")
ok, why = P.check_sentence("Most of a [[OWN_RETURN]]-type return must come from earnings.", r)
assert not ok and "article" in why
assert P.check_sentence("The filings show that [[OWN_13F_HOLDERS_CHANGE_PHRASE]].", rules)[0]
assert P.check_sentence("My scenario is [[OWN_RETURN]].", r)[0]
def test_two_one_variable_break_even_solves_never_render_as_a_joint_requirement(good):
"""Each break-even solve moves ONE variable and holds the other: the relation is a frontier of alternatives."""
run, _ = good
ex = run.state.relations.get("state")["items"]["REL_EXPECTATIONS_STORY"]
txt = ex["rendered"].lower()
assert ex["conjunctive"] is False
for joint in ("more revenue at maturity and a higher", "revenue and a higher margin", "both higher revenue", "and a higher operating margin"):
assert joint not in txt, joint
assert "revenue/margin combination" in txt and "alternatively" in txt
rc, mc = ex["revenue_case"], ex["margin_case"]
assert rc["held_constant_metric"] == "operating margin at maturity" and rc["held_constant_value"] == pytest.approx(0.45)
assert rc["basis"] == "break-even basis" and "holding the operating margin at the break-even basis of 45.0%" in txt
assert mc["held_constant_metric"] == "revenue path of the adopted story" and "holding the story's revenue constant" in txt
assert (rc["required_revenue"], rc["comparison_revenue"]) == (31.5e9, 25.2e9)
assert (mc["required_margin"], mc["comparison_margin"]) == (0.55, 0.494)
# the basis mismatch stays visible: EXPECTATIONS_BASIS warns and the user limitation is published
chk = next(c for v in run.validations for c in v.checks if c.id == "EXPECTATIONS_BASIS")
assert chk.status == "WARNING" and "45.0%" in chk.detail
def test_a_source_row_naming_a_guidance_concept_cannot_silently_become_not_found():
"""SPGI 8-K 2026-07-28: the outlook table has Interest Expense and Tax Rate rows; a proposal that drops them (output
budget, model variance) must re-propose them and, failing that, mark them FOUND_BUT_REJECTED — never NOT_FOUND."""
table = (T.TABLE + "\nInterest Expense, net | $365 to $385 million | $390 to $410 million"
"\nTax Rate | 22% to 23% | 22% to 23%")
rel = dict(T.RELEASE, text=table)
sec = table
cands = {x["concept"]: x["quote"] for x in G.candidate_rows(sec)}
assert cands["INTEREST_EXPENSE"].startswith("Interest Expense, net |") and cands["TAX_RATE"].startswith("Tax Rate |")
calls = []
def truncated_then_focused(system, user):
calls.append(user)
if len(calls) == 1: # first pass: the tail of the table is lost
return {"metrics": copy.deepcopy(T.PROPOSED["metrics"])}
return {"metrics": [
{"concept": "INTEREST_EXPENSE", "label": "Interest Expense, net (GAAP)", "period_label": "full year 2026", "low": 365,
"high": 385, "unit": "USD_MILLION", "basis": "GAAP", "quote": "Interest Expense, net | $365 to $385 million | $390 to $410 million"},
{"concept": "TAX_RATE", "label": "Tax Rate (GAAP)", "period_label": "full year 2026", "low": 22, "high": 23,
"unit": "PERCENT", "basis": "GAAP", "quote": "Tax Rate | 22% to 23% | 22% to 23%"}]}
st = G.extract(rel, sec, truncated_then_focused)
assert len(calls) == 2 and "Interest Expense, net" in calls[1] and "Tax Rate" in calls[1]
assert st["concept_states"]["INTEREST_EXPENSE"]["state"] == G.AVAILABLE
assert st["concept_states"]["TAX_RATE"]["state"] == G.AVAILABLE
assert {"INTEREST_EXPENSE", "TAX_RATE"} <= set(st["diagnostics"]["coverage_pass"]["requested"])
def never_extracts(system, user):
return {"metrics": copy.deepcopy(T.PROPOSED["metrics"])}
st2 = G.extract(rel, sec, never_extracts)
for c in ("INTEREST_EXPENSE", "TAX_RATE"):
assert st2["concept_states"][c]["state"] == G.FOUND_BUT_REJECTED, c
assert st2["concept_states"][c]["source_row"]
# a quote-less proposal is a found-but-rejected row, not an absence
s3 = G.build_state(rel, {"metrics": copy.deepcopy(T.PROPOSED["metrics"]) + [{"concept": "TAX_RATE", "low": 22, "high": 23,
"unit": "PERCENT", "quote": ""}]})
assert s3["concept_states"]["TAX_RATE"]["state"] == G.FOUND_BUT_REJECTED
# and the data-stage validator fails a candidate row left NOT_FOUND
import validation as V
d, st_ = _story_with_breakeven()
bad = dict(st2, concept_states=dict(st2["concept_states"], TAX_RATE={"state": G.NOT_FOUND}))
d["company_guidance"] = bad
run, payload, failed = spgi_run(data=d, story=st_)
assert failed == "DATA"
chk = next(c for c in run.validations[0].checks if c.id == "GUIDANCE_EPISTEMIC_STATE")
assert chk.status == "FAIL" and "TAX_RATE" in chk.detail
# ── prose / trigger cleanup (run 38fe4e9814b0ea98) ───────────────────────────────────────────────────────
def test_trend_conclusions_need_an_approved_trend_relationship(good):
run, _ = good
cat, rules = _rules(run)
rs = run.state.relations.get("state")
assert not any(k.startswith("TREND_") for k in rs["items"]) # every trailing → forward pair crosses the separation
rej = {r["relation_id"]: r for r in rs["trends_rejected"]}
assert rej["TREND_REVENUE"]["perimeter_compatible"] is False and rej["TREND_REVENUE"]["allowed"] is False
s = "The engine forecasts revenue of [[FORECAST_REVENUE]] next twelve months, implying deceleration."
ok, why = P.check_sentence(s, rules)
assert not ok and why.startswith("TREND_RELATION_REQUIRED")
assert _audit_ids(run, _llm_frags(run, s))["TREND_RELATION_REQUIRED"].status == "FAIL"
# a typed change claim carries a level change itself; the fixture's compatible issuer gets a real trend relation
g = next(n for n, m in enumerate(run.state.guidance.company["metrics"], 1)
if m["concept"] == "OPERATING_MARGIN_CHANGE" and m["accounting_basis"] == "GAAP")
assert P.check_sentence(f"Management guides to GAAP operating margin expansion of [[GUIDANCE_{g}]].", rules)[0]
vrun, _p, _f = F.run_pipeline()
vcat, vrules = _rules(vrun)
assert "TREND_OPERATING_MARGIN" in vrules.trends_allowed
rel = vrules.trends_allowed["TREND_OPERATING_MARGIN"].lower()
assert P.check_sentence(f"On the engine's forecast, [[TREND_OPERATING_MARGIN]], so the margin is {rel}.", vrules)[0]
@pytest.mark.parametrize("sentence", ["Institutional sponsorship is weakening.", "Ownership support is fading among holders.",
"Holder breadth is bearish for the stock.", "Institutional flows are constructive.",
"Positioning among tracked investors is supportive."])
def test_descriptive_only_ownership_cannot_carry_an_interpretation(good, sentence):
run, _ = good
cat, rules = _rules(run)
assert rules.interpretation["13F"]["permission"] == EV.DESCRIPTIVE_ONLY
ok, why = P.check_sentence(sentence, rules)
assert not ok and why.startswith("EVIDENCE_INTERPRETATION")
assert _audit_ids(run, [P.Fragment(sentence, origin="llm")])["EVIDENCE_INTERPRETATION"].status == "FAIL"
def test_descriptive_ownership_facts_remain_allowed(good):
run, _ = good
cat, rules = _rules(run)
assert P.check_sentence("[[OWN_13F_HOLDERS]] institutions reported positions, and [[OWN_13F_HOLDERS_CHANGE_PHRASE]].", rules)[0]
def test_ownership_change_is_a_complete_clause_that_no_verb_can_precede(good):
run, _ = good
cat, rules = _rules(run)
r = cat["OWN_13F_HOLDERS_CHANGE_PHRASE"].rendered
assert r.startswith("the number of institutions reporting a position ") and " by " in r
ok, why = P.check_sentence("Institutional holders declined by [[OWN_13F_HOLDERS_CHANGE_PHRASE]].", rules)
assert not ok and why.startswith("RELATION_CLAUSE_GRAMMAR")
def test_trigger_already_satisfied_is_omitted_and_fails_validation():
import state_builder as S
import validation as V
run, payload, failed = spgi_run()
eps = run.state.value("forecast.eps")
t = {"metric_id": "next-twelve-month EPS", "condition_operator": ">=", "threshold": f"${eps:.2f}",
"scenario_destination": "BULL", "source_claim": "CONSENSUS_EPS_NTM"}
typed = S._typed_trigger(t, run.state)
assert typed["currently_satisfied"] is True and ("FORECAST_EPS", pytest.approx(eps)) in [(c, v) for c, v in typed["current_values"]]
llm = F.llm_result()
llm["thesis_triggers"] = [dict(t, polarity="POSITIVE_FOR_EQUITY", status="OK"),
{"metric_id": "forward P/E", "condition_operator": "<=", "threshold": "1.0x", "source_claim": "PE_FORWARD",
"scenario_destination": "BEAR", "status": "OK"}]
run2, _p2, failed2 = spgi_run(llm=llm)
assert failed2 is None
pub = run2.state.scenarios.get("triggers")
assert all(not x["currently_satisfied"] for x in pub) and not any(x["metric_id"] == "next-twelve-month EPS" for x in pub)
assert any("already satisfied" in d["reason"] for d in run2.state.scenarios.get("triggers_dropped"))
st = run2.state
try:
st.scenarios._data["triggers"] = list(pub) + [typed]
chk = next(c for c in V.validate_analysis(st).checks if c.id == "TRIGGER_NOT_ALREADY_SATISFIED")
assert chk.status == "FAIL"
finally:
st.scenarios._data["triggers"] = pub
def test_relationship_render_modes_come_from_one_typed_relation(good):
run, _ = good
r = run.state.relations.get("state")["items"]["REL_PRICE_MARKET_COMPOSITE"]
m = r["renderings"]
word = "below" if r["relation_type"] == RL.BELOW else "above"
size = f"{r['magnitude'] * 100:.1f}%"
assert m["FULL"] == r["rendered"] and m["CLAUSE"] == f"{size} {word} the $172.00 market composite"
assert m["COMPACT"] == f"{size} {word} the market composite" # never "below market pricing" (the price IS the market's)
cat, rules = _rules(run)
assert P.check_sentence("The price is [[REL_PRICE_MARKET_COMPOSITE__COMPACT]].", rules)[0]
assert P.check_sentence("Shares trade [[REL_PRICE_MARKET_COMPOSITE__CLAUSE]].", rules)[0]
ok, why = P.check_sentence("The market composite is [[REL_PRICE_MARKET_COMPOSITE__COMPACT]].", rules)
assert not ok and why.startswith(("RELATION_IDENTITY", "CLAIM_QUALIFIER_DOUBLED"))
# a tampered mode rendering fails the relation check
st = run.state
rs = st.relations.get("state")
bad = dict(rs["items"]["REL_PRICE_MARKET_COMPOSITE"], renderings=dict(m, COMPACT=f"{size} {'above' if word == 'below' else 'below'} the market composite"))
try:
st.relations._data["state"] = dict(rs, items=dict(rs["items"], REL_PRICE_MARKET_COMPOSITE=bad))
assert any(k == "REL1" and "render mode" in p for k, p in RL.check(st))
finally:
st.relations._data["state"] = rs
@pytest.mark.parametrize("sentence", ["[[REL_TARGET_PRICE]] suggests meaningful upside.",
"The engine's EPS is [[REL_FORECAST_EPS_CONSENSUS]].",
"The gap is real per [[REL_EXPECTATIONS_STORY]].",
"Today, [[REL_PRICE_INTRINSIC]], [[REL_PRICE_MECHANICAL]], and [[REL_PRICE_MARKET_COMPOSITE]]."])
def test_full_relationship_clauses_keep_clause_positions(good, sentence):
run, _ = good
cat, rules = _rules(run)
ok, why = P.check_sentence(sentence, rules)
# "The engine's EPS is [[REL_FORECAST_EPS_CONSENSUS]]" also restates the relation's own subject words, which the
# doubling rule may report first; the grammar audit must still fail on its own
assert not ok and why.startswith(("RELATION_CLAUSE_GRAMMAR", "CLAIM_QUALIFIER_DOUBLED"))
assert _audit_ids(run, _llm_frags(run, sentence))["RELATION_CLAUSE_GRAMMAR"].status == "FAIL"
def test_well_formed_relationship_clauses_pass(good):
run, _ = good
cat, rules = _rules(run)
assert P.check_sentence("Today, [[REL_PRICE_INTRINSIC]], while [[REL_PRICE_MARKET_COMPOSITE]].", rules)[0]
assert P.check_sentence("The mean analyst target sits [[REL_TARGET_PRICE__COMPACT]], which argues for patience.", rules)[0]
def test_guidance_value_must_be_named_by_its_own_concept():
"""Live SPGI (567098321128f07b): "private-markets growth from a more than $7.00B base" used the share-repurchase value."""
d, st = _story_with_breakeven()
q = "The Company now expects to repurchase more than $7 billion in shares in total in 2026."
prop = copy.deepcopy(T.PROPOSED)
prop["metrics"].append({"concept": "SHARE_REPURCHASE", "label": "Share repurchases", "period_label": "2026", "low": 7000,
"high": None, "unit": "USD_MILLION", "operator": "GREATER_THAN", "basis": "GAAP", "quote": q})
d["company_guidance"] = G.build_state(dict(T.RELEASE, text=T.TABLE + " " + q), prop)
run, payload, failed = spgi_run(data=d, story=st)
cat, rules = _rules(run)
rows = run.state.guidance.company["metrics"]
rp = next(i for i, m in enumerate(rows, 1) if m["concept"] == "SHARE_REPURCHASE")
rg = next(i for i, m in enumerate(rows, 1) if m["concept"] == "REVENUE_GROWTH")
mg = next(i for i, m in enumerate(rows, 1) if m["concept"] == "OPERATING_MARGIN_CHANGE" and m["accounting_basis"] == "GAAP")
ma = next(i for i, m in enumerate(rows, 1) if m["concept"] == "OPERATING_MARGIN_CHANGE" and m["accounting_basis"] == "ADJUSTED")
bad = f"The earnings call touted private-markets growth from a [[GUIDANCE_{rp}]] base."
ok, why = P.check_sentence(bad, rules)
assert not ok and why.startswith("GUIDANCE_LABEL_BINDING")
assert _audit_ids(run, _llm_frags(run, bad))["GUIDANCE_LABEL_BINDING"].status == "FAIL"
for good_s in (f"Management plans share repurchases of [[GUIDANCE_{rp}]] in 2026.",
f"The company guides revenue growth of [[GUIDANCE_{rg}]] and GAAP operating margin expansion of [[GUIDANCE_{mg}]].",
f"Guidance calls for [[GUIDANCE_{ma}]] of adjusted margin expansion."):
assert P.check_sentence(good_s, rules)[0], good_s
# a value placed where another concept is named is caught even when its own name appears elsewhere in the sentence
ok, why = P.check_sentence(f"Buybacks are large, and revenue growth of [[GUIDANCE_{rp}]] follows.", rules)
assert not ok and why.startswith("GUIDANCE_LABEL_BINDING")
# ── publication-integrity cleanup (run a9a20ec4ae73b891) ─────────────────────────────────────────────────
def _summary_audit(run, frags):
import report_model as RM
cat = C.build_catalog(run.state, include_deferred=True)
rep = RM.build_user_report(run.state, cat, {"summary": frags, "thesis": [], "risks": [], "seats": {}}, [])
a = RM.audit(rep, run.state, RM.render_markdown(rep), RM.render_html(rep))
return {c.id: c for c in a.checks}
def test_final_claim_only_rejects_a_stale_summary_number(good):
run, _ = good
pm = C.claim_for(run.state, C.BY_ID["PM_RETURN_PHRASE"])
ok = _summary_audit(run, [P.Fragment("The panel's view is ", origin="llm"), P.Fragment(pm.rendered, "PM_RETURN_PHRASE", pm.value, "llm"),
P.Fragment(".", origin="llm")])
assert ok["FINAL_CLAIM_ONLY"].status == "PASS"
stale = "a 5.7% probability-weighted gain" if "5.7%" not in pm.rendered else "a 9.9% probability-weighted gain"
typed = _summary_audit(run, [P.Fragment(f"The outcome is {stale}.", origin="llm")])
assert typed["FINAL_CLAIM_ONLY"].status == "FAIL" # authored number
wrong = _summary_audit(run, [P.Fragment("The outcome is ", origin="llm"), P.Fragment(stale, "PM_RETURN_PHRASE", 0.057, "llm"),
P.Fragment(".", origin="llm")])
assert wrong["FINAL_CLAIM_ONLY"].status == "FAIL" # claim fragment with a stale value
def test_relation_logic_rejects_alternative_cases_stated_jointly(good):
run, _ = good
cat, rules = _rules(run)
ex = run.state.relations.get("state")["items"]["REL_EXPECTATIONS_STORY"]
assert ex["logical_structure"] == "ALTERNATIVE_CONDITIONALS"
s = "The market prices in break-even assumptions that demand [[BE_REVENUE]] in revenue and [[BE_MARGIN]] margins."
ok, why = P.check_sentence(s, rules)
assert not ok and why.startswith("RELATION_LOGIC_PRESERVATION")
assert _audit_ids(run, _llm_frags(run, s))["RELATION_LOGIC_PRESERVATION"].status == "FAIL"
assert P.check_sentence("In short, [[REL_EXPECTATIONS_STORY]].", rules)[0]
st = run.state
rs = st.relations.get("state")
joint = dict(ex, rendered=ex["rendered"].replace("; alternatively, ", " and "))
try:
st.relations._data["state"] = dict(rs, items=dict(rs["items"], REL_EXPECTATIONS_STORY=joint))
assert any(k == "REL1" and "alternative" in p for k, p in RL.check(st))
finally:
st.relations._data["state"] = rs
def test_ownership_change_is_owned_by_its_phrase(good):
run, _ = good
cat, rules = _rules(run)
for s in ("Institutional holders declined by [[OWN_13F_HOLDERS_CHANGE_PHRASE]].",
"Institutional holders declined by [[OWN_13F_HOLDERS_CHANGE]].", "Institutional holders declined by 51 fewer institutions."):
assert not P.check_sentence(s, rules)[0], s
ids = _audit_ids(run, _llm_frags(run, "Holders changed: [[OWN_13F_HOLDERS_CHANGE]]."))
assert ids["OWNERSHIP_CHANGE_PHRASE"].status == "FAIL"
assert P.check_sentence("The filings show that [[OWN_13F_HOLDERS_CHANGE_PHRASE]].", rules)[0]
def test_valuation_direction_words_need_their_relation(good):
run, _ = good
cat, rules = _rules(run)
s = "Expected return should land near the neutral case, modestly tilted by the mechanical DCF discount."
ok, why = P.check_sentence(s, rules)
assert not ok and why.startswith("VALUATION_RELATION_LANGUAGE")
assert _audit_ids(run, [P.Fragment(s, origin="llm")])["VALUATION_RELATION_LANGUAGE"].status == "FAIL"
rel = run.state.relations.get("state")["items"]["REL_PRICE_MECHANICAL"]["relation_type"]
right, wrong = ("premium", "discount") if rel == RL.ABOVE else ("discount", "premium")
# (the relation's own rendering already says "premium to the mechanical earnings DCF value": the direction word is
# tested against the FULL clause, which a restated noun phrase would double)
assert P.check_sentence(f"The stock trades at a {right} to the mechanical DCF: [[REL_PRICE_MECHANICAL]].", rules)[0]
ok, why = P.check_sentence(f"The stock trades at a {wrong} to the mechanical DCF: [[REL_PRICE_MECHANICAL]].", rules)
assert not ok and "contradicts" in why
@pytest.mark.parametrize("rid", ["REL_PRICE_INTRINSIC", "REL_PRICE_MECHANICAL", "REL_PRICE_MARKET_COMPOSITE", "REL_TARGET_PRICE",
"REL_FORECAST_EPS_CONSENSUS", "REL_BREAK_EVEN_REVENUE_STORY", "REL_BREAK_EVEN_MARGIN_STORY"])
def test_every_render_mode_has_a_grammar_fixture(good, rid):
run, _ = good
cat, rules = _rules(run)
subj = {"REL_TARGET_PRICE": "The mean analyst target", "REL_FORECAST_EPS_CONSENSUS": "The engine's EPS forecast",
"REL_BREAK_EVEN_REVENUE_STORY": "The required revenue", "REL_BREAK_EVEN_MARGIN_STORY": "The required margin"}.get(rid, "The price")
good_s = {"FULL": f"Today, [[{rid}]].", "PREDICATE": f"{subj} [[{rid}__PREDICATE]].", "CLAUSE": f"{subj} sits [[{rid}__CLAUSE]].",
"COMPACT": f"{subj} is [[{rid}__COMPACT]].", "NOUN_PHRASE": f"{subj} stands at [[{rid}__NOUN_PHRASE]]."}
bad_s = {"FULL": f"{subj} is [[{rid}]].", "PREDICATE": f"{subj} is [[{rid}__PREDICATE]].",
"CLAUSE": f"{subj} [[{rid}__CLAUSE]].", "COMPACT": f"{subj} [[{rid}__COMPACT]] than the other side."}
for mode, s in good_s.items():
cid = rid if mode == "FULL" else f"{rid}__{mode}"
if cid not in cat:
continue # e.g. NOUN_PHRASE exists only for valuation and EPS relations
assert P.check_sentence(s, rules)[0], (mode, s, P.check_sentence(s, rules)[1])
for mode, s in bad_s.items():
assert not P.check_sentence(s, rules)[0], (mode, s)
ok, why = P.check_sentence("The engine's forecast [[REL_FORECAST_EPS_CONSENSUS__COMPACT]] than sell-side consensus.", rules)
assert not ok and why.startswith("RELATION_CLAUSE_GRAMMAR")
def test_momentum_evidence_cannot_establish_fundamental_deceleration(good):
run, _ = good
cat, rules = _rules(run)
s = ("Mature-growth lifecycle stage and a price below the 200-day moving average ([[SMA200]]) support the bear's "
"deceleration concern.")
ok, why = P.check_sentence(s, rules)
assert not ok and why.startswith("EVIDENCE_DOMAIN_SCOPE")
assert _audit_ids(run, _llm_frags(run, s))["EVIDENCE_DOMAIN_SCOPE"].status == "FAIL"
assert EV.evidence_domain("SMA200") == EV.MOMENTUM and EV.EVIDENCE_DOMAINS[EV.MOMENTUM]["allowed_target_domains"] == (EV.MOMENTUM, "TECHNICAL_SETUP")
assert P.check_sentence("The price sits below the 200-day moving average ([[SMA200]]), a weak near-term technical setup.", rules)[0]
def test_trigger_on_a_guidance_range_encodes_its_position_not_a_scalar(good):
import state_builder as S
run, _ = good
rows = run.state.guidance.company["metrics"]
g = next(i for i, m in enumerate(rows, 1) if m["concept"] == "REVENUE_GROWTH")
lo, hi = rows[g - 1]["low"], rows[g - 1]["high"]
t = {"metric_id": "revenue growth", "condition_operator": ">=", "threshold": f"{hi * 100:.1f}%", "source_claim": f"GUIDANCE_{g}"}
x = S._typed_trigger(t, run.state)
assert x["current_value"] is None and x["current_range"]["kind"] == "GUIDANCE_RANGE"
assert x["range_position"] == "AT_RANGE_ENDPOINT" and x["currently_satisfied"] is False
y = S._typed_trigger(dict(t, threshold=f"{(lo - 0.01) * 100:.1f}%"), run.state)
assert y["range_position"] == "SATISFIED_ACROSS_RANGE" and y["currently_satisfied"] is True
def test_the_growth_that_reaches_break_even_revenue_belongs_to_the_revenue_case(good):
"""Live SPGI (8f110e6c5292709b): "[[REL_BREAK_EVEN_REVENUE_GROWTH]] … and [[REL_BREAK_EVEN_MARGIN_STORY]]" read as joint."""
run, _ = good
cat, rules = _rules(run)
s = ("The break-even analysis indicates that [[REL_BREAK_EVEN_REVENUE_GROWTH]], and [[REL_BREAK_EVEN_MARGIN_STORY]], so "
"any pressure would create downside.")
ok, why = P.check_sentence(s, rules)
assert not ok and why.startswith("RELATION_LOGIC_PRESERVATION")
assert _audit_ids(run, _llm_frags(run, s))["RELATION_LOGIC_PRESERVATION"].status == "FAIL"
def test_claims_repeated_or_written_back_to_back_are_rejected(good):
"""Live SPGI (8f110e6c5292709b): "4 new and 2 adding 1 1 4 2"."""
run, _ = good
cat, rules = _rules(run)
s = ("Among tracked holders, [[OWN_13F_TRIMS]] trimmed and [[OWN_13F_EXITS]] exited versus [[OWN_13F_NEW]] new and "
"[[OWN_13F_ADDS]] adding [[OWN_13F_TRIMS]] [[OWN_13F_EXITS]] [[OWN_13F_NEW]] [[OWN_13F_ADDS]].")
assert not P.check_sentence(s, rules)[0]
assert not P.check_sentence("Tracked investors: [[OWN_13F_NEW]] [[OWN_13F_ADDS]] moved.", rules)[0]
assert P.check_sentence("Among tracked holders, [[OWN_13F_NEW]] opened, [[OWN_13F_ADDS]] added, [[OWN_13F_TRIMS]] "
"trimmed and [[OWN_13F_EXITS]] exited.", rules)[0]
# live 5bcb332c678ba99f: "16 tracked long-term investors hold, with 1 exiting" (4 opened, 2 added omitted)
ok, why = P.check_sentence("Tracked investors hold, with [[OWN_13F_EXITS]] exiting last quarter.", rules)
assert not ok and "breakdown" in why, why
def test_predicate_form_must_directly_follow_its_subject(good):
"""Live SPGI (9c959623b5147027): "the price still shows is 9.0% below market pricing"."""
run, _ = good
cat, rules = _rules(run)
ok, why = P.check_sentence("But the price still shows [[REL_PRICE_MARKET_COMPOSITE__PREDICATE]].", rules)
assert not ok and why.startswith("RELATION_CLAUSE_GRAMMAR")
assert P.check_sentence("But the price [[REL_PRICE_MARKET_COMPOSITE__PREDICATE]].", rules)[0]
assert _audit_ids(run, _llm_frags(run, "But the price still shows [[REL_PRICE_MARKET_COMPOSITE__PREDICATE]]."))[
"RELATION_CLAUSE_GRAMMAR"].status == "FAIL"
def test_a_unit_typed_next_to_a_claim_that_renders_it_is_rejected(good):
"""Live SPGI (f565f581d2d0256b): "a $$72.00B TAM"."""
run, _ = good
cat, rules = _rules(run)
ok, why = P.check_sentence("The story assumes a $[[STORY_TAM]] TAM.", rules)
assert not ok and why.startswith("CLAIM_QUALIFIER_DOUBLED")
assert not P.check_sentence("The forward P/E is [[PE_FORWARD]]x.", rules)[0]
assert P.check_sentence("The story assumes a [[STORY_TAM]] market at maturity.", rules)[0]
assert _audit_ids(run, _llm_frags(run, "The story assumes a $[[STORY_TAM]] TAM."))["CLAIM_QUALIFIER_DOUBLED"].status == "FAIL"
def test_a_sentence_referring_back_to_a_removed_sentence_is_removed(good):
"""Live SPGI (f565f581d2d0256b): "Either the company more than doubles … [removed]. … Neither is credible." dangled."""
run, _ = good
cat, rules = _rules(run)
text = ("Either the company more than doubles continuing revenue in 10 years or expands margins 10.9 points. "
"The forward P/E of [[PE_FORWARD]] demands flawless execution. Neither is credible.")
frags, rej = P.ingest(text, rules, "bear.key_point.4")
out = "".join(f.text for f in frags)
assert "Neither is credible" not in out and "demands flawless execution" in out
assert any(r.reason.startswith("DANGLING_REFERENCE") for r in rej)
# without a removed antecedent, a sentence may still open by pointing back
frags2, rej2 = P.ingest("The forward P/E is [[PE_FORWARD]]. This leaves little margin for error.", rules, "x")
assert not rej2 and "This leaves" in "".join(f.text for f in frags2)
# ── publication-semantics cleanup: corruption tests A–F ──────────────────────────────────────────────────
def test_A_alternative_solves_paraphrased_as_joint_requirement_fail(good):
run, _ = good
cat, rules = _rules(run)
for s_ in ("The price requires [[BE_REVENUE]] revenue and [[BE_MARGIN]] margin.",
"The price requires both higher revenue and higher margins than the adopted story.",
"Today's valuation needs revenue plus margin well beyond the story."):
ok, why = P.check_sentence(s_, rules)
assert not ok and why.startswith("RELATION_LOGIC_PRESERVATION"), s_
assert P.check_sentence("The price requires revenue well above the story's.", rules)[0]
assert P.check_sentence("In short, [[REL_EXPECTATIONS_STORY]].", rules)[0]
ex = run.state.relations.get("state")["items"]
assert ex["REL_EXPECTATIONS_STORY"]["logical_structure"] == RL.ALTERNATIVE_CONDITIONALS
assert all(r.get("logical_structure") in RL.LOGICAL_STRUCTURES for r in ex.values())
def test_B_valuation_direction_needs_explicit_sides_and_the_relations_direction(good):
run, _ = good
cat, rules = _rules(run)
rel = run.state.relations.get("state")["items"]["REL_PRICE_MECHANICAL"]["relation_type"]
hi, lo = ("above", "below") if rel == RL.ABOVE else ("below", "above")
ok, why = P.check_sentence("Expected return is modestly tilted by the mechanical DCF discount.", rules)
assert not ok and why.startswith("VALUATION_RELATION_LANGUAGE") and "no subject" in why
assert P.check_sentence(f"The price is {hi} the mechanical DCF.", rules)[0]
assert P.check_sentence(f"The mechanical DCF sits {lo} the current price.", rules)[0]
ok, why = P.check_sentence(f"The mechanical DCF sits {hi} the current price.", rules)
assert not ok and "contradicts" in why
def test_C_full_clause_inside_a_predicate_and_full_sentence_slot(good):
run, _ = good
cat, rules = _rules(run)
assert not P.check_sentence("The engine's forecast is [[REL_FORECAST_EPS_CONSENSUS]].", rules)[0]
assert cat["REL_PRICE_MARKET_COMPOSITE__FULL_SENTENCE"].rendered.endswith(".")
# FULL_SENTENCE fills a template slot that takes a whole sentence; authors are not offered it
ok, why = P.check_sentence("[[REL_PRICE_MARKET_COMPOSITE__FULL_SENTENCE]]", rules)
assert not ok and why.startswith("RELATION_CLAUSE_GRAMMAR")
assert P.relation_slot_violation("[[REL_PRICE_MARKET_COMPOSITE__FULL_SENTENCE]]", rules.rel_subjects) is None
assert P.relation_slot_violation("Today [[REL_PRICE_MARKET_COMPOSITE__FULL_SENTENCE]]", rules.rel_subjects)
def test_D_momentum_cannot_establish_margin_or_revenue_conclusions(good):
run, _ = good
cat, rules = _rules(run)
for s_ in ("Weak momentum below the 200-day average points to margin deterioration.",
"Shares below the 200-day moving average ([[SMA200]]) signal revenue deceleration."):
ok, why = P.check_sentence(s_, rules)
assert not ok and why.startswith("EVIDENCE_DOMAIN_SCOPE"), s_
assert EV.conclusion_domain("margin deterioration") == EV.MARGIN_C
def test_E_historical_segment_count_cannot_describe_the_continuing_company(good):
run, _ = good
cat, rules = _rules(run)
assert run.state.value("valuation.segment_count_filed") == 5 and run.state.value("valuation.segment_count_continuing") == 4
assert run.state.valuation.segment_count_filed.perimeter == "pre-separation"
for s_ in ("The business now spans four businesses, with five reportable segments covering [[SEGMENTS_COVERAGE]] of revenue.",
"The business now spans [[SEGMENT_COUNT_CONTINUING]] businesses, with [[SEGMENT_COUNT_FILED]] reportable segments."):
ok, why = P.check_sentence(s_, rules)
assert not ok and why.startswith("SEGMENT_PERIMETER_LANGUAGE"), s_
assert P.check_sentence("Before the separation, [[SEGMENT_COUNT_FILED]] reportable segments covered [[SEGMENTS_COVERAGE]] "
"of revenue.", rules)[0]
assert P.check_sentence("The continuing company has [[SEGMENT_COUNT_CONTINUING]] businesses.", rules)[0]
s_ = "The business now spans [[SEGMENT_COUNT_CONTINUING]] businesses, with [[SEGMENT_COUNT_FILED]] reportable segments."
assert _audit_ids(run, _llm_frags(run, s_))["SEGMENT_PERIMETER_LANGUAGE"].status == "FAIL"
def test_F_descriptive_only_insider_evidence_cannot_carry_selling_pressure(good):
run, _ = good
cat, rules = _rules(run)
assert rules.interpretation["insider"]["permission"] == EV.DESCRIPTIVE_ONLY
for s_ in ("No insider selling pressure is evident.", "Insider support is solid.", "Ownership support looks durable."):
ok, why = P.check_sentence(s_, rules)
assert not ok and why.startswith("EVIDENCE_INTERPRETATION"), s_
assert P.check_sentence("No material discretionary insider sale was identified.", rules)[0]
def test_frontier_framing_is_a_valid_paraphrase_and_full_sentence_is_template_only(good):
"""Live SPGI (c213229ef3909d8c): "a more demanding revenue and margin combination" is the frontier's own framing;
the PM used __FULL_SENTENCE tokens as clauses, so that form is no longer offered to authors."""
run, _ = good
cat, rules = _rules(run)
assert P.check_sentence("The price requires a revenue and margin combination beyond the story's.", rules)[0]
assert P.check_sentence("The price needs either more revenue or a higher margin than the story.", rules)[0]
assert not P.check_sentence("The price requires both higher revenue and higher margins than the adopted story.", rules)[0]
assert "REL_PRICE_INTRINSIC__FULL_SENTENCE" not in rules.offered
assert "REL_PRICE_INTRINSIC__FULL_SENTENCE" not in C.persona_catalog_text(C.build_catalog(run.state, include_deferred=False), run.state)
def test_empty_summary_and_pm_sections_fall_back_to_typed_template_sentences(good):
"""Live SPGI (ce149f0e7abc97e5): every authored summary and PM sentence was (correctly) rejected, leaving empty sections."""
import report_model as RM
run, _ = good
cat = C.build_catalog(run.state, include_deferred=True)
rep = RM.build_user_report(run.state, cat, {"summary": [], "thesis": [], "risks": [], "seats": {}}, [])
md = RM.render_markdown(rep)
brief = md[md.index("## In brief"):md.index("## The analyst debate")]
assert run.state.relations.get("state")["items"]["REL_PRICE_INTRINSIC"]["renderings"]["FULL_SENTENCE"] in brief
assert "The portfolio manager weights the bear case at" in md
a = RM.audit(rep, run.state, md, RM.render_html(rep))
assert a.passed, [(c.id, c.detail) for c in a.checks if c.status == "FAIL"]
def test_title_case_strong_rating_is_checked_and_empty_seats_say_so(good):
"""Live SPGI (c11694dec842419a): the HOLD bull wrote "I maintain a Strong Buy"; the neutral seat had no publishable prose."""
import report_model as RM
run, _ = good
cat = C.build_catalog(run.state, include_deferred=True)
lab = (run.state.scenarios.get("bull") or {}).get("label")
r = run.rules(cat, own="SCENARIO_BULL_RETURN_PHRASE", own_label=lab)
other = "Strong Sell" if lab != "STRONG SELL" else "Strong Buy"
ok, why = P.check_sentence(f"I maintain a {other} with modestly trimmed upside.", r)
assert not ok and "rating" in why
rep = RM.build_user_report(run.state, cat, {"summary": [], "thesis": [], "risks": [], "seats": {}}, [])
md = RM.render_markdown(rep)
assert md.count("This analyst's written reasoning is not shown") == 3
a = RM.audit(rep, run.state, md, RM.render_html(rep))
assert a.passed, [(c.id, c.detail) for c in a.checks if c.status == "FAIL"]
# ── evidence-lineage / publication acceptance: corruption tests A–G ──────────────────────────────────────
def test_A_numeric_prose_absent_from_the_final_registry_fails(good):
import report_model as RM
run, _ = good
ids = _audit_ids(run, [P.Fragment("Interest expense guidance of $390M–$410M consumes cash flow.", origin="llm")])
assert ids["FINAL_MATERIAL_CLAIM_LINEAGE"].status == "FAIL"
# a guidance claim fragment for a row the final state does not contain
n = len(run.state.guidance.company["metrics"]) + 5
ids = _audit_ids(run, [P.Fragment("Interest expense guidance of ", origin="llm"),
P.Fragment("$390M–$410M", f"GUIDANCE_{n}", None, "llm"), P.Fragment(".", origin="llm")])
assert ids["FINAL_MATERIAL_CLAIM_LINEAGE"].status == "FAIL"
# every guidance value an author cites is in the published table
cat = C.build_catalog(run.state, include_deferred=True)
rep = RM.build_user_report(run.state, cat, {"summary": [], "thesis": [], "risks": [], "seats": {}}, [])
table = {f.claim_id for b in rep.blocks if b.kind == "table" and b.section == "forecast" for f in b.fragments() if f.claim_id}
assert all(f"GUIDANCE_{i}" in table for i, _m in enumerate(run.state.guidance.company["metrics"], 1))
def test_B_internal_qa_status_in_user_prose_fails(good):
run, _ = good
cat, rules = _rules(run)
for s_ in ("The engine forecasts revenue of [[FORECAST_REVENUE]], but forecast coherence is unverifiable.",
"The forecast is unverified.", "The claim registry shows a different figure."):
ok, why = P.check_sentence(s_, rules)
assert not ok and why.startswith("PUBLIC_LANGUAGE_ONLY"), s_
assert _audit_ids(run, [P.Fragment("Forecast coherence is unverifiable.", origin="llm")])["PUBLIC_LANGUAGE_ONLY"].status == "FAIL"
import brief as B
assert "coherence" not in B.persona_brief(run.state, C.build_catalog(run.state, include_deferred=False)).lower()
def test_C_descriptive_insider_evidence_cannot_be_market_pressure(good):
run, _ = good
cat, rules = _rules(run)
assert "MARKET_PRESSURE" not in rules.evidence_uses["INSIDER"]["allowed_conclusion_types"]
for s_ in ("Insider filings show [[INSIDER_DISCRETIONARY_NET]] of purchases, so there is no selling pressure.",
"[[INSIDER_DISCRETIONARY_NET]] of insider purchases is a tailwind.", "Insider buying signals management confidence."):
ok, why = P.check_sentence(s_, rules)
assert not ok and (why.startswith("EVIDENCE_CONCLUSION_SCOPE") or why.startswith("EVIDENCE_INTERPRETATION")), s_
assert P.check_sentence("Discretionary insider purchases totaled [[INSIDER_DISCRETIONARY_NET]] and were immaterial relative "
"to market capitalization.", rules)[0]
def test_D_beta_cannot_establish_transient_drawdowns(good):
run, _ = good
cat, rules = _rules(run)
s_ = "With a bottom-up beta of [[DMD_BETA]], drawdowns are likely transient entry points rather than permanent impairment."
ok, why = P.check_sentence(s_, rules)
assert not ok and why.startswith("EVIDENCE_CONCLUSION_SCOPE") and "beta" in why
assert _audit_ids(run, _llm_frags(run, s_))["EVIDENCE_CONCLUSION_SCOPE"].status == "FAIL"
assert P.check_sentence("The bottom-up beta of [[DMD_BETA]] is an input to the discount rate.", rules)[0]
def test_E_third_party_headline_cannot_become_issuer_fundamental_evidence(good):
run, _ = good
cat, rules = _rules(run)
s_ = "Recent headlines on ratings actions underscore the company's moat and secular growth."
ok, why = P.check_sentence(s_, rules)
assert not ok and why.startswith("NEWS_EVIDENCE_SCOPE")
assert _audit_ids(run, [P.Fragment(s_, origin="llm")])["NEWS_EVIDENCE_SCOPE"].status == "FAIL"
assert P.check_sentence("Recent headlines primarily concern ratings or index actions involving other issuers.", rules)[0]
assert P.check_sentence("Recent headlines are not catalysts for the company.", rules)[0]
def test_F_lifecycle_descriptor_duplicated_around_its_token_fails(good):
run, _ = good
cat, rules = _rules(run)
stage = cat["LIFECYCLE_STAGE"].rendered
ok, why = P.check_sentence(f"The company carries a {stage} classification [[LIFECYCLE_STAGE]].", rules)
assert not ok and why.startswith("CLAIM_QUALIFIER_DOUBLED")
assert P.check_sentence("The life-cycle classification is [[LIFECYCLE_STAGE]].", rules)[0]
ok, why = P.check_sentence("Shares are [[RETURN_12M_PHRASE]] over twelve months.", rules)
assert not ok and why.startswith("CLAIM_QUALIFIER_DOUBLED")
def test_G_historical_segment_count_cannot_describe_the_continuing_company(good):
run, _ = good
cat, rules = _rules(run)
for s_ in ("The company now has [[SEGMENT_COUNT_FILED]] reportable segments covering [[SEGMENTS_COVERAGE]] of revenue.",
"The company reports five segments today."):
ok, why = P.check_sentence(s_, rules)
assert not ok and why.startswith("SEGMENT_PERIMETER_LANGUAGE"), s_
assert P.check_sentence("Before the separation, the consolidated company reported [[SEGMENT_COUNT_FILED]] segments.", rules)[0]
def test_ordinary_words_are_not_internal_qa_status(good):
"""Live SPGI (withheld 1790534165): R5 read the ordinary word "fail" as the QA status FAIL."""
import report_model as RM
assert RM.QA_VOCAB.search("the check returned FAIL") and RM.QA_VOCAB.search("status PASS")
assert not RM.QA_VOCAB.search("margins could fail to expand if issuance slows")
def test_descriptive_evidence_cannot_be_a_reason_and_clause_phrases_capitalise(good):
"""Live SPGI (aac508e43deeb009): "What would make this wrong: the number of institutions … fell by 51 …"."""
import dataclasses
run, _ = good
cat, rules = _rules(run)
RD = dataclasses.replace(rules, directional_context=True)
ok, why = P.check_sentence("[[OWN_13F_HOLDERS_CHANGE_PHRASE]].", RD)
assert not ok and why.startswith("EVIDENCE_INTERPRETATION")
assert not P.check_sentence("Recent headlines concern index actions on other issuers.", RD)[0]
assert P.check_sentence("[[OWN_13F_HOLDERS_CHANGE_PHRASE]].", rules)[0] # descriptive in a descriptive place
frags = P.to_fragments("[[OWN_13F_HOLDERS_CHANGE_PHRASE]].", rules)
assert frags[0].text[0].isupper()
def test_a_failed_guidance_extraction_is_not_an_absence():
calls = []
def fails(system, user):
calls.append(1)
return None
st = G.extract(dict(T.RELEASE), T.TABLE, fails)
assert len(calls) >= 2 and st.get("extraction_failed") and "did not return" in st["reason"]
assert "states no forward guidance" not in st["reason"]
# ── acceptance pass (live SPGI f5bdea53e59785f0): four defect classes the audit passed ──────────────────────────
def test_1_ownership_others_count_against_the_tracked_count_not_the_name_sample(good):
"""Live: "16 tracked long-term investors … including [6 names] and 2 others" (the snapshot keeps 8 names)."""
import validation as V
run, _ = good
f = run.state.ownership.get("f13")
assert len(f["guru_names_shown"]) + f["guru_names_others"] == int(f["n_gurus"])
assert f["guru_names_short"].endswith(f"and {f['guru_names_others']} others")
chk = {c.id: c for c in V.validate_data(run.state).checks}["OWNERSHIP_NAME_COUNT"]
assert chk.status == "PASS"
fresh, _p, _f = spgi_run() # a separate run: the fixture state stays intact
corrupt = fresh.state
object.__setattr__(corrupt, "_frozen", False)
corrupt._sealed.discard("ownership") if hasattr(corrupt._sealed, "discard") else None
corrupt.ownership.set("f13", dict(f, guru_names_others=2))
chk = {c.id: c for c in V.validate_data(corrupt).checks}["OWNERSHIP_NAME_COUNT"]
assert chk.status == "FAIL" and chk.blocking
def test_2_relation_clause_cannot_follow_which_nor_restate_its_operands(good):
"""Live: "Buybacks at [[PRICE]], [[REL_PRICE_INTRINSIC]], destroy value" and "… of [[FORECAST_EPS]], which
[[REL_FORECAST_EPS_CONSENSUS]]"."""
run, _ = good
cat, rules = _rules(run)
for s_, rule in (("Buybacks at [[PRICE]], [[REL_PRICE_INTRINSIC]], destroy value per dollar spent.", "CLAIM_QUALIFIER_DOUBLED"),
("Buybacks now, [[REL_PRICE_INTRINSIC]], destroy value per dollar spent.", "RELATION_CLAUSE_GRAMMAR"),
("The forward P/E is [[PE_FORWARD]], which [[REL_FORECAST_EPS_CONSENSUS]].", "RELATION_CLAUSE_GRAMMAR"),
("Engine EPS of [[FORECAST_EPS]] is high: [[REL_FORECAST_EPS_CONSENSUS]].", "CLAIM_QUALIFIER_DOUBLED")):
ok, why = P.check_sentence(s_, rules)
assert not ok and why.startswith(rule), (s_, why)
for s_ in ("[[REL_PRICE_INTRINSIC]], leaving little margin of safety.",
"Because buybacks continue, [[REL_PRICE_MARKET_COMPOSITE]].",
"Per the engine, [[REL_FORECAST_EPS_CONSENSUS]], which leaves room for estimate revisions."):
assert P.check_sentence(s_, rules)[0], s_
def test_3a_a_fundamental_versus_what_the_market_prices_in_follows_the_expectations_relation(good):
"""Live: "Growth is stronger than the market is pricing in" while the price requires MORE than the story."""
run, _ = good
cat, rules = _rules(run)
assert rules.expectations_polarity == RL.MORE_DEMANDING
for s_ in ("Growth is stronger than the market is pricing in.",
"The market is pricing in less growth than the business will deliver."):
ok, why = P.check_sentence(s_, rules)
assert not ok and why.startswith("EXPECTATIONS_POLARITY"), (s_, why)
assert P.check_sentence("The price requires growth beyond the adopted story.", rules)[0] # a faithful paraphrase
import dataclasses
ok, why = P.check_sentence("Growth is stronger than the market is pricing in.",
dataclasses.replace(rules, expectations_polarity=None))
assert not ok and "no expectations relation" in why
def test_3b_conservative_or_aggressive_only_as_the_typed_assessment_states(good):
"""Live: "a conservative failure probability of [[STORY_FAILURE]]" (0.0%: no haircut at all)."""
import dataclasses
run, _ = good
cat, rules = _rules(run)
R = dataclasses.replace(rules, assessments={"STORY_FAILURE": {"assessment": RL.NO_HAIRCUT, "basis": "0%"},
"STORY_WACC": {"assessment": RL.AGGRESSIVE, "basis": "below benchmark"}})
for s_ in ("The story uses a conservative failure probability of [[STORY_FAILURE]].",
"The initial cost of capital of [[STORY_WACC]] is conservative.",
"These conservative discounting inputs leave room for error.",
"The story's margin of [[STORY_MARGIN]] is cautious."): # unassessed: neither word
ok, why = P.check_sentence(s_, R)
assert not ok and why.startswith("ASSUMPTION_ASSESSMENT"), (s_, why)
assert P.check_sentence("The initial cost of capital of [[STORY_WACC]] is aggressive.", R)[0]
assert P.check_sentence("The failure probability of [[STORY_FAILURE]] is not conservative.", R)[0]
a = RL.assess_assumptions(run.state)
assert a["STORY_FAILURE"]["assessment"] in (RL.NO_HAIRCUT, RL.CONSERVATIVE)
def test_3c_stock_performance_needs_its_signed_return_claim(good):
"""Live: "after a weak year for the stock" with no return claim."""
import dataclasses
run, _ = good
cat, rules = _rules(run)
R = dataclasses.replace(rules, stock_returns={"RETURN_12M_PHRASE": -0.118, "RETURN_3M_PHRASE": 0.04})
for s_ in ("The market is undervaluing the business after a weak year for the stock.",
"The shares rallied after a strong year, [[RETURN_12M_PHRASE]].",
"After a weak year for the stock, [[RETURN_3M_PHRASE]]."):
ok, why = P.check_sentence(s_, R)
assert not ok and why.startswith("STOCK_PERFORMANCE_CLAIM"), (s_, why)
assert P.check_sentence("The stock is [[RETURN_12M_PHRASE]], a weak year for the shares.", R)[0]
assert P.check_sentence("If the shares sell off, buybacks become cheaper.", R)[0] # a conditional
TRANSCRIPT = ("to see strong growth this year. Thanks for the question.\n"
"Operator: We will now take our final question from Owen Lau with Clear Street.\n"
"Owen Lau: It just appears that it's hard to find a good way to monetize it or expand the revenue contribution.\n"
"Martina Cheung: We saw a base of about six hundred million in private markets revenues across the organization.")
def test_4a_a_call_quote_keeps_its_speakers_role(good):
"""Live: "The earnings call acknowledges … — 'hard to find a good way to monetize it' per the analyst question"."""
import dataclasses
run, _ = good
cat, rules = _rules(run)
class _Stub: # the two state paths TranscriptEvidence reads
def get(self, path):
return {"context.transcript_excerpt": TRANSCRIPT,
"leadership.roles": {"CEO": {"name": "Martina Cheung"}}}.get(path)
turns = EV.transcript_turns(_Stub())
roles = {t["speaker"]: t["role"] for t in turns if t["speaker"]}
assert roles == {"Operator": EV.OPERATOR, "Owen Lau": EV.ANALYST, "Martina Cheung": EV.MANAGEMENT}
R = dataclasses.replace(rules, transcript_turns=turns)
for s_ in ("The earnings call acknowledges monetization is difficult: 'hard to find a good way to monetize it'.",
"Management concedes it is 'hard to find a good way to monetize it'.",
"Management said on the call that 'private markets will be the fastest growing segment'."):
ok, why = P.check_sentence(s_, R)
assert not ok and why.startswith("QUOTE_ATTRIBUTION"), (s_, why)
assert P.check_sentence("An analyst on the call noted it is 'hard to find a good way to monetize it'.", R)[0]
assert P.check_sentence("The story calls the ratings arm a 'toll-bridge' business.", R)[0] # a scare quote
def test_4b_a_figure_stays_on_its_own_financial_statement(good):
"""Live: "Balance sheet carries deal-related amortization of [[GUIDANCE_9]]"."""
run, _ = good
cat, rules = _rules(run)
am = next((c for c, v in rules.statements.items() if v == G.INCOME_STATEMENT and c.startswith("GUIDANCE_")), None)
assert am, "fixture guidance has an income-statement row"
concept = (cat[am].value or {}).get("concept", "").lower().replace("_", " ")
ok, why = P.check_sentence(f"The balance sheet carries {concept} of [[{am}]].", rules)
assert not ok and why.startswith("STATEMENT_DOMAIN"), why
ok, why = P.check_sentence("The balance sheet holds revenue of [[REVENUE_TTM]].", rules)
assert not ok and why.startswith("STATEMENT_DOMAIN"), why
assert P.check_sentence("Revenue of [[REVENUE_TTM]] runs through the income statement.", rules)[0]
def test_new_rules_are_report_audits_too(good):
run, _ = good
ids = _audit_ids(run, _llm_frags(run, "The market is undervaluing the business after a weak year for the stock."))
assert ids["STOCK_PERFORMANCE_CLAIM"].status == "FAIL"
for cid in ("ASSUMPTION_ASSESSMENT", "QUOTE_ATTRIBUTION", "STATEMENT_DOMAIN"):
assert cid in ids
# ── remaining items from live SPGI 792f743c8943f0f8 ───────────────────────────────────────────────────────────
def test_5_predicate_and_list_claims_keep_their_render_slot(good):
"""Live: "With down 11.8% over twelve months already showing…" and "…; 2 shares withheld for tax (not directional)
shows minimal materiality"."""
run, _ = good
cat, rules = _rules(run)
ok, why = P.check_sentence("With [[RETURN_12M_PHRASE]] already showing exhaustion, sentiment is poor.", rules)
assert not ok and why.startswith("RELATION_CLAUSE_GRAMMAR") and "predicate" in why, why
for s_ in ("The shares are [[RETURN_12M_PHRASE]].", "With shares [[RETURN_12M_PHRASE]], buybacks buy more stock."):
assert P.check_sentence(s_, rules)[0], s_
lst = next((c for c, x in cat.items() if x.stype == C.TEXT and ";" in str(x.rendered or "")), None)
if lst:
ok, why = P.check_sentence(f"Insiders recorded [[{lst}]] shows little.", rules)
assert not ok and "list" in why, why
assert P.check_sentence(f"Insider activity: [[{lst}]].", rules)[0]
def test_6_expected_return_direction_matches_its_sign(good):
"""Live PM: "Expected return should sit close to neutral, tilted slightly bearish" at +6.9%."""
run, _ = good
cat, _r = _rules(run)
R = run.rules(cat, "PM_RETURN_PHRASE", run.state.scenarios.get("pm_label"))
v = cat["PM_RETURN_PHRASE"].value
bad = "tilted slightly bearish" if v > 0 else "tilted slightly bullish"
ok, why = P.check_sentence(f"Expected return should sit close to neutral, {bad}, because the gaps disagree.", R)
assert not ok and why.startswith("RETURN_DIRECTION_CONSISTENCY"), why
good_w = "tilted slightly bullish" if v > 0 else "tilted slightly bearish"
assert P.check_sentence(f"Expected return should sit close to neutral, {good_w}.", R)[0]
assert P.check_sentence("Closing toward the story value implies downside.", R)[0] # not the return's direction
def test_7_close_to_fair_value_needs_an_anchor_inside_the_band(good):
"""Live summary: "The shares trade close to fair value: the price is 12.6% above the story value…"."""
import dataclasses
run, _ = good
cat, rules = _rules(run)
far = dataclasses.replace(rules, valuation_magnitudes={"REL_PRICE_INTRINSIC": 0.126, "REL_PRICE_MARKET_COMPOSITE": 0.09})
for s_ in ("The shares trade close to fair value: [[REL_PRICE_INTRINSIC]].", "The stock looks fairly valued."):
ok, why = P.check_sentence(s_, far)
assert not ok and why.startswith("VALUATION_RELATION_LANGUAGE"), (s_, why)
near = dataclasses.replace(rules, valuation_magnitudes={"REL_PRICE_INTRINSIC": 0.03})
assert P.check_sentence("The shares trade close to fair value: [[REL_PRICE_INTRINSIC]].", near)[0]
assert P.check_sentence("The shares are far from fair value: [[REL_PRICE_INTRINSIC]].", far)[0]
def test_8_no_comparison_across_metric_families_without_a_relation(good):
"""Live bull: "Earnings power is set to compound faster than revenue"."""
run, _ = good
cat, rules = _rules(run)
for s_ in ("Earnings power is set to compound faster than revenue.", "Margins grow ahead of sales.",
"Free cash flow should outpace earnings."):
ok, why = P.check_sentence(s_, rules)
assert not ok and why.startswith("RELATION_REQUIRED_FOR_PAIR"), (s_, why)
assert P.check_sentence("Revenue growth could slow if issuance weakens.", rules)[0]
def test_9_trigger_names_its_guidance_basis_and_composite_labels_its_inputs():
"""Live: "If operating margin expansion rises to 360 bps" (the GAAP row; adjusted is 35–60 bps) and "20.6x trailing,
on adjusted EPS" for a multiple applied to next-twelve-month EPS."""
import report_model as RM
class _C:
def __init__(self, v):
self.value = v
cat = {"GUIDANCE_9": _C({"concept": "OPERATING_MARGIN_CHANGE", "accounting_basis": "GAAP", "low": 0.0335, "high": 0.036}),
"GUIDANCE_10": _C({"concept": "OPERATING_MARGIN_CHANGE", "accounting_basis": "ADJUSTED", "low": 0.0035, "high": 0.006}),
"GUIDANCE_2": _C({"concept": "REVENUE_GROWTH", "accounting_basis": "GAAP", "low": 0.059, "high": 0.079}),
"GUIDANCE_3": _C({"concept": "REVENUE_GROWTH", "accounting_basis": "ADJUSTED", "low": 0.059, "high": 0.079})}
assert C.guidance_basis_label(cat, "GUIDANCE_9") == "GAAP" and C.guidance_basis_label(cat, "GUIDANCE_10") == "adjusted"
assert C.guidance_basis_label(cat, "GUIDANCE_2") is None # same figure on both bases
# live f(1790537766): a trigger whose source is an ordinary metric claim (a float), not a guidance row
cat["FORECAST_REVENUE_GROWTH"] = _C(0.068)
cat["GUIDANCE_X"] = _C(0.05)
assert C.guidance_basis_label(cat, "FORECAST_REVENUE_GROWTH") is None and C.guidance_basis_label(cat, "GUIDANCE_X") is None
assert C.guidance_basis_label(cat, "GUIDANCE_9") == "GAAP"
lab = RM._method_basis({"multiple_basis": "TRAILING_ADJUSTED", "raw_input_basis": "NTM adjusted EPS (x)"})
assert "next-12-month adjusted EPS" in lab and "trailing basis" in lab and not lab.startswith(" trailing")
def test_10_segment_build_reconciles_and_a_cap_is_disclosed():
"""Live: segment margins weight to 52.7% while the valuation uses 49.4% (capped), with no disclosure."""
import validation as V
st = T.spgi_story()
st["assumptions"].update(revenue_N_usd_b=13.0, margin_target=0.494)
run, payload, failed = F.run_pipeline(data=T.spgi_data(), forecast_over={"story": st})
assert failed is None, failed
chk = {c.id: c for v in run.validations for c in v.checks}
assert chk["SEGMENT_BUILD_RECONCILES"].status == "PASS"
assert "is capped at" in payload["markdown"] and "(storyteller's proposal)" in payload["markdown"]
corrupt = run.state
object.__setattr__(corrupt, "_frozen", False)
corrupt._sealed.discard("valuation") if hasattr(corrupt._sealed, "discard") else None
story = dict(corrupt.valuation.get("story"))
story["assumptions"] = dict(story["assumptions"], margin_cap=0.40)
corrupt.valuation.set("story", story)
assert {c.id: c for c in V.validate_forecast(corrupt).checks}["SEGMENT_BUILD_RECONCILES"].status == "FAIL"
# ── false positives and misses from live SPGI 0eb399e58f07a852 ────────────────────────────────────────────────
def test_11_clause_scoped_rules_do_not_cross_clauses(good):
"""Live PM/neutral prose rejected: a direction bound to the wrong anchor across 'while', a negated conditional read
as a trend, a description read as a joint requirement, a pronoun subject, a PP-modified predicate subject."""
import dataclasses
run, _ = good
cat, rules = _rules(run)
R = dataclasses.replace(rules, valuation_relations={"REL_PRICE_INTRINSIC": RL.ABOVE, "REL_PRICE_MARKET_COMPOSITE": RL.BELOW})
for s_ in ("The price sits below market pricing while it sits above story value.",
"The stock is not stretched against market pricing even as it sits above story value.",
"Earnings support some upside even if the multiple does not expand.",
"A franchise with very high margins and steady revenue growth, but the stock already prices in strong economics.",
"The engine's forecast for the next twelve months [[REL_FORECAST_EPS_CONSENSUS__PREDICATE]]."):
ok, why = P.check_sentence(s_, R)
assert ok, (s_, why)
ok, why = P.check_sentence("The price sits above market pricing while it sits below story value.", R)
assert not ok and why.startswith("VALUATION_RELATION_LANGUAGE"), why
assert P._split("It operates in Financial Svcs. (Non-bank & Insurance) at high margins. Next.") == \
["It operates in Financial Svcs. (Non-bank & Insurance) at high margins.", "Next."]
def test_12_issuer_subject_nonbreaking_hyphen_and_count_units(good):
"""Live bull: "…that the market is undervaluing after a year of underperformance", "mature‑growth … [[LIFECYCLE_STAGE]]"
(U+2011) and "over [[STORY_YEARS]],"."""
import dataclasses
run, _ = good
cat, rules = _rules(run)
R = dataclasses.replace(rules, company_words=("acme data",))
ok, why = P.check_sentence("Acme Data is undervalued after a year of underperformance.", R)
assert not ok and why.startswith("STOCK_PERFORMANCE_CLAIM"), why
ok, why = P.check_sentence(f"A {cat['LIFECYCLE_STAGE'].rendered.replace('-', chr(0x2011))} compounder in the "
"[[LIFECYCLE_STAGE]] phase.", rules)
assert not ok and why.startswith("CLAIM_QUALIFIER_DOUBLED"), why
ok, why = P.check_sentence("Revenue reaches [[STORY_REVENUE_MATURITY]] over [[STORY_YEARS]], a long runway.", rules)
assert not ok and "count of years" in why, why
for s_ in ("Revenue reaches [[STORY_REVENUE_MATURITY]] over [[STORY_YEARS]] years.",
"A [[STORY_YEARS]]-year runway to maturity."):
assert P.check_sentence(s_, rules)[0], s_
# ── live SPGI efa889205cd5c53d ────────────────────────────────────────────────────────────────────────────────
def test_13_cut_sentences_unnamed_figures_named_counterparts_and_count_agreement(good):
"""Live: "…leaving limited…" (a cut key point), "requires [[BE_REVENUE]] at maturity", "upside to
[[FORECAST_REVENUE]]", "[[GUIDANCE_n]] …, far below the break-even revenue growth the price demands", "1 trims"."""
import predict_monthly as pm
run, _ = good
cat, rules = _rules(run)
long_one = "Risks include " + ", ".join(f"item {i}" for i in range(120)) + ", leaving limited room."
assert pm._coerce_key_points([{"topic": "Risks", "point": long_one}])[0]["point"] == long_one
ok, why = P.check_sentence("The story's cost of capital starts low and moves to a terminal rate with growth, leaving limited…", rules)
assert not ok and why.startswith("INCOMPLETE_SENTENCE"), why
for s_ in ("Today's price requires [[BE_REVENUE]] at maturity.", "New products can add upside to [[FORECAST_REVENUE]]."):
ok, why = P.check_sentence(s_, rules)
assert not ok and why.startswith("CLAIM_LABEL_BINDING"), (s_, why)
assert P.check_sentence("Today's price requires revenue of [[BE_REVENUE]] at maturity.", rules)[0]
g = next((c for c in cat if c.startswith("GUIDANCE_") and (cat[c].value or {}).get("concept") == "ORGANIC_REVENUE_GROWTH"), None)
if g and rules.pair_fn(g, "BE_REVENUE_CAGR") and rules.pair_fn(g, "BE_REVENUE_CAGR")[0] != "OK":
ok, why = P.check_sentence(f"Organic growth guidance of [[{g}]] sits far below the growth the price demands.", rules)
assert not ok and why.startswith("RELATION_REQUIRED_FOR_PAIR"), why
one = next((c for c in ("OWN_13F_TRIMS", "OWN_13F_EXITS", "OWN_13F_ADDS", "OWN_13F_NEW")
if c in cat and cat[c].value is not None and round(cat[c].value) == 1), None)
if one:
rest = [c for c in ("OWN_13F_NEW", "OWN_13F_ADDS", "OWN_13F_TRIMS", "OWN_13F_EXITS") if c != one]
tail = ", ".join(f"[[{c}]] moved" for c in rest)
ok, why = P.check_sentence(f"Last quarter there were [[{one}]] exits, and {tail}.", rules)
assert not ok and "plural" in why, why
def test_16_named_forecast_comparison_and_rebuttal_topics(good):
"""Live 5bcb332c678ba99f: "Guided 2026 adjusted EPS of $17.50–$17.75 could reset earnings power higher than our
forecast assumes" (the forecast is $20.52; the pair is not comparable), and a rebuttal topic — the OPPOSING claim —
shown as the author's own ("Limited earnings surprise potential justifies a SELL")."""
import report_model as RM
run, _ = good
cat, rules = _rules(run)
g = next((c for c in cat if c.startswith("GUIDANCE_") and (cat[c].value or {}).get("concept") == "EPS"), None)
if g and rules.pair_fn(g, "FORECAST_EPS") and rules.pair_fn(g, "FORECAST_EPS")[0] != "OK":
ok, why = P.check_sentence(f"Guided EPS of [[{g}]] could reset earnings power higher than our forecast assumes.", rules)
assert not ok and why.startswith("RELATION_REQUIRED_FOR_PAIR"), why
seats = {"bull_open": {"reasoning": [], "key_points": []},
"bull": {"reasoning": [], "key_points": [[P.Fragment("Limited upside justifies a SELL: the beat rate says otherwise.")]]}}
rep = RM.build_user_report(run.state, cat, {"summary": [], "thesis": [], "risks": [], "seats": seats, "stories": []}, [])
md = RM.render_markdown(rep)
assert "The bear's claim — Limited upside justifies a SELL" in md
# ── live SPGI 1c32779f5e1d3e46 ────────────────────────────────────────────────────────────────────────────────
def test_14_management_speech_needs_its_own_quoted_words(good):
"""Live bear: "The CEO acknowledged on the call that monetization remains challenging" — the CEO said "how we're
monetizing and growing in this space … very constructive"; the difficulty was an analyst's question."""
import dataclasses
run, _ = good
cat, rules = _rules(run)
turns = [{"speaker": "Owen Lau", "role": EV.ANALYST, "text": EV._norm("It's hard to find a good way to monetize it.")},
{"speaker": "Martina Cheung", "role": EV.MANAGEMENT,
"text": EV._norm("These are examples of how we're monetizing and growing in this space.")}]
R = dataclasses.replace(rules, transcript_turns=turns)
for s_ in ("The CEO acknowledged on the call that monetization remains challenging.",
"Management noted that private markets are hard to monetize.",
"The CEO said it is 'hard to find a good way to monetize it'."):
ok, why = P.check_sentence(s_, R)
assert not ok and why.startswith("QUOTE_ATTRIBUTION"), (s_, why)
assert P.check_sentence("The CEO said the company is 'monetizing and growing in this space'.", R)[0]
def test_15_verbal_magnitudes_spelled_counts_and_unnamed_multiples(good):
"""Live: "…supports high-teens EPS growth", "one trimmed … two adds and four new positions", "At [[PE_FORWARD]], …"."""
import dataclasses
run, _ = good
cat, rules = _rules(run)
ok, why = P.check_sentence("The forecast supports high-teens EPS growth.", rules)
assert not ok and why.startswith("VERBAL_MAGNITUDE"), why
fg = cat.get("LIFECYCLE_FORWARD_GROWTH")
if fg is not None and isinstance(fg.value, float) and 0.03 <= fg.value <= 0.10:
assert P.check_sentence("A franchise with durable single-digit growth and elite profitability.", rules)[0]
ok, why = P.check_sentence("Tracked investors made two adds and four new positions.", rules)
assert not ok and why.startswith("FINAL_MATERIAL_CLAIM_LINEAGE"), why
assert P.check_sentence("Beats came in over the last eight quarters.", rules)[0] # a window, not a count
ok, why = P.check_sentence("At [[PE_FORWARD]], the stock offers no margin of safety.", rules)
assert not ok and why.startswith("CLAIM_LABEL_BINDING"), why
assert P.check_sentence("At a forward P/E of [[PE_FORWARD]], the stock offers no margin of safety.", rules)[0]
def test_17_a_sourced_level_trigger_is_not_matched_to_a_guided_change_by_its_words():
"""Live SPGI 1790539998 withheld by S15: "operating margin below 45.0%" (source: the story's maturity margin, a LEVEL)
was word-matched to "Operating Profit Margin expansion" guidance (a bps CHANGE)."""
llm = F.llm_result()
llm["thesis_triggers"] = [{"metric_id": "operating margin", "condition_operator": "<=", "threshold": "45%",
"source_claim": "STORY_MARGIN", "scenario_destination": "BEAR",
"polarity": "NEGATIVE_FOR_EQUITY", "status": "OK"}]
run, _p, failed = T.run_spgi(llm=llm)
s15 = T._status(run, "S15")
assert s15[0] in ("PASS", "NOT_APPLICABLE"), s15
def test_18_universal_valuation_claims_and_paraphrase_then_token(good):
"""Live 27b06a16e6b17439: "The stock is expensive on every anchor" (the price is 9.0% below the market composite) and
"Today's price requires a demanding setup: [[REL_EXPECTATIONS_STORY]]" ("today's price requires" twice)."""
import dataclasses
run, _ = good
cat, rules = _rules(run)
R = dataclasses.replace(rules, valuation_relations={"REL_PRICE_INTRINSIC": RL.ABOVE, "REL_PRICE_MARKET_COMPOSITE": RL.BELOW})
ok, why = P.check_sentence("Valuation: The stock is expensive on every anchor.", R)
assert not ok and why.startswith("VALUATION_RELATION_LANGUAGE"), why
allr = dataclasses.replace(rules, valuation_relations={"REL_PRICE_INTRINSIC": RL.ABOVE, "REL_PRICE_MARKET_COMPOSITE": RL.ABOVE})
assert P.check_sentence("Valuation: The stock is expensive on every anchor.", allr)[0]
if "REL_EXPECTATIONS_STORY" in cat:
ok, why = P.check_sentence("Today's price requires a demanding setup: [[REL_EXPECTATIONS_STORY]].", rules)
assert not ok and why.startswith("CLAIM_QUALIFIER_DOUBLED"), why
def test_19_spread_assessments_consequences_of_descriptive_evidence_and_compounds(good):
"""Live f38c1e26f57229fd: "…failure probability of [[STORY_FAILURE]] …, suggesting downside is framed conservatively",
"insiders recorded … purchases of [[INSIDER_DISCRETIONARY_NET]], aligning leadership with shareholders", "CEO Cheung
cites growing traction", and the PM's "Story value premium: The price [[REL_PRICE_INTRINSIC__PREDICATE]]" (rejected)."""
import dataclasses
run, _ = good
cat, rules = _rules(run)
R = dataclasses.replace(rules, assessments={"STORY_FAILURE": {"assessment": RL.NO_HAIRCUT, "basis": "0%"}})
ok, why = P.check_sentence("The story uses a failure probability of [[STORY_FAILURE]], suggesting downside is framed "
"conservatively.", R)
assert not ok and why.startswith("ASSUMPTION_ASSESSMENT"), why
if "INSIDER_DISCRETIONARY_NET" in cat and (rules.interpretation.get("insider") or {}).get("permission") == EV.DESCRIPTIVE_ONLY:
ok, why = P.check_sentence("Insiders bought [[INSIDER_DISCRETIONARY_NET]] on the open market, aligning leadership with "
"owners.", rules)
assert not ok and why.startswith("EVIDENCE_INTERPRETATION"), why
Rt = dataclasses.replace(rules, transcript_turns=[{"speaker": "A", "role": EV.MANAGEMENT, "text": "we see growing traction"}])
ok, why = P.check_sentence("CEO Cheung cites growing traction but offers no revenue contribution.", Rt)
assert not ok and why.startswith("QUOTE_ATTRIBUTION"), why
rel = run.state.relations.get("state")["items"]["REL_PRICE_INTRINSIC"]["relation_type"]
word = "premium" if rel == RL.ABOVE else "discount"
assert P.check_sentence(f"Story value {word}: The price [[REL_PRICE_INTRINSIC__PREDICATE]].", rules)[0]
def test_20_boundary_repeats_stage_terms_elliptical_subjects_and_coordinated_predicates(good):
"""Live 0c5f1ba9a03231bc: "Today, [[REL_EXPECTATIONS_STORY]]" ("Today, today's …"), "S&P Global combines durable high
growth" for a mature-growth company, and PM prose rejected by false positives: "Shares trade below market pricing but
above the story value", "The price [[X__PREDICATE]] yet [[Y__PREDICATE]]", "Engine EPS for the next twelve months
[[REL_FORECAST_EPS_CONSENSUS__PREDICATE]]"."""
import dataclasses
run, _ = good
cat, rules = _rules(run)
if "REL_EXPECTATIONS_STORY" in cat and str(cat["REL_EXPECTATIONS_STORY"].rendered).lower().startswith("today"):
ok, why = P.check_sentence("Today, [[REL_EXPECTATIONS_STORY]].", rules)
assert not ok and why.startswith("CLAIM_QUALIFIER_DOUBLED"), why
stage = str(cat["LIFECYCLE_STAGE"].rendered) if "LIFECYCLE_STAGE" in cat else ""
if stage == "mature-growth":
R = dataclasses.replace(rules, company_words=("acme data",))
ok, why = P.check_sentence("Acme Data combines durable high growth with elite profitability.", R)
assert not ok and "life-cycle stage" in why, why
assert P.check_sentence("Acme Data sees high growth in private markets.", R)[0]
vr = dataclasses.replace(rules, valuation_relations={"REL_PRICE_INTRINSIC": RL.ABOVE, "REL_PRICE_MARKET_COMPOSITE": RL.BELOW})
assert P.check_sentence("Shares trade below market pricing but above the story value.", vr)[0]
ok, why = P.check_sentence("Shares trade above market pricing but below the story value.", vr)
assert not ok and why.startswith("VALUATION_RELATION_LANGUAGE"), why
for s_ in ("The price [[REL_PRICE_INTRINSIC__PREDICATE]] yet [[REL_PRICE_MARKET_COMPOSITE__PREDICATE]].",
"Engine EPS for the next twelve months [[REL_FORECAST_EPS_CONSENSUS__PREDICATE]]."):
ok, why = P.check_sentence(s_, rules)
assert ok, (s_, why)
def test_21_price_moves_return_bands_pronoun_objects_and_raw_ids(good):
"""Live 54c37e3e26ddaa90: "multiple compression to [[MARKET_COMPOSITE]] or below" ($443.00 is above the $403.30
price), "supporting high-teens total return" (the bull case is +22.0%), PM prose rejected for "the market composite sits
above it" and "the composite-vs-price discount", and a raw "REL_EXPECTATIONS_STORY" typed as text."""
import dataclasses
run, _ = good
cat, rules = _rules(run)
price = cat["PRICE"].value
above = next((c for c, x in cat.items() if x.stype == C.CURRENCY and c != "PRICE" and isinstance(x.value, float)
and x.value > price * 1.02 and c in (rules.offered or cat)), None)
if above:
ok, why = P.check_sentence(f"The bear case needs only modest multiple compression to [[{above}]] or below.", rules)
assert not ok and "would be a rise" in why, why
ok, why = P.check_sentence("Operating leverage supports high-teens total return.", rules)
fits = any(0.15 <= abs(cat[c].value) <= 0.21 for c in ("SCENARIO_BULL_RETURN", "PM_RETURN", "HEADLINE_RETURN",
"TARGET_VS_PRICE", "RETURN_12M", "SCENARIO_BEAR_RETURN",
"SCENARIO_NEUTRAL_RETURN", "RETURN_3M")
if c in cat and isinstance(cat[c].value, float))
assert ok == fits, why
vr = dataclasses.replace(rules, valuation_relations={"REL_PRICE_INTRINSIC": RL.ABOVE, "REL_PRICE_MARKET_COMPOSITE": RL.BELOW})
for s_ in ("The price sits between the anchors: the story value is well below price while the market composite sits above it.",
"I lean slightly bullish on the composite-vs-price discount."):
ok, why = P.check_sentence(s_, vr)
assert ok, (s_, why)
ok, why = P.check_sentence("REL_EXPECTATIONS_STORY shows the price needs more than the story.", rules)
assert not ok and why.startswith("PUBLIC_LANGUAGE_ONLY"), why
def test_22_counts_after_participles_coordinated_predicates_and_direction_adjectives(good):
"""Live 734d958b38a57d17 false positives: "[[OWN_13F_HOLDERS_CHANGE_PHRASE]], leaving [[OWN_13F_HOLDERS]] institutional
holders", "the price [[REL_PRICE_INTRINSIC__PREDICATE]] and [[REL_PRICE_MECHANICAL__PREDICATE]]", "the composite
that argues cheap"."""
import dataclasses
run, _ = good
cat, rules = _rules(run)
assert P.check_sentence("[[OWN_13F_HOLDERS_CHANGE_PHRASE]], leaving [[OWN_13F_HOLDERS]] institutional holders.", rules)[0]
ok, why = P.check_sentence("The price [[REL_PRICE_INTRINSIC__PREDICATE]] and [[REL_PRICE_MECHANICAL__PREDICATE]].", rules)
assert ok, why
vr = dataclasses.replace(rules, valuation_relations={"REL_PRICE_INTRINSIC": RL.ABOVE, "REL_PRICE_MARKET_COMPOSITE": RL.BELOW})
assert P.check_sentence("The value anchors show the price rich, outweighing the composite that argues cheap.", vr)[0]
ok, why = P.check_sentence("The value anchors show the price rich, outweighing the composite that argues expensive.", vr)
assert not ok and "contradicts" in why, why
def test_23_high_or_low_about_an_assumption_is_an_assessment(good):
"""Live 3e73c4a91f1b9dae bull rebuttal: "…a terminal cost of capital of [[STORY_WACC_STABLE]], which is high for a
dominant, asset-light oligopolist" (it equals the mature-market benchmark: NEUTRAL)."""
import dataclasses
run, _ = good
cat, rules = _rules(run)
R = dataclasses.replace(rules, assessments={"STORY_WACC_STABLE": {"assessment": RL.NEUTRAL, "basis": "equal to benchmark"},
"STORY_WACC": {"assessment": RL.AGGRESSIVE, "basis": "below benchmark"}})
for s_ in ("The story uses a terminal cost of capital of [[STORY_WACC_STABLE]], which is high for a dominant franchise.",
"The initial cost of capital of [[STORY_WACC]] is high."):
ok, why = P.check_sentence(s_, R)
assert not ok and why.startswith("ASSUMPTION_ASSESSMENT"), (s_, why)
assert P.check_sentence("The initial cost of capital of [[STORY_WACC]] is low.", R)[0] # low WACC = aggressive
assert P.check_sentence("A high-quality franchise with an initial cost of capital of [[STORY_WACC]].", R)[0]
def test_24_trigger_basis_is_not_written_twice(good):
"""A guidance-sourced trigger is named by the guidance row's own label, period and basis, each once (the subject no
longer comes from the author's metric string, which could repeat the basis)."""
import report_model as RM
run, _ = good
cat = C.build_catalog(run.state, include_deferred=True)
g = next((c for c in sorted(cat) if c.startswith("GUIDANCE_") and "__" not in c and C.guidance_basis_label(cat, c)), None)
if g is None:
pytest.skip("no guidance row with a basis in the fixture")
subj, b = RM.trigger_subject(run.state, cat, g), C.guidance_basis_label(cat, g)
# the row's own label (scope qualifiers kept), its period and basis once each
rows = run.state.guidance.get("company")["metrics"]
assert subj.startswith(str(rows[int(g.split("_")[1]) - 1]["label"])), subj
assert subj.lower().count(b.lower()) == 1 and "_" not in subj and not subj.startswith("company guidance"), subj
def test_25_predicate_verb_agrees_with_its_subject(good):
"""Live 4c2a118fbd859fc7: "the shares is 9.0% below market pricing"."""
run, _ = good
cat, rules = _rules(run)
r = str(cat["REL_PRICE_MARKET_COMPOSITE__PREDICATE"].rendered)
if r.startswith("is "):
ok, why = P.check_sentence("Despite quality, the shares [[REL_PRICE_MARKET_COMPOSITE__PREDICATE]].", rules)
assert not ok and "plural" in why, why
assert P.check_sentence("Despite quality, the price [[REL_PRICE_MARKET_COMPOSITE__PREDICATE]].", rules)[0]
def test_26_pipeline_notes_never_reach_an_author(good):
"""Live 6ffe4638943f601d bear: "restated historicals show 8 quarters of data-provider discrepancies" — the report's own
note about its DATA PROVIDER turned into a company risk. Fixed at the input: the reconciliation's basis text and notes
(typed DATA_PIPELINE, allowed only in the limitations section) never appear in any author-facing prompt. No word-list
rule: "S&P Global is the leading data provider" is a true company statement and must stay publishable."""
import brief as B
run, _ = good
cat, rules = _rules(run)
rec = run.state.historical.get("reconciliation") or {}
for n in rec.get("pipeline_notes") or []:
assert n["subject"] == "DATA_PIPELINE" and n["allowed_uses"] == ["REPORT_LIMITATIONS"]
for text in (B.persona_brief(run.state, cat), B.summary_prompt(run.state, cat)):
assert rec.get("basis") and rec["basis"] not in text
for lim in rec.get("user_limitations") or []:
assert lim not in text, lim
assert P.check_sentence("S&P Global is the leading data provider to capital markets.", rules)[0]
"""SPGI relationship-layer remediation (forensic package SPGI_6901d42876e85a17): RelationshipClaim, typed break-even
and story metrics, model-evidence scope, guidance epistemic state, semantic returns and typed blend sources.
Corruption tests A–K each assert the named validator fails. Structure only: no SPGI rating or value is asserted.
"""
import copy
import os
import sys
import pytest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import catalog as C # noqa: E402
import evidence as EV # noqa: E402
import guidance as G # noqa: E402
import pipeline_fixtures as F # noqa: E402
import prose as P # noqa: E402
import relations as RL # noqa: E402
import test_spgi_remediation as T # noqa: E402 (SPGI-shaped data, story and guidance fixtures)
def _story_with_breakeven(bm=0.45, be_rev=31.5, be_margin=0.55):
d = T.spgi_data()
start = d["story_base"]["revenue"]
st = T.spgi_story(base=d["story_base"])
st["assumptions"].update(revenue_N_usd_b=25.2, tam_usd_b=85.0, years=10, margin_target=0.494)
st["breakeven"] = {"price": 403.3, "story_value": 333.11, "story_revenue_N_usd_b": 25.2, "story_margin": bm, "years": 10,
"revenue_needed_N_usd_b": be_rev, "rev_cagr_needed": round((be_rev * 1e9 / start) ** 0.1 - 1.0, 4),
"revenue_needed_reachable": True, "margin_needed": be_margin, "margin_needed_reachable": True}
return d, st
def spgi_run(llm=None, summary=None, panel_visible=False, data=None, story=None, panel_stats=True):
d, st = _story_with_breakeven()
d = data or d
st = story or st
pf_ = F.pred_fund(); pf_.update(revenue_annual=11.8e9, eps_annual=4.3)
over = {"pred_fund": pf_, "story": st}
import policy as pol
if panel_visible:
over["panel_rel"] = dict(pol.panel_reliability(None), state="LIMITED", persona_visibility=True, usable_for_argument=True,
rank_ic=0.05, ic_positive_share=0.62)
over["panel_row"] = {"decile": 8, "xs_rank_pct": 80.0, "pred_excess_12m": 0.05}
elif panel_stats:
# as in the live SPGI run: the rank model is UNRELIABLE and hidden, yet its statistics exist in the state
over["panel_rel"] = dict(pol.panel_reliability(None), state="UNRELIABLE", persona_visibility=False, rank_ic=0.0346,
ic_positive_share=0.688)
return F.run_pipeline(data=d, forecast_over=over, llm=llm, summary=summary)
def good():
run, payload, failed = spgi_run()
assert failed is None, [(v.stage, c.id, c.detail) for v in run.validations for c in v.blocking]
return run, payload
def _rules(run):
cat = C.build_catalog(run.state, include_deferred=True)
return cat, run.rules(cat)
def _audit_ids(run, extra_text_frags):
"""Run the report audit on the run's report plus one authored (llm) sentence."""
import report_model as RM
cat = C.build_catalog(run.state, include_deferred=True)
rep = RM.build_user_report(run.state, cat, {"summary": [], "thesis": [], "risks": [], "seats": {}}, [])
rep.blocks.append(RM.Block("para", "summary", extra_text_frags))
a = RM.audit(rep, run.state, RM.render_markdown(rep), RM.render_html(rep))
return {c.id: c for c in a.checks}
def _llm_frags(run, sentence):
"""Fragments of an authored sentence, as ingestion would build them, bypassing the ingestion checks (a corruption)."""
cat = C.build_catalog(run.state, include_deferred=True)
frags = P.to_fragments(sentence, P.Rules(catalog=cat))
for f in frags:
f.origin = "llm"
return frags
# ── component: RelationshipClaim ──────────────────────────────────────────────────────────────────────
def test_relationships_have_explicit_sides_operator_and_magnitude(good):
run, _ = good
items = run.state.relations.get("state")["items"]
r = items["REL_PRICE_MARKET_COMPOSITE"]
assert (r["lhs_claim_id"], r["rhs_claim_id"]) == ("PRICE", "MARKET_COMPOSITE")
comp, price = run.state.value("pricing.market_composite"), run.state.value("market.price")
assert r["relation_type"] == (RL.BELOW if price < comp else RL.ABOVE)
assert r["magnitude"] == pytest.approx(abs(price / comp - 1)) and r["magnitude_definition"] == RL.RATIO_GAP
assert r["rendered"].startswith("the price (") and "the market composite (" in r["rendered"]
for cid in ("REL_PRICE_INTRINSIC", "REL_PRICE_MECHANICAL", "REL_FORECAST_EPS_CONSENSUS", "REL_TARGET_PRICE",
"REL_BREAK_EVEN_REVENUE_STORY", "REL_BREAK_EVEN_MARGIN_STORY", "REL_BREAK_EVEN_REVENUE_GROWTH", "REL_EXPECTATIONS_STORY"):
assert cid in items, cid
assert RL.check(run.state) == []
def test_break_even_relations_are_typed_and_bound(good):