FabledScribe/src/scribe/services/reply_preferences.py

"""The operator's own preferences for a completion report, at the moment it is written.

WHY THIS EXISTS (milestone 409 step 4)

The reporting-back skill ships DEFAULT shapes. An operator will want some of
them different ("my completion reports also say how it was tested", "decisions
as a numbered list"), and those adjustments are `preference` records. The gap
is the query: prompt-time retrieval matches the OPERATOR'S MESSAGE, and a shape
preference is about the REPLY. "Fix the flaky test" never retrieves "completion
reports should say how it was tested", so the preference is on file and never
arrives.

THE DECISION (operator, 2026-09-14, logged on the step): two deliveries, split
by reply kind.

  - A COMPLETION REPORT has a moment the server can see — a task closing — so
    the server retrieves for it and hands the matches back beside
    `report_back`. That is this module.
  - EVERY OTHER REPLY KIND (a finding, a decision, a handoff…) has no tool call
    in front of it, so the reporting-back skill asks: it tells the agent to
    `search(content_type="rule")` for that kind before writing.

Loading reply-shape preferences at session start was the rejected third option:
it is a small copy of the preloading milestone 394 retired, and just as
unmeasurable.

HOW A PREFERENCE SAYS IT IS ABOUT COMPLETION REPORTS

By its trigger, which is what it already has — no tag, no new column. A
preference's `when_to_apply` dominates its embedded document, so one written
for this moment ("writing the report after finishing a task") resembles
COMPLETION_QUERY below, and one about anything else does not. That keeps
delivery entirely in retrieval, as 394 decided, and leaves an operator nothing
new to learn: a preference reaches the completion report the same way every
other record reaches its moment.

THE BAR IS THE PROMPT ARM'S, AND IT IS NOT YET EARNED HERE

A fixed query against triggers is a different score distribution from an
operator's message against the same documents. Starting at the prompt arm's
setting is the value with evidence behind it, and an operator's tuning of that
bar reaches this too. Every call logs under its own source,
`report_preference`, so step 6 can read this surface's near misses apart from
the prompt arm's before anyone moves the number.
"""
from __future__ import annotations

import logging
import time

from scribe.services.embeddings import semantic_search_rules
from scribe.services.plugin_context import (
    PROMPTRULE_DEFAULT_THRESHOLD,
    PROMPTRULE_THRESHOLD_KEY,
)
from scribe.services.retrieval_telemetry import record_retrieval
from scribe.services.rule_usage import record_rule_surfaced
from scribe.services.settings import get_setting

logger = logging.getLogger(__name__)

SOURCE = "report_preference"

# Written in the vocabulary of the MOMENT, because that is what a trigger is
# written in and what this query is scored against. Domain-neutral on purpose
# (rule #115): a writing project or a home-infrastructure project closes tasks
# too, and its operator's preferences must match as well as a developer's.
COMPLETION_QUERY = (
    "writing the completion report to the operator after finishing a task — "
    "how that reply should be laid out and what it should include"
)

# A handful, not a menu. More than a few shape preferences for ONE kind of
# reply would contradict each other before they helped; the limit is here to
# keep one noisy corpus from turning a status change into a wall of text.
LIMIT = 3


async def _threshold(user_id: int) -> float:
    try:
        value = float(await get_setting(
            user_id, PROMPTRULE_THRESHOLD_KEY, str(PROMPTRULE_DEFAULT_THRESHOLD)))
    except (TypeError, ValueError):
        value = PROMPTRULE_DEFAULT_THRESHOLD
    return min(1.0, max(0.0, value))


async def completion_preferences(user_id: int, *, project_id: int | None = None) -> list[dict]:
    """The operator's preferences for a completion report, best match first.

    KIND-FILTERED, for the reason `_reserve_slot_for_preference` gives: a
    binding rule that happened to resemble the query would otherwise ride out
    under a key that says "how the operator likes this written", which is a
    claim about force the record does not make.

    Every call is logged, the empty ones included: a surface that records only
    the calls it liked reports a flawless clear-rate however badly its bar is
    set. An install with no preferences at all logs nothing, because no search
    ran (`searched` stays False) — that is not a decline.

    Fails open to an empty list: this decorates a write that has already
    happened, and a lookup that errors must not turn it into a failure.
    """
    try:
        threshold = await _threshold(user_id)
        report: dict = {}
        t0 = time.perf_counter()
        hits = await semantic_search_rules(
            user_id, COMPLETION_QUERY, limit=LIMIT, threshold=threshold,
            kind="preference", report=report, project_id=project_id,
        )
        hits = [(score, rule) for score, rule in hits if rule.kind == "preference"]
        record_retrieval(
            user_id=user_id, source=SOURCE, query=COMPLETION_QUERY,
            threshold=threshold, limit=LIMIT, project_id=project_id,
            is_task=None, results=hits,
            best_available=report.get("best_available_score"),
            best_available_id=report.get("best_available_id"),
            searched=bool(report.get("searched", True)),
            duration_ms=(time.perf_counter() - t0) * 1000.0,
        )
        if not hits:
            return []
        # RANKED: this surface chose what it showed, so the name is in
        # rule_usage.RANKED_SOURCES and its hits count toward pull-through.
        record_rule_surfaced(
            user_id=user_id, rule_ids=[rule.id for _s, rule in hits], source=SOURCE,
        )
        return [
            {"id": rule.id, "title": rule.title, "statement": rule.statement, "kind": "preference"}
            for _score, rule in hits
        ]
    except Exception:  # noqa: BLE001 - a decoration never breaks its payload
        logger.warning("completion preference lookup failed", exc_info=True)
        return []