teleo-infrastructure/tests/test_verify_leo_unseen_reasoning_chain.py

218 lines
9.3 KiB
Python

from __future__ import annotations
import copy
from scripts import verify_leo_unseen_reasoning_chain as verifier
CLAIM_ID = "fd159490-280d-4ede-84ef-faa169cff766"
def execution_manifest(execution_sha256: str, previous: str | None, *, first: bool) -> dict:
return {
"execution_sha256": execution_sha256,
"attribution": {"status": "complete"},
"session_boundary": {
"conversation": {
"previous_execution_sha256": previous,
"prior_turn_state_bound": True,
"history_prefix_preserved": True,
"starts_from_empty_conversation": first,
}
},
}
def passing_report() -> dict:
replies = [
(
"The claim fd159490 has confidence 0.9. Its grounds row is no_source_pointer and the only other support "
"is an internal document with no external primary source. First narrower test: test one tradition."
),
(
"Assumptions versus observed support: the categorical leap is assumed. Falsifier: one contrary case. "
"PROPOSAL A is empirical. PROPOSAL B is structural. Both are staged only; nothing applied. fd159490."
),
(
"Revised Proposal A separates the mechanism from the outcome. What would resolve the disagreement: "
"Ostrom, Governing the Commons, as the primary per-case source. Stage pending_review, then human review; "
"approved is not apply, and applied_at must be non-null. Nothing staged or changed here."
),
]
fingerprints = {
"fingerprint_sha256": "f" * 64,
"table_count": 39,
"total_rows": 52167,
}
first_sha = "a" * 64
second_sha = "b" * 64
third_sha = "c" * 64
return {
"mode": "live_vps_postmerge_restart_unseen_reasoning_chain",
"remote_run_id": "run-1",
"db_counts_changed": False,
"db_fingerprint_unchanged": True,
"db_fingerprint_before": fingerprints,
"db_fingerprint_after": fingerprints,
"safety_gate": {"status": "pass"},
"execution_manifest_summary": {
"all_turns_attribution_complete": True,
"harness_source": {"git_head": "d" * 40},
},
"executed_behavior_manifest": {
"teleo_infrastructure_runtime": {"git_head": "1" * 40}
},
"service_before_after": {
"before": {"MainPID": "42", "NRestarts": "0"},
"after": {"MainPID": "42", "NRestarts": "0"},
"unchanged_from_preexisting_live_readback": True,
},
"results": [
{
"prompt_id": verifier.PROMPTS[0]["id"],
"prompt": verifier.PROMPTS[0]["message"],
"reply": replies[0],
"database_tool_trace": {
"database_tool_call_proven": True,
"database_retrieval_receipt_proven": True,
"database_tool_calls_read_only": True,
"calls": [
{
"database_invocations": [{"subcommand": "context"}],
"result": {"row_ids": [CLAIM_ID]},
}
]
},
"execution_manifest": execution_manifest(first_sha, None, first=True),
},
{
"prompt_id": verifier.PROMPTS[1]["id"],
"prompt": verifier.PROMPTS[1]["message"],
"reply": replies[1],
"database_tool_trace": {
"database_tool_call_proven": True,
"database_retrieval_receipt_proven": True,
"database_tool_calls_read_only": True,
"calls": [
{
"database_invocations": [
{"subcommand": "show"},
{"subcommand": "evidence"},
],
"result": {"row_ids": [CLAIM_ID]},
}
]
},
"execution_manifest": execution_manifest(second_sha, first_sha, first=False),
},
{
"prompt_id": verifier.PROMPTS[2]["id"],
"prompt": verifier.PROMPTS[2]["message"],
"reply": replies[2],
"database_tool_trace": {"calls": []},
"execution_manifest": execution_manifest(third_sha, second_sha, first=False),
},
],
}
def test_unseen_reasoning_chain_passes_only_with_db_challenge_revision_and_apply_boundary() -> None:
receipt = verifier.verify_report(passing_report(), source_report_sha256="e" * 64)
assert receipt["status"] == "pass"
assert all(receipt["checks"].values())
assert all(receipt["outcomes"].values())
assert receipt["schema"] == "livingip.leoUnseenReasoningChainReceipt.v2"
assert receipt["selected_knowledge_object_refs"] == ["fd159490"]
assert receipt["selected_claim_ids"] == [CLAIM_ID]
assert receipt["execution_sha256_chain"] == ["a" * 64, "b" * 64, "c" * 64]
assert receipt["deployed_git_head"] == "1" * 40
def test_unseen_reasoning_chain_accepts_belief_context_then_search_reinspection() -> None:
report = passing_report()
report["results"][0]["reply"] = (
"The selected coordination bottleneck belief has confidence 0.85 and zero evidence rows. What supports it "
"in the DB: nothing. The nearest claim has no_source_pointer. First narrower claim to test: measure "
"coordination handoff friction rather than technology capability."
)
report["results"][0]["database_tool_trace"]["calls"][0]["result"]["row_ids"] = [
"7da4851c-0aac-4084-a99f-baeeeb3aa23e"
]
report["results"][1]["reply"] = (
"Claim c0000000 is the coordination bottleneck belief and still has zero evidence. Three assumptions are "
"doing the work, and the listed falsifier is escape-proof. Two proposals for review, not applies:\n\n"
"1. A measurable handoff-friction mechanism.\n"
"2. A scoped coordination-technology capability comparison."
)
report["results"][1]["database_tool_trace"]["calls"][0]["database_invocations"] = [
{"subcommand": "search"}
]
report["results"][1]["database_tool_trace"]["calls"][0]["result"]["row_ids"] = [
"16b2c565-52a4-48c0-a9e7-109ce4793c66"
]
report["results"][2]["reply"] = (
"Revised proposal (not an apply) stays at the mechanism rather than the business outcome. What would "
"resolve the disagreement is a randomized or quasi-experimental study, such as a METR replication. It "
"enters pending_review, requires human review and approval, then a separate apply; only non-null applied_at "
"proves the write. Until then it is staged intent, not knowledge."
)
receipt = verifier.verify_report(report, source_report_sha256="e" * 64)
assert receipt["status"] == "pass"
assert all(receipt["checks"].values())
assert all(receipt["outcomes"].values())
assert receipt["selected_knowledge_object_refs"] == ["c0000000"]
assert receipt["selected_claim_ids"] == []
def test_unseen_reasoning_chain_rejects_a_challenge_that_does_not_reinspect_evidence() -> None:
report = copy.deepcopy(passing_report())
report["results"][1]["database_tool_trace"]["calls"][0]["database_invocations"] = [
{"subcommand": "show"}
]
receipt = verifier.verify_report(report, source_report_sha256="e" * 64)
assert receipt["status"] == "fail"
assert receipt["checks"]["same_subject_reinspected_after_challenge"] is False
assert receipt["outcomes"]["survives_a_claim_body_and_evidence_challenge"] is False
def test_unseen_reasoning_chain_rejects_unreceipted_or_writable_reinspection() -> None:
report = copy.deepcopy(passing_report())
report["results"][1]["database_tool_trace"]["database_retrieval_receipt_proven"] = False
report["results"][1]["database_tool_trace"]["database_tool_calls_read_only"] = False
receipt = verifier.verify_report(report, source_report_sha256="e" * 64)
assert receipt["status"] == "fail"
assert receipt["checks"]["same_subject_reinspected_after_challenge"] is False
def test_unseen_reasoning_chain_rejects_unrelated_subject_drift() -> None:
report = copy.deepcopy(passing_report())
report["results"][0]["reply"] = (
"The chosen axiom has confidence 0.85, zero evidence, and no_source_pointer."
)
report["results"][1]["reply"] = (
"Claim c0000000 concerns orbital manufacturing. Its assumption has an escape-proof falsifier. "
"Proposal A and Proposal B are staged only for review; nothing applied."
)
receipt = verifier.verify_report(report, source_report_sha256="e" * 64)
assert receipt["status"] == "fail"
assert receipt["checks"]["same_subject_reinspected_after_challenge"] is False
def test_unseen_reasoning_chain_rejects_broken_conversation_parent_hash() -> None:
report = copy.deepcopy(passing_report())
report["results"][2]["execution_manifest"]["session_boundary"]["conversation"][
"previous_execution_sha256"
] = "0" * 64
receipt = verifier.verify_report(report, source_report_sha256="e" * 64)
assert receipt["status"] == "fail"
assert receipt["checks"]["all_turn_execution_manifests_complete"] is False