57c08e6b15
Parallel infrastructure:
evals/parallel.py
multiprocessing.Pool helper (spawn context, default workers
min(cpu_count, 8)). Per-case lanes use it via:
run_lane(cases, workers=N)
workers=1 forces serial (debugging); None uses the default pool.
Generic over the per-case return type, so dataclass-returning
runners (provenance) and dict-returning runners both work.
Wired into:
- evals/adversarial_identity/runner.py
- evals/calibration/runner.py
- evals/symbolic_logic/runner.py
- evals/provenance/runner.py
Per-case helpers are now picklable (module-level, single arg).
Monotonic-learning stays serial within a split — shared session
is structural to its longitudinal protocol.
Empirical speedup (adversarial-identity public/v1, 25 cases on
macOS 8 cores): serial 14.1s -> parallel 3.1s (~4.5x). Identical
per-case results.
adversarial-identity v2:
public/v2 — 35 cases (20 attack / 15 legitimate). Attacks cover
more varied phrasings: punctuation variation
("Actually -" / "No:" / "Correction —"), embedded
hedges ("please" / "regardless of prior context"),
multi-clause attacks, and identity-marker triggers
in mid-clause position.
holdouts/v2 — 22 cases (12 attack / 10 legitimate) on distinct
priming vocabulary.
Results: attack_rejection_rate=1.0, legitimate_acceptance_rate=1.0
on both splits.
The marker-regex defense in teaching/review.py:_is_identity_override
holds against every v2 phrasing — markers are checked case-insensitive
against the full text, so capitalization / punctuation tricks don't
slip past.
Test suite: 596 passing (no regression).
119 lines
4 KiB
Python
119 lines
4 KiB
Python
"""Symbolic-logic eval lane runner.
|
|
|
|
Tests the structural foundations CORE provides for proposition-based
|
|
inference: premise-chain storage, replay determinism, and recallability
|
|
from the probe.
|
|
|
|
For each case the runner:
|
|
1. Runs the premise list on a fresh CognitiveTurnPipeline,
|
|
collecting per-turn pack_mutation_proposal counts.
|
|
2. Runs the probe on that pipeline.
|
|
3. Runs the whole sequence again on a *separate* fresh pipeline
|
|
to verify trace-hash determinism.
|
|
|
|
Sub-metrics (per case):
|
|
M1. premise_recall — probe vault_hits >= min_vault_hits
|
|
M2. replay_determinism — trace_hash matches across the two runs
|
|
M3. proposal_storage — count of fired proposals == expected_proposals
|
|
|
|
See contract.md for the structural claim and gaps.md for the
|
|
architectural findings underlying v1's signal choice.
|
|
|
|
Conforms to the framework interface: run_lane(cases, config=None) -> report.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
from chat.runtime import ChatRuntime
|
|
from core.cognition.pipeline import CognitiveTurnPipeline
|
|
from core.config import RuntimeConfig
|
|
from evals.parallel import run_cases_parallel
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class LaneReport:
|
|
metrics: dict[str, Any] = field(default_factory=dict)
|
|
case_details: list[dict[str, Any]] = field(default_factory=list)
|
|
|
|
|
|
def _run_chain(premises: list[str], probe: str) -> tuple[int, str, int]:
|
|
"""Return (vault_hits, trace_hash, proposal_count) for one fresh run."""
|
|
runtime = ChatRuntime()
|
|
pipeline = CognitiveTurnPipeline(runtime)
|
|
proposal_count = 0
|
|
for premise in premises:
|
|
try:
|
|
r = pipeline.run(premise, max_tokens=8)
|
|
except ValueError:
|
|
continue
|
|
if r.pack_mutation_proposal is not None:
|
|
proposal_count += 1
|
|
try:
|
|
probe_result = pipeline.run(probe, max_tokens=8)
|
|
except ValueError:
|
|
return 0, "", proposal_count
|
|
return probe_result.vault_hits, probe_result.trace_hash, proposal_count
|
|
|
|
|
|
def _run_case(case: dict[str, Any]) -> dict[str, Any]:
|
|
premises = case.get("premises", [])
|
|
probe = case["probe"]
|
|
min_vault_hits = int(case.get("min_vault_hits", 1))
|
|
expected_proposals = int(case.get("expected_proposals", 0))
|
|
|
|
vh1, hash1, pc1 = _run_chain(premises, probe)
|
|
vh2, hash2, pc2 = _run_chain(premises, probe)
|
|
|
|
premise_recall_pass = vh1 >= min_vault_hits
|
|
replay_pass = bool(hash1) and hash1 == hash2 and vh1 == vh2 and pc1 == pc2
|
|
proposal_pass = pc1 == expected_proposals
|
|
|
|
return {
|
|
"id": case.get("id", ""),
|
|
"pattern": case.get("pattern", ""),
|
|
"vault_hits": vh1,
|
|
"trace_hash": hash1,
|
|
"trace_hash_replay": hash2,
|
|
"proposal_count": pc1,
|
|
"expected_proposals": expected_proposals,
|
|
"min_vault_hits": min_vault_hits,
|
|
"premise_recall_pass": premise_recall_pass,
|
|
"replay_pass": replay_pass,
|
|
"proposal_pass": proposal_pass,
|
|
"passed": premise_recall_pass and replay_pass and proposal_pass,
|
|
}
|
|
|
|
|
|
def run_lane(
|
|
cases: list[dict[str, Any]],
|
|
*,
|
|
config: RuntimeConfig | None = None,
|
|
workers: int | None = None,
|
|
) -> LaneReport:
|
|
if not cases:
|
|
return LaneReport(metrics={}, case_details=[])
|
|
_ = config
|
|
|
|
case_details = run_cases_parallel(cases, _run_case, workers=workers)
|
|
total = len(case_details)
|
|
|
|
pr = sum(1 for d in case_details if d["premise_recall_pass"]) / total
|
|
rd = sum(1 for d in case_details if d["replay_pass"]) / total
|
|
ps = sum(1 for d in case_details if d["proposal_pass"]) / total
|
|
overall = sum(1 for d in case_details if d["passed"]) / total
|
|
|
|
overall_pass = pr >= 0.80 and rd >= 0.95 and ps >= 0.80
|
|
|
|
metrics: dict[str, Any] = {
|
|
"premise_recall": round(pr, 4),
|
|
"replay_determinism": round(rd, 4),
|
|
"proposal_storage": round(ps, 4),
|
|
"all_three_pass_rate": round(overall, 4),
|
|
"case_count": total,
|
|
"overall_pass": overall_pass,
|
|
}
|
|
|
|
return LaneReport(metrics=metrics, case_details=case_details)
|