From 36a4af2c061e4316568551657bca001496a2bad7 Mon Sep 17 00:00:00 2001 From: TJ Baker <1617679+zaridan@users.noreply.github.com> Date: Tue, 14 Jul 2026 14:54:37 -0700 Subject: [PATCH 1/2] fix: make off-brand variant selection hash-seed-invariant `_pick_off_brand_variant_id` indexed the variant list with the builtin `hash(conv_id)`, which Python salts per process via PYTHONHASHSEED. Two otherwise-identical `rforge generate --dry-run --smoke` runs therefore produced different `off_brand_variant_id`s (and a different `planted_quality.jsonl` / `planted_quality_hash`), violating the documented bit-identical determinism guarantee. The function's own docstring already promised deterministic selection. Switch to a stable blake2b digest. Adds a subprocess regression test asserting the pick is identical across PYTHONHASHSEED values. Pre-existing bug, surfaced while adding contact modeling (whose corpus must be seed-reproducible). Co-Authored-By: Claude Opus 4.8 --- resonantforge/layer1/quality_plan_injector.py | 10 +++- tests/test_injector_offbrand_determinism.py | 57 +++++++++++++++++++ 2 files changed, 66 insertions(+), 1 deletion(-) create mode 100644 tests/test_injector_offbrand_determinism.py diff --git a/resonantforge/layer1/quality_plan_injector.py b/resonantforge/layer1/quality_plan_injector.py index cb40e79..5148adf 100644 --- a/resonantforge/layer1/quality_plan_injector.py +++ b/resonantforge/layer1/quality_plan_injector.py @@ -1,5 +1,6 @@ """Quality plan injector — plants rubric-dimension targets into the organic event log.""" +import hashlib import logging import math import random @@ -123,9 +124,16 @@ def _pick_off_brand_variant_id(conv_id: str, off_brand_variants: dict) -> str: Uses sorted key order so dict insertion order never affects selection. Hash is computed over the conv_id string to give uniform distribution across the variant set without introducing an external RNG dependency. + + The digest MUST be a stable hash (blake2b), not the builtin ``hash()`` — + ``hash()`` on ``str`` is salted per process by PYTHONHASHSEED, which made the + chosen off-brand variant (and therefore ``planted_quality.jsonl``) differ + across otherwise-identical runs, breaking the ``--dry-run --smoke`` + bit-identical determinism guarantee. """ keys = sorted(off_brand_variants.keys()) - return keys[hash(conv_id) % len(keys)] + digest = hashlib.blake2b(conv_id.encode("utf-8"), digest_size=8).digest() + return keys[int.from_bytes(digest, "big") % len(keys)] def _render_off_brand_directive(variant_spec: OffBrandVariantSpec) -> str: diff --git a/tests/test_injector_offbrand_determinism.py b/tests/test_injector_offbrand_determinism.py new file mode 100644 index 0000000..bc99f51 --- /dev/null +++ b/tests/test_injector_offbrand_determinism.py @@ -0,0 +1,57 @@ +""" +Regression test: off-brand variant selection must be stable across processes. + +`_pick_off_brand_variant_id` previously indexed with the builtin `hash(conv_id)`, +which Python salts per process via PYTHONHASHSEED. That made the chosen off-brand +variant — and therefore `planted_quality.jsonl` — differ across otherwise +identical `--dry-run --smoke` runs, breaking the bit-identical determinism +guarantee. It now uses a stable blake2b digest. This test pins that: the picked +variant for a fixed conv_id is identical under different PYTHONHASHSEED values. +""" +from __future__ import annotations + +import os +import subprocess +import sys + +from resonantforge.layer1.quality_plan_injector import _pick_off_brand_variant_id + +_VARIANTS = {"aggressive": 1, "clinical_detached": 1, "robotic": 1, "verbose": 1} + +# Inline script: print the picked variant for a set of conv_ids. Run under two +# different PYTHONHASHSEED values; the output must be byte-identical. +_SCRIPT = ( + "from resonantforge.layer1.quality_plan_injector import _pick_off_brand_variant_id as p;" + "v={'aggressive':1,'clinical_detached':1,'robotic':1,'verbose':1};" + "print(','.join(p(f'conv_evt_{i:05d}', v) for i in range(50)))" +) + + +def _run(hashseed: str) -> str: + env = {**os.environ, "PYTHONHASHSEED": hashseed} + out = subprocess.run( + [sys.executable, "-c", _SCRIPT], + capture_output=True, text=True, env=env, check=True, + ) + return out.stdout.strip() + + +def test_offbrand_pick_is_hashseed_invariant(): + a = _run("0") + b = _run("1") + c = _run("12345") + assert a == b == c, "off-brand variant selection varies with PYTHONHASHSEED" + # And it must be non-trivial (actually distributes across variants). + assert len(set(a.split(","))) > 1 + + +def test_offbrand_pick_matches_stable_digest(): + # Locks the algorithm to the stable digest, so a revert to builtin hash() is caught. + import hashlib + + keys = sorted(_VARIANTS.keys()) + for i in range(20): + conv_id = f"conv_evt_{i:05d}" + digest = hashlib.blake2b(conv_id.encode("utf-8"), digest_size=8).digest() + expected = keys[int.from_bytes(digest, "big") % len(keys)] + assert _pick_off_brand_variant_id(conv_id, _VARIANTS) == expected From 8eb8db1303c401b8c1fd319b7f918c1a2b258b58 Mon Sep 17 00:00:00 2001 From: TJ Baker <1617679+zaridan@users.noreply.github.com> Date: Tue, 14 Jul 2026 14:54:59 -0700 Subject: [PATCH 2/2] =?UTF-8?q?feat(intelligence):=20model=20customer=20co?= =?UTF-8?q?ntacts=20as=20first-class=20entities=20=E2=80=94=20relationship?= =?UTF-8?q?=20churn=20signals=20(RFORGE-93)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds first-class customer-contact modeling so the Resonant IQ engine's two relationship churn detectors — `relationship_champion_at_risk` (Rule 4) and `relationship_single_threaded` (Rule 5) — can be scored against planted ground truth. Both require tracking specific customer contacts across many conversations, which the corpus previously could not express. Schema (schemas.py) - Contact (account_id, contact_id, name, role, is_champion, engagement span) + ContactRole enum. - ContactRollup + DaySnapshot rollup fields (distinct_active_contacts_60d, distinct_active_contacts_prior_180d, total_contacts_all_time, champion_rollups) — trailing-window aggregates pre-rolled like health_score. - ConversationStartedPayload / ConversationRecord contact attribution. - RelationshipLabel + RelationshipDetector ground-truth schema. - Manifest: contact_count, relationship_label_count, hashes, and relationship_signal_distribution telemetry. Simulation (layer1/contact_planner.py, state_machine.py, sim_events.py) - ContactPlanner: a deterministic post-simulation pass (seed-derived RNG that never perturbs the event stream). Knows each account's realized lifespan, so planted conditions land relative to the account's actual "now". Generates contacts, attributes every conversation to a contact, enriches snapshots, and emits labels. Wired into StateMachine.simulate(). - Plants both positives (champion silence ≥30d and champion frequency <1/mo; narrowing to a single active thread from ≥3 prior) and near-miss decoys (active champion; narrowed-to-two; small-account guard) mirroring the RFORGE-73 distractor discipline, so false-positive rate is measurable. - Labels are computed by pure Python mirrors of the engine's Rule 4/5 logic (champion_at_risk_fires / single_threaded_fires) against the actual rollup — ground truth by construction. Prose (pipeline.py) - Conversations attributed to the specific customer contact; contact name/role threaded into the prose prompt and ConversationRecord. Artifacts + CLI - New contacts.jsonl and relationship_labels.jsonl (hashed in the manifest). - validate / inspect / stats extended to cover both. Validator (validators/relationship.py) - Account-level feature-extractor + rule-engine that independently recomputes windowed contact features from raw attribution and agrees with the planted ground truth — extends disagreement-ledger coverage to relationship signals. Determinism / audit - `rforge generate --dry-run --smoke` is bit-identical across independent processes (verified end-to-end; determinism property test extended to cover contacts + relationship labels). - docs/relationship-signal-distribution-audit.md audits the planted positive/negative/decoy distribution for both detectors (60-account dry-run). 17 new contact-modeling tests; full suite green (501 passed). Co-Authored-By: Claude Opus 4.8 --- .../relationship-signal-distribution-audit.md | 192 +++++ resonantforge/cli.py | 21 + resonantforge/layer1/contact_planner.py | 755 ++++++++++++++++++ resonantforge/layer1/sim_events.py | 13 + resonantforge/layer1/state_machine.py | 25 + resonantforge/pipeline.py | 65 +- resonantforge/schemas.py | 143 ++++ resonantforge/validators/relationship.py | 238 ++++++ tests/test_contact_modeling.py | 331 ++++++++ tests/test_properties.py | 10 + 10 files changed, 1791 insertions(+), 2 deletions(-) create mode 100644 docs/relationship-signal-distribution-audit.md create mode 100644 resonantforge/layer1/contact_planner.py create mode 100644 resonantforge/validators/relationship.py create mode 100644 tests/test_contact_modeling.py diff --git a/docs/relationship-signal-distribution-audit.md b/docs/relationship-signal-distribution-audit.md new file mode 100644 index 0000000..8673abc --- /dev/null +++ b/docs/relationship-signal-distribution-audit.md @@ -0,0 +1,192 @@ +# ResonantForge — Relationship-Signal Planted Distribution Audit + +**Corpus audited:** `saas` profile, `--dry-run --accounts 60 --months 8 --seed 42` +(60 accounts, 240 simulated days, generator v0.2.0) +**Signals covered:** `relationship_champion_at_risk` (engine Rule 4), +`relationship_single_threaded` (engine Rule 5) + +> These two detectors track *specific customer contacts across many +> conversations*, so they could not be exercised until the corpus modeled +> customer-side contacts as first-class entities. This audit reports the planted +> positive / negative / decoy distribution for both, so the benchmark harness can +> compute precision/recall/F1 and a false-positive rate per detector. + +The audit is fully reproducible in `--dry-run` (no LLM cost): contacts, +attribution, snapshot rollups, and labels are all deterministic from `--seed`. + +--- + +## Section 0 — What was planted, and how the ground truth is guaranteed + +Each account is assigned exactly one *relationship scenario* (positive, decoy, or +null). A scenario is only assigned to an account whose realized conversation +distribution can actually express it, so a planted positive genuinely fires and a +planted decoy genuinely does not. Labels are then computed from the account's +*actual* final-day contact rollup via pure Python mirrors of the engine's Rule 4 +/ Rule 5 logic (`champion_at_risk_fires` / `single_threaded_fires` in +`resonantforge/layer1/contact_planner.py`). The label is therefore, by +construction, exactly what a correct detector must output — not an aspiration. + +Two ground-truth labels are emitted per account (one per detector) to +`relationship_labels.jsonl`; the account's contacts go to `contacts.jsonl`; and +every `DaySnapshot` carries the pre-rolled trailing-window contact aggregates +(`distinct_active_contacts_60d`, `distinct_active_contacts_prior_180d`, +`total_contacts_all_time`, `champion_rollups`) so a scorer can evaluate the rules +without replaying events. + +**Engine thresholds mirrored (verified against +`resonantiq/src/lib/intelligence/churn-detectors.ts`):** + +| Detector | Fires when | +|---|---| +| Rule 4 champion-at-risk | account has ≥1 champion contact AND a champion (with known recency) is unseen ≥30d **OR** engages <1×/month (touches over trailing 3mo ÷ 3) | +| Rule 5 single-threaded | `totalContactsAllTime > 2` AND exactly **1** distinct contact active in the last 60d AND **≥3** distinct contacts active in the prior 60–240d window | + +--- + +## Section 1 — Scenario distribution (per account, n=60) + +| Scenario | Accounts | % | Plants for | +|---|---:|---:|---| +| `single_threaded_positive` | 10 | 16.7% | Rule 5 positive | +| `champion_silence_positive` | 8 | 13.3% | Rule 4 positive (silence branch) | +| `champion_lowfreq_positive` | 5 | 8.3% | Rule 4 positive (frequency branch) | +| `champion_active_decoy` | 7 | 11.7% | Rule 4 **decoy** (near-miss) | +| `single_threaded_decoy_narrowed` | 6 | 10.0% | Rule 5 **decoy** (narrowed to 2) | +| `single_threaded_decoy_guard` | 4 | 6.7% | Rule 5 **decoy** (small-account guard) | +| `null_multithreaded` | 14 | 23.3% | negative control (multi-threaded, healthy) | +| `sparse_single` | 6 | 10.0% | negative control (1-contact account) | + +Both Rule 4 positive branches are represented, so the OR in the detector is +exercised on both sides. + +--- + +## Section 2 — Per-detector label distribution + +Every account contributes one label per detector (n=60 each). + +### 2a. `relationship_champion_at_risk` (Rule 4) + +| Label | Count | % | +|---|---:|---:| +| Positive (should fire) | 13 | 21.7% | +| Negative (should not fire) | 47 | 78.3% | +|   — of which **decoy** (near-miss) | 7 | 11.7% | +|   — of which plain null | 40 | 66.7% | + +Positive : negative = **0.28 : 1** (positives are the minority — the correct +direction for a churn signal; most accounts should not trip it). +**20 / 60 accounts have a champion contact at all** (13 positives + 7 active +decoys), so the decoys are true FP tests: they *have* a champion, but the +champion is engaged, so a correct detector must stay silent. + +### 2b. `relationship_single_threaded` (Rule 5) + +| Label | Count | % | +|---|---:|---:| +| Positive (should fire) | 10 | 16.7% | +| Negative (should not fire) | 50 | 83.3% | +|   — of which **decoy** (near-miss) | 10 | 16.7% | +|   — of which plain null | 40 | 66.7% | + +Positive : negative = **0.20 : 1**. The 10 decoys split into the two distinct +near-miss shapes below. + +--- + +## Section 3 — Decoy inspection (false-positive traps) + +Decoys mirror the RFORGE-73 paraphrase-trap discipline: cases engineered to +*look* like a fire while being genuine negatives, so the FP rate is measurable. +Representative examples from the audited corpus (as-of day 239): + +### Rule 4 decoy — `champion_active_decoy` (acct_034) + +> Champion last seen **2 days ago**, engagement **≈1.33/month** → clear of both +> the 30-day silence bound and the 1×/month frequency bound. **Must NOT fire.** + +The account has a champion (so Rule 4 has a champion to evaluate), but the +champion is actively engaged — the single most common false positive a naive +detector would make. Frequency is planted comfortably above the boundary +(≥1.33, not exactly 1.0) so a one-touch ingestion drift can't flip the label. + +### Rule 5 decoy A — `single_threaded_decoy_narrowed` (acct_024) + +> `total=4, active_60d=2, prior_60_240d=3` → narrowed from 4 threads to **2**, +> not 1. **Must NOT fire.** + +A genuine relationship contraction that stops one contact short of the trigger — +the hardest single-threading false positive. + +### Rule 5 decoy B — `single_threaded_decoy_guard` (acct_030) + +> `total=2, active_60d=1, prior_60_240d=2` → effectively single-threaded, but the +> account only ever had **2** contacts, so the engine's `totalContactsAllTime > 2` +> guard suppresses it. **Must NOT fire.** + +Tests the small-account guard specifically: a detector that ignores the guard +would false-fire here. + +--- + +## Section 4 — Positive inspection (gold examples) + +### Rule 4 positive — silence branch — `champion_silence_positive` (acct_011) + +> Champion last seen **238 days ago**, frequency **0.0/month** → silence fires. +> `expected_fire=true`. + +### Rule 4 positive — frequency branch — `champion_lowfreq_positive` (acct_019) + +> Champion last seen **1 day ago** (not silent) but only **0.33/month** → the +> frequency branch fires independently of silence. `expected_fire=true`. + +### Rule 5 positive — `single_threaded_positive` (acct_001) + +> `total=4, active_60d=1, prior_60_240d=3` → exactly one recent thread, three +> distinct in the prior window, guard satisfied. `expected_fire=true`. The final +> `DaySnapshot` rollup for acct_001 carries these same three values, so a scorer +> reading the pre-roll reaches the identical verdict. + +--- + +## Section 5 — Integrity checks + +All enforced by `tests/test_contact_modeling.py` (15 tests) and +`tests/test_properties.py::test_determinism`: + +- **Determinism:** `contacts.jsonl`, `relationship_labels.jsonl`, and + `snapshots.jsonl` are byte-identical across two same-seed runs + (`--dry-run --smoke` stays reproducible). +- **Label ↔ evidence consistency:** every label's `expected_fire` equals the + detector mirror applied to its own recorded `evidence` (0 mismatches across + 120 labels). +- **Snapshot ↔ label consistency:** the final `DaySnapshot` per account carries + contact-rollup fields matching the single-threaded label's evidence. +- **Attribution completeness:** every emitted contact engages ≥1 conversation; + every `CONVERSATION_STARTED` event is attributed to an existing contact; + `total_contacts_all_time` equals the account's contact-row count. +- **Independent validator agreement:** the account-level relationship validator + (`resonantforge/validators/relationship.py`) recomputes windowed contact + features straight from raw attribution (not the pre-roll) and agrees with the + planted ground truth on every account × detector. + +--- + +## Section 6 — Consumption notes for the benchmark harness + +- **Ground truth:** `relationship_labels.jsonl` — one record per (account, + detector). Use `expected_fire` as the label; `is_decoy` marks the near-miss + negatives for a separate FP-rate cut. +- **Contacts:** `contacts.jsonl` — first-class contact rows + (`is_champion`, role, engagement span). Map to the engine's `contacts` table. +- **Pre-rolled features:** each `snapshots.jsonl` row carries the trailing-window + aggregates; the last snapshot per account is the as-of-"now" state the labels + are evaluated against (`as_of_day_index`). +- **Base rate:** positives are ~17–22% per detector by design (churn signals are + the minority). For a balanced training cut, sample decoys and plain nulls + separately using `is_decoy` and `scenario`. +- The `manifest.json` field `relationship_signal_distribution` carries the + per-detector positive/negative/decoy counts and the scenario tally shown in + Sections 1–2, so the distribution is auditable straight from the manifest. diff --git a/resonantforge/cli.py b/resonantforge/cli.py index 06bb05e..64ed605 100644 --- a/resonantforge/cli.py +++ b/resonantforge/cli.py @@ -288,6 +288,8 @@ def validate(corpus_dir: Path, profile: str) -> None: ("conversations.jsonl", manifest_data.get("conversation_count", 0)), ("planted_quality.jsonl", manifest_data.get("planted_quality_count", 0)), ("corrections.jsonl", manifest_data.get("corrections_count", 0)), + ("contacts.jsonl", manifest_data.get("contact_count", 0)), + ("relationship_labels.jsonl", manifest_data.get("relationship_label_count", 0)), ] for filename, expected_count in jsonl_specs: fpath = profile_dir / filename @@ -385,6 +387,8 @@ def inspect(corpus_dir: Path, profile: str, limit: int) -> None: ("events.jsonl", "Events"), ("conversations.jsonl", "Conversations"), ("corrections.jsonl", "Corrections"), + ("contacts.jsonl", "Contacts"), + ("relationship_labels.jsonl", "Relationship labels"), ] for filename, label in artifacts: @@ -512,9 +516,24 @@ def _section(title: str) -> None: _section("Artifacts") _row("Corrections", manifest_data.get("corrections_count", "")) _row("Agents", manifest_data.get("agent_count", "")) + _row("Contacts", manifest_data.get("contact_count", "")) + _row("Relationship labels", manifest_data.get("relationship_label_count", "")) _row("KB docs", manifest_data.get("knowledge_base_doc_count", "")) _row("KB chunks", manifest_data.get("knowledge_base_chunk_count", "")) + # Relationship-signal planted distribution (per-detector positive/negative/decoy). + rel_dist = manifest_data.get("relationship_signal_distribution", {}) + if rel_dist: + _section("Relationship signals (planted)") + for det in ("relationship_champion_at_risk", "relationship_single_threaded"): + b = rel_dist.get(det) + if b: + _row( + det, + f"{b.get('positive', 0)} positive · {b.get('negative', 0)} negative " + f"({b.get('decoy', 0)} decoy) / {b.get('total', 0)}", + ) + # Quality rates. _section("Quality rates") _row("Prose fact violation rate", f"{manifest_data.get('prose_fact_violation_rate', 0):.3f}") @@ -532,6 +551,8 @@ def _section(title: str) -> None: ("tenant_config_hash", "Tenant config"), ("agent_fixtures_hash", "Agent fixtures"), ("corrections_hash", "Corrections"), + ("contacts_hash", "Contacts"), + ("relationship_labels_hash", "Relationship labels"), ] for field_name, label in hash_keys: h = manifest_data.get(field_name, "") diff --git a/resonantforge/layer1/contact_planner.py b/resonantforge/layer1/contact_planner.py new file mode 100644 index 0000000..70d65b8 --- /dev/null +++ b/resonantforge/layer1/contact_planner.py @@ -0,0 +1,755 @@ +""" +Customer-contact modeling: generate first-class contacts, attribute +conversations to them, plant the two relationship churn conditions the +Resonant IQ engine detects, and emit ground-truth labels. + +Why this is a post-simulation pass +----------------------------------- +The two engine detectors this feeds — ``relationship_champion_at_risk`` +(Rule 4) and ``relationship_single_threaded`` (Rule 5) — evaluate an account's +contact-engagement graph as-of "now" (the account's final day). Planting them +therefore requires knowing each account's *realized* lifespan, which is only +known after the day-by-day simulation has run (accounts churn early at emergent, +health-driven times). So contacts are planned in a single pass over the finished +event/snapshot stream rather than woven into the day loop. + +Determinism +----------- +The planner draws from a seed-derived ``random.Random`` (salted so it never +perturbs the state machine's own event RNG stream). Given the same seed and the +same simulation output, it produces byte-identical contacts, attribution, +snapshot rollups, and labels. + +Ground truth by construction +---------------------------- +Every account is assigned one exclusive *relationship scenario* (positive, +decoy, or null), but scenarios are only assigned to accounts whose realized +conversation distribution can actually express them — so a planted positive +genuinely fires and a planted decoy genuinely does not. Labels are then computed +from the *actual* resulting rollup via :func:`champion_at_risk_fires` / +:func:`single_threaded_fires`, which are line-for-line mirrors of the engine's +detector logic. The label is thus, by construction, exactly what a correct +detector must output. +""" + +from __future__ import annotations + +import random +from dataclasses import dataclass, field + +from resonantforge.schemas import ( + SimEvent, + DaySnapshot, + SimEventType, + Contact, + ContactRole, + ContactRollup, + RelationshipLabel, + RelationshipDetector, +) + +# --------------------------------------------------------------------------- +# Thresholds — mirror the engine (churn-detectors.ts + cross-stream-helpers.ts). +# Kept here as the single source of truth for both attribution targeting and the +# label-generating mirror functions, so the two can never drift apart. +# --------------------------------------------------------------------------- + +CHAMPION_SILENCE_DAYS = 30 # Rule 4: days-since-seen ≥ this → fires +RECENT_WINDOW_DAYS = 60 # Rule 5: distinct contacts in trailing 60d +PRIOR_WINDOW_START_DAYS = 60 # Rule 5: prior window is [60, 240) days ago +PRIOR_WINDOW_END_DAYS = 240 +SINGLE_THREAD_MIN_ALLTIME = 3 # Rule 5 guard: totalContactsAllTime > 2 +# Per-contact engagement frequency window: 3 average months (3 × 30.44d), then +# the touch count is divided by 3 to yield touches/month (RIQAPP-186). +FREQUENCY_WINDOW_DAYS = 91 +FREQUENCY_WINDOW_MONTHS = 3.0 + +# Salt mixed into the seed so contact planning is deterministic but independent +# of the state machine's event RNG (planning must not shift the event stream). +_CONTACT_RNG_SALT = 0x00C0FFEE + +# Attribution safety margins — keep planted signal values clear of the engine's +# exact window boundaries so an off-by-one day at ingestion can't flip a label. +_CHAMPION_SILENT_BY = 40 # silent-champion convs are ≥40d old (≥30 with margin) +_CHAMPION_ACTIVE_WITHIN = 15 # active-champion decoy: latest champion conv ≤15d old +_CHAMPION_LOWFREQ_WITHIN = 29 # low-freq champion: latest champion conv ≤29d (still <30) + + +# --------------------------------------------------------------------------- +# Pure detector mirrors — the ground-truth oracle for both label generation and +# the relationship validator. These MUST stay faithful to the TypeScript engine. +# --------------------------------------------------------------------------- + + +def champion_at_risk_fires(champion_rollups: list[ContactRollup]) -> bool: + """ + Mirror of ``detectChampionSilence`` (churn-detectors.ts Rule 4). + + Fires when the account has ≥1 champion contact and, for a champion with a + known last-seen, either it hasn't been seen in ≥30 days OR its engagement + frequency is below 1/month. Champions never seen (last_seen is None) are + skipped, exactly as the engine skips contacts with no live recency. + """ + if not champion_rollups: + return False + for c in champion_rollups: + if c.last_seen_day_index is None or c.days_since_seen is None: + continue + silent = c.days_since_seen >= CHAMPION_SILENCE_DAYS + engagement_low = c.engagement_frequency is not None and c.engagement_frequency < 1 + if silent or engagement_low: + return True + return False + + +def single_threaded_fires( + total_contacts_all_time: int, + distinct_active_contacts_60d: int, + distinct_active_contacts_prior_180d: int, +) -> bool: + """ + Mirror of ``detectSingleThreadRisk`` (churn-detectors.ts Rule 5). + + Suppressed unless the account ever had >2 contacts. Fires only when exactly + one contact was active in the last 60 days but ≥3 distinct contacts were + active in the prior 60–240-day window. + """ + if total_contacts_all_time <= 2: + return False + if distinct_active_contacts_60d != 1: + return False + if distinct_active_contacts_prior_180d < SINGLE_THREAD_MIN_ALLTIME: + return False + return True + + +# --------------------------------------------------------------------------- +# Rollup computation — reused for every DaySnapshot AND for the final-day label. +# --------------------------------------------------------------------------- + + +@dataclass +class _ContactState: + """Mutable per-contact bookkeeping while attributing conversations.""" + + contact: Contact + conv_days: list[int] = field(default_factory=list) # sorted ascending + + +def _distinct_in_window( + states: dict[str, _ContactState], day: int, lo_days_ago: int, hi_days_ago: int +) -> int: + """ + Count distinct contacts with ≥1 conversation whose age (in days before + ``day``) falls in [lo_days_ago, hi_days_ago]. Ages are inclusive on both + ends; callers pass window bounds that already avoid the exact engine + boundaries for planted cases. + """ + lo_day = day - hi_days_ago + hi_day = day - lo_days_ago + return sum( + 1 + for st in states.values() + if any(lo_day <= d <= hi_day for d in st.conv_days if d <= day) + ) + + +def compute_rollup_at( + states: dict[str, _ContactState], champion_ids: list[str], day: int +) -> tuple[int, int, int, list[ContactRollup]]: + """ + Compute the four contact-rollup snapshot fields as-of ``day``. + + Returns ``(distinct_60d, distinct_prior_180d, total_all_time, + champion_rollups)``. Trailing windows end on ``day``; ``total_all_time`` is + the count of contacts that have engaged on or before ``day``. + """ + distinct_60d = _distinct_in_window(states, day, 0, RECENT_WINDOW_DAYS) + # Prior window: 60–240 days ago → ages in [61, 240] to sit strictly before + # the recent window (which owns age 0..60). + distinct_prior = _distinct_in_window( + states, day, PRIOR_WINDOW_START_DAYS + 1, PRIOR_WINDOW_END_DAYS + ) + total_all_time = sum( + 1 for st in states.values() if any(d <= day for d in st.conv_days) + ) + + rollups: list[ContactRollup] = [] + for cid in champion_ids: + st = states[cid] + seen = [d for d in st.conv_days if d <= day] + if not seen: + rollups.append( + ContactRollup( + contact_id=st.contact.contact_id, + name=st.contact.name, + role=st.contact.role, + is_champion=True, + last_seen_day_index=None, + days_since_seen=None, + engagement_frequency=None, + ) + ) + continue + last_seen = max(seen) + touches_in_window = sum(1 for d in seen if day - d <= FREQUENCY_WINDOW_DAYS) + freq = touches_in_window / FREQUENCY_WINDOW_MONTHS + rollups.append( + ContactRollup( + contact_id=st.contact.contact_id, + name=st.contact.name, + role=st.contact.role, + is_champion=True, + last_seen_day_index=last_seen, + days_since_seen=day - last_seen, + engagement_frequency=round(freq, 4), + ) + ) + return distinct_60d, distinct_prior, total_all_time, rollups + + +# --------------------------------------------------------------------------- +# Deterministic name generation +# --------------------------------------------------------------------------- + +_FIRST_NAMES = [ + "Avery", "Jordan", "Riley", "Morgan", "Casey", "Taylor", "Quinn", "Rowan", + "Sydney", "Devon", "Harper", "Reese", "Emerson", "Finley", "Marlowe", + "Priya", "Diego", "Mei", "Omar", "Nadia", "Kenji", "Zara", "Lucas", "Ingrid", +] +_LAST_NAMES = [ + "Chen", "Patel", "Nguyen", "Okafor", "Rossi", "Kowalski", "Silva", "Haddad", + "Larsson", "Mbeki", "Fischer", "Romano", "Delgado", "Ivanov", "Yamamoto", + "Ahmed", "Novak", "Costa", "Reyes", "Bauer", "Kim", "Dubois", "Singh", +] + +_ROLE_CYCLE = [ + ContactRole.DECISION_MAKER, + ContactRole.DAY_TO_DAY_USER, + ContactRole.TECHNICAL_ADMIN, + ContactRole.ECONOMIC_BUYER, + ContactRole.EXECUTIVE_SPONSOR, +] + + +# --------------------------------------------------------------------------- +# Scenario catalogue +# --------------------------------------------------------------------------- + +# (scenario_id, target_fraction, is_decoy_for) +# is_decoy_for names the detector a decoy superficially resembles; None for +# positives and plain nulls. +_SCENARIO_PLAN: list[tuple[str, float]] = [ + ("single_threaded_positive", 0.16), + ("champion_silence_positive", 0.14), + ("champion_lowfreq_positive", 0.08), + ("single_threaded_decoy_narrowed", 0.10), + ("single_threaded_decoy_guard", 0.07), + ("champion_active_decoy", 0.11), + # remainder → null_multithreaded / sparse_single +] + +_DECOY_SCENARIOS = { + "single_threaded_decoy_narrowed", + "single_threaded_decoy_guard", + "champion_active_decoy", +} + + +@dataclass +class _AccountConvStats: + """Per-account conversation distribution used for scenario eligibility.""" + + account_id: str + final_day: int + conv_events: list[SimEvent] # CONVERSATION_STARTED, sorted by (day_index, event_id) + n_recent: int # convs with age ≤ 60d + n_prior: int # convs with age in [61, 240] + earliest_day: int + latest_day: int + n_last91: int # convs with age ≤ 91d + + +class ContactPlanner: + """ + Plans contacts, attributes conversations, enriches snapshots, and emits + ground-truth relationship labels over a finished simulation. + """ + + def __init__( + self, + seed: int, + accounts: list, # list[AccountSkeleton] — duck-typed to avoid import cycle + events: list[SimEvent], + snapshots: list[DaySnapshot], + ): + self.rng = random.Random(seed ^ _CONTACT_RNG_SALT) + self.seed = seed + self._account_ids = [a.account_id for a in accounts] + self.events = events + self.snapshots = snapshots + self.contacts: list[Contact] = [] + self.labels: list[RelationshipLabel] = [] + # Filled during plan(); keyed by account_id. + self._states_by_account: dict[str, dict[str, _ContactState]] = {} + self._champions_by_account: dict[str, list[str]] = {} + + # ------------------------------------------------------------------ + # Public entry point + # ------------------------------------------------------------------ + + def plan(self) -> None: + """Run the full contact-planning pass, mutating events + snapshots.""" + stats = self._scan_accounts() + scenarios = self._assign_scenarios(stats) + for account_id in self._account_ids: + if account_id not in stats: + continue # account with no snapshots (churned day 0) — nothing to plan + self._plan_account(stats[account_id], scenarios[account_id]) + self._enrich_snapshots() + self._emit_labels(stats, scenarios) + + # ------------------------------------------------------------------ + # Scan + # ------------------------------------------------------------------ + + def _scan_accounts(self) -> dict[str, _AccountConvStats]: + final_day: dict[str, int] = {} + for s in self.snapshots: + d = final_day.get(s.account_id, -1) + if s.day_index > d: + final_day[s.account_id] = s.day_index + + convs_by_account: dict[str, list[SimEvent]] = {} + for e in self.events: + if e.event_type == SimEventType.CONVERSATION_STARTED: + convs_by_account.setdefault(e.account_id, []).append(e) + + stats: dict[str, _AccountConvStats] = {} + for account_id, D in final_day.items(): + convs = sorted( + convs_by_account.get(account_id, []), + key=lambda e: (e.day_index, e.event_id), + ) + if not convs: + # No conversations — account still gets an (empty) contact set and + # trivially-negative labels; represent with a stats stub. + stats[account_id] = _AccountConvStats( + account_id=account_id, final_day=D, conv_events=[], + n_recent=0, n_prior=0, earliest_day=D, latest_day=D, n_last91=0, + ) + continue + days = [e.day_index for e in convs] + n_recent = sum(1 for d in days if D - d <= RECENT_WINDOW_DAYS) + n_prior = sum(1 for d in days if PRIOR_WINDOW_START_DAYS + 1 <= D - d <= PRIOR_WINDOW_END_DAYS) + n_last91 = sum(1 for d in days if D - d <= FREQUENCY_WINDOW_DAYS) + stats[account_id] = _AccountConvStats( + account_id=account_id, + final_day=D, + conv_events=convs, + n_recent=n_recent, + n_prior=n_prior, + earliest_day=min(days), + latest_day=max(days), + n_last91=n_last91, + ) + return stats + + # ------------------------------------------------------------------ + # Scenario assignment (deterministic, eligibility-gated) + # ------------------------------------------------------------------ + + def _eligible(self, scenario: str, st: _AccountConvStats) -> bool: + D = st.final_day + if scenario == "single_threaded_positive": + return st.n_prior >= 3 and st.n_recent >= 1 + if scenario == "champion_silence_positive": + # Need at least one conv old enough to attribute to a silent champion. + return st.conv_events != [] and st.earliest_day <= D - _CHAMPION_SILENT_BY + if scenario == "champion_lowfreq_positive": + # A recent conv for the champion (<30d) plus ≥1 other recent conv so + # the account stays multi-threaded and single-thread doesn't fire. + has_recent_for_champion = any( + D - e.day_index <= _CHAMPION_LOWFREQ_WITHIN for e in st.conv_events + ) + return has_recent_for_champion and st.n_recent >= 2 + if scenario == "single_threaded_decoy_narrowed": + return st.n_prior >= 3 and st.n_recent >= 2 + if scenario == "single_threaded_decoy_guard": + return st.n_recent >= 1 and st.n_prior >= 1 + if scenario == "champion_active_decoy": + # Champion must be safely clear of BOTH fire branches: seen recently + # (≤15d) AND frequency comfortably > 1/mo. Needs ≥5 touches in the + # trailing 91d so the champion can hold ≥4 (freq ≈1.33) while a second + # contact keeps a recent touch (distinct_60d ≥ 2, no single-thread). + has_active = any(D - e.day_index <= _CHAMPION_ACTIVE_WITHIN for e in st.conv_events) + return st.n_last91 >= 5 and has_active + return False + + def _assign_scenarios(self, stats: dict[str, _AccountConvStats]) -> dict[str, str]: + n = len(self._account_ids) + assigned: dict[str, str] = {} + # Assign in a fixed scenario priority order; within each, take eligible + # unassigned accounts in account_id order up to the scenario's quota. + ordered_ids = sorted(stats.keys()) + for scenario, frac in _SCENARIO_PLAN: + quota = round(frac * n) + if quota <= 0: + continue + taken = 0 + for account_id in ordered_ids: + if taken >= quota: + break + if account_id in assigned: + continue + if self._eligible(scenario, stats[account_id]): + assigned[account_id] = scenario + taken += 1 + # Remainder → null control. Multi-contact-capable accounts become + # null_multithreaded; the rest are sparse_single. + for account_id in ordered_ids: + if account_id in assigned: + continue + st = stats[account_id] + if st.n_prior >= 2 and st.n_recent >= 2: + assigned[account_id] = "null_multithreaded" + else: + assigned[account_id] = "sparse_single" + return assigned + + # ------------------------------------------------------------------ + # Per-account planning: build contacts + attribute conversations + # ------------------------------------------------------------------ + + def _make_contact( + self, account_id: str, index: int, is_champion: bool, scenario: str + ) -> Contact: + first = _FIRST_NAMES[self.rng.randrange(len(_FIRST_NAMES))] + last = _LAST_NAMES[self.rng.randrange(len(_LAST_NAMES))] + role = ContactRole.DECISION_MAKER if is_champion else _ROLE_CYCLE[index % len(_ROLE_CYCLE)] + return Contact( + contact_id=f"contact_{account_id}_{index:02d}", + account_id=account_id, + name=f"{first} {last}", + role=role, + is_champion=is_champion, + relationship_scenario=scenario, + ) + + def _plan_account(self, st: _AccountConvStats, scenario: str) -> None: + account_id = st.account_id + D = st.final_day + convs = st.conv_events + + # Decide contact roster size + champion designation per scenario, then + # produce an attribution mapping event_id -> contact_index. + if scenario == "single_threaded_positive": + contacts = [self._make_contact(account_id, i, False, scenario) for i in range(4)] + attribution = self._attr_single_threaded_positive(convs, D) + elif scenario == "champion_silence_positive": + contacts = [ + self._make_contact(account_id, 0, True, scenario), + self._make_contact(account_id, 1, False, scenario), + self._make_contact(account_id, 2, False, scenario), + ] + attribution = self._attr_champion_silence_positive(convs, D) + elif scenario == "champion_lowfreq_positive": + contacts = [ + self._make_contact(account_id, 0, True, scenario), + self._make_contact(account_id, 1, False, scenario), + self._make_contact(account_id, 2, False, scenario), + ] + attribution = self._attr_champion_lowfreq_positive(convs, D) + elif scenario == "single_threaded_decoy_narrowed": + contacts = [self._make_contact(account_id, i, False, scenario) for i in range(4)] + attribution = self._attr_single_threaded_decoy_narrowed(convs, D) + elif scenario == "single_threaded_decoy_guard": + contacts = [self._make_contact(account_id, i, False, scenario) for i in range(2)] + attribution = self._attr_single_threaded_decoy_guard(convs, D) + elif scenario == "champion_active_decoy": + contacts = [ + self._make_contact(account_id, 0, True, scenario), + self._make_contact(account_id, 1, False, scenario), + ] + attribution = self._attr_champion_active_decoy(convs, D) + elif scenario == "null_multithreaded": + contacts = [self._make_contact(account_id, i, False, scenario) for i in range(3)] + attribution = self._attr_round_robin(convs, n_contacts=3) + else: # sparse_single + contacts = [self._make_contact(account_id, 0, False, scenario)] + attribution = {e.event_id: 0 for e in convs} + + # Build contact state, drop contacts that ended up with zero convs so the + # emitted roster equals the set of contacts that actually engaged (keeps + # total_contacts_all_time == len(contacts), matching the engine's row + # count where every ingested contact has engagement). + used_indices = set(attribution.values()) + states: dict[str, _ContactState] = {} + index_to_contact: dict[int, Contact] = {} + for idx, c in enumerate(contacts): + if idx in used_indices: + states[c.contact_id] = _ContactState(contact=c) + index_to_contact[idx] = c + + for e in convs: + idx = attribution[e.event_id] + contact = index_to_contact[idx] + states[contact.contact_id].conv_days.append(e.day_index) + self._attribute_event(e, contact) + + for st_c in states.values(): + st_c.conv_days.sort() + c = st_c.contact + c.engagement_count = len(st_c.conv_days) + c.first_seen_day_index = st_c.conv_days[0] if st_c.conv_days else -1 + c.last_seen_day_index = st_c.conv_days[-1] if st_c.conv_days else -1 + + self.contacts.extend(st_c.contact for st_c in states.values()) + self._states_by_account[account_id] = states + self._champions_by_account[account_id] = [ + cid for cid, s in states.items() if s.contact.is_champion + ] + + def _attribute_event(self, e: SimEvent, contact: Contact) -> None: + """Stamp a conversation event's payload with its customer contact.""" + e.payload["customer_name"] = contact.name + e.payload["contact_id"] = contact.contact_id + e.payload["contact_role"] = contact.role.value + e.payload["is_champion_contact"] = contact.is_champion + + # ------------------------------------------------------------------ + # Attribution strategies (return event_id -> contact_index) + # ------------------------------------------------------------------ + + @staticmethod + def _age(e: SimEvent, D: int) -> int: + return D - e.day_index + + def _attr_round_robin(self, convs: list[SimEvent], n_contacts: int) -> dict[str, int]: + return {e.event_id: i % n_contacts for i, e in enumerate(convs)} + + def _attr_single_threaded_positive(self, convs: list[SimEvent], D: int) -> dict[str, int]: + """ + Recent-window convs → the single surviving thread (contact 0); prior + convs → round-robin across contacts 1..3 so ≥3 distinct appear in the + 60–240d window. Result: distinct_60d == 1, distinct_prior ≥ 3. + """ + out: dict[str, int] = {} + prior_rr = 0 + prior_contacts = [1, 2, 3] + for e in convs: + if self._age(e, D) <= RECENT_WINDOW_DAYS: + out[e.event_id] = 0 + else: + out[e.event_id] = prior_contacts[prior_rr % len(prior_contacts)] + prior_rr += 1 + return out + + def _attr_champion_silence_positive(self, convs: list[SimEvent], D: int) -> dict[str, int]: + """ + Champion (contact 0) gets only old convs (age ≥ 40d), and only those; + everything else is split across non-champion contacts 1..2 so the account + stays alive and multi-threaded. Champion last-seen ≥40d ago → fires. + """ + out: dict[str, int] = {} + other_rr = 0 + champion_assigned = False + for e in convs: + age = self._age(e, D) + if age >= _CHAMPION_SILENT_BY: + # Give the champion its engagement, but keep some old convs on + # others too so the prior window isn't champion-only. Assign the + # first eligible old conv to the champion, alternate the rest. + if not champion_assigned: + out[e.event_id] = 0 + champion_assigned = True + else: + out[e.event_id] = 1 + (other_rr % 2) + other_rr += 1 + else: + out[e.event_id] = 1 + (other_rr % 2) + other_rr += 1 + # Safety: if no conv was old enough for the champion (shouldn't happen — + # eligibility guarantees earliest ≤ D-40), fall back to giving contact 0 + # the earliest conv. + if not champion_assigned and convs: + out[convs[0].event_id] = 0 + return out + + def _attr_champion_lowfreq_positive(self, convs: list[SimEvent], D: int) -> dict[str, int]: + """ + Champion (contact 0) gets exactly ONE recent conv (age ≤ 29d) and nothing + else, so its frequency over the trailing 3 months is 1/3 < 1 while its + last-seen is <30d (not silent). Fires via the frequency branch only. + Everything else → non-champion contacts 1..2. + """ + out: dict[str, int] = {} + # Pick the champion's single conv: the most recent conv within the + # low-freq window (largest day_index with age ≤ 29d). + champion_event_id = None + for e in convs: # convs sorted ascending by day → last match is most recent + if self._age(e, D) <= _CHAMPION_LOWFREQ_WITHIN: + champion_event_id = e.event_id + other_rr = 0 + for e in convs: + if e.event_id == champion_event_id: + out[e.event_id] = 0 + else: + out[e.event_id] = 1 + (other_rr % 2) + other_rr += 1 + return out + + def _attr_single_threaded_decoy_narrowed(self, convs: list[SimEvent], D: int) -> dict[str, int]: + """ + Near-miss: narrows to TWO active threads (not one). Recent convs split + across contacts 0 and 1; prior convs round-robin across 1..3 so ≥3 + distinct appear in the prior window. distinct_60d == 2 → must NOT fire. + """ + out: dict[str, int] = {} + recent_rr = 0 + prior_rr = 0 + prior_contacts = [1, 2, 3] + for e in convs: + if self._age(e, D) <= RECENT_WINDOW_DAYS: + out[e.event_id] = recent_rr % 2 # contacts 0 and 1 + recent_rr += 1 + else: + out[e.event_id] = prior_contacts[prior_rr % len(prior_contacts)] + prior_rr += 1 + return out + + def _attr_single_threaded_decoy_guard(self, convs: list[SimEvent], D: int) -> dict[str, int]: + """ + Near-miss: an account that IS effectively single-threaded (1 recent + contact, was 2) but only ever had TWO contacts — the engine's guard + (totalContactsAllTime > 2) suppresses it. distinct_60d == 1, + distinct_prior == 2, total == 2 → must NOT fire. + """ + out: dict[str, int] = {} + prior_rr = 0 + for e in convs: + if self._age(e, D) <= RECENT_WINDOW_DAYS: + out[e.event_id] = 0 + else: + out[e.event_id] = prior_rr % 2 # contacts 0 and 1 → 2 distinct prior + prior_rr += 1 + return out + + def _attr_champion_active_decoy(self, convs: list[SimEvent], D: int) -> dict[str, int]: + """ + Near-miss: the account HAS a champion (so Rule 4 has champions to check), + but the champion is actively engaged — seen ≤15d ago with ≥3 touches over + the trailing 3 months (frequency ≥ 1). Must NOT fire. + """ + out: dict[str, int] = {} + # Champion (contact 0) holds the NEWEST touch (so days_since is small) plus + # 3 more of the trailing-91d touches → ≥4 touches, freq ≈1.33/mo (clear of + # the <1 boundary). The 2nd-newest touch goes to contact 1 so the account + # keeps ≥2 distinct recent contacts and single-thread cannot fire. + recent91 = [e.event_id for e in convs if self._age(e, D) <= FREQUENCY_WINDOW_DAYS] + if len(recent91) >= 5: + champion_set = {recent91[-1]} | set(recent91[-5:-2]) # newest + 3 older-within-91d + else: # eligibility guarantees ≥5, but stay safe + champion_set = set(recent91[-1:]) + for e in convs: + out[e.event_id] = 0 if e.event_id in champion_set else 1 + return out + + # ------------------------------------------------------------------ + # Snapshot enrichment + # ------------------------------------------------------------------ + + def _enrich_snapshots(self) -> None: + for s in self.snapshots: + states = self._states_by_account.get(s.account_id) + if not states: + continue + champions = self._champions_by_account.get(s.account_id, []) + d60, dprior, total, rollups = compute_rollup_at(states, champions, s.day_index) + s.distinct_active_contacts_60d = d60 + s.distinct_active_contacts_prior_180d = dprior + s.total_contacts_all_time = total + s.champion_rollups = rollups + + # ------------------------------------------------------------------ + # Ground-truth label emission + # ------------------------------------------------------------------ + + def _emit_labels( + self, stats: dict[str, _AccountConvStats], scenarios: dict[str, str] + ) -> None: + for account_id in sorted(stats.keys()): + D = stats[account_id].final_day + scenario = scenarios[account_id] + states = self._states_by_account.get(account_id, {}) + champions = self._champions_by_account.get(account_id, []) + d60, dprior, total, rollups = compute_rollup_at(states, champions, D) + + champ_fire = champion_at_risk_fires(rollups) + single_fire = single_threaded_fires(total, d60, dprior) + + champ_evidence = { + "has_champion": bool(rollups), + "champions": [r.model_dump() for r in rollups], + } + single_evidence = { + "total_contacts_all_time": total, + "distinct_active_contacts_60d": d60, + "distinct_active_contacts_prior_180d": dprior, + } + + self.labels.append( + RelationshipLabel( + account_id=account_id, + detector=RelationshipDetector.CHAMPION_AT_RISK, + expected_fire=champ_fire, + is_decoy=(scenario == "champion_active_decoy" and not champ_fire), + scenario=scenario, + as_of_day_index=D, + rationale=self._champion_rationale(scenario, rollups, champ_fire), + evidence=champ_evidence, + ) + ) + self.labels.append( + RelationshipLabel( + account_id=account_id, + detector=RelationshipDetector.SINGLE_THREADED, + expected_fire=single_fire, + is_decoy=( + scenario + in ("single_threaded_decoy_narrowed", "single_threaded_decoy_guard") + and not single_fire + ), + scenario=scenario, + as_of_day_index=D, + rationale=self._single_rationale(scenario, total, d60, dprior, single_fire), + evidence=single_evidence, + ) + ) + + @staticmethod + def _champion_rationale(scenario: str, rollups: list, fires: bool) -> str: + if not rollups: + return "Account has no champion contact; Rule 4 cannot fire." + r = rollups[0] + if fires: + if r.days_since_seen is not None and r.days_since_seen >= CHAMPION_SILENCE_DAYS: + return f"Champion last seen {r.days_since_seen}d ago (≥30) — silence fires Rule 4." + return f"Champion engagement {r.engagement_frequency}/mo (<1) — low-frequency fires Rule 4." + if scenario == "champion_active_decoy": + return ( + f"Decoy: champion active (seen {r.days_since_seen}d ago, " + f"{r.engagement_frequency}/mo) — must NOT fire." + ) + return "Champion present and engaged; Rule 4 does not fire." + + @staticmethod + def _single_rationale(scenario: str, total: int, d60: int, dprior: int, fires: bool) -> str: + base = f"total={total}, active_60d={d60}, prior_60_240d={dprior}" + if fires: + return f"Single-threaded: {base} (1 recent, ≥3 prior, >2 all-time) — fires Rule 5." + if scenario == "single_threaded_decoy_narrowed": + return f"Decoy: narrowed to {d60} active threads, not 1 ({base}) — must NOT fire." + if scenario == "single_threaded_decoy_guard": + return f"Decoy: only {total} contacts all-time — guard suppresses ({base})." + return f"Not single-threaded ({base}); Rule 5 does not fire." diff --git a/resonantforge/layer1/sim_events.py b/resonantforge/layer1/sim_events.py index 4f70aaf..960f262 100644 --- a/resonantforge/layer1/sim_events.py +++ b/resonantforge/layer1/sim_events.py @@ -21,6 +21,16 @@ SimEventType.SCORE_CORRECTION: ["agent_id", "criterion", "original_score", "corrected_score"], } +# Optional payload fields — permitted (and documented) but not required. Enrichers +# that run after emission add these post-hoc. CONVERSATION_STARTED carries the +# customer-contact attribution stamped by ContactPlanner (contact_id + role) and +# the coverage-backfill target-gate hint. Keeping these here documents the full +# payload vocabulary without forcing them at emission time, when they aren't yet +# known (contacts are attributed once each account's realized lifespan is known). +SIM_EVENT_OPTIONAL_FIELDS: dict[str, list[str]] = { + SimEventType.CONVERSATION_STARTED: ["contact_id", "contact_role", "backfill_target_gate"], +} + def validate_event_payload(event_type: SimEventType, payload: dict) -> list[str]: """ @@ -29,6 +39,9 @@ def validate_event_payload(event_type: SimEventType, payload: dict) -> list[str] Returns a list of missing required field names; an empty list means the payload is valid. Callers can use this to assert correctness during testing or to raise structured errors before emitting events into the stream. + + Optional fields (see ``SIM_EVENT_OPTIONAL_FIELDS``) are never reported as + missing — they are enrichment fields added after emission. """ required = SIM_EVENT_REGISTRY.get(event_type, []) return [f for f in required if f not in payload] diff --git a/resonantforge/layer1/state_machine.py b/resonantforge/layer1/state_machine.py index 71d10fc..211f3be 100644 --- a/resonantforge/layer1/state_machine.py +++ b/resonantforge/layer1/state_machine.py @@ -13,9 +13,12 @@ SimEventType, LifecycleStage, HealthState, + Contact, + RelationshipLabel, ) from resonantforge.layer1.sim_events import validate_event_payload from resonantforge.layer1.clocks import ClockRegistry +from resonantforge.layer1.contact_planner import ContactPlanner if TYPE_CHECKING: from resonantforge.layer1.coverage_backfill import CoverageBackfill @@ -138,6 +141,11 @@ def __init__( self.base_date = date(2025, 7, 1) # simulation starts 2025-07-01 self.events: list[SimEvent] = [] self.snapshots: list[DaySnapshot] = [] + # Customer-contact modeling outputs — populated by the ContactPlanner + # pass at the end of simulate(). Read by the pipeline for artifact + # emission (contacts.jsonl + relationship_labels.jsonl). + self.contacts: list[Contact] = [] + self.relationship_labels: list[RelationshipLabel] = [] self._event_counter = 0 self._snapshot_counter = 0 @@ -275,6 +283,23 @@ def simulate(self) -> tuple[list[SimEvent], list[DaySnapshot]]: for account in accounts: self._simulate_account(account) + # Customer-contact modeling: a single deterministic post-pass over the + # finished stream. It needs each account's realized lifespan (emergent + # from churn), so it runs after the day loop, not inside it. It generates + # contacts, attributes each conversation to a customer contact (mutating + # CONVERSATION_STARTED payloads), enriches DaySnapshots with the contact + # rollup, and emits ground-truth relationship labels. Uses a seed-derived + # RNG that never touches the event RNG stream above. + planner = ContactPlanner( + seed=self.seed, + accounts=accounts, + events=self.events, + snapshots=self.snapshots, + ) + planner.plan() + self.contacts = planner.contacts + self.relationship_labels = planner.labels + return self.events, self.snapshots def _simulate_account(self, account: AccountSkeleton) -> None: diff --git a/resonantforge/pipeline.py b/resonantforge/pipeline.py index 8d37693..910cd55 100644 --- a/resonantforge/pipeline.py +++ b/resonantforge/pipeline.py @@ -25,7 +25,7 @@ from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from pathlib import Path -from typing import TYPE_CHECKING, Callable +from typing import TYPE_CHECKING, Any, Callable from resonantforge.agents.generator import generate_agents from resonantforge.corrections.generator import generate_corrections @@ -218,6 +218,12 @@ def _build_prompt( channel = conv_event.payload.get("surface_channel", "chat") customer = conv_event.payload.get("customer_name", f"Customer_{account_id}") agent_id = conv_event.payload.get("agent_id", "agent_unknown") + # Contact role (present once ContactPlanner has attributed the conversation to + # a specific customer contact) — gives the prose a consistent speaker identity. + contact_role = conv_event.payload.get("contact_role") + customer_line = ( + f"{customer} ({contact_role.replace('_', ' ')})" if contact_role else customer + ) health_info = "" if month_summary: @@ -239,7 +245,7 @@ def _build_prompt( user_prompt = ( f"Generate a customer-support conversation on the {channel} channel." - f"\nCustomer: {customer}" + f"\nCustomer: {customer_line}" f"\nAgent ID: {agent_id}" f"\nAccount: {account_id}" f"{health_info}" @@ -327,6 +333,12 @@ def _make_conversation_record( started_at = conv_event.timestamp ended_at = started_at + timedelta(minutes=10) + # Customer-contact attribution stamped onto the event payload by the + # ContactPlanner. None on pre-contact-modeling corpora. + contact_id = conv_event.payload.get("contact_id") + contact_role = conv_event.payload.get("contact_role") + contact_name = conv_event.payload.get("customer_name") if contact_id else None + return ConversationRecord( conversation_id=conv_id, account_id=account_id, @@ -341,6 +353,10 @@ def _make_conversation_record( tone_variant=None, planted_constraint=quality_plan.planted_constraint if quality_plan is not None else None, planted_contradiction=quality_plan.planted_contradiction if quality_plan is not None else None, + contact_id=contact_id, + contact_name=contact_name, + contact_role=contact_role, + is_champion_contact=bool(conv_event.payload.get("is_champion_contact", False)), ) @@ -1687,6 +1703,46 @@ def _cb(status: str) -> None: planted_quality_hash = _sha256_jsonl(plan_lines) atomic_write_jsonl(profile_dir / "planted_quality.jsonl", plan_lines) + # Serialise customer contacts (Phase 1 data). Contacts are the first-class + # customer-side entities the relationship churn detectors track. + contacts = sm.contacts + contact_lines = [c.model_dump_json() for c in contacts] + contacts_hash = _sha256_jsonl(contact_lines) + atomic_write_jsonl(profile_dir / "contacts.jsonl", contact_lines) + + # Serialise relationship-signal ground-truth labels (Phase 1 data). Two labels + # per account (champion-at-risk + single-threaded), computed by pure mirrors + # of the engine's Rules 4 & 5 against the planted contact-engagement graph. + relationship_labels = sm.relationship_labels + rel_label_lines = [lab.model_dump_json() for lab in relationship_labels] + relationship_labels_hash = _sha256_jsonl(rel_label_lines) + atomic_write_jsonl(profile_dir / "relationship_labels.jsonl", rel_label_lines) + + # Relationship planted-distribution telemetry for the audit: per-detector + # counts of positive / negative / decoy labels and scenario tallies. + _rel_dist: dict[str, Any] = {} + for lab in relationship_labels: + det = lab.detector.value + bucket = _rel_dist.setdefault( + det, {"positive": 0, "negative": 0, "decoy": 0, "total": 0} + ) + bucket["total"] += 1 + if lab.expected_fire: + bucket["positive"] += 1 + else: + bucket["negative"] += 1 + if lab.is_decoy: + bucket["decoy"] += 1 + # Per-account scenario tally (each account has two labels sharing a scenario). + _rel_scenarios: dict[str, int] = {} + _seen_scen_accounts: set[str] = set() + for lab in relationship_labels: + if lab.account_id in _seen_scen_accounts: + continue + _seen_scen_accounts.add(lab.account_id) + _rel_scenarios[lab.scenario] = _rel_scenarios.get(lab.scenario, 0) + 1 + _rel_dist["scenarios"] = _rel_scenarios + # _kb_version already computed post-contamination in Phase 2. # domain_distribution_observed: count CONVERSATION_STARTED events per domain. @@ -1725,6 +1781,8 @@ def _cb(status: str) -> None: knowledge_base_chunk_count=kb_chunk_count, agent_count=len(agent_profiles), corrections_count=len(corrections), + contact_count=len(contacts), + relationship_label_count=len(relationship_labels), events_hash=events_hash, snapshots_hash=snapshots_hash, conversations_hash=conversations_hash, @@ -1733,6 +1791,9 @@ def _cb(status: str) -> None: tenant_config_hash=tenant_config_hash, agent_fixtures_hash=agents_hash, corrections_hash=corrections_hash, + contacts_hash=contacts_hash, + relationship_labels_hash=relationship_labels_hash, + relationship_signal_distribution=_rel_dist, prose_fact_violation_rate=skip_tracker.prose_fact_rate, validator_rule_failure_rate=skip_tracker.quality_rule_rate, disagreement_rate=skip_tracker.disagreement_rate, diff --git a/resonantforge/schemas.py b/resonantforge/schemas.py index d374127..cbc8666 100644 --- a/resonantforge/schemas.py +++ b/resonantforge/schemas.py @@ -59,6 +59,78 @@ class HealthState(str, Enum): CHURNED = "churned" +class ContactRole(str, Enum): + """ + Coarse classification of a customer-side contact's function on the account. + + ``role`` is descriptive provenance for the synthetic relationship graph; the + Resonant IQ engine's champion detector keys off the separate + ``Contact.is_champion`` boolean, not this enum. Any role may be flagged a + champion, though in practice the advocate is usually a decision-maker or a + hands-on power user. + """ + + ECONOMIC_BUYER = "economic_buyer" + DECISION_MAKER = "decision_maker" + DAY_TO_DAY_USER = "day_to_day_user" + TECHNICAL_ADMIN = "technical_admin" + EXECUTIVE_SPONSOR = "executive_sponsor" + + +class Contact(BaseModel): + """ + A first-class customer-side contact on an account. + + Contacts are the people the support/CS team engages with, distinct from the + tenant's own support agents (``AgentProfile``). Each conversation is + attributed to exactly one contact (``ConversationRecord.contact_id``), which + lets the Resonant IQ engine track specific relationships over time — the + prerequisite for the ``relationship_champion_at_risk`` and + ``relationship_single_threaded`` churn detectors. + + ``is_champion`` marks the internal advocate driving adoption. ``first_seen_day_index`` + / ``last_seen_day_index`` bound the contact's realized engagement window in + simulation-day terms (both ``-1`` if the contact never appears on a + conversation). ``engagement_count`` is the total attributed conversations. + ``relationship_scenario`` records which planted scenario placed this contact, + for corpus provenance and audit. + """ + + contact_id: str # deterministic, e.g. "contact_acct_001_02" + account_id: str + name: str + role: ContactRole + is_champion: bool = False + first_seen_day_index: int = -1 + last_seen_day_index: int = -1 + engagement_count: int = 0 + relationship_scenario: str = "null_multithreaded" + + +class ContactRollup(BaseModel): + """ + Per-champion contact state rolled up as-of a specific DaySnapshot day. + + Mirrors the fields the Resonant IQ champion detector (Rule 4) reads so a + downstream scorer can evaluate the rule without replaying the event stream. + ``days_since_seen`` and ``engagement_frequency`` are relative to the owning + snapshot's day (the account's "now" when the snapshot is the last one). + Both are ``None`` if the champion has not yet engaged by that day. + + ``engagement_frequency`` is touches/month over the trailing 3-month window + (``CONTACT_FREQUENCY_WINDOW_DAYS``), matching the engine's per-contact + frequency definition. + """ + + contact_id: str + name: str + role: ContactRole + is_champion: bool + last_seen_day_index: Optional[int] = None + days_since_seen: Optional[int] = None + engagement_frequency: Optional[float] = None + + class SimEvent(BaseModel): """ A single discrete event emitted by the simulation engine. @@ -99,6 +171,17 @@ class DaySnapshot(BaseModel): active_agents: list[str] # agent IDs active this day payment_status: Literal["current", "overdue", "failed"] renewal_days_remaining: Optional[int] = None + # Contact-relationship rollup (RForge contact modeling). Pre-computed + # trailing-window aggregates over per-contact conversation attribution so + # downstream scorers can evaluate the engine's relationship churn detectors + # (Rules 4 & 5) without replaying events — mirroring how health_score is + # pre-rolled. All windows are trailing and end on this snapshot's day, so the + # final snapshot per account carries the account's as-of-"now" state. + # Defaults keep pre-contact-modeling constructors valid. + distinct_active_contacts_60d: int = 0 # distinct contacts on convs in trailing 60d + distinct_active_contacts_prior_180d: int = 0 # distinct contacts on convs 60–240d ago + total_contacts_all_time: int = 0 # distinct contacts engaged on/before this day + champion_rollups: list[ContactRollup] = Field(default_factory=list) # one per champion, as-of this day @field_validator("health_score") @classmethod @@ -128,6 +211,11 @@ class ConversationStartedPayload(BaseModel): agent_id: str domain: str # e.g. "billing", "api", "refunds" intent: list[str] = Field(default_factory=list) # e.g. ["how_to", "feature_request"] + # Contact attribution — populated by ContactPlanner after simulation, so the + # transcript's customer identity is the specific customer contact that + # participated. Optional because the raw event is emitted before attribution. + contact_id: Optional[str] = None + contact_role: Optional[str] = None class ConversationRecord(BaseModel): @@ -155,6 +243,14 @@ class ConversationRecord(BaseModel): tone_variant: Optional[str] = None # brand voice variant id if contaminated; None if dominant planted_constraint: Optional[str] = None # normalized constraint phrase for overgeneralization events planted_contradiction: Optional[PlantedContradiction] = None # fact/negation pair for contradicted:exact events + # Customer-contact attribution (RForge contact modeling). Identifies which + # customer contact participated, so transcript evidence is consistent with + # the planted relationship state. None on corpora generated before contact + # modeling. + contact_id: Optional[str] = None + contact_name: Optional[str] = None + contact_role: Optional[str] = None + is_champion_contact: bool = False class AccuracyLabel(BaseModel): @@ -570,6 +666,46 @@ class DisagreementRecord(BaseModel): notes: Optional[str] = None +# --------------------------------------------------------------------------- +# Section 5b — Relationship-signal ground-truth labels +# --------------------------------------------------------------------------- + + +class RelationshipDetector(str, Enum): + """The two engine churn detectors that require per-contact relationship state.""" + + CHAMPION_AT_RISK = "relationship_champion_at_risk" + SINGLE_THREADED = "relationship_single_threaded" + + +class RelationshipLabel(BaseModel): + """ + Ground-truth label for one account × one relationship detector, evaluated at + the corpus "now" (the account's final snapshot day). + + Labels are produced by the ContactPlanner using pure Python mirrors of the + engine's Rule 4 / Rule 5 logic, so ``expected_fire`` is by construction what + a correct detector must output given the planted contact-engagement graph. + The benchmark harness reads these to compute per-detector precision/recall/F1. + + ``is_decoy`` marks near-miss negatives — accounts engineered to look like a + fire (champion who dipped but is still active; an account that narrowed to two + threads but not one) that a correct detector must NOT fire on. They make the + false-positive rate measurable, mirroring the RFORGE-73 paraphrase-trap + distractors. ``evidence`` carries the exact window values the decision rests + on, keyed the same as the engine's ``AccountChurnData`` fields. + """ + + account_id: str + detector: RelationshipDetector + expected_fire: bool # True = should fire (positive); False = should not (negative/decoy) + is_decoy: bool = False # near-miss negative that superficially looks like a fire + scenario: str # planted scenario id, e.g. "single_threaded_positive" + as_of_day_index: int # the final snapshot day this label is evaluated at + rationale: str + evidence: dict[str, Any] = Field(default_factory=dict) + + # --------------------------------------------------------------------------- # Section 6 — Agent profile schemas (Section 11.2) # --------------------------------------------------------------------------- @@ -789,6 +925,8 @@ class Manifest(BaseModel): knowledge_base_chunk_count: int agent_count: int corrections_count: int + contact_count: int = 0 + relationship_label_count: int = 0 # Hashes (SHA-256 hex of corresponding NDJSON output files) events_hash: str snapshots_hash: str @@ -798,6 +936,11 @@ class Manifest(BaseModel): tenant_config_hash: str agent_fixtures_hash: str corrections_hash: str + contacts_hash: str = "" + relationship_labels_hash: str = "" + # Relationship-signal planted distribution telemetry — per-detector counts of + # positive / negative / decoy / null labels, for the distribution audit. + relationship_signal_distribution: dict[str, Any] = Field(default_factory=dict) # Skip / disagreement rates prose_fact_violation_rate: float validator_rule_failure_rate: float diff --git a/resonantforge/validators/relationship.py b/resonantforge/validators/relationship.py new file mode 100644 index 0000000..c41e78e --- /dev/null +++ b/resonantforge/validators/relationship.py @@ -0,0 +1,238 @@ +# resonantforge/validators/relationship.py +""" +Relationship-signal validator — account-level, deterministic (no LLM). + +This is the relationship counterpart to the per-conversation dimension +validators. It follows the same "feature extractor + rule engine" split, but the +unit of validation is an *account × detector* rather than a conversation: + +- **Feature extractor** (:func:`extract_relationship_features`) independently + recomputes the windowed contact-engagement features (distinct active contacts + in the trailing 60d / prior 60–240d windows, total contacts, per-champion + recency and frequency) straight from the attributed ``ConversationRecord`` + stream — *not* from the pre-rolled ``DaySnapshot`` fields. Recomputing from raw + attribution is the point: it cross-checks that the snapshot pre-roll and the + planted labels agree with a second, independent implementation. + +- **Rule engine** (:func:`validate_account_relationships`) applies the shared + detector oracle (``champion_at_risk_fires`` / ``single_threaded_fires`` — the + same thresholds the engine and the label generator use) and compares the + extracted outcome against the planted ``RelationshipLabel.expected_fire``. A + mismatch is a FAIL: the corpus's own ground truth disagrees with an + independent recompute, which is exactly what the disagreement ledger exists to + surface. + +Only the feature extraction is re-implemented here; the thresholds stay in +``contact_planner`` so they cannot drift between planting and validation. +""" +from __future__ import annotations + +from resonantforge.schemas import ( + Contact, + ContactRollup, + ConversationRecord, + RelationshipLabel, + RelationshipDetector, + DimensionVerdict, + ValidationVerdict, +) +from resonantforge.layer1.contact_planner import ( + champion_at_risk_fires, + single_threaded_fires, + RECENT_WINDOW_DAYS, + PRIOR_WINDOW_START_DAYS, + PRIOR_WINDOW_END_DAYS, + FREQUENCY_WINDOW_DAYS, + FREQUENCY_WINDOW_MONTHS, +) + + +class RelationshipFeatures: + """Extracted account-level relationship features as-of a day.""" + + def __init__( + self, + account_id: str, + as_of_day_index: int, + distinct_active_contacts_60d: int, + distinct_active_contacts_prior_180d: int, + total_contacts_all_time: int, + champion_rollups: list[ContactRollup], + ): + self.account_id = account_id + self.as_of_day_index = as_of_day_index + self.distinct_active_contacts_60d = distinct_active_contacts_60d + self.distinct_active_contacts_prior_180d = distinct_active_contacts_prior_180d + self.total_contacts_all_time = total_contacts_all_time + self.champion_rollups = champion_rollups + + def as_summary(self) -> dict: + return { + "distinct_active_contacts_60d": self.distinct_active_contacts_60d, + "distinct_active_contacts_prior_180d": self.distinct_active_contacts_prior_180d, + "total_contacts_all_time": self.total_contacts_all_time, + "champions": [r.model_dump() for r in self.champion_rollups], + } + + # Convenience: apply the shared oracle to the extracted features. + def champion_fires(self) -> bool: + return champion_at_risk_fires(self.champion_rollups) + + def single_threaded_fires(self) -> bool: + return single_threaded_fires( + self.total_contacts_all_time, + self.distinct_active_contacts_60d, + self.distinct_active_contacts_prior_180d, + ) + + +def _contact_days( + conversations: list[ConversationRecord], + day_by_conv: dict[str, int], +) -> dict[str, list[int]]: + """Map contact_id → sorted list of conversation day-indices it participated in.""" + days: dict[str, list[int]] = {} + for c in conversations: + if not c.contact_id: + continue + day = day_by_conv.get(c.conversation_id) + if day is None: + continue + days.setdefault(c.contact_id, []).append(day) + for v in days.values(): + v.sort() + return days + + +def extract_relationship_features( + account_id: str, + contacts: list[Contact], + conversations: list[ConversationRecord], + day_by_conv: dict[str, int], + as_of_day_index: int, +) -> RelationshipFeatures: + """ + Recompute the account's relationship features from raw attribution. + + Args: + account_id: Account to extract for. + contacts: All Contact rows for this account. + conversations: All ConversationRecords for this account (must carry + ``contact_id`` attribution). + day_by_conv: Map conversation_id → simulation day-index (from the + trigger event; conversations don't store day-index + directly). + as_of_day_index: The day the windows end on (the account's "now"). + """ + contact_days = _contact_days(conversations, day_by_conv) + champion_ids = [c.contact_id for c in contacts if c.is_champion] + role_by_id = {c.contact_id: c.role for c in contacts} + name_by_id = {c.contact_id: c.name for c in contacts} + D = as_of_day_index + + def _distinct(lo_days_ago: int, hi_days_ago: int) -> int: + lo_day = D - hi_days_ago + hi_day = D - lo_days_ago + return sum( + 1 + for days in contact_days.values() + if any(lo_day <= d <= hi_day for d in days if d <= D) + ) + + distinct_60d = _distinct(0, RECENT_WINDOW_DAYS) + distinct_prior = _distinct(PRIOR_WINDOW_START_DAYS + 1, PRIOR_WINDOW_END_DAYS) + total_all_time = sum(1 for days in contact_days.values() if any(d <= D for d in days)) + + rollups: list[ContactRollup] = [] + for cid in champion_ids: + seen = [d for d in contact_days.get(cid, []) if d <= D] + if not seen: + rollups.append( + ContactRollup( + contact_id=cid, name=name_by_id.get(cid, ""), + role=role_by_id[cid], is_champion=True, + last_seen_day_index=None, days_since_seen=None, + engagement_frequency=None, + ) + ) + continue + last_seen = max(seen) + touches = sum(1 for d in seen if D - d <= FREQUENCY_WINDOW_DAYS) + rollups.append( + ContactRollup( + contact_id=cid, name=name_by_id.get(cid, ""), + role=role_by_id[cid], is_champion=True, + last_seen_day_index=last_seen, days_since_seen=D - last_seen, + engagement_frequency=round(touches / FREQUENCY_WINDOW_MONTHS, 4), + ) + ) + + return RelationshipFeatures( + account_id=account_id, + as_of_day_index=D, + distinct_active_contacts_60d=distinct_60d, + distinct_active_contacts_prior_180d=distinct_prior, + total_contacts_all_time=total_all_time, + champion_rollups=rollups, + ) + + +def _validate_one( + features: RelationshipFeatures, + detector: RelationshipDetector, + label: RelationshipLabel | None, +) -> DimensionVerdict: + if detector == RelationshipDetector.CHAMPION_AT_RISK: + computed = features.champion_fires() + else: + computed = features.single_threaded_fires() + + dimension = detector.value + if label is None: + return DimensionVerdict( + dimension=dimension, + verdict=ValidationVerdict.SKIP, + target="unlabeled", + signals_summary=features.as_summary(), + ) + + passed = computed == label.expected_fire + return DimensionVerdict( + dimension=dimension, + verdict=ValidationVerdict.PASS if passed else ValidationVerdict.FAIL, + target=f"expected_fire={label.expected_fire}", + signals_summary={ + **features.as_summary(), + "computed_fire": computed, + "expected_fire": label.expected_fire, + "scenario": label.scenario, + "is_decoy": label.is_decoy, + }, + ) + + +def validate_account_relationships( + account_id: str, + contacts: list[Contact], + conversations: list[ConversationRecord], + day_by_conv: dict[str, int], + labels: list[RelationshipLabel], + as_of_day_index: int, +) -> list[DimensionVerdict]: + """ + Validate both relationship detectors for one account. + + Returns two DimensionVerdicts (champion-at-risk, single-threaded). A FAIL + means the independently-extracted features disagree with the planted label — + a corpus-integrity signal for the disagreement ledger. + """ + features = extract_relationship_features( + account_id, contacts, conversations, day_by_conv, as_of_day_index + ) + label_by_detector = {lab.detector: lab for lab in labels if lab.account_id == account_id} + return [ + _validate_one(features, RelationshipDetector.CHAMPION_AT_RISK, + label_by_detector.get(RelationshipDetector.CHAMPION_AT_RISK)), + _validate_one(features, RelationshipDetector.SINGLE_THREADED, + label_by_detector.get(RelationshipDetector.SINGLE_THREADED)), + ] diff --git a/tests/test_contact_modeling.py b/tests/test_contact_modeling.py new file mode 100644 index 0000000..430de5c --- /dev/null +++ b/tests/test_contact_modeling.py @@ -0,0 +1,331 @@ +""" +Tests for first-class customer-contact modeling and the two relationship churn +signals it makes testable (`relationship_champion_at_risk`, +`relationship_single_threaded`). + +Coverage: +- The pure detector mirrors match the engine's Rule 4 / Rule 5 semantics. +- The ContactPlanner is deterministic (contacts, labels, snapshot rollups). +- Every planted positive genuinely fires; every decoy and null genuinely does + not, and decoys are flagged as such. +- Labels are internally consistent with their own evidence and with the + as-of-now snapshot rollup. +- Contact attribution is complete and well-formed. +- The relationship validator agrees with the planted ground truth (independent + recompute from raw attribution). +""" +from __future__ import annotations + +from collections import Counter, defaultdict + +import pytest + +from resonantforge.layer1.state_machine import StateMachine +from resonantforge.layer1.contact_planner import ( + champion_at_risk_fires, + single_threaded_fires, +) +from resonantforge.layer1.sim_events import validate_event_payload +from resonantforge.profiles.saas import SaaSProfile +from resonantforge.schemas import ( + SimEventType, + ContactRole, + ContactRollup, + RelationshipDetector, + ConversationRecord, +) +from resonantforge.validators.relationship import validate_account_relationships + + +SEED = 42 +N_ACCOUNTS = 40 +N_MONTHS = 6 + + +@pytest.fixture(scope="module") +def sim(): + """A single simulated corpus, reused across tests in this module.""" + sm = StateMachine(seed=SEED, num_accounts=N_ACCOUNTS, num_months=N_MONTHS, profile=SaaSProfile()) + events, snapshots = sm.simulate() + return sm, events, snapshots + + +def _champion(days_since, freq, seen=True): + return ContactRollup( + contact_id="c", name="X", role=ContactRole.DECISION_MAKER, is_champion=True, + last_seen_day_index=(100 if seen else None), + days_since_seen=(days_since if seen else None), + engagement_frequency=(freq if seen else None), + ) + + +# --------------------------------------------------------------------------- +# Pure detector mirrors +# --------------------------------------------------------------------------- + + +def test_champion_mirror_no_champions_never_fires(): + assert champion_at_risk_fires([]) is False + + +def test_champion_mirror_silence_fires(): + assert champion_at_risk_fires([_champion(30, 5.0)]) is True # exactly 30d + assert champion_at_risk_fires([_champion(45, 5.0)]) is True + assert champion_at_risk_fires([_champion(29, 5.0)]) is False # <30 and freq ok + + +def test_champion_mirror_low_frequency_fires(): + assert champion_at_risk_fires([_champion(5, 0.5)]) is True # active but <1/mo + assert champion_at_risk_fires([_champion(5, 1.0)]) is False # exactly 1/mo is fine + assert champion_at_risk_fires([_champion(5, 2.0)]) is False + + +def test_champion_mirror_never_seen_is_skipped(): + # A champion with no recency is skipped (engine guard) — does not fire. + assert champion_at_risk_fires([_champion(0, 0.0, seen=False)]) is False + + +def test_single_threaded_mirror_semantics(): + # Fires: >2 all-time, exactly 1 recent, >=3 prior. + assert single_threaded_fires(4, 1, 3) is True + # Guard: <=2 all-time suppresses. + assert single_threaded_fires(2, 1, 3) is False + # Must be exactly 1 recent. + assert single_threaded_fires(4, 0, 3) is False + assert single_threaded_fires(4, 2, 3) is False + # Prior must be >=3. + assert single_threaded_fires(4, 1, 2) is False + + +# --------------------------------------------------------------------------- +# Determinism +# --------------------------------------------------------------------------- + + +def test_planner_determinism(): + def run(): + sm = StateMachine(seed=SEED, num_accounts=N_ACCOUNTS, num_months=N_MONTHS, profile=SaaSProfile()) + sm.simulate() + return sm + + a, b = run(), run() + assert [c.model_dump_json() for c in a.contacts] == [c.model_dump_json() for c in b.contacts] + assert [l.model_dump_json() for l in a.relationship_labels] == [ + l.model_dump_json() for l in b.relationship_labels + ] + assert [s.model_dump_json() for s in a.snapshots] == [s.model_dump_json() for s in b.snapshots] + + +# --------------------------------------------------------------------------- +# Planted positives fire, decoys / nulls do not +# --------------------------------------------------------------------------- + + +def _labels_by(sm): + out = defaultdict(dict) # account_id -> {detector: label} + for lab in sm.relationship_labels: + out[lab.account_id][lab.detector] = lab + return out + + +def test_positives_fire(sim): + sm, _, _ = sim + by = _labels_by(sm) + seen_scenarios = Counter() + for account_id, labs in by.items(): + champ = labs[RelationshipDetector.CHAMPION_AT_RISK] + single = labs[RelationshipDetector.SINGLE_THREADED] + scenario = champ.scenario + seen_scenarios[scenario] += 1 + if scenario == "single_threaded_positive": + assert single.expected_fire is True + assert single.is_decoy is False + elif scenario in ("champion_silence_positive", "champion_lowfreq_positive"): + assert champ.expected_fire is True + assert champ.is_decoy is False + # The larger corpus must actually contain each positive scenario. + assert seen_scenarios["single_threaded_positive"] >= 1 + assert seen_scenarios["champion_silence_positive"] >= 1 + assert seen_scenarios["champion_lowfreq_positive"] >= 1 + + +def test_decoys_do_not_fire_but_are_flagged(sim): + sm, _, _ = sim + by = _labels_by(sm) + champ_decoys = single_decoys = 0 + for labs in by.values(): + champ = labs[RelationshipDetector.CHAMPION_AT_RISK] + single = labs[RelationshipDetector.SINGLE_THREADED] + if champ.scenario == "champion_active_decoy": + # Near-miss: account HAS a champion but it's active → must not fire. + assert champ.expected_fire is False + assert champ.is_decoy is True + assert champ.evidence["has_champion"] is True + champ_decoys += 1 + if single.scenario in ("single_threaded_decoy_narrowed", "single_threaded_decoy_guard"): + assert single.expected_fire is False + assert single.is_decoy is True + single_decoys += 1 + assert champ_decoys >= 1 + assert single_decoys >= 1 + + +def test_nulls_do_not_fire(sim): + sm, _, _ = sim + by = _labels_by(sm) + for labs in by.values(): + for lab in labs.values(): + if lab.scenario in ("null_multithreaded", "sparse_single"): + assert lab.expected_fire is False + assert lab.is_decoy is False + + +# --------------------------------------------------------------------------- +# Label / evidence / rollup consistency +# --------------------------------------------------------------------------- + + +def test_labels_consistent_with_their_evidence(sim): + sm, _, _ = sim + for lab in sm.relationship_labels: + ev = lab.evidence + if lab.detector == RelationshipDetector.SINGLE_THREADED: + got = single_threaded_fires( + ev["total_contacts_all_time"], + ev["distinct_active_contacts_60d"], + ev["distinct_active_contacts_prior_180d"], + ) + else: + got = champion_at_risk_fires([ContactRollup(**r) for r in ev["champions"]]) + assert got == lab.expected_fire, (lab.account_id, lab.detector, lab.scenario) + + +def test_final_snapshot_rollup_matches_single_thread_label(sim): + sm, events, snapshots = sim + final_snap = {} + for s in snapshots: + cur = final_snap.get(s.account_id) + if cur is None or s.day_index > cur.day_index: + final_snap[s.account_id] = s + single_labels = { + lab.account_id: lab + for lab in sm.relationship_labels + if lab.detector == RelationshipDetector.SINGLE_THREADED + } + for account_id, lab in single_labels.items(): + snap = final_snap[account_id] + assert snap.day_index == lab.as_of_day_index + assert snap.distinct_active_contacts_60d == lab.evidence["distinct_active_contacts_60d"] + assert snap.distinct_active_contacts_prior_180d == lab.evidence["distinct_active_contacts_prior_180d"] + assert snap.total_contacts_all_time == lab.evidence["total_contacts_all_time"] + + +# --------------------------------------------------------------------------- +# Contact attribution integrity +# --------------------------------------------------------------------------- + + +def test_every_contact_engaged_and_referenced(sim): + sm, events, _ = sim + contact_ids = {c.contact_id for c in sm.contacts} + # Every emitted contact engaged at least once. + for c in sm.contacts: + assert c.engagement_count >= 1 + assert c.first_seen_day_index >= 0 + assert c.last_seen_day_index >= c.first_seen_day_index + # Every attributed conversation event references an emitted contact. + attributed = 0 + for e in events: + if e.event_type == SimEventType.CONVERSATION_STARTED: + cid = e.payload.get("contact_id") + assert cid is not None, f"conversation {e.event_id} was not attributed" + assert cid in contact_ids + assert e.payload.get("customer_name") # contact name set + assert e.payload.get("contact_role") + attributed += 1 + assert attributed > 0 + + +def test_total_contacts_all_time_equals_contact_rows(sim): + sm, _, snapshots = sim + contacts_per_account = Counter(c.account_id for c in sm.contacts) + final_snap = {} + for s in snapshots: + cur = final_snap.get(s.account_id) + if cur is None or s.day_index > cur.day_index: + final_snap[s.account_id] = s + for account_id, n_contacts in contacts_per_account.items(): + # Every emitted contact engages, so the final rollup's total equals the + # number of contact rows (matches the engine's contact-row count). + assert final_snap[account_id].total_contacts_all_time == n_contacts + + +# --------------------------------------------------------------------------- +# Relationship validator agrees with ground truth +# --------------------------------------------------------------------------- + + +def test_validator_agrees_with_planted_labels(sim): + sm, events, snapshots = sim + # Build ConversationRecords + day map from CONVERSATION_STARTED events. + day_by_conv: dict[str, int] = {} + convs_by_account: dict[str, list[ConversationRecord]] = defaultdict(list) + for e in events: + if e.event_type != SimEventType.CONVERSATION_STARTED: + continue + conv_id = f"conv_{e.event_id}" + day_by_conv[conv_id] = e.day_index + convs_by_account[e.account_id].append( + ConversationRecord( + conversation_id=conv_id, + account_id=e.account_id, + agent_id=e.payload.get("agent_id", "a"), + surface_channel="intercom", + started_at=e.timestamp, + ended_at=e.timestamp, + turn_count=1, + prose="", + trigger_event_id=e.event_id, + contact_id=e.payload.get("contact_id"), + contact_name=e.payload.get("customer_name"), + contact_role=e.payload.get("contact_role"), + ) + ) + contacts_by_account: dict[str, list] = defaultdict(list) + for c in sm.contacts: + contacts_by_account[c.account_id].append(c) + final_day = {} + for s in snapshots: + final_day[s.account_id] = max(final_day.get(s.account_id, -1), s.day_index) + + checked = 0 + for account_id in contacts_by_account: + verdicts = validate_account_relationships( + account_id=account_id, + contacts=contacts_by_account[account_id], + conversations=convs_by_account[account_id], + day_by_conv=day_by_conv, + labels=sm.relationship_labels, + as_of_day_index=final_day[account_id], + ) + for v in verdicts: + assert v.verdict.value == "pass", ( + account_id, v.dimension, v.signals_summary + ) + checked += 1 + assert checked > 0 + + +# --------------------------------------------------------------------------- +# sim_events optional fields +# --------------------------------------------------------------------------- + + +def test_contact_fields_not_required_by_validate_payload(): + # A CONVERSATION_STARTED payload without contact fields is still valid + # (they are enrichment fields added post-emission). + missing = validate_event_payload( + SimEventType.CONVERSATION_STARTED, + {"surface_channel": "intercom", "agent_id": "a", "customer_name": "n", "domain": "d"}, + ) + assert missing == [] diff --git a/tests/test_properties.py b/tests/test_properties.py index 6d74671..2c2b9a6 100644 --- a/tests/test_properties.py +++ b/tests/test_properties.py @@ -129,6 +129,16 @@ def _run(output_root: Path) -> Manifest: "snapshots_hash differs between two runs with the same seed — " f"run1={m1.snapshots_hash[:16]}… run2={m2.snapshots_hash[:16]}…" ) + # Contact modeling must be deterministic too: contacts.jsonl and + # relationship_labels.jsonl are seed-derived and must be byte-identical. + assert m1.contacts_hash == m2.contacts_hash, ( + "contacts_hash differs between two runs with the same seed — " + f"run1={m1.contacts_hash[:16]}… run2={m2.contacts_hash[:16]}…" + ) + assert m1.relationship_labels_hash == m2.relationship_labels_hash, ( + "relationship_labels_hash differs between two runs with the same seed — " + f"run1={m1.relationship_labels_hash[:16]}… run2={m2.relationship_labels_hash[:16]}…" + ) # ---------------------------------------------------------------------------