teleo-infrastructure/tests/test_leo_turn_execution_manifest.py

from __future__ import annotations

import copy

from scripts import leo_turn_execution_manifest as manifest


def behavior_manifest() -> dict:
    return {
        "behavior_sha256": "a" * 64,
        "hermes_runtime": {"git_head": "b" * 40, "source_tree": {"sha256": "c" * 64}},
        "teleo_infrastructure_runtime": {
            "git_head": "d" * 40,
            "source_tree": {"sha256": "e" * 64},
        },
        "components": {
            "runtime_identity": {"content": {"sha256": "f" * 64}},
            "procedural_skills": {"content": {"sha256": "1" * 64}},
        },
    }


def retrieval_receipt() -> dict:
    return {
        "schema": "livingip.teleoKbRetrievalReceipt.v1",
        "query_sha256": "2" * 64,
        "semantic_context_sha256": "3" * 64,
        "artifact_state_sha256": "4" * 64,
        "claim_ids": ["claim-b", "claim-a"],
        "source_ids": ["source-a"],
        "counts": {"claims": 2, "context_rows": 1},
        "read_consistency": {
            "status": "stable_wal_marker",
            "attempts": 1,
            "database": "teleo",
            "database_user": "runtime",
            "system_identifier": "system-1",
            "wal_lsn_before": "0/123",
            "wal_lsn_after": "0/123",
        },
    }


def turn_result() -> dict:
    return {
        "turn": 1,
        "prompt_id": "OOS-01",
        "prompt": "Challenge the weak acquisition claim.",
        "reply": "The current evidence is shallow; here is a narrower candidate claim.",
        "started_at_utc": "2026-07-14T12:00:00+00:00",
        "ended_at_utc": "2026-07-14T12:00:02+00:00",
        "database_context_trace": [
            {
                "event": "pre_llm_call",
                "contract_ids": ["reply_budget", "source_link_audit"],
                "retrieval_receipt": retrieval_receipt(),
            }
        ],
        "database_tool_trace": {"database_tool_call_count": 0, "calls": []},
        "model_call_trace": [
            {
                "event": "post_api_request",
                "api_call_count": 1,
                "api_mode": "openai_chat_completions",
                "model": "anthropic/claude-sonnet-4-6",
                "provider": "openrouter",
                "response_model": "anthropic/claude-sonnet-4-6",
                "finish_reason": "stop",
                "message_count": 4,
                "assistant_content_chars": 120,
                "assistant_tool_call_count": 0,
                "task_id_sha256": "5" * 64,
                "session_id_sha256": "6" * 64,
                "base_url_sha256": "7" * 64,
                "usage": {"input_tokens": 100, "output_tokens": 20, "ignored": "secret-shaped"},
                "provider_response_id": None,
                "provider_response_id_status": "not_exposed_by_hermes_post_api_request_hook",
            }
        ],
    }


def database_fingerprint() -> dict:
    marker = {
        "database": "teleo",
        "database_user": "runtime",
        "system_identifier": "system-1",
        "wal_lsn": "0/123",
    }
    return {
        "schema": "livingip.teleoCanonicalDatabaseFingerprint.v1",
        "status": "ok",
        "fingerprint_sha256": "9" * 64,
        "table_count": 39,
        "total_rows": 52167,
        "table_rows_sha256": "a" * 64,
        "structure_sha256": "b" * 64,
        "read_consistency": {"status": "stable_wal_marker", "before": marker, "after": marker},
    }


def build(result: dict | None = None, *, fingerprint: dict | None = None) -> dict:
    selected_fingerprint = database_fingerprint() if fingerprint is None else fingerprint
    return manifest.build_turn_manifest(
        result=result or turn_result(),
        behavior_manifest=behavior_manifest(),
        session_key="agent:main:telegram:group:private",
        source={"platform": "telegram", "chat_type": "group"},
        suite_mode="live_vps_gatewayrunner_temp_profile",
        remote_run_id="run-1",
        db_counts_before={"public.claims": 1837},
        db_counts_after={"public.claims": 1837},
        db_counts_changed=False,
        db_fingerprint_before=selected_fingerprint,
        db_fingerprint_after=selected_fingerprint,
        posted_to_telegram=False,
        mutates_kb_by_harness=False,
        harness_source={"git_head": "8" * 40, "tracked_tree_clean": True},
    )


def test_turn_manifest_binds_model_runtime_session_and_exact_database_read() -> None:
    value = build()

    assert value["attribution"]["status"] == "complete"
    assert value["replay_claim"]["status"] == "bounded_recorded_inputs_pinned"
    assert value["model_execution"]["calls"][0]["response_model"] == "anthropic/claude-sonnet-4-6"
    assert value["model_execution"]["calls"][0]["usage"] == {"input_tokens": 100, "output_tokens": 20}
    assert value["canonical_database"]["binding_status"] == "retrieval_receipt_bound"
    assert value["canonical_database"]["fingerprint_before"]["table_count"] == 39
    assert value["canonical_database"]["fingerprint_unchanged"] is True
    receipt = value["canonical_database"]["context_retrieval_receipts"][0]
    assert receipt["read_consistency"]["system_identifier"] == "system-1"
    assert receipt["read_consistency"]["wal_lsn_before"] == receipt["read_consistency"]["wal_lsn_after"]
    assert value["delivery_and_safety"]["posted_to_telegram"] is False
    assert value["delivery_and_safety"]["kb_mutation_by_harness"] is False
    assert manifest.validate_turn_manifest(value) == []

    encoded = str(value)
    assert "Challenge the weak acquisition claim" not in encoded
    assert "narrower candidate claim" not in encoded
    assert "agent:main:telegram" not in encoded
    assert "secret-shaped" not in encoded


def test_turn_manifest_detects_missing_actual_model_call() -> None:
    result = turn_result()
    result["model_call_trace"] = []

    value = build(result)

    assert value["attribution"]["status"] == "incomplete"
    assert "actual_model_call" in value["attribution"]["missing_required_bindings"]
    assert value["replay_claim"]["local_runtime_inputs_reconstructible"] is False


def test_turn_manifest_hash_detects_tampering() -> None:
    value = build()
    tampered = copy.deepcopy(value)
    tampered["turn"]["reply_sha256"] = "0" * 64

    assert manifest.validate_turn_manifest(tampered) == ["execution_sha256_mismatch"]


def test_database_question_requires_a_stable_full_database_fingerprint() -> None:
    value = build(fingerprint={"status": "error"})

    assert value["attribution"]["status"] == "incomplete"
    assert "database_fingerprint_before" in value["attribution"]["missing_required_bindings"]
    assert "database_fingerprint_after" in value["attribution"]["missing_required_bindings"]
    assert value["canonical_database"]["fingerprint_unchanged"] is False