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


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_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