yau-plant-assistant/api/stub.py

"""No-LLM stub mode. OFF by default, and it must stay that way.

WHY THIS EXISTS
---------------
Azure OpenAI access is pending, and so is `imh`. Without a model there is no
classifier and no prose, so `POST /ask` cannot return anything at all - which
means the entire chain either side of the model is unproven: the browser, Caddy,
Authelia, the API, entity resolution, Cube, retrieval, the contracts, the
banners, the error paths. That is a lot of untested surface to leave until the
day the key arrives, and it is all testable now.

So this module substitutes the two steps that need a model:

  * classification  -> the class is supplied by the caller (a dropdown in the
                       UI). Not guessed by keywords: a crude keyword classifier
                       would produce a *plausible* label, and a plausible wrong
                       label is the failure this system is built to avoid.
                       Choosing explicitly is honest, and it lets each branch be
                       driven deliberately.
  * prose           -> a fixed placeholder string per class. Deterministic, and
                       written to satisfy the contract for its class.

WHAT IT PROVES
--------------
The transport chain end to end. Entity resolution. Cube queries and their
numbers. Retrieval plumbing and citation assembly. The fixture banner. The
scope banners. Contract validation - including its failures, because a stub
payload goes through `enforce_contract()` exactly like a generated one and a
422 is a real result, not a bug in the stub.

WHAT IT DOES NOT PROVE
----------------------
Anything about the model. Whether the classifier would label the question
correctly - the whole point is that a human labelled it. Whether retrieval finds
the RIGHT chunk: stub mode retrieves lexically because embedding the question
needs the model, and lexical hits are not vector hits. And no prose quality,
because there is no prose.

It also cannot produce `prerequisites_verbatim` for a Procedural answer. Pulling
the prerequisites out of a retrieved chunk is an extraction task, and doing it
with a regex would be exactly the "synthesised from fragments" failure the
Procedural contract exists to prevent. So the list comes back empty and the
answer says so.

THE SAFETY RULES STILL RUN
--------------------------
`apply_safety_rules()` still runs over the forced classification. Contracts are
still validated after assembly and before returning. Retrieval still filters
superseded revisions. Nothing here is a bypass - it is the same pipeline with
the model-shaped holes filled by constants.

TURNING IT OFF AGAIN
--------------------
`NO_LLM_STUB=false` in ~/ai/api.env, and restart ai-api. Every answer produced
in this mode carries `stub_mode: true` and a banner in the same place, and for
the same reason, as the fixture banner: an answer nobody generated is otherwise
indistinguishable from one that was.
"""

from __future__ import annotations

import logging
from typing import Any

from classifier import Classification, apply_safety_rules
from config import settings
from contracts import QuestionClass

log = logging.getLogger("stub")

BANNER = (
    "NO-LLM STUB MODE. No language model was called. The class was chosen by "
    "hand, retrieval was lexical rather than semantic, and this text is a fixed "
    "placeholder - not a generated answer. The figures, citations and document "
    "identities below are real query results."
)


def enabled() -> bool:
    return settings().no_llm_stub


def classify(question: str, forced_class: str | None) -> Classification:
    """Take the class from the caller and put it through the real safety rules.

    Confidence is 1.0 because a person chose it, not because anything was
    inferred. The rules still run: a forced HISTORICAL with no time window
    still becomes UNCLEAR, which is worth seeing.
    """
    try:
        klass = QuestionClass(forced_class) if forced_class else QuestionClass.UNCLEAR
    except ValueError:
        log.warning("unknown forced class %r - falling back to unclear", forced_class)
        klass = QuestionClass.UNCLEAR

    raw = Classification(
        question_class=klass,
        confidence=1.0,
        alternatives={},
        entities={},
        missing_context=[] if forced_class else ["what you are asking about"],
    )
    result = apply_safety_rules(raw, settings().classifier_confidence_threshold)
    result.downgraded_reason = (
        (result.downgraded_reason + "; ") if result.downgraded_reason else ""
    ) + "class supplied by the caller - NO_LLM_STUB is on, nothing was classified"
    return result


# ---------------------------------------------------------------------------
# One payload builder per class. Each returns what the generate() step would
# have returned, minus the prose. The factual fields are attached from evidence
# by the caller in agent.py, exactly as they are for a generated answer.
# ---------------------------------------------------------------------------


def _historical(question: str, evidence: dict[str, Any]) -> dict[str, Any]:
    row_count = evidence.get("row_count", 0)
    window = evidence.get("time_window", {}).get("description", "the window queried")
    if row_count == 0:
        # Must say so, and must not state a quantity. The contract checks both.
        answer = (
            f"{BANNER} The query returned no records for {window}."
        )
    else:
        # Row counts and figures live in the structured fields, which the UI
        # renders. Keeping them out of the prose keeps this string honest -
        # it is not a summary of anything.
        answer = (
            f"{BANNER} The query ran and returned rows for {window}. "
            "The figures are in the rows and query fields, not in this text."
        )
    return {"answer": answer}


def _reference(question: str, evidence: dict[str, Any]) -> dict[str, Any]:
    if not evidence.get("citations"):
        answer = (
            f"{BANNER} No documents in the active set matched this question."
        )
    else:
        answer = (
            f"{BANNER} The documents below matched lexically. Whether they "
            "actually answer the question is not established - that is the "
            "judgement a language model would have made."
        )
    return {"answer": answer}


def _procedural(question: str, evidence: dict[str, Any]) -> dict[str, Any]:
    """Identity only. No prerequisites, and emphatically no steps.

    The procedure identity is assembled from the retrieved chunk's own
    metadata - document number, revision, effective date - which is database
    content, not generated text. That is the part of a Procedural answer that
    matters most, and it is the part that needs no model at all.
    """
    chunks = evidence.get("chunks") or []
    if not chunks:
        return {
            "answer": (
                f"{BANNER} No controlled procedure was found in the active "
                "document set for this question."
            ),
            "procedure": None,
            "prerequisites_verbatim": [],
        }

    top = chunks[0]
    procedure = {
        "doc_number": top.get("doc_number") or top.get("source_file"),
        "title": top.get("section_title") or top.get("source_file"),
        "revision": top.get("revision") or "unknown",
        "effective_date": top.get("effective_date"),
        "authorising_role": None,
        "controlled_copy_location": (
            "Controlled copy location is not held in the document index - "
            "obtain the controlled copy through document control."
        ),
    }
    answer = (
        f"{BANNER} A candidate controlled procedure was located and is "
        "identified below. Prerequisites were not extracted: quoting them "
        "verbatim is an extraction step this mode does not perform, and "
        "reconstructing them from the retrieved text is precisely what this "
        "class forbids. Work from the controlled copy."
    )
    return {"answer": answer, "procedure": procedure, "prerequisites_verbatim": []}


def _advisory(question: str, evidence: dict[str, Any]) -> dict[str, Any]:
    """Evidence rows come from Cube. The deferral is the whole point.

    Note what is NOT here: no number is selected, ranked or described as
    suitable. The stub could not do that even if asked - it has no way to form
    an opinion, which makes this the one class where a stub behaves almost
    exactly as the real thing should.
    """
    rows = evidence.get("rows") or []
    window = evidence.get("time_window", {}).get("description", "the window queried")

    def _interesting(row: dict[str, Any]) -> list[str]:
        # Ordering only. It decides which column is shown, never what it says -
        # the value is whatever Cube returned.
        keys = [
            k for k, v in row.items()
            if isinstance(v, (int, float)) and not isinstance(v, bool)
        ]
        preferred = [
            k for k in keys
            if any(w in k.lower() for w in ("rate", "discharge", "level", "count"))
        ]
        return preferred + [k for k in keys if k not in preferred]

    items: list[dict[str, Any]] = []
    for row in rows[:6]:
        for key in _interesting(row):
            value = row[key]
            items.append(
                {
                    "description": f"{key} over {window}",
                    "metric": key,
                    "value": float(value),
                    "unit": "",
                    "sample_size": len(rows),
                }
            )
            break

    if items:
        answer = (
            f"{BANNER} What the history shows over {window} is set out in the "
            "evidence below. No operating parameter has been selected from it."
        )
    else:
        answer = (
            f"{BANNER} There are no records of station operations for {window}."
        )
    return {
        "answer": answer,
        "evidence": items,
        "documented_limits": [],
        "deferral": (
            "This system does not recommend setpoints or operating parameters. "
            "That decision needs a competent person with sight of current "
            "equipment condition and concurrent operations. In this mode it "
            "could not make a recommendation even if it were permitted to - no "
            "model was called."
        ),
    }


BUILDERS = {
    QuestionClass.HISTORICAL: _historical,
    QuestionClass.REFERENCE: _reference,
    QuestionClass.PROCEDURAL: _procedural,
    QuestionClass.ADVISORY: _advisory,
}


def generated_fields(
    klass: QuestionClass, question: str, evidence: dict[str, Any]
) -> dict[str, Any]:
    """Stand in for the model's half of the payload. Deterministic."""
    builder = BUILDERS.get(klass)
    if builder is None:
        raise ValueError(f"no stub builder for {klass}")
    return builder(question, evidence)