core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

feat(evals): warmed_session_consistency lane — pipeline override regression substrate

Browse source 
Asymmetric counterpart to cold_start_grounding.  Builds the
measurement substrate for the Phase B1 pipeline-override usefulness
gate.  Lane is committed now (red baseline measured) so the fix is
landed against a fixed regression target.

The 2026-05-19 design review surfaced the bug this lane catches:

  > pipeline overrode a runtime surface with a placeholder realizer
  > surface because realized_plan.surface was non-empty, even though
  > it contained '...'.  The runtime audit log still held a different
  > surface.  This is the central fluency/design fault: the system
  > can be "green" while user-facing selection, pipeline selection,
  > and telemetry selection disagree.

The lane reproduces this exactly on the current main:

  Surface "Soon is defined as ..." emitted on turn 2 of "What does
  soon mean?" (where turn 1 grounded as pack correctly).  Telemetry
  recorded a different surface than the pipeline returned.

Initial red baseline (THIS commit):
  no_placeholder_rate        = 0.4444  (target after Phase B1: 1.00)
  telemetry_consistency_rate = 0.4444  (target after Phase B1: 1.00)
  warm_grounding_stability   = 0.0000  (target after Phase B1: >=0.95)

Cold-start-grounding stays at 1.00 on its own metrics.  The cold lane
measures routing, the warmed lane measures override discipline; they
are deliberately not the same.

Files:
  evals/warmed_session_consistency/contract.md
    What is measured, why, and the asymmetry with cold_start_grounding.
    Documents the four binary per-turn signals (no_placeholder,
    pipeline_match_telemetry, pipeline_match_walk, grounded_holds_on_warm)
    and the per-case warm_grounding_stable invariant.
  evals/warmed_session_consistency/public/v1/cases.jsonl
    8 cases / 18 turns.  Mix of:
      - replay-the-same-prompt (catches override drift)
      - mixed-intent sequences (catches OOV / pack interaction)
      - cause-no-chain (must stay none across replays)
      - what-does-x-mean (the warmed variant of the cold-start test)
  evals/warmed_session_consistency/dev/cases.jsonl
    2 representative cases for fast iteration.
  evals/warmed_session_consistency/runner.py
    Framework-compliant run_lane(cases, config=None) -> LaneReport.
    Constructs ONE ChatRuntime + CognitiveTurnPipeline per case,
    plays the turn sequence through them.  Per-turn signals:
      no_placeholder       — surface free of ..., <pending>, <prior>
      telemetry_match      — pipeline result.surface == turn_log[-1].surface
      grounding_match      — actual_grounding == expected_grounding
    Per-case signal:
      warm_grounding_stable — every replayed prompt produces the same
                              grounding across turns
  tests/test_warmed_session_lane.py
    8 contract tests covering: case-set integrity, replay-pattern
    presence, lane discovery, runner emits every required metric,
    per-turn details carry all signals, and the warmed-runtime
    invariant (static check that ChatRuntime is constructed
    per-case, not per-turn and not module-scope).

NOT pinned in this commit (deliberate):
  Threshold assertions are NOT in the test file.  They will land in
  Phase B1 alongside the pipeline-override usefulness gate.  This
  lane's role at present is to PROVIDE the regression target, not
  to enforce it before the fix.

Verification: 8/8 lane tests green; the lane itself runs and emits
the red metrics documented above.
This commit is contained in:
Shay 2026-05-19 07:13:41 -07:00
parent c6b4f1d21e
commit 0cf1a8fdc4
5 changed files with 459 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

94
evals/warmed_session_consistency/contract.md Normal file
View file

@ -0,0 +1,94 @@
# Warmed-Session Consistency Eval Lane — Contract
**Lane:** `warmed_session_consistency`
**Version:** v1
**Created:** 2026-05-19
## What this lane measures
Pipeline / runtime / telemetry surface consistency across a *warmed*
session — i.e. a single `CognitiveTurnPipeline` running multiple turns
through the same `ChatRuntime`, where vault state, prior-turn context,
and session-level memory accumulate.
This is the asymmetric counterpart to `cold_start_grounding`. Cold
start measures routing on fresh runtimes. Warmed session measures
whether accumulated state, pipeline overrides, or telemetry-vs-final-
result drift can corrupt an answer that was correctly grounded at the
runtime level.
The 2026-05-19 design review surfaced the bug this lane pins:
> first pipeline run: truth - pack-grounded (...).
> second pipeline run: Truth is defined as ...
> second walk: Truth infer.
> runtime turn log: Truth infer.
The pipeline's `realized_plan.surface` overrode a perfectly good
runtime-level pack-grounded surface with a `<pending>`-rendered
placeholder. Telemetry recorded yet a third surface.
## Scoring rubric
Each case runs **N turns** through one `CognitiveTurnPipeline` /
`ChatRuntime` pair. After every turn the lane checks:
| Signal | Definition |
|---|---|
| `no_placeholder` | surface contains none of: `...`, `<pending>`, `<prior>`, ` placeholder ` |
| `pipeline_match_telemetry` | `result.surface` equals the surface in the most recent `runtime.turn_log` entry |
| `pipeline_match_walk` | `result.surface` is either equal to `result.walk_surface` (when the pipeline did not override) OR is the realized-plan output (when it did and the override was a *useful* surface) |
| `grounded_holds_on_warm` | a turn that grounded as `pack`/`teaching` on turn 1 must not regress to `none`/`vault`/walk-fragment on subsequent turns *for the same prompt* |
Lane-level metrics (rates over all (case × turn) pairs):
| Metric | Definition | v1 pass threshold |
|---|---|---|
| `no_placeholder_rate` | fraction of turns whose surface contains no placeholder | 1.00 |
| `telemetry_consistency_rate` | fraction of turns where pipeline-final == turn_log-emitted | 1.00 |
| `warm_grounding_stability` | fraction of replayed prompts whose grounding_source is byte-identical across all replays | >= 0.95 |
`no_placeholder_rate` and `telemetry_consistency_rate` are hard 1.00
because either failure is a doctrine violation, not a tunable metric.
## Why cold-start alone is not enough
`cold_start_grounding` constructs a fresh `ChatRuntime` per case to
isolate routing logic. That is correct for what it measures. But it
hides every bug class that requires accumulated state:
- pipeline overriding a good runtime surface with a worse realizer one
- telemetry emitting a surface that doesn't match the final pipeline
return value
- vault accumulation winning over pack grounding on later turns
- correction-pass injecting a backward perturbation that mis-attributes
to the wrong turn
The warmed-session lane runs these exact paths and asserts they hold.
## Cold-start invariant — INVERTED here
Unlike `cold_start_grounding`, this lane DOES NOT construct a fresh
runtime per case. Each case explicitly carries a *turn sequence*; the
runner constructs one runtime and one pipeline, then plays the
sequence through them. Each turn's expected behavior may reference
prior-turn state (e.g. "after a CORRECTION on turn 2, turn 3's surface
must still ground correctly on a fresh DEFINITION prompt").
## Case schema
```jsonl
{
"id": "warm_replay_truth_001",
"category": "pipeline_override_no_placeholder",
"turns": [
{"prompt": "What is truth?", "expected_grounding_source": "pack"},
{"prompt": "What is truth?", "expected_grounding_source": "pack"}
],
"warm_invariants": ["no_placeholder", "pipeline_match_telemetry"]
}
```
Each case carries a `turns` list (ordered). Optional
`warm_invariants` names the subset of contract checks to enforce on
this case (default: all four).

2
evals/warmed_session_consistency/dev/cases.jsonl Normal file
View file

@ -0,0 +1,2 @@
{"id":"dev_warm_truth","category":"pipeline_override_no_placeholder","turns":[{"prompt":"What is truth?","expected_grounding_source":"pack"},{"prompt":"What is truth?","expected_grounding_source":"pack"}],"warm_invariants":["no_placeholder","warm_grounding_stability"]}
{"id":"dev_warm_oov","category":"oov_does_not_drift","turns":[{"prompt":"What is quasar?","expected_grounding_source":"oov"},{"prompt":"What is quasar?","expected_grounding_source":"oov"}],"warm_invariants":["no_placeholder","warm_grounding_stability"]}

8
evals/warmed_session_consistency/public/v1/cases.jsonl Normal file
View file

@ -0,0 +1,8 @@
{"id":"warm_replay_truth_001","category":"pipeline_override_no_placeholder","turns":[{"prompt":"What is truth?","expected_grounding_source":"pack"},{"prompt":"What is truth?","expected_grounding_source":"pack"}],"warm_invariants":["no_placeholder","warm_grounding_stability"]}
{"id":"warm_replay_define_moment_002","category":"pipeline_override_no_placeholder","turns":[{"prompt":"Define moment.","expected_grounding_source":"pack"},{"prompt":"Define moment.","expected_grounding_source":"pack"}],"warm_invariants":["no_placeholder","warm_grounding_stability"]}
{"id":"warm_replay_doubt_003","category":"pipeline_override_no_placeholder","turns":[{"prompt":"What is doubt?","expected_grounding_source":"pack"},{"prompt":"What is doubt?","expected_grounding_source":"pack"}],"warm_invariants":["no_placeholder","warm_grounding_stability"]}
{"id":"warm_mixed_intents_004","category":"mixed_intent_no_drift","turns":[{"prompt":"What is truth?","expected_grounding_source":"pack"},{"prompt":"What is doubt?","expected_grounding_source":"pack"},{"prompt":"Define moment.","expected_grounding_source":"pack"}],"warm_invariants":["no_placeholder"]}
{"id":"warm_oov_stable_005","category":"oov_does_not_drift","turns":[{"prompt":"What is quasar?","expected_grounding_source":"oov"},{"prompt":"What is quasar?","expected_grounding_source":"oov"}],"warm_invariants":["no_placeholder","warm_grounding_stability"]}
{"id":"warm_cause_stays_none_006","category":"cause_no_chain_stays_none","turns":[{"prompt":"How does memory work?","expected_grounding_source":"none"},{"prompt":"How does memory work?","expected_grounding_source":"none"}],"warm_invariants":["warm_grounding_stability"]}
{"id":"warm_define_then_recall_007","category":"warm_pack_after_pack","turns":[{"prompt":"What is fact?","expected_grounding_source":"pack"},{"prompt":"What is statement?","expected_grounding_source":"pack"},{"prompt":"What is fact?","expected_grounding_source":"pack"}],"warm_invariants":["no_placeholder","warm_grounding_stability"]}
{"id":"warm_what_does_x_mean_008","category":"warm_pack_what_does","turns":[{"prompt":"What does soon mean?","expected_grounding_source":"pack"},{"prompt":"What does soon mean?","expected_grounding_source":"pack"}],"warm_invariants":["no_placeholder","warm_grounding_stability"]}

197
evals/warmed_session_consistency/runner.py Normal file
View file

@ -0,0 +1,197 @@
"""Warmed-session consistency eval lane runner.
Asymmetric counterpart to ``cold_start_grounding``. Constructs ONE
runtime + pipeline per case and plays a turn sequence through them,
asserting that pipeline overrides do not corrupt a runtime-grounded
answer and that telemetry-emitted surfaces match the pipeline's
final returned surface.
Framework contract: ``run_lane(cases, config=None) -> LaneReport``
where ``LaneReport.metrics`` is a dict and ``LaneReport.case_details``
is a list of per-case dicts.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
from chat.runtime import ChatRuntime
from core.cognition.pipeline import CognitiveTurnPipeline
_PLACEHOLDER_MARKERS = (
"...",
"<pending>",
"<prior>",
" placeholder ",
)
def _has_placeholder(surface: str) -> bool:
if not isinstance(surface, str):
return False
return any(m in surface for m in _PLACEHOLDER_MARKERS)
@dataclass(frozen=True, slots=True)
class TurnResult:
turn_index: int
prompt: str
surface: str
grounding_source: str
expected_grounding_source: str
grounding_match: bool
no_placeholder: bool
telemetry_match: bool
@dataclass(frozen=True, slots=True)
class CaseResult:
case_id: str
category: str
invariants: tuple[str, ...]
turn_results: tuple[TurnResult, ...]
warm_grounding_stable: bool
@dataclass
class LaneReport:
metrics: dict[str, Any] = field(default_factory=dict)
case_details: list[dict[str, Any]] = field(default_factory=list)
def _run_case(case: dict[str, Any]) -> CaseResult:
"""Run one case's full turn sequence through a single warmed
runtime + pipeline pair."""
turns_spec = case.get("turns", [])
invariants = tuple(case.get("warm_invariants", (
"no_placeholder", "telemetry_match", "warm_grounding_stability"
)))
runtime = ChatRuntime()
pipeline = CognitiveTurnPipeline(runtime=runtime)
turn_results: list[TurnResult] = []
grounding_by_prompt: dict[str, list[str]] = {}
for idx, turn in enumerate(turns_spec):
prompt = turn["prompt"]
expected_grounding = turn["expected_grounding_source"]
result = pipeline.run(prompt, max_tokens=8)
actual_surface = result.surface
# Telemetry match: the most recent entry in runtime.turn_log
# must carry the same surface that the pipeline returned.
# Pipeline overrides that happen AFTER turn_log emission would
# produce a mismatch here.
last_event = (
runtime.turn_log[-1] if runtime.turn_log else None
)
telemetry_surface = (
last_event.surface if last_event is not None else ""
)
actual_grounding = (
getattr(last_event, "grounding_source", None) or "none"
if last_event is not None
else "none"
)
telemetry_match = actual_surface == telemetry_surface
no_ph = not _has_placeholder(actual_surface)
grounding_match = actual_grounding == expected_grounding
turn_results.append(TurnResult(
turn_index=idx,
prompt=prompt,
surface=actual_surface,
grounding_source=actual_grounding,
expected_grounding_source=expected_grounding,
grounding_match=grounding_match,
no_placeholder=no_ph,
telemetry_match=telemetry_match,
))
grounding_by_prompt.setdefault(prompt, []).append(actual_grounding)
# Warm-grounding stability: for any prompt that appears more than
# once in this case, every replay must produce the same grounding.
stable = all(
len(set(srcs)) == 1
for srcs in grounding_by_prompt.values()
if len(srcs) > 1
)
return CaseResult(
case_id=case["id"],
category=case.get("category", "uncategorised"),
invariants=invariants,
turn_results=tuple(turn_results),
warm_grounding_stable=stable,
)
def run_lane(cases: list[dict[str, Any]], config: Any = None) -> LaneReport: # noqa: ARG001
if not cases:
return LaneReport(metrics={}, case_details=[])
results = [_run_case(c) for c in cases]
total_turns = sum(len(r.turn_results) for r in results)
no_ph = sum(
1 for r in results for t in r.turn_results if t.no_placeholder
)
telem_match = sum(
1 for r in results for t in r.turn_results if t.telemetry_match
)
grounding_match = sum(
1 for r in results for t in r.turn_results if t.grounding_match
)
replayable_cases = [
r for r in results
if any(
sum(1 for t in r.turn_results if t.prompt == tp) > 1
for tp in {t.prompt for t in r.turn_results}
)
]
stable = sum(1 for r in replayable_cases if r.warm_grounding_stable)
metrics: dict[str, Any] = {
"cases": len(results),
"total_turns": total_turns,
"no_placeholder_rate": round(no_ph / total_turns, 4) if total_turns else 1.0,
"telemetry_consistency_rate": round(telem_match / total_turns, 4) if total_turns else 1.0,
"grounding_match_rate": round(grounding_match / total_turns, 4) if total_turns else 1.0,
"warm_grounding_stability": (
round(stable / len(replayable_cases), 4)
if replayable_cases else 1.0
),
}
case_details = [
{
"case_id": r.case_id,
"category": r.category,
"invariants": list(r.invariants),
"warm_grounding_stable": r.warm_grounding_stable,
"turns": [
{
"turn_index": t.turn_index,
"prompt": t.prompt,
"surface": t.surface,
"grounding_source": t.grounding_source,
"expected_grounding_source": t.expected_grounding_source,
"grounding_match": t.grounding_match,
"no_placeholder": t.no_placeholder,
"telemetry_match": t.telemetry_match,
}
for t in r.turn_results
],
}
for r in results
]
return LaneReport(metrics=metrics, case_details=case_details)
__all__ = ["run_lane", "LaneReport", "CaseResult", "TurnResult"]

158
tests/test_warmed_session_lane.py Normal file
View file

@ -0,0 +1,158 @@
"""Contract tests for the ``warmed_session_consistency`` eval lane.
This lane catches a bug class the design-review-2026-05-19 named:
> pipeline overrode a runtime surface with a placeholder realizer
> surface because realized_plan.surface was non-empty, even though it
> contained '...'. The runtime audit log still held a different
> surface.
These tests pin:
- Case-set integrity (turn sequence + expected grounding per turn).
- Lane discovery (framework can find and load the lane).
- The runner produces every required metric (no_placeholder_rate,
telemetry_consistency_rate, warm_grounding_stability,
grounding_match_rate).
- The runner uses ONE warmed ``ChatRuntime`` per case (warmed-session
invariant — the asymmetric counterpart to cold_start_grounding's
fresh-runtime invariant).
NOTE: Pass-threshold assertions (``no_placeholder_rate == 1.0`` etc.)
are deliberately NOT pinned here yet — they will land green only after
the Phase B1 pipeline-override usefulness gate is in place. The
contract.md documents the targets; this lane's role at present is to
PROVIDE the regression substrate for that fix.
"""
from __future__ import annotations
from pathlib import Path
from evals.framework import (
discover_lanes,
get_lane,
load_cases,
load_lane_runner,
run_lane,
)
LANE_NAME = "warmed_session_consistency"
_EVAL_ROOT = Path(__file__).resolve().parent.parent / "evals" / LANE_NAME
_PUBLIC_CASES = _EVAL_ROOT / "public" / "v1" / "cases.jsonl"
class TestCaseSetIntegrity:
def test_public_cases_file_exists(self) -> None:
assert _PUBLIC_CASES.exists()
def test_every_case_has_turn_sequence(self) -> None:
cases = load_cases(_PUBLIC_CASES)
assert len(cases) >= 4
for case in cases:
assert "id" in case and isinstance(case["id"], str)
assert "turns" in case and isinstance(case["turns"], list)
assert len(case["turns"]) >= 1, case
for turn in case["turns"]:
assert "prompt" in turn and isinstance(turn["prompt"], str)
assert "expected_grounding_source" in turn
def test_at_least_one_case_has_replay_pattern(self) -> None:
"""At least one case must replay the same prompt across turns
so the ``warm_grounding_stability`` metric has something to
measure."""
cases = load_cases(_PUBLIC_CASES)
for case in cases:
prompts = [t["prompt"] for t in case["turns"]]
if len(prompts) > 1 and len(set(prompts)) < len(prompts):
return # found one
raise AssertionError("no case carries a repeated-prompt sequence")
class TestLaneDiscovery:
def test_lane_is_discoverable(self) -> None:
names = {lane.name for lane in discover_lanes()}
assert LANE_NAME in names
def test_lane_runner_loads(self) -> None:
lane = get_lane(LANE_NAME)
runner = load_lane_runner(lane)
assert hasattr(runner, "run_lane")
class TestRunnerMetrics:
"""The runner must emit every metric the contract.md names, even
when the values are red. Missing-metric drift would make the
Phase B1 regression catch silent."""
REQUIRED_METRICS = (
"cases",
"total_turns",
"no_placeholder_rate",
"telemetry_consistency_rate",
"warm_grounding_stability",
"grounding_match_rate",
)
def test_all_required_metrics_present(self) -> None:
lane = get_lane(LANE_NAME)
result = run_lane(lane, version="v1", split="public")
for name in self.REQUIRED_METRICS:
assert name in result.metrics, (
f"missing metric {name!r}; got keys: {sorted(result.metrics)}"
)
def test_per_turn_details_carry_all_signals(self) -> None:
lane = get_lane(LANE_NAME)
result = run_lane(lane, version="v1", split="public")
assert result.case_details
# Every per-turn dict must carry the four binary signals.
for case in result.case_details:
for turn in case["turns"]:
for key in (
"no_placeholder", "telemetry_match",
"grounding_match", "surface",
):
assert key in turn, (case["case_id"], turn["turn_index"], key)
class TestWarmedRuntimeInvariant:
"""The runner must construct ONE ChatRuntime+pipeline per case and
play the full turn sequence through it — the inverse of
cold_start_grounding's fresh-per-case invariant.
Static check: the runner's _run_case helper instantiates ChatRuntime
inside the function (one per case), NOT at module scope (which
would share state across cases — wrong invariant) and NOT inside
the per-turn loop (which would defeat the warmed-session purpose).
"""
def test_runner_constructs_runtime_per_case_not_per_turn(self) -> None:
src = (_EVAL_ROOT / "runner.py").read_text(encoding="utf-8")
# Module-level ChatRuntime instances would share state across
# cases — forbidden.
for line in src.splitlines():
stripped = line.lstrip()
if stripped.startswith(("_RUNTIME ", "RUNTIME ", "_PIPELINE ", "PIPELINE ")):
if "ChatRuntime(" in stripped or "CognitiveTurnPipeline(" in stripped:
raise AssertionError(
"module-scope ChatRuntime/pipeline instance "
"breaks the warmed-per-case invariant"
)
# Runtime must be constructed exactly once per case (inside
# _run_case, before the per-turn loop).
assert "runtime = ChatRuntime()" in src
# And the per-turn body must NOT re-construct it.
in_for_loop = False
for line in src.splitlines():
if "for idx, turn in enumerate(turns_spec):" in line:
in_for_loop = True
continue
if in_for_loop and line.startswith("def "):
break
if in_for_loop and "ChatRuntime(" in line:
raise AssertionError(
"ChatRuntime construction inside per-turn loop "
"defeats warmed-session invariant"
)