tests/test_crowd.py

"""Offline eval harness for the crowd gate.

Runs with no API key and no network: the LLM judge and embeddings are replaced by
the deterministic `HeuristicSoundnessJudge` and `HashingEmbedder`. Verifies both
the pure scoring math and the end-to-end gating behaviour against fixtures.

Run:  python tests/test_crowd.py     (or: python -m pytest tests/test_crowd.py)
"""
import sys
import tempfile
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from crowd import (  # noqa: E402
    Contribution, CrowdGate, GateConfig, HashingEmbedder, HeuristicSoundnessJudge,
    ContributionStore, js_divergence, aggregate, novelty, cosine,
)

# --- fixtures --------------------------------------------------------------- #
TARGET = "US data-center AI capex exceeds $400B in 2026"
PRIOR_P = 0.7
PRIOR_RATIONALE = ("Hyperscaler capital expenditure keeps rising because compute "
                   "demand from large models compounds; therefore capex should exceed "
                   "the threshold as buildout continues across cloud providers.")

ECHO = Contribution(
    id="c_echo", target_id="t1", contributor_id="u_echo", probability=0.7,
    # a restatement of the consensus thesis (same viewpoint). The offline test
    # embedder is lexical, so this is worded close to PRIOR_RATIONALE; a real
    # embeddings backend flags paraphrase too, not just shared vocabulary.
    argument=("Hyperscaler capital expenditure keeps rising because compute demand "
              "from large models compounds; therefore capex will exceed the threshold "
              "as buildout continues across cloud providers, since order books remain "
              "full and given announced budgets the rising trend should clearly hold."),
    evidence_urls=["http://example.com/earnings"])

DISSENT_A = Contribution(
    id="c_dissentA", target_id="t1", contributor_id="u_grid", probability=0.35,
    argument=("However, an energy and power-grid bottleneck will cap deployments, "
              "because interconnection queues and transformer shortages mean physical "
              "buildout lags announced budgets, so realized spending falls short."),
    evidence_urls=["http://example.com/grid-report"])

DISSENT_B = Contribution(  # near-duplicate of A -> should collapse
    id="c_dissentB", target_id="t1", contributor_id="u_grid2", probability=0.33,
    argument=("However, power-grid and energy bottlenecks will cap deployments, "
              "because interconnection queues and transformer shortages mean buildout "
              "lags budgets, so realized spending falls short of plans."),
    evidence_urls=["http://example.com/grid2"])

NOISE = Contribution(
    id="c_noise", target_id="t1", contributor_id="u_troll", probability=0.1,
    argument="No. Wrong. Bubble.", evidence_urls=["http://example.com/x"])

NO_EVIDENCE = Contribution(
    id="c_noev", target_id="t1", contributor_id="u_eloquent", probability=0.4,
    argument=("The market will correct because valuations are stretched, however "
              "depreciation schedules imply writedowns, therefore reported capex "
              "may be revised downward as utilization disappoints over the year."),
    evidence_urls=[])


def _gate(k=5):
    return CrowdGate(HashingEmbedder(), HeuristicSoundnessJudge(),
                     cfg=GateConfig(k=k))


def _run(contribs, k=5):
    return _gate(k).process(TARGET, "t1", PRIOR_P, PRIOR_RATIONALE, contribs)


# --- pure math tests -------------------------------------------------------- #
def test_js_divergence_properties():
    assert abs(js_divergence(0.5, 0.5)) < 1e-9            # identical -> 0
    assert js_divergence(0.99, 0.01) > 0.9                # opposite -> ~1
    assert abs(js_divergence(0.2, 0.8) - js_divergence(0.8, 0.2)) < 1e-9  # symmetric
    print("ok  js_divergence properties")


def test_aggregate_bounds_and_direction():
    # one strong dissent at 0.3 should pull a 0.7 prior downward
    out = aggregate(0.7, [(0.3, 1.0)], prior_weight=1.0)
    assert 0.3 < out < 0.7
    # no contributions -> unchanged
    assert abs(aggregate(0.7, []) - 0.7) < 1e-9
    # always in [0,1]
    assert 0.0 <= aggregate(0.9, [(0.95, 5.0)], extremize=2.0) <= 1.0
    print("ok  aggregate bounds + direction")


def test_cosine_self_is_one():
    v = HashingEmbedder().embed(["energy grid transformer shortage"])[0]
    assert abs(cosine(v, v) - 1.0) < 1e-6
    print("ok  cosine self-similarity")


# --- behavioural gate tests ------------------------------------------------- #
def test_no_evidence_rejected():
    res = _run([NO_EVIDENCE])
    d = {x.contribution_id: x for x in res.decisions}["c_noev"]
    assert not d.admitted and d.reason == "no_evidence"
    print("ok  eloquent-but-unsourced rejected at evidence gate")


def test_noise_rejected():
    res = _run([NOISE])
    d = {x.contribution_id: x for x in res.decisions}["c_noise"]
    assert not d.admitted and d.reason == "weak_argument"
    print("ok  short contrarian noise rejected at soundness gate")


def test_echo_rejected_as_redundant():
    res = _run([ECHO])
    d = {x.contribution_id: x for x in res.decisions}["c_echo"]
    assert not d.admitted and d.reason == "redundant"
    print("ok  consensus echo rejected as redundant (low entropy)")


def test_well_argued_dissent_admitted_and_moves_forecast():
    res = _run([DISSENT_A])
    d = {x.contribution_id: x for x in res.decisions}["c_dissentA"]
    assert d.admitted and d.weight > 0
    assert res.aggregate_probability < PRIOR_P  # dissent pulled it down
    print(f"ok  well-argued dissent admitted, forecast {PRIOR_P} -> "
          f"{res.aggregate_probability}")


def test_near_duplicate_dissent_collapses():
    res = _run([DISSENT_A, DISSENT_B])
    dd = {x.contribution_id: x for x in res.decisions}
    assert dd["c_dissentA"].admitted
    assert not dd["c_dissentB"].admitted and dd["c_dissentB"].reason == "redundant"
    print("ok  near-duplicate dissent collapsed to one representative (sparsity)")


def test_full_mix_selects_only_informative():
    res = _run([ECHO, DISSENT_A, DISSENT_B, NOISE, NO_EVIDENCE])
    assert res.selected == ["c_dissentA"]
    reasons = {x.contribution_id: x.reason for x in res.decisions}
    assert reasons["c_echo"] == "redundant"
    assert reasons["c_noise"] == "weak_argument"
    assert reasons["c_noev"] == "no_evidence"
    assert reasons["c_dissentB"] == "redundant"
    print("ok  full mix -> 1 of 5 admitted; the rest correctly rejected")


def test_sparsity_cap():
    # three genuinely distinct, well-argued views; k=2 keeps only two
    extra = Contribution(
        id="c_demand", target_id="t1", contributor_id="u_demand", probability=0.9,
        argument=("Moreover, enterprise adoption accelerates because inference demand "
                  "compounds, therefore providers over-provision capacity, thus pushing "
                  "spend above the threshold even if some projects slip."),
        evidence_urls=["http://example.com/adoption"])
    res = _run([DISSENT_A, extra, NO_EVIDENCE, NOISE], k=1)
    assert len(res.selected) == 1
    print("ok  sparsity cap k enforced (only k non-zero parameters)")


def test_determinism():
    a = _run([ECHO, DISSENT_A, DISSENT_B, NOISE])
    b = _run([ECHO, DISSENT_A, DISSENT_B, NOISE])
    assert a.aggregate_probability == b.aggregate_probability
    assert a.selected == b.selected
    print("ok  deterministic across runs")


def test_track_record_skill():
    with tempfile.TemporaryDirectory() as tmp:
        store = ContributionStore(Path(tmp) / "c.jsonl")
        # cold start -> neutral
        assert store.contributor_skill("u_new") == 0.5
        # a sharp contributor: confident-and-right across 3 targets
        for i in range(3):
            store.add(Contribution(id=f"s{i}", target_id=f"t{i}",
                                   contributor_id="u_sharp", probability=0.9,
                                   argument="x", evidence_urls=["u"]))
            store.resolve_target(f"t{i}", True)
        assert store.contributor_skill("u_sharp") > 0.9
        # a poorly calibrated one
        for i in range(3):
            store.add(Contribution(id=f"b{i}", target_id=f"q{i}",
                                   contributor_id="u_bad", probability=0.1,
                                   argument="x", evidence_urls=["u"]))
            store.resolve_target(f"q{i}", True)
        assert store.contributor_skill("u_bad") < 0.2
    print("ok  contributor skill earned from realised Brier (cold-start neutral)")


def main():
    fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
    for fn in fns:
        fn()
    print(f"\nALL {len(fns)} CROWD-GATE TESTS PASSED")


if __name__ == "__main__":
    main()