feat(feedback): close the miss→gap loop, quarantine noise, admin correct path

CHANGELOG.md

@@ -7,6 +7,33 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed — the miss→gap flywheel now runs itself (feedback loop fixes)
+
+- **A failed market query now automatically becomes ranked demand.** When the
+  resolver (HTTP `/resolve`), the MCP `resolve_market`, or `verify_market_claim`
+  comes up empty for a signed-in workspace, the miss is recorded in the
+  sourcing-gap queue on its own — repeats of the same question bump the gap's
+  hit count instead of piling up. Previously a miss either filed a bug-style
+  feedback row or relied on the agent volunteering a follow-up call. An actual
+  internal error still files feedback: an exception is a bug report, a miss is
+  a demand signal — and a gap is only ever a demand signal, never a number.
+- **Feedback noise is quarantined, never deleted.** Throwaway probes ("test",
+  "asdf", two-character queries) and flags that merely echo the canned no-data
+  message are stored as `wontfix` with severity `noise`, so the open triage
+  queue stays clean while every submission remains auditable.
+- **`strata admin correct <feedback_id>` closes the correction loop honestly.**
+  It shows the flagged primitive's current sourced value side-by-side with the
+  complaint, applies the corrected number org-private by default (via the
+  existing apply path), and `--shared` promotes through the existing
+  shared-bank fence — which still refuses modeled/analyst-tier numbers (Q20),
+  pointing the operator at the honest fallback (correct org-private + file a
+  sourcing gap for a real source).
+- **The backlog is visible.** Admin `whoami` reports the count of open
+  primitive corrections, and `admin_list_feedback` correction rows carry a
+  `correction_path` — the exact apply → promote → commit chain (or, when the
+  flag has no usable payload, a template showing what a promotable correction
+  needs).
+
 ### Changed — web app trimmed to landing + auth + OAuth consent (CLI/MCP direction lock)
 
 The web frontend now serves only the public landing page, the Clerk sign-in /

backend/src/strata/api/routes/resolve.py

@@ -30,8 +30,10 @@
 from strata.api.ratelimit import enforce_rate_limit
 from strata.api.responses import RATE_LIMITED, UNAUTHORIZED, merge
 from strata.db.audit import record_audit_safe
-from strata.db.feedback import record_feedback_safe
+from strata.db.feedback import CANNED_MISS_BODY, record_feedback_safe
+from strata.db.sourcing_gaps import record_gap_safe
 from strata.models.feedback import FeedbackKind, TargetType
+from strata.models.sourcing_gap import GapSource
 from strata.registry import Registry
 from strata.resolver import RunOptions, parse_query, run, search_primitive
 from strata.resolver.events import PipelineStatus, ResolverResult
@@ -198,24 +200,20 @@ async def generate() -> AsyncIterator[str]:
                 "Pick the nearest curated concept below, or retry once a backend "
                 "is configured."
                 if llm_unavailable
-                else "Could not find enough market data for this query."
+                else CANNED_MISS_BODY
             )
-            # Capture the dead-end ONLY when it's a real market-data gap — a query
-            # the bank couldn't answer (Phase F backlog). An LLM-backend-unavailable
-            # / fallback-parse failure is an infra/config problem, not a missing
-            # number; flooding triage with it would bury the actual data gaps, so
-            # skip the auto-flag for that branch (the user still gets the honest
-            # message + curated-concept fallback above).
+            # A real market-data miss is DEMAND for a missing number, not a bug
+            # report — it lands in the de-duplicated sourcing-gap queue (a repeat
+            # of the same query bumps hit_count via make_dedup_key normalization)
+            # instead of the feedback triage queue. The internal-error branch
+            # above stays feedback: an exception is a bug report. An LLM-backend-
+            # unavailable / fallback-parse failure is an infra/config problem,
+            # not a missing number, so it records neither.
             if not llm_unavailable:
-                await record_feedback_safe(
+                await record_gap_safe(
                     org_id=principal.org_id,
-                    user_id=principal.user_id,
-                    kind=FeedbackKind.FAILED_QUERY,
-                    target_type=TargetType.QUERY,
-                    body=message,
+                    source=GapSource.RESOLVE_MISS,
                     query_text=body.query[:2048],
-                    severity="warning",
-                    suggested_value={"reason": result.error} if result.error else None,
                 )
             yield _sse("error", {"message": message, "step": "search"})
             yield _sse("done", {})

backend/src/strata/cli.py

@@ -20,6 +20,7 @@
 from strata.evaluator import EvaluatorError, size_concept
 from strata.loader import LoaderError, load_data_dir
 from strata.models import QuestionCategory, ResolvedAnswer, TAMOutput, Unit
+from strata.models.feedback import FeedbackEvent
 from strata.registry import Registry, RegistryError
 from strata.report import (
     Branding,
@@ -1532,7 +1533,7 @@ def feedback_list(org: str = _FB_ORG_OPT, status: str | None = _FB_STATUS_OPT) -
 
     from strata.db.engine import org_session_scope
     from strata.db.feedback import list_feedback
-    from strata.models.feedback import FeedbackEvent, FeedbackStatus
+    from strata.models.feedback import FeedbackStatus
 
     status_filter = FeedbackStatus(status) if status else None
 
@@ -1759,6 +1760,22 @@ async def _run() -> bool:
     False, "--validate-only", help="Validate + print the shared YAML but don't write/resolve."
 )
 _ADMIN_ID_ARG = typer.Argument(..., help="Feedback row id to promote.")
+_ADMIN_CORRECT_ID_ARG = typer.Argument(..., help="Feedback row id to correct.")
+_ADMIN_CORRECT_PAYLOAD_OPT = typer.Option(
+    None,
+    "--payload",
+    help="YAML file with the corrected primitive (value + unit + source + tier). "
+    "Falls back to the flag's own suggested_value.",
+)
+_ADMIN_CORRECT_SHARED_OPT = typer.Option(
+    False,
+    "--shared",
+    help="Promote into the SHARED bank (runs the shared-bank fence; a modeled / "
+    "analyst-tier number is refused — Q20). Default is org-private.",
+)
+_ADMIN_CORRECT_YES_OPT = typer.Option(
+    False, "--yes", "-y", help="Apply without the interactive confirmation."
+)
 
 
 @admin_app.command("feedback")
@@ -1772,7 +1789,7 @@ def admin_feedback(
 
     from strata.db.engine import admin_session_scope
     from strata.db.feedback import admin_list_feedback
-    from strata.models.feedback import FeedbackEvent, FeedbackKind, FeedbackStatus
+    from strata.models.feedback import FeedbackKind, FeedbackStatus
 
     status_filter = FeedbackStatus(status) if status else None
     kind_filter = FeedbackKind(kind) if kind else None
@@ -1849,6 +1866,177 @@ async def _run() -> dict[str, Any]:
     typer.echo("  Validate + seed: `strata validate` then `strata db seed`.")
 
 
+def _echo_correction_context(flag: FeedbackEvent, data_dir: Path) -> None:
+    """Show the flagged primitive's CURRENT sourced value next to the complaint
+    so the operator decides with both numbers in view — never blind-applies."""
+    from strata.models.feedback import TargetType
+
+    tgt = f"{flag.target_type.value}:{flag.target_id}" if flag.target_id else flag.target_type.value
+    typer.echo(f"Flag #{flag.id} [{flag.status.value}] org={flag.org_id} kind={flag.kind.value}")
+    typer.echo(f"  Target:    {tgt}")
+    typer.echo(f"  Complaint: {flag.body}")
+    if flag.target_type is TargetType.PRIMITIVE and flag.target_id:
+        current = None
+        try:
+            reg = Registry.from_loaded(load_data_dir(data_dir))
+            current = reg.atomic.get(flag.target_id)
+        except Exception:
+            # No readable data dir here — say so rather than pretend.
+            typer.echo(f"  Current:   (could not load the registry from {data_dir})")
+        else:
+            if current is not None:
+                typer.echo(
+                    f"  Current:   {current.value} [{current.source.tier.value}] "
+                    f"{current.source.url}"
+                )
+            else:
+                typer.echo(f"  Current:   {flag.target_id!r} not in the shared registry")
+    if flag.suggested_value:
+        src_raw = flag.suggested_value.get("source")
+        src: dict[str, Any] = src_raw if isinstance(src_raw, dict) else {}
+        typer.echo(
+            f"  Proposed:  {flag.suggested_value.get('value')} [{src.get('tier')}] {src.get('url')}"
+        )
+    else:
+        typer.echo("  Proposed:  (none on the flag — pass --payload <yaml>)")
+
+
+async def _admin_correct_run(
+    feedback_id: int,
+    *,
+    payload: dict[str, Any] | None,
+    shared: bool,
+    data_dir: Path,
+    assume_yes: bool,
+) -> dict[str, Any]:
+    """The whole correct flow on ONE event loop (the async engine binds its pool
+    to the loop, so a second ``asyncio.run`` in the same process would break).
+    Reuses the existing apply / promote paths — no new fence is built here."""
+    import sys
+
+    import yaml
+
+    from strata.db.engine import admin_session_scope, org_session_scope
+    from strata.db.feedback import (
+        FeedbackNotApplicableError,
+        admin_get_feedback_row,
+        admin_promote_payload,
+        apply_feedback_correction,
+    )
+    from strata.models.feedback import FeedbackStatus
+
+    async with admin_session_scope(_ADMIN_ORG) as session:
+        row = await admin_get_feedback_row(session, feedback_id)
+        if row is None:
+            raise FeedbackNotApplicableError(f"no feedback {feedback_id}")
+        flag = FeedbackEvent.model_validate(row)
+
+    _echo_correction_context(flag, data_dir)
+
+    if not assume_yes and sys.stdin.isatty():
+        target = "the SHARED bank" if shared else f"org-private ({flag.org_id})"
+        typer.confirm(f"Apply this correction to {target}?", abort=True)
+
+    if not shared:
+        # Default: org-private via the existing apply path — the workspace gets
+        # the fix; any banner/fence the apply path enforces stays intact.
+        async with org_session_scope(flag.org_id) as session:
+            event, prim = await apply_feedback_correction(
+                session,
+                feedback_id,
+                org_id=flag.org_id,
+                created_by="cli-admin",
+                primitive_payload=payload,
+            )
+        return {
+            "mode": "org-private",
+            "org_id": flag.org_id,
+            "slug": prim.slug,
+            "primitive_id": prim.id,
+            "status": event.status.value,
+        }
+
+    # --shared: an explicit payload is applied org-private FIRST so the promote
+    # reads the fenced primitive the admin actually reviewed, not a stale one.
+    if payload is not None and flag.status is not FeedbackStatus.RESOLVED:
+        async with org_session_scope(flag.org_id) as session:
+            await apply_feedback_correction(
+                session,
+                feedback_id,
+                org_id=flag.org_id,
+                created_by="cli-admin",
+                primitive_payload=payload,
+            )
+    async with admin_session_scope(_ADMIN_ORG) as session:
+        shared_payload, from_org = await admin_promote_payload(session, feedback_id, resolve=True)
+    slug = str(shared_payload["slug"])
+    dest = data_dir / "primitives" / f"{slug}.yaml"
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    dest.write_text(yaml.safe_dump(shared_payload, sort_keys=False), encoding="utf-8")
+    return {"mode": "shared", "org_id": from_org, "slug": slug, "path": str(dest)}
+
+
+@admin_app.command("correct")
+def admin_correct(
+    feedback_id: int = _ADMIN_CORRECT_ID_ARG,
+    payload_file: Path | None = _ADMIN_CORRECT_PAYLOAD_OPT,
+    shared: bool = _ADMIN_CORRECT_SHARED_OPT,
+    data_dir: Path = _DATA_DIR_OPT,
+    yes: bool = _ADMIN_CORRECT_YES_OPT,
+) -> None:
+    """Close one correction flag honestly: show the flagged primitive's CURRENT
+    sourced value side-by-side with the complaint, then apply the corrected
+    number ORG-PRIVATE (default, via the existing apply path) or promote it
+    through the existing shared-bank fence (--shared). A modeled / analyst-tier
+    number is refused at the shared door (Q20) — the honest fallback is an
+    org-private correction plus a sourcing gap for a real source."""
+    import yaml
+
+    from strata.db.feedback import FeedbackNotApplicableError
+    from strata.guards.fence import ModeledFenceError
+
+    payload: dict[str, Any] | None = None
+    if payload_file is not None:
+        loaded: Any = yaml.safe_load(payload_file.read_text(encoding="utf-8"))
+        if not isinstance(loaded, dict):
+            typer.echo("ERROR: --payload must be a YAML mapping (one primitive).", err=True)
+            raise typer.Exit(code=1)
+        payload = {str(k): v for k, v in loaded.items()}  # pyright: ignore[reportUnknownVariableType]
+
+    try:
+        outcome = asyncio.run(
+            _admin_correct_run(
+                feedback_id, payload=payload, shared=shared, data_dir=data_dir, assume_yes=yes
+            )
+        )
+    except typer.Abort:
+        raise
+    except ModeledFenceError as exc:
+        typer.echo(f"REFUSED by the shared-bank fence (Q20): {exc}", err=True)
+        typer.echo(
+            "  Honest path: apply it org-private (drop --shared) so the workspace gets "
+            "the fix with the modeled banner intact, and file a sourcing gap "
+            "(`strata gaps record`) so a real source gets prioritized.",
+            err=True,
+        )
+        raise typer.Exit(code=1) from exc
+    except (FeedbackNotApplicableError, ValueError) as exc:
+        typer.echo(f"ERROR: {exc}", err=True)
+        raise typer.Exit(code=1) from exc
+    except Exception as exc:
+        typer.echo(f"ERROR correcting feedback: {exc}", err=True)
+        raise typer.Exit(code=1) from exc
+
+    if outcome["mode"] == "shared":
+        typer.echo(f"✓ Promoted flag {feedback_id} (from {outcome['org_id']}) → {outcome['path']}")
+        typer.echo("  Validate + seed: `strata validate` then `strata db seed`.")
+    else:
+        typer.echo(
+            f"✓ Corrected org-private for {outcome['org_id']}: primitive "
+            f"{outcome['slug']!r} (id={outcome['primitive_id']}); flag {outcome['status']}."
+        )
+
+
 traces_app = typer.Typer(
     name="traces",
     help="MCP session traces: list captured sessions, inspect ordered calls, export "

backend/src/strata/db/feedback.py

@@ -16,7 +16,7 @@
 import logging
 from typing import Any
 
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -51,6 +51,27 @@ class FeedbackNotApplicableError(ValueError):
 # an anonymous flag is attempted best-effort and silently dropped — never a 500.
 _ANONYMOUS_ORG = "public"
 
+# The canned message the resolve surfaces emit on a market-data miss. The miss
+# itself now lands in the sourcing-gap queue, so a flag whose body merely echoes
+# this message carries no triage signal beyond the gap row that already exists.
+CANNED_MISS_BODY = "Could not find enough market data for this query."
+
+# Throwaway probe strings people type to poke the form/tool — zero triage signal.
+_NOISE_TOKENS = frozenset({"test", "testing", "asdf", "foo"})
+
+
+def _is_noise(query_text: str | None, body: str, kind: FeedbackKind) -> bool:
+    """Quarantine predicate. A noise flag is still INSERTED — just born
+    ``wontfix`` with severity ``noise`` so the open queue stays clean while the
+    row remains auditable (never dropped, never deleted)."""
+    if body.strip() == CANNED_MISS_BODY:
+        return True
+    # A failed-query flag is judged by what was asked; everything else by what
+    # was said about it.
+    probe = query_text if (kind is FeedbackKind.FAILED_QUERY and query_text) else body
+    s = probe.strip().lower()
+    return s in _NOISE_TOKENS or len(s) <= 2
+
 
 async def record_feedback(
     session: AsyncSession,
@@ -71,6 +92,11 @@ async def record_feedback(
     The session-match guard refuses a spoofed/misrouted ``org_id`` even under a
     BYPASSRLS (superuser) role, where the RLS ``WITH CHECK`` would not catch it."""
     await assert_org_matches_session(session, org_id)
+    # Only a default-status flag is quarantined — an explicit triage decision
+    # (e.g. a re-filed row already marked triaged) is never second-guessed.
+    if status is FeedbackStatus.OPEN and _is_noise(query_text, body, kind):
+        status = FeedbackStatus.WONTFIX
+        severity = "noise"
     row = FeedbackEventRow(
         org_id=org_id,
         user_id=user_id,
@@ -218,6 +244,21 @@ async def admin_list_feedback(
     return [FeedbackEvent.model_validate(r) for r in rows]
 
 
+async def admin_count_open_corrections(session: AsyncSession) -> int:
+    """Cross-org count of OPEN ``primitive_correction`` flags (admin session) —
+    the number ``whoami`` surfaces so the correction backlog is visible from the
+    very first call an admin makes."""
+    stmt = (
+        select(func.count())
+        .select_from(FeedbackEventRow)
+        .where(
+            FeedbackEventRow.kind == FeedbackKind.PRIMITIVE_CORRECTION.value,
+            FeedbackEventRow.status == FeedbackStatus.OPEN.value,
+        )
+    )
+    return int((await session.execute(stmt)).scalar_one())
+
+
 async def admin_get_feedback_row(
     session: AsyncSession, feedback_id: int
 ) -> FeedbackEventRow | None: