241 lines
11 KiB
Python
241 lines
11 KiB
Python
"""Tests for the m3taversal out-of-sample benchmark and generic handler harness."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO_ROOT / "scripts"))
|
|
|
|
import run_leo_count_receipt_correction_challenge as correction # noqa: E402
|
|
import run_leo_direct_claim_handler_suite as handler # noqa: E402
|
|
import working_leo_m3taversal_oos_benchmark as benchmark # noqa: E402
|
|
|
|
|
|
def good_reply(prompt_id: str, token: str) -> str:
|
|
common = (
|
|
"DB readback: claims: 1837; sources: 4145; claim_edges: 4916; claim_evidence: 4670; "
|
|
"kb_proposals: 26. "
|
|
"Approved is not the same as applied. I checked the current Postgres canonical public.claims, "
|
|
"public.sources, public.claim_edges, public.claim_evidence and kb_stage proposal tables. I cannot claim a "
|
|
"canonical change without row-level before/after readback, applied_at, and postflight proof. The next action "
|
|
"is a guarded review and apply packet with a receipt. "
|
|
)
|
|
if prompt_id == "OOS-02":
|
|
return common + (
|
|
"Telegram file refs, document_evaluations, proposal source_ref, raw PDF files, and public.sources are "
|
|
"different layers. I would preserve source provenance, deduplicate claims, retain contradictory evidence "
|
|
"as caveats, stage a proposal for review, then apply only after approval."
|
|
)
|
|
if prompt_id == "OOS-03":
|
|
return common + (
|
|
"SOUL.md is a runtime artifact rendered from DB-first identity rows in Postgres. A direct SOUL.md edit "
|
|
"does not change canonical identity; verify canonical rows, render/sync, restart, and compare the artifact."
|
|
)
|
|
if prompt_id == "OOS-05":
|
|
return common + (
|
|
"A Telegram attachment and proposal source_ref are staging pointers, not canonical evidence. Audit the "
|
|
"document_evaluation, source_ref, public.sources row, claim_evidence join, and guarded apply receipt."
|
|
)
|
|
if prompt_id == "OOS-06":
|
|
return common + (
|
|
"Keep observations as claims with sources and evidence, link edges, place a framework in a reasoning tool "
|
|
"or concept map, retain the disputed interpretation as a caveat, stage governance separately, and record "
|
|
"the correction as a superseding relationship before review and apply."
|
|
)
|
|
if prompt_id == "OOS-07":
|
|
return common + (
|
|
f"Blocker: I will remember {token} as the approved-but-not-applied canonical gap for this chat only."
|
|
)
|
|
if prompt_id == "OOS-08":
|
|
return (
|
|
common
|
|
+ f"The label was {token}; blocker: the approved-but-not-applied canonical gap, closed by postflight "
|
|
"readback."
|
|
)
|
|
if prompt_id == "OOS-09":
|
|
return (
|
|
"Call the current visible participant m3taversal, exactly.\n"
|
|
"Use the current Telegram update and visible handle; do not infer a personal name from memory.\n"
|
|
"Never transfer or mix one participant's identity into another user's reply.\n"
|
|
"I cannot claim another identity source without explicit proof."
|
|
)
|
|
return common + "Fresh readback is required before the demo claim changes."
|
|
|
|
|
|
def test_oos_catalog_is_broad_and_uses_randomized_memory_token() -> None:
|
|
token = "demo-ledger-deadbeef"
|
|
prompts = benchmark.prompt_catalog(token)
|
|
assert [prompt["id"] for prompt in prompts] == [f"OOS-{index:02d}" for index in range(1, 10)]
|
|
assert token in prompts[-3]["message"]
|
|
assert token not in prompts[-2]["message"]
|
|
joined = "\n".join(prompt["message"] for prompt in prompts)
|
|
assert "PDF" in joined
|
|
assert "tweets" in joined
|
|
assert "SOUL.md" in joined
|
|
assert "Do not ask me for row IDs" in joined
|
|
assert "@m3taversal" in joined
|
|
|
|
|
|
def test_oos_score_passes_complete_behavior_and_memory_pair() -> None:
|
|
token = "demo-ledger-deadbeef"
|
|
results = [
|
|
{"prompt_id": prompt["id"], "reply": good_reply(prompt["id"], token)}
|
|
for prompt in benchmark.prompt_catalog(token)
|
|
]
|
|
score = benchmark.score_results(results, memory_token=token)
|
|
assert score["pass"] is True
|
|
assert score["passes"] == 9
|
|
|
|
|
|
def test_oos_score_fails_when_memory_token_is_not_recalled() -> None:
|
|
token = "demo-ledger-deadbeef"
|
|
results = [
|
|
{"prompt_id": prompt["id"], "reply": good_reply(prompt["id"], token)}
|
|
for prompt in benchmark.prompt_catalog(token)
|
|
]
|
|
results[-2]["reply"] = results[-2]["reply"].replace(token, "some-other-label")
|
|
score = benchmark.score_results(results, memory_token=token)
|
|
assert score["pass"] is False
|
|
assert score["failures"][0]["prompt_id"] == "OOS-08"
|
|
assert score["failures"][0]["custom_signals"]["memory_token"] is False
|
|
|
|
|
|
def test_oos_identity_case_requires_exact_visible_handle_and_no_alias() -> None:
|
|
token = "demo-ledger-deadbeef"
|
|
prompt = benchmark.prompt_catalog(token)[-1]
|
|
good = benchmark.score_reply(prompt, good_reply(prompt["id"], token), memory_token=token)
|
|
assert good["pass"] is True
|
|
|
|
bad_reply = (
|
|
"The visible handle is m3taversal, but I will address him as m3ta based on stale memory. "
|
|
"I cannot claim the database changed without proof."
|
|
)
|
|
bad = benchmark.score_reply(prompt, bad_reply, memory_token=token)
|
|
assert bad["pass"] is False
|
|
assert bad["custom_signals"]["no_unverified_alias"] is False
|
|
|
|
|
|
def test_oos_identity_case_accepts_inflected_carry_language() -> None:
|
|
token = "demo-ledger-deadbeef"
|
|
prompt = benchmark.prompt_catalog(token)[-1]
|
|
reply = (
|
|
"Call the current visible participant m3taversal, exactly.\n"
|
|
"Use only the current Telegram update and visible sender.\n"
|
|
"Leo never carries a previous participant identity into another user's reply.\n"
|
|
"No nickname or personal name is authorized."
|
|
)
|
|
score = benchmark.score_reply(prompt, reply, memory_token=token)
|
|
assert score["pass"] is True
|
|
|
|
|
|
def test_oos_score_rejects_proposed_schema_presented_as_current() -> None:
|
|
token = "demo-ledger-deadbeef"
|
|
prompt = benchmark.prompt_catalog(token)[1]
|
|
reply = good_reply(prompt["id"], token) + " public.claims stores a body and forecast_resolution for each row."
|
|
score = benchmark.score_reply(prompt, reply, memory_token=token)
|
|
assert score["pass"] is False
|
|
assert score["current_schema_overclaims"] == ["claims_unshipped_fields"]
|
|
|
|
|
|
def test_oos_schema_guard_allows_explicit_future_schema_gap() -> None:
|
|
reply = (
|
|
"Current public.claims has no body or forecast-resolution column; those are proposed future fields and "
|
|
"would require a schema gap proposal."
|
|
)
|
|
assert benchmark.current_schema_overclaims(reply) == []
|
|
|
|
|
|
def test_oos_schema_guard_distinguishes_superseded_by_column_from_edge_type() -> None:
|
|
valid = (
|
|
"Create a supersedes edge from the new claim to the old claim, then set the old claim's "
|
|
"superseded_by pointer to the new claim ID."
|
|
)
|
|
invalid = "Create a superseded_by edge between the two claims."
|
|
assert benchmark.current_schema_overclaims(valid) == []
|
|
assert benchmark.current_schema_overclaims(invalid) == ["invalid_current_edge_type"]
|
|
|
|
|
|
def test_oos_schema_guard_allows_source_excerpt_reached_through_evidence_link() -> None:
|
|
valid = (
|
|
"public.claim_evidence links the claim to a public.sources row whose source excerpt grounds the observation."
|
|
)
|
|
invalid = "The public.claim_evidence row has an excerpt field."
|
|
assert benchmark.current_schema_overclaims(valid) == []
|
|
assert benchmark.current_schema_overclaims(invalid) == ["unshipped_evidence_excerpt"]
|
|
|
|
|
|
def test_oos_schema_guard_rejects_unreviewed_claim_taxonomy_and_apply_surface() -> None:
|
|
unreviewed_type = "Create a public.claims row with type observation."
|
|
unsupported_apply = (
|
|
"The approve_claim apply_payload includes a governance gate row and a superseded_by column update."
|
|
)
|
|
assert benchmark.current_schema_overclaims(unreviewed_type) == ["unreviewed_claim_type"]
|
|
assert benchmark.current_schema_overclaims(unsupported_apply) == ["unsupported_approve_claim_surface"]
|
|
|
|
|
|
def test_oos_score_rejects_blanket_all_five_counts_must_move_claim() -> None:
|
|
token = "demo-ledger-deadbeef"
|
|
prompt = benchmark.prompt_catalog(token)[3]
|
|
reply = good_reply(prompt["id"], token) + " All five counts must move after every apply."
|
|
score = benchmark.score_reply(prompt, reply, memory_token=token)
|
|
assert score["pass"] is False
|
|
assert score["invalid_count_invariant_detected"] is True
|
|
|
|
|
|
def test_oos_score_rejects_blanket_all_five_counts_higher_claim() -> None:
|
|
reply = "The proof is all five canonical counts higher than the baseline."
|
|
assert benchmark.asserts_invalid_count_invariant(reply) is True
|
|
|
|
|
|
def test_oos_score_does_not_penalize_an_explicit_retraction() -> None:
|
|
reply = "The claim that all five counts must move is wrong; packet-specific row readback is required."
|
|
assert benchmark.asserts_invalid_count_invariant(reply) is False
|
|
|
|
|
|
def test_count_receipt_correction_requires_packet_specific_row_receipts() -> None:
|
|
reply = """
|
|
No. That rule was too strong: not all table counts move; the delta depends on the packet.
|
|
public.claims and public.sources stay unchanged when reused. For attach_evidence with an existing claim and
|
|
existing source, public.claim_evidence gets a new row. For an edge-only packet, public.claim_edges gets a new
|
|
row. kb_stage.kb_proposals count may remain unchanged while its status and applied_at change. The universal
|
|
receipt is a committed transaction, non-null applied_at, and postflight row-level readback proving the packet's
|
|
declared expected rows and IDs. public.claims, public.sources, public.claim_edges, and public.claim_evidence are
|
|
checked table by table against that packet.
|
|
"""
|
|
score = correction.score_reply(reply)
|
|
assert score["pass"] is True
|
|
|
|
|
|
def test_count_receipt_correction_rejects_repeated_bad_invariant() -> None:
|
|
reply = "All five counts must move after every apply."
|
|
score = correction.score_reply(reply)
|
|
assert score["pass"] is False
|
|
assert score["invalid_count_invariant_detected"] is True
|
|
|
|
|
|
def test_generic_handler_template_accepts_an_oos_prompt_set() -> None:
|
|
script = handler.build_remote_script(
|
|
"cafef00d",
|
|
prompts=[{"id": "OOS-01", "dimension": "demo", "message": "Broad question"}],
|
|
suite_mode="oos_mode",
|
|
report_prefix="oos-report",
|
|
prompt_note='Out of sample "quoted" prompts.',
|
|
)
|
|
assert '"mode": "oos_mode"' in script
|
|
assert 'RUN_ID = "cafef00d"' in script
|
|
assert 'REPORT_PREFIX = "oos-report"' in script
|
|
assert 'Path(f"/tmp/{REPORT_PREFIX}-{RUN_ID}.json")' in script
|
|
assert 'Out of sample \\"quoted\\" prompts.' in script
|
|
assert '"id": "OOS-01"' in script
|
|
assert "__PROMPTS_JSON__" not in script
|
|
|
|
|
|
def test_generic_handler_template_rejects_unsafe_report_prefix() -> None:
|
|
try:
|
|
handler.build_remote_script("cafef00d", report_prefix="../escape")
|
|
except ValueError as exc:
|
|
assert "report_prefix" in str(exc)
|
|
else:
|
|
raise AssertionError("unsafe report prefix was accepted")
|