CIAgent
1a46606827
Task 3-2-01: prompts/grading.py (grader-v1 rubric, 4 criteria x 0-4 anchors, a-4 churn
advisory) + grading/engine.py — GradingEngine with the BINDING G-4 gate-first ordering
(INCOMPLETE_FLOODED -> gaps -> empty; LLM unreachable for gated traces; first-class
UNGRADABLE_* GradeRecords, model="none" provenance), digest-only prompts (D-028; planted
marker proven absent from all provider messages), D-020 reused via one module-direct
import of agents/structured (grep-auditable). RubricScore validated per-criterion.
Task 3-2-02: corpus/trace_fixtures.py (D-021-aligned archetype IDs) + ordering-contract
calibration test (strong>=lazy on process; strong>struggling on correctness; digest
feature separation asserted deterministically).
54 grading tests green; suite 272 green; ruff clean.
---ci---
phase: 3
milestone: v0.3
status: execute
requirements: {covered: [REQ-3-004], partial: []}
---/ci---
362 lines
14 KiB
Python
362 lines
14 KiB
Python
"""Grading calibration CONTRACT test (Task 3-2-02, REQ-3-004, D-021).
|
|
|
|
HONEST SCOPE — what this test is and is NOT:
|
|
|
|
This is a CALIBRATION CONTRACT test, not an LLM quality test. The mock
|
|
provider scripts archetype-mapped rubric scores for each fixture; the
|
|
assertions verify that:
|
|
|
|
1. the ENGINE pipeline (gate → digest → prompt → D-020 → validate →
|
|
persist) runs each archetype end-to-end and preserves the score
|
|
ordering the scripts impose;
|
|
2. the fixtures' digest FEATURES actually separate the archetypes
|
|
deterministically (the non-LLM half of the calibration: paste-and-run
|
|
vs iterative debugging vs never-passing MUST produce observably
|
|
different digests — otherwise no rubric could ever separate them);
|
|
3. fixture IDs are valid corpus-aligned strings (D-021).
|
|
|
|
It does NOT prove the real LLM scores archetypes in this order — that
|
|
requires live-model evaluation, which is out of scope for v0.3 tests
|
|
(tests NEVER call the cloud; conftest enforces). What the contract pins
|
|
is the ORDERING the rubric + digest must eventually enforce:
|
|
|
|
strong.process_quality >= lazy.process_quality
|
|
strong.correctness > struggling.correctness
|
|
|
|
If a future prompt/digest change makes the scripted mapping unachievable
|
|
by ANY model (e.g. the digest stops separating the archetypes), test 2
|
|
fails — that is the calibration signal.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from ai_service.corpus.telemetry import LAB_SCENARIOS
|
|
from ai_service.corpus.trace_fixtures import (
|
|
CALIBRATION_LEARNER_ID,
|
|
LAZY_BUILDER,
|
|
STRONG_BUILDER,
|
|
STRUGGLING_BUILDER,
|
|
TRACE_FIXTURES,
|
|
digest_of,
|
|
)
|
|
from ai_service.grading.engine import GradingEngine
|
|
from ai_service.grading.store import SQLiteGradeStore
|
|
from ai_service.telemetry.ingest import TraceIntegrityMap
|
|
from ai_service.telemetry.store import SQLiteTraceStore
|
|
|
|
#: Archetype-mapped rubric scripts (the ORDERING CONTRACT in numbers).
|
|
#: strong: iterative verified building — highest process quality.
|
|
#: lazy: ends pass with zero iteration — the a-4 negative (churn with no
|
|
#: test progress caps process_quality at 1; here it's paste-churn +
|
|
#: one late test, so 1). correctness lands high (final pass, fast).
|
|
#: struggling: never passes — lowest correctness.
|
|
ARCHETYPE_SCORES: dict[str, dict] = {
|
|
"strong": {
|
|
"criteria": {
|
|
"process_quality": 4,
|
|
"correctness": 4,
|
|
"debugging_discipline": 4,
|
|
"test_usage": 4,
|
|
},
|
|
"strengths": ["tight edit-test loops; every failure cycle closed"],
|
|
"gaps": ["session ended without a final full-suite run"],
|
|
"verdict": "mastered",
|
|
},
|
|
"lazy": {
|
|
"criteria": {
|
|
"process_quality": 1, # a-4: bulk paste + churn, one late test
|
|
"correctness": 3,
|
|
"debugging_discipline": 1,
|
|
"test_usage": 1,
|
|
},
|
|
"strengths": ["final tests pass on the first run"],
|
|
"gaps": ["one bulk paste with zero verified increments"],
|
|
"verdict": "developing",
|
|
},
|
|
"struggling": {
|
|
"criteria": {
|
|
"process_quality": 2,
|
|
"correctness": 0, # never passes
|
|
"debugging_discipline": 1, # cycles never close
|
|
"test_usage": 3,
|
|
},
|
|
"strengths": ["persisted through repeated failures"],
|
|
"gaps": ["the same failure recurred three times"],
|
|
"verdict": "not_yet",
|
|
},
|
|
}
|
|
|
|
|
|
class ArchetypeProvider:
|
|
"""Mock provider scripting archetype-mapped scores keyed by fixture.
|
|
|
|
Keying on the digest embedded in the user turn (not on call order)
|
|
keeps the mapping robust to gate-path noise and asserts the prompt
|
|
really carries the fixture's digest. Records the last user turn so
|
|
tests can re-assert D-028 (digest in prompt; raw events out).
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self.digest_to_archetype: dict[str, str] = {}
|
|
self.calls = 0
|
|
self.last_user_prompt: str = ""
|
|
|
|
def register(self, fixture) -> None:
|
|
self.digest_to_archetype[digest_of(fixture).model_dump_json()] = (
|
|
fixture.archetype
|
|
)
|
|
|
|
async def chat(self, messages, *, model, temperature=0.7, response_format=None):
|
|
self.calls += 1
|
|
user = next((m for m in reversed(messages) if m.role == "user"), None)
|
|
assert user is not None
|
|
self.last_user_prompt = user.content
|
|
archetype = None
|
|
for digest_json, arch in self.digest_to_archetype.items():
|
|
if digest_json in user.content:
|
|
archetype = arch
|
|
break
|
|
assert archetype is not None, "fixture digest not found in prompt"
|
|
return json.dumps(ARCHETYPE_SCORES[archetype])
|
|
|
|
async def stream_chat(self, messages, *, model, temperature=0.7, response_format=None):
|
|
yield await self.chat(
|
|
messages, model=model, temperature=temperature, response_format=response_format
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def trace_store(tmp_path: Path) -> SQLiteTraceStore:
|
|
store = SQLiteTraceStore(db_path=tmp_path / "traces.db")
|
|
yield store
|
|
store.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def grade_store(tmp_path: Path) -> SQLiteGradeStore:
|
|
store = SQLiteGradeStore(db_path=tmp_path / "grades.db")
|
|
yield store
|
|
store.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def provider() -> ArchetypeProvider:
|
|
p = ArchetypeProvider()
|
|
for fixture in (STRONG_BUILDER, LAZY_BUILDER, STRUGGLING_BUILDER):
|
|
p.register(fixture)
|
|
return p
|
|
|
|
|
|
@pytest.fixture
|
|
def engine(trace_store, grade_store, provider) -> GradingEngine:
|
|
return GradingEngine(
|
|
trace_store, grade_store, TraceIntegrityMap(), provider, model="gemma4:31b"
|
|
)
|
|
|
|
|
|
async def _grade_fixture(engine, trace_store, fixture) -> dict:
|
|
for event in fixture.events:
|
|
trace_store.append(event)
|
|
record = await engine.grade(fixture.learner_id, fixture.task_id)
|
|
assert record.verdict == "GRADED", (
|
|
f"fixture {fixture.fixture_id} did not grade cleanly: {record.verdict}"
|
|
)
|
|
return record.scores
|
|
|
|
|
|
class TestFixtureAlignment:
|
|
"""D-021: fixture IDs align with the v0.2 corpus scenario IDs."""
|
|
|
|
def test_every_fixture_id_is_corpus_scenario_scoped(self):
|
|
for fixture in TRACE_FIXTURES.values():
|
|
assert fixture.fixture_id.count("::") == 1
|
|
scenario_part, trace_part = fixture.fixture_id.split("::")
|
|
assert scenario_part in LAB_SCENARIOS, (
|
|
f"{scenario_part!r} is not a v0.2 lab scenario id"
|
|
)
|
|
assert trace_part.endswith("-trace")
|
|
# double-brace-free single slug
|
|
assert "-" in trace_part
|
|
|
|
def test_every_fixture_aligns_to_an_existing_scenario(self):
|
|
assert set(f.aligned_scenario_id for f in TRACE_FIXTURES.values()) <= set(
|
|
LAB_SCENARIOS
|
|
)
|
|
|
|
def test_fixture_ids_are_valid_strings(self):
|
|
for fixture in TRACE_FIXTURES.values():
|
|
assert fixture.fixture_id and isinstance(fixture.fixture_id, str)
|
|
assert fixture.learner_id == CALIBRATION_LEARNER_ID
|
|
assert fixture.learner_id.startswith("learner-00")
|
|
|
|
def test_lazy_alignment_is_the_flagged_scenario(self):
|
|
# v0.2's paste-and-run archetype is the flagged lab scenario (D-021).
|
|
assert LAZY_BUILDER.aligned_scenario_id == "lab-scenario-flagged"
|
|
assert LAZY_BUILDER.archetype == "lazy"
|
|
|
|
def test_all_three_archetypes_present(self):
|
|
assert {f.archetype for f in TRACE_FIXTURES.values()} == {
|
|
"strong",
|
|
"lazy",
|
|
"struggling",
|
|
}
|
|
|
|
def test_events_are_valid_telemetry_events(self):
|
|
from ai_service.telemetry.models import TelemetryEvent
|
|
|
|
for fixture in TRACE_FIXTURES.values():
|
|
assert len(fixture.events) > 0
|
|
for e in fixture.events:
|
|
assert isinstance(e, TelemetryEvent)
|
|
assert e.seq >= 0
|
|
# contiguous seqs: a grading trace must clear the G-4 gate
|
|
seqs = [e.seq for e in fixture.events]
|
|
assert seqs == list(range(len(seqs))), (
|
|
f"{fixture.fixture_id} has seq gaps — it would trip the gate"
|
|
)
|
|
|
|
|
|
class TestDigestSeparatesArchetypes:
|
|
"""The non-LLM half of calibration: digests differ in the RIGHT direction.
|
|
|
|
If these deterministic assertions ever fail, no rubric — however good —
|
|
could separate the archetypes: the digest would be lossy in the wrong
|
|
places. These are the features the a-4 advisory and the rubric anchors
|
|
speak about, computed over REAL trace fixtures.
|
|
"""
|
|
|
|
def test_strong_shows_iterative_cycles_lazy_shows_none(self):
|
|
strong = digest_of(STRONG_BUILDER)
|
|
lazy = digest_of(LAZY_BUILDER)
|
|
assert strong.error_fix_cycles >= 1
|
|
assert lazy.error_fix_cycles == 0
|
|
|
|
def test_strong_tests_early_lazy_tests_late(self):
|
|
strong = digest_of(STRONG_BUILDER)
|
|
lazy = digest_of(LAZY_BUILDER)
|
|
assert strong.first_test_pass_offset_s is not None
|
|
assert lazy.first_test_pass_offset_s is not None
|
|
assert strong.first_test_pass_offset_s < lazy.first_test_pass_offset_s
|
|
|
|
def test_struggling_never_passes(self):
|
|
struggling = digest_of(STRUGGLING_BUILDER)
|
|
assert struggling.final_test_status == "fail"
|
|
assert struggling.test_pass_count == 0
|
|
assert struggling.test_fail_count >= 3
|
|
|
|
def test_strong_and_struggling_both_end(self):
|
|
strong = digest_of(STRONG_BUILDER)
|
|
assert strong.final_test_status == "pass"
|
|
assert strong.test_fail_count >= 1 # honest failure, then closed
|
|
|
|
def test_lazy_churns_without_test_progress(self):
|
|
"""a-4 anchor: bulk edit + command churn with (almost) no test signal."""
|
|
lazy = digest_of(LAZY_BUILDER)
|
|
assert lazy.edit_count >= 2
|
|
assert lazy.test_fail_count == 0
|
|
assert lazy.command_count >= 2
|
|
assert lazy.final_test_status == "pass" # ...and still one late test
|
|
|
|
|
|
class TestOrderingContract:
|
|
"""The scripted ORDERING CONTRACT through the full engine pipeline."""
|
|
|
|
async def test_strong_at_least_lazy_on_process_quality(
|
|
self, engine, trace_store
|
|
):
|
|
strong = await _grade_fixture(engine, trace_store, STRONG_BUILDER)
|
|
lazy = await _grade_fixture(engine, trace_store, LAZY_BUILDER)
|
|
assert (
|
|
strong["criteria"]["process_quality"]
|
|
>= lazy["criteria"]["process_quality"]
|
|
)
|
|
|
|
async def test_strong_above_struggling_on_correctness(self, engine, trace_store):
|
|
strong = await _grade_fixture(engine, trace_store, STRONG_BUILDER)
|
|
struggling = await _grade_fixture(engine, trace_store, STRUGGLING_BUILDER)
|
|
assert strong["criteria"]["correctness"] > struggling["criteria"]["correctness"]
|
|
|
|
async def test_full_ordering_across_all_three_archetypes(
|
|
self, engine, trace_store, grade_store
|
|
):
|
|
scores = {
|
|
arch: await _grade_fixture(engine, trace_store, f)
|
|
for arch, f in (
|
|
("strong", STRONG_BUILDER),
|
|
("lazy", LAZY_BUILDER),
|
|
("struggling", STRUGGLING_BUILDER),
|
|
)
|
|
}
|
|
# the required ordering contract (task 3-2-02):
|
|
assert (
|
|
scores["strong"]["criteria"]["process_quality"]
|
|
>= scores["lazy"]["criteria"]["process_quality"]
|
|
)
|
|
assert (
|
|
scores["strong"]["criteria"]["correctness"]
|
|
> scores["struggling"]["criteria"]["correctness"]
|
|
)
|
|
# Deliberate NON-ordering, asserted to document the calibration
|
|
# intent: struggling > lazy on process quality. Process quality is
|
|
# not outcome — the struggling builder iterated (edits between
|
|
# runs, three test runs), the lazy builder only pasted once and
|
|
# verified once (a-4 caps that at 1). A rubric that scores a
|
|
# never-passing iterator BELOW a paste-and-runner on process is
|
|
# miscalibrated on its own anchors.
|
|
assert (
|
|
scores["struggling"]["criteria"]["process_quality"]
|
|
> scores["lazy"]["criteria"]["process_quality"]
|
|
)
|
|
# verdicts also ordered: mastered >= developing > not_yet
|
|
assert scores["strong"]["verdict"] == "mastered"
|
|
assert scores["lazy"]["verdict"] == "developing"
|
|
assert scores["struggling"]["verdict"] == "not_yet"
|
|
|
|
# all three persisted under the calibration learner, latest-state
|
|
grades = grade_store.list_for_learner(CALIBRATION_LEARNER_ID)
|
|
assert len(grades) == 3
|
|
assert {g.task_id for g in grades} == {
|
|
"task-calibration-strong",
|
|
"task-calibration-lazy",
|
|
"task-calibration-struggling",
|
|
}
|
|
|
|
async def test_prompt_carries_fixture_digest_not_raw_events(
|
|
self, engine, trace_store, provider
|
|
):
|
|
"""D-028 on the calibration path: digest in the prompt, raw events
|
|
never — the lazy fixture plants distinctive payload strings
|
|
("eval.py", the bulk 240-line paste) that must NOT appear."""
|
|
for event in LAZY_BUILDER.events:
|
|
trace_store.append(event)
|
|
await engine.grade(LAZY_BUILDER.learner_id, LAZY_BUILDER.task_id)
|
|
|
|
assert provider.calls == 1
|
|
prompt = provider.last_user_prompt
|
|
# the fixture's digest reached the LLM...
|
|
assert "PROCESS TRACE DIGEST (JSON):" in prompt
|
|
assert '"error_fix_cycles"' in prompt
|
|
# ...but no raw trace material or identity did (D-028)
|
|
assert "eval.py" not in prompt
|
|
assert LAZY_BUILDER.learner_id not in prompt
|
|
assert LAZY_BUILDER.task_id not in prompt
|
|
|
|
async def test_struggling_digests_reach_engine_without_gate(
|
|
self, engine, trace_store, provider
|
|
):
|
|
for event in STRUGGLING_BUILDER.events:
|
|
trace_store.append(event)
|
|
record = await engine.grade(
|
|
STRUGGLING_BUILDER.learner_id, STRUGGLING_BUILDER.task_id
|
|
)
|
|
assert record.verdict == "GRADED" # honest struggle is gradable
|
|
assert record.scores["criteria"]["correctness"] == 0
|
|
# a never-passing session still yields a real verdict — the grade
|
|
# is low, not missing
|
|
assert record.scores["verdict"] == "not_yet"
|