teleo-infrastructure/tests/test_working_leo_open_ended_benchmark.py

"""Tests for the open-ended Working Leo benchmark."""

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 working_leo_open_ended_benchmark as bench  # noqa: E402


def test_prompts_are_open_ended_and_not_id_led():
    assert len(bench.OPEN_ENDED_PROMPTS) >= 5
    joined = "\n".join(prompt["message"] for prompt in bench.OPEN_ENDED_PROMPTS)
    assert "00957f6c-9883-4015-95a4-6b09367efb0e" not in joined
    assert "c167933e-d513-4f43-9335-d5d8aeb259f2" not in joined
    assert "same state as last night" in joined


def test_cory_style_scenarios_are_broad_outcome_cases():
    assert len(bench.CORY_STYLE_OUTCOME_SCENARIOS) >= 9
    joined = "\n".join(prompt["message"] for prompt in bench.CORY_STYLE_OUTCOME_SCENARIOS)
    assert "00957f6c-9883-4015-95a4-6b09367efb0e" not in joined
    assert "c167933e-d513-4f43-9335-d5d8aeb259f2" not in joined
    assert "proposal ID" not in joined
    assert "partner demo" in joined
    assert "They're the same state as last night" in joined
    assert "SOUL.md" in joined
    assert "decision matrix" in joined
    assert "document artifacts" in joined
    assert {prompt["runtime"] for prompt in bench.CORY_STYLE_OUTCOME_SCENARIOS} == {
        "disposable_clone_or_sandbox_first"
    }


def test_prompt_catalog_keeps_live_default_smaller_than_cory_style_suite():
    assert bench.prompt_catalog() == bench.OPEN_ENDED_PROMPTS
    full_catalog = bench.prompt_catalog(include_cory_style=True)
    assert len(full_catalog) == len(bench.OPEN_ENDED_PROMPTS) + len(bench.CORY_STYLE_OUTCOME_SCENARIOS)
    assert any(prompt["id"] == "CS-01" for prompt in full_catalog)


def test_scores_good_cory_style_answer_as_pass():
    prompt = bench.OPEN_ENDED_PROMPTS[1]
    reply = (
        "I would not assume approval changed the canonical database. I need to split proposed, pending_review, "
        "approved, applied, and not applied. I would query kb_stage plus canonical public.claims, public.sources, "
        "public.claim_edges, and public.claim_evidence, then give row-level before/after counts and row ids. "
        "Until that readback exists I cannot claim it applied. The next action is a guarded staging or apply packet "
        "with preflight, clone rehearsal, and postflight proof."
    )
    result = bench.score_reply(prompt, reply)
    assert result["pass"] is True


def test_scores_live_style_open_ended_triage_answer_as_pass():
    prompt = bench.OPEN_ENDED_PROMPTS[0]
    reply = (
        "Cory likely means the agents did not advance the graph state beyond yesterday: same claims, same edges, "
        "same proposal queue. I would first check kb_stage.kb_proposals for new proposals since last night, then "
        "check whether agent processes actually executed through gateway logs or scheduler logs. If proposals exist "
        "but the canonical graph is unchanged, that is the approved-not-applied gap. What counts as fixed is new "
        "canonical rows in public.claims, public.claim_edges, or public.claim_evidence with created_at after the "
        "baseline and provenance back to a stage to approve to apply chain. A full-green run means agent fires, "
        "proposal lands in kb_stage, reviewer approves, apply tool writes canonical rows, and readback confirms. "
        "Proposed means pending_review in kb_stage only. Approved means approval is a human gate, not a write. "
        "Applied means public.* rows exist; the only ground truth is whether the row is present in the canonical "
        "table, and the canonical row is the authoritative test. Approved with zero canonical effect means the graph "
        "is unchanged."
    )
    result = bench.score_reply(prompt, reply)
    assert result["pass"] is True


def test_scores_overconfident_answer_as_fail():
    prompt = bench.OPEN_ENDED_PROMPTS[1]
    reply = "Yes, I already applied the changes to the database. We talked about it yesterday, so the KB is updated."
    result = bench.score_reply(prompt, reply)
    assert result["pass"] is False
    assert result["overclaim_detected"] is True


def test_score_results_requires_all_prompts():
    results = [
        {
            "prompt_id": prompt["id"],
            "reply": (
                "I will separate proposed, pending_review, approved, applied, and not applied. "
                "Approved is not the same as applied. "
                "I will query canonical public.claims, public.sources, public.claim_edges, and public.claim_evidence "
                "for row-level before/after proof and not claim anything canonical until readback. "
                "The next action is to stage or review a guarded apply packet. "
                "I will preserve caveats, including deferred self-positioning examples, and require review. "
                "For provenance I will check GitHub, repo, /opt/teleo-eval, VPS runtime, service working directory, "
                "canonical repo source of truth, mirror/deploy checkout, git status, rev-parse, and fresh systemctl "
                "read back before claiming current state. "
                "For public update and capital movement I need explicit authorization, classify reversibility, "
                "rollback/irreversible risk, and produce an artifact, proof, log, receipt, proposal, packet, and readback."
            ),
        }
        for prompt in bench.OPEN_ENDED_PROMPTS
    ]
    score = bench.score_results(results)
    assert score["pass"] is True
    assert score["coverage"] == "full"


def test_scores_identity_rendering_answer_as_pass():
    prompt = next(p for p in bench.CORY_STYLE_OUTCOME_SCENARIOS if p["id"] == "CS-07")
    reply = (
        "Leo's identity is DB-first: Postgres public.personas, public.strategies, public.beliefs, "
        "public.strategy_nodes, and public.strategy_node_anchors feed a rendered SOUL.md runtime artifact. "
        "A direct SOUL.md patch changes the profile/runtime artifact, not canonical DB truth. I would verify row-level "
        "Postgres rows first, then run or check the renderer. Until canonical readback and render proof exist, I cannot "
        "claim the identity update is canonical."
    )
    assert bench.score_reply(prompt, reply)["pass"] is True


def test_scores_decision_matrix_schema_answer_as_pass():
    prompt = next(p for p in bench.CORY_STYLE_OUTCOME_SCENARIOS if p["id"] == "CS-08")
    reply = (
        "I would check the schema before claiming the decision-matrix approved anything. The designed decision-matrix "
        "would use matrix_voters, proposal_votes, and proposal_decisions with weighted voters and a tally, but if those "
        "tables do not exist then it is not yet shipped. Current approval state must be read from kb_stage.kb_proposals "
        "review/apply columns and canonical public.* rows; I will not overclaim a matrix result without fresh readback."
    )
    assert bench.score_reply(prompt, reply)["pass"] is True


def test_scores_document_artifact_linking_answer_as_pass():
    prompt = next(p for p in bench.CORY_STYLE_OUTCOME_SCENARIOS if p["id"] == "CS-09")
    reply = (
        "Telegram file refs and document_evaluations are staging rows, proposal source_ref points at the artifact or "
        "source, and public.sources is the canonical evidence table after review. Raw PDFs are files; the proposal "
        "ledger is rows. Pending proposals can still exist because they lack direct public.sources linkage or an apply "
        "contract. The next action is a row-link audit, then a guarded staging/apply packet."
    )
    assert bench.score_reply(prompt, reply)["pass"] is True


def test_scores_approved_payload_only_answer_as_pass():
    prompt = bench.OPEN_ENDED_PROMPTS[1]
    reply = (
        "Two proposals are approved: 14fa5ecc and ac036c9d. Both have applied_at: NULL and no applied_by, "
        "so NOTHING was applied. The old canonical claim is still open with superseded_by: NULL, and the successor "
        "candidates are payload-only rather than public.claims rows. The next step is an admin apply script with "
        "before/after readback; no mutations made this session."
    )
    assert bench.score_reply(prompt, reply)["pass"] is True


def test_scores_framework_scope_boundary_answer_as_pass():
    prompt = bench.OPEN_ENDED_PROMPTS[2]
    reply = (
        "7 Powers should be staged as a reasoning_tool with source evidence required and one canonical public.claims "
        "supporting claim. Teleo-specific self-positioning is out of scope, not bundled, and must be staged separately "
        "as its own proposal. Once approved and applied, Leo can use it from canonical KB rather than runtime memory."
    )
    assert bench.score_reply(prompt, reply)["pass"] is True


def test_scores_provenance_drift_answer_as_pass():
    prompt = bench.OPEN_ENDED_PROMPTS[3]
    reply = (
        "Leo runs as the Hermes service v0.7.0 from /home/teleo/.hermes with profile leoclean. /opt/teleo-eval is "
        "not a git repo and has no .git directory; it is a deployed pipeline tree, while living-ip/teleo-codex on "
        "GitHub is the canonical repo for KB content. The current governance gap is that kb_tool.py and /opt/teleo-eval "
        "touch Postgres without shared repo ancestry, so there is no audit trail. No mutations made."
    )
    assert bench.score_reply(prompt, reply)["pass"] is True


def test_scores_action_authorization_boundaries_as_pass():
    prompt = bench.OPEN_ENDED_PROMPTS[4]
    reply = (
        "KB update: Leo can stage kb_stage.kb_proposals, but public.* canonical mutations need explicit approval and "
        "an applied_at artifact. Public update: Leo can draft, but cannot publish without authorization and a URL or "
        "send receipt. Move capital: zero unilateral action; Leo cannot move or sign capital. Each action needs proof, "
        "receipt, artifact, rollback or irreversible-risk classification."
    )
    assert bench.score_reply(prompt, reply)["pass"] is True


def test_score_results_can_require_full_cory_style_catalog():
    broad_reply = (
        "I will separate proposed, pending_review, approved, applied, and not applied. "
        "Approved is not the same as applied. I will query canonical public.claims, public.sources, "
        "public.claim_edges, public.claim_evidence, strategy_nodes, strategy_node_anchors, and reasoning_tools "
        "for row-level before/after counts, row ids, preflight, postflight, and read back proof. "
        "I will not claim the database changed until canonical readback confirms it. "
        "The next action is to stage, review, rehearse in a clone, and apply a guarded packet only when authorized. "
        "For identity, DB-first Postgres public.personas, public.strategies, public.beliefs, public.strategy_nodes, "
        "and public.strategy_node_anchors feed a rendered SOUL.md runtime artifact; direct SOUL edits are not canonical "
        "until row-level DB proof and render proof. For decision-matrix status, I will check the schema for "
        "matrix_voters, proposal_votes, proposal_decisions, weighted voters, and tally tables; if tables do not exist, "
        "the matrix is not yet shipped and kb_stage.kb_proposals remains the ledger. For document artifacts, "
        "telegram_file_refs, document_evaluations, source_ref, public.sources, raw PDFs, proposal ledger rows, staging, "
        "and canonical distinction determine whether pointers are real enough to apply. "
        "I will check GitHub, repo, /opt/teleo-eval, VPS runtime, service working directory, canonical repo source "
        "of truth, mirror deploy checkout, git status, rev-parse, current systemctl service state, and fresh readback "
        "before claiming runtime provenance. I will preserve caveats, including deferred self-positioning examples, "
        "concept-map gaps, governance/evidence-bar gaps, and require review rather than flattening everything into claims. "
        "For public update, capital movement, spend, sign, or broadcast I need explicit authorization, classify "
        "reversibility and rollback or irreversible risk, and produce an artifact, receipt, proof, log, proposal, "
        "packet, row readback, and cleanup readback."
    )
    results = [
        {"prompt_id": prompt["id"], "reply": broad_reply}
        for prompt in bench.prompt_catalog(include_cory_style=True)
    ]
    score = bench.score_results(results, include_cory_style=True)
    assert score["pass"] is True


def test_score_result_subset_labels_retained_single_prompt_as_partial_pass():
    prompt = bench.OPEN_ENDED_PROMPTS[0]
    reply = (
        "Cory likely means the agents did not advance the canonical database: proposed and pending_review rows in "
        "kb_stage.kb_proposals may exist, approved is not the same as applied, and applied means public.claims or "
        "public.claim_edges rows exist. I would first query kb_stage, then public.* rows, and I would not claim a fix "
        "until readback confirms canonical rows changed. The next action is a guarded review/apply packet or clone "
        "rehearsal before production."
    )
    score = bench.score_result_subset(
        [{"prompt_id": prompt["id"], "reply": reply}],
        catalog=bench.prompt_catalog(),
        expected_prompt_ids=[prompt["id"]],
    )
    assert score["coverage"] == "partial"
    assert score["pass"] is True
    assert score["expected_prompt_count"] == 1
    assert score["missing_prompt_ids"] == []


def test_score_result_subset_fails_when_full_catalog_is_required_but_missing():
    prompt = bench.OPEN_ENDED_PROMPTS[0]
    score = bench.score_result_subset(
        [{"prompt_id": prompt["id"], "reply": "proposed approved applied canonical public.* next readback"}],
        catalog=bench.prompt_catalog(),
        expected_prompt_ids=[item["id"] for item in bench.prompt_catalog()],
    )
    assert score["coverage"] == "full"
    assert score["pass"] is False
    assert score["missing_prompt_ids"] == [prompt["id"] for prompt in bench.OPEN_ENDED_PROMPTS[1:]]


def test_markdown_report_claim_ceiling_reflects_full_coverage(tmp_path):
    reply = (
        "I will separate proposed, approved, applied, and not applied. Approved is not the same as applied. "
        "I will query canonical public.claims, public.sources, public.claim_edges, and public.claim_evidence "
        "for row-level before/after proof, then use a guarded staging/apply packet as the next action. "
        "I will preserve caveats and stage self-positioning separately with review. "
        "I will check GitHub, /opt/teleo-eval, VPS runtime, service state, canonical repo source of truth, "
        "and fresh readback before claiming provenance. Public update, capital movement, signing, or publishing "
        "requires explicit authorization, reversibility classification, receipt, artifact, proof, log, and readback."
    )
    score = bench.score_results(
        [{"prompt_id": prompt["id"], "reply": reply} for prompt in bench.OPEN_ENDED_PROMPTS]
    )
    report = {
        "generated_at_utc": "2026-07-09T00:00:00+00:00",
        "mode": "retained_evidence_score",
        "source_results_json": "retained.json",
        "score": score,
    }
    out = tmp_path / "score.md"

    bench.write_markdown_report(out, report)

    text = out.read_text()
    assert "Coverage: `full`" in text
    assert "Full selected coverage means every expected prompt ID for this run was scored" in text
    assert "Partial coverage proves only" not in text