teleo-infrastructure/hermes-agent/runtime/leoclean-nosend/profile-sources/operational_contracts.py

"""SQL-free response contracts for the bounded GCP no-send staging slice."""

from __future__ import annotations

import re
from typing import Any

SUPPORTED_DATABASE_MODES = frozenset({"claim_evidence_challenge", "review_only_candidate"})


def _reply_budget(query: str) -> dict[str, Any]:
    normalized = re.sub(r"[-_/]+", " ", query.lower())
    match = re.search(
        r"\b(?:under|within|at most|no more than|maximum(?: of)?|max(?:imum)?(?: of)?)\s+"
        r"(?P<words>\d{2,4})\s+words?\b",
        normalized,
    )
    hard_max = max(20, min(220, int(match.group("words")))) if match else 220
    return {
        "id": "reply_budget",
        "target_words": min(170, hard_max),
        "hard_max_words": hard_max,
        "shape": (
            "no intro; at most three compact bullets covering mapping, review/apply, and receipt/limitation; "
            "omission is better than exceeding the budget"
        ),
    }


def classify_contract_mode(query: str) -> str:
    lowered = query.lower()
    existing_claim_audit = any(
        term in lowered
        for term in (
            "inspect one exact claim",
            "exact claim body and its evidence",
            "one retrieved canonical claim",
            "linked source or evidence id",
        )
    )
    candidate = bool(
        any(
            term in lowered
            for term in (
                "candidate claim",
                "reviewable candidate",
                "candidate only",
                "revised exact claim body",
                "review-only proposal",
                "review only proposal",
            )
        )
        and any(term in lowered for term in ("draft", "propose", "revise", "rewrite", "narrow", "return", "turn"))
        and not existing_claim_audit
    )
    if candidate:
        return "review_only_candidate"
    challenge = bool(
        any(term in lowered for term in ("claim", "exact body", "claim body"))
        and any(term in lowered for term in ("evidence", "source"))
        and any(
            term in lowered
            for term in ("challenge", "unsupported", "narrower", "assumption", "what supports", "falsif")
        )
    )
    return "claim_evidence_challenge" if challenge else "general"


def operational_contracts(query: str) -> list[dict[str, Any]]:
    result = [_reply_budget(query)]
    if classify_contract_mode(query) == "claim_evidence_challenge":
        result.append(
            {
                "id": "claim_evidence_challenge",
                "required_sections": [
                    "one retrieved canonical claim ID and exact claim body",
                    "one linked source or evidence ID and exact evidence excerpt",
                    "one inference not established by that evidence",
                    "one narrower evidence-bounded revision",
                ],
                "boundary": (
                    "the claim body and evidence excerpt are distinct database values; do not paraphrase either "
                    "as if it were the other"
                ),
                "mutation_policy": "read-only; proposed wording remains a candidate until separately staged and reviewed",
            }
        )
    return result


def _bounded_exact_excerpt(value: Any, *, max_words: int = 45) -> str:
    text = str(value or "").strip()
    words = list(re.finditer(r"\S+", text))
    if len(words) <= max_words:
        return text
    return text[: words[max_words - 1].end()].rstrip()


def bind_claim_evidence_challenge(
    contracts: list[dict[str, Any]],
    claims: list[dict[str, Any]],
    evidence: dict[str, list[dict[str, Any]]],
) -> None:
    contract = next((item for item in contracts if item.get("id") == "claim_evidence_challenge"), None)
    if contract is None:
        return
    for claim in claims:
        claim_id = str(claim.get("id") or "")
        claim_text = str(claim.get("text") or "").strip()
        if not claim_id or not claim_text:
            continue
        for row in evidence.get(claim_id, []):
            source_id = str(row.get("source_id") or "")
            excerpt = _bounded_exact_excerpt(row.get("excerpt"))
            if not source_id or not excerpt:
                continue
            contract["binding_status"] = "bound"
            contract["selected_claim"] = {"id": claim_id, "text": claim_text}
            contract["selected_evidence"] = {
                "source_id": source_id,
                "excerpt": excerpt,
                "role": row.get("role"),
            }
            return
    contract["binding_status"] = "unavailable"


def compile_operational_response(contracts: list[dict[str, Any]]) -> str | None:
    challenge = next((item for item in contracts if item.get("id") == "claim_evidence_challenge"), None)
    if challenge is None:
        return None
    claim = challenge.get("selected_claim") or {}
    evidence = challenge.get("selected_evidence") or {}
    claim_id = str(claim.get("id") or "")
    claim_text = str(claim.get("text") or "")
    source_id = str(evidence.get("source_id") or "")
    excerpt = str(evidence.get("excerpt") or "")
    if challenge.get("binding_status") != "bound" or not all((claim_id, claim_text, source_id, excerpt)):
        return (
            "Live database readback returned no claim/source pair with a non-empty evidence excerpt, so an exact "
            "claim-versus-evidence challenge cannot be completed from this result. No identifiers or support were "
            "inferred, and no database write was made. Next proof-changing follow-up: broaden the read-only query "
            "until one canonical claim and linked source excerpt are returned, then rerun the challenge."
        )
    return (
        f"Claim ID: {claim_id}\n"
        f'Exact claim body: "{claim_text}"\n\n'
        f"Source ID: {source_id}\n"
        f'Exact evidence excerpt: "{excerpt}"\n\n'
        "Unsupported leap: this evidence excerpt does not establish every broader causal, trend, or generality "
        "assertion in the claim body.\n\n"
        "Narrower revision: the linked source establishes only the excerpted observation; any broader inference "
        "remains unproven by this evidence row alone. No database write was made."
    )