From b69dd53bc62fac7938ef487a61aaf63d72f0c4bb Mon Sep 17 00:00:00 2001 From: Arkash Jain Date: Wed, 10 Jun 2026 10:56:08 -0500 Subject: [PATCH] =?UTF-8?q?feat(feedback):=20close=20the=20miss=E2=86=92ga?= =?UTF-8?q?p=20loop,=20quarantine=20noise,=20admin=20correct=20path?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two independent model reviews converged on the same flywheel fixes: - record_gap_safe (fire-and-forget twin of record_feedback_safe); the HTTP /resolve miss and the MCP resolve_market / verify_market_claim misses now auto-record sourcing gaps (dedup bumps hit_count). Internal errors stay feedback — an exception is a bug report, a miss is a demand signal. - Feedback noise (test/asdf/short probes, canned-miss echoes) is quarantined as wontfix + severity=noise — inserted and auditable, never dropped. - strata admin correct : current-vs-proposed side-by-side, org-private apply by default, --shared through the existing fence (modeled/analyst tier refused with the honest fallback spelled out). No new fence built. - Admin whoami gains open_corrections; admin_list_feedback correction rows carry a correction_path (apply → promote → commit, or a submit template). 1228 backend tests green; pyright strict 0 errors on src; ruff clean. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 27 +++ backend/src/strata/api/routes/resolve.py | 28 ++- backend/src/strata/cli.py | 192 ++++++++++++++++- backend/src/strata/db/feedback.py | 43 +++- backend/src/strata/db/sourcing_gaps.py | 55 +++++ backend/src/strata/mcp/admin_tools.py | 66 +++++- backend/src/strata/mcp/tools.py | 72 ++++++- backend/src/strata/models/sourcing_gap.py | 7 +- backend/tests/test_admin_feedback.py | 239 ++++++++++++++++++++++ backend/tests/test_feedback.py | 150 +++++++++++++- backend/tests/test_sourcing_gaps.py | 108 +++++++++- 11 files changed, 954 insertions(+), 33 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 28cf595..c795cd0 100644 --- a/CHANGELOG.md +++ b/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 ` 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 / diff --git a/backend/src/strata/api/routes/resolve.py b/backend/src/strata/api/routes/resolve.py index de67ade..43c988c 100644 --- a/backend/src/strata/api/routes/resolve.py +++ b/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", {}) diff --git a/backend/src/strata/cli.py b/backend/src/strata/cli.py index 96db00a..5939da7 100644 --- a/backend/src/strata/cli.py +++ b/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 )") + + +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 " diff --git a/backend/src/strata/db/feedback.py b/backend/src/strata/db/feedback.py index e4d62a5..b32f712 100644 --- a/backend/src/strata/db/feedback.py +++ b/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: diff --git a/backend/src/strata/db/sourcing_gaps.py b/backend/src/strata/db/sourcing_gaps.py index 35746ae..01a6791 100644 --- a/backend/src/strata/db/sourcing_gaps.py +++ b/backend/src/strata/db/sourcing_gaps.py @@ -12,6 +12,7 @@ from __future__ import annotations +import logging import re from sqlalchemy import func, select @@ -24,6 +25,13 @@ _WS = re.compile(r"\s+") +_log = logging.getLogger(__name__) + +# The anonymous tenant sentinel (mirrors ``db/feedback.py`` / the audit writer). +# The closed RLS policy rejects an anon-org row by design, so an anonymous miss +# is dropped silently — never a 500 on the request that observed it. +_ANONYMOUS_ORG = "public" + def make_dedup_key(*, query_text: str | None, resource_target: str | None) -> str: """Collapse a miss onto a stable identity. A named ``resource_target`` (the @@ -87,6 +95,53 @@ async def record_gap( return SourcingGap.model_validate(row) +async def record_gap_safe( + *, + org_id: str | None, + source: GapSource, + query_text: str | None = None, + concept_slug: str | None = None, + role: str | None = None, + lens: str | None = None, + unit_hint: str | None = None, + resource_target: str | None = None, + note: str | None = None, +) -> int | None: + """Fire-and-forget gap write on its own org-scoped session (mirrors + ``record_feedback_safe``). Swallows (but logs) any failure so recording a + miss can never break the request that observed it. Returns the row id, or + ``None`` when the write was dropped (anonymous org has no tenant bucket + under the closed RLS policy). + + Resolve-vs-verify phrasings of the same market can land two rows — the + dedup key is the normalized query text and the two surfaces see different + strings. Acceptable: both still rank as demand for the same missing number. + """ + if not org_id or org_id == _ANONYMOUS_ORG: + return None + try: + from strata.db.engine import org_session_scope + + async with org_session_scope(org_id) as session: + gap = await record_gap( + session, + org_id=org_id, + source=source, + query_text=query_text, + concept_slug=concept_slug, + role=role, + lens=lens, + unit_hint=unit_hint, + resource_target=resource_target, + note=note, + ) + # org_session_scope commits on exit (session.begin()). + return gap.id + except Exception: + _log.warning("sourcing_gap.write_failed source=%s", source.value) + return None + + async def list_gaps( session: AsyncSession, *, diff --git a/backend/src/strata/mcp/admin_tools.py b/backend/src/strata/mcp/admin_tools.py index d49d5e8..1bcc9f8 100644 --- a/backend/src/strata/mcp/admin_tools.py +++ b/backend/src/strata/mcp/admin_tools.py @@ -51,6 +51,60 @@ def _require_admin() -> dict[str, Any] | None: return None +def _correction_path(row: dict[str, Any]) -> list[dict[str, Any]]: + """Next-actions to close a ``primitive_correction``. Honest: the steps differ + on whether the flag actually carries a promotable payload — without one, the + only real next action is a better submission, not a promote that will fail.""" + fid = row.get("id") + if row.get("suggested_value"): + return [ + { + "tool": "apply_feedback", + "arguments": {"feedback_id": fid}, + "description": "create the org-private corrected primitive (fenced, Q20)", + }, + { + "tool": "admin_promote_correction", + "arguments": {"feedback_id": fid}, + "description": "re-fence for the SHARED bank and return the committable YAML", + }, + { + "cli": f"strata admin promote {fid}", + "description": "write data/primitives/.yaml in a repo checkout", + }, + ] + return [ + { + "tool": "submit_feedback", + "description": ( + "this flag has no promotable correction; a promotable one needs a FULL " + "primitive payload in suggested_value (value + unit + as_of + geo + a " + "sourced, hash-pinned source — modeled/analyst tier cannot cross the " + "shared fence)" + ), + "arguments": { + "payload": { + "kind": "primitive_correction", + "target_type": "primitive", + "target_id": row.get("target_id") or "", + "body": "", + "suggested_value": { + "kind": "atomic_primitive", + "slug": "", + "value": "", + "unit": {"kind": "", "dimension": ""}, + "source": { + "tier": "", + "url": "", + "snapshot_sha256": "<64-hex content pin>", + }, + }, + } + }, + } + ] + + async def admin_list_feedback( reg: Registry, *, @@ -87,10 +141,18 @@ async def admin_list_feedback( org_id=org_id, limit=max(1, min(limit, 500)), ) + feedback: list[dict[str, Any]] = [] + for r in rows: + d = r.model_dump(mode="json") + # Corrections carry their closing path so the operator (or the agent + # driving the queue) never has to guess the apply → promote sequence. + if r.kind is FeedbackKind.PRIMITIVE_CORRECTION: + d["correction_path"] = _correction_path(d) + feedback.append(d) return { "admin": True, - "count": len(rows), - "feedback": [r.model_dump(mode="json") for r in rows], + "count": len(feedback), + "feedback": feedback, } diff --git a/backend/src/strata/mcp/tools.py b/backend/src/strata/mcp/tools.py index 3e92c9e..8dbf586 100644 --- a/backend/src/strata/mcp/tools.py +++ b/backend/src/strata/mcp/tools.py @@ -729,10 +729,56 @@ def _tam_result(reg: Registry, tam_dict: dict[str, Any]) -> CallToolResult: ) -def _h_whoami(reg: Registry, _args: dict[str, Any]) -> dict[str, Any]: +def _h_whoami(reg: Registry, _args: dict[str, Any]) -> dict[str, Any] | Awaitable[dict[str, Any]]: + from ..api.auth import is_admin + from .context import current_principal + + p = current_principal.get() + # Async only for an admin caller — the backlog count needs a DB round-trip; + # everyone else keeps the synchronous, DB-free diagnostic. + if p is not None and is_admin(p): + return _whoami_with_backlog(reg) return whoami(reg) +async def _whoami_with_backlog(reg: Registry) -> dict[str, Any]: + """Admin callers also see the open primitive-correction backlog, so the + correction loop is visible from the first call of a session.""" + from ..db.engine import admin_session_scope + from ..db.feedback import admin_count_open_corrections + from .context import current_org_id + + out = whoami(reg) + try: + async with admin_session_scope(current_org_id()) as session: + out["open_corrections"] = await admin_count_open_corrections(session) + except Exception: + # A DB outage must not break the connection diagnostic; unknown is honest. + out["open_corrections"] = None + return out + + +async def _record_miss_gap(out: dict[str, Any], *, source: str, query_text: str) -> dict[str, Any]: + """Auto-record a miss as ranked demand under the caller's org (best-effort: + the payload is returned unchanged when the write is dropped). The agent's + follow-up report_sourcing_gap call enriches the SAME row — the upsert + coalesces on the dedup key, so auto + manual never duplicate.""" + from ..db.sourcing_gaps import record_gap_safe + from ..models.sourcing_gap import GapSource + from .context import current_principal + + p = current_principal.get() + if p is None: + return out + gap_id = await record_gap_safe( + org_id=p.org_id, source=GapSource(source), query_text=query_text[:2048] + ) + if gap_id is not None: + out["gap_recorded"] = True + out["gap_id"] = gap_id + return out + + def _h_search(reg: Registry, args: dict[str, Any]) -> dict[str, Any]: return search_concepts( reg, @@ -764,14 +810,27 @@ def _h_size_tam(reg: Registry, args: dict[str, Any]) -> CallToolResult: ) -def _h_resolve_market(reg: Registry, args: dict[str, Any]) -> CallToolResult: +def _h_resolve_market( + reg: Registry, args: dict[str, Any] +) -> CallToolResult | Awaitable[dict[str, Any]]: + from .context import current_principal + out = resolve_market(reg, query=args["query"]) if out.get("matched"): return _tam_result(reg, out) - return tool_result(out) + # A miss with a tenant auto-records the gap (async); the next_action hint + # stays in the payload — the agent's explicit report enriches the same row. + # Over stdio (no principal) the path stays synchronous and write-free. + if current_principal.get() is None: + return tool_result(out) + return _record_miss_gap(out, source="resolve_miss", query_text=str(args["query"])) -def _h_verify_market_claim(reg: Registry, args: dict[str, Any]) -> CallToolResult: +def _h_verify_market_claim( + reg: Registry, args: dict[str, Any] +) -> CallToolResult | Awaitable[dict[str, Any]]: + from .context import current_principal + out = verify_market_claim( reg, claim=args["claim"], @@ -779,6 +838,11 @@ def _h_verify_market_claim(reg: Registry, args: dict[str, Any]) -> CallToolResul asserted_unit=args.get("asserted_unit"), as_of=args.get("as_of"), ) + # An unverifiable claim is the same demand signal as a resolve miss — record + # it when there is a tenant to attribute it to (a verify-side phrasing may + # land a second row next to a resolve-side one; see record_gap_safe). + if out.get("verdict") == "unverifiable" and current_principal.get() is not None: + return _record_miss_gap(out, source="unverifiable_claim", query_text=str(args["claim"])) links: list[dict[str, Any]] = [] next_actions: list[dict[str, Any]] = [] if out.get("strata_range") is not None and out.get("state_hash"): diff --git a/backend/src/strata/models/sourcing_gap.py b/backend/src/strata/models/sourcing_gap.py index 32f9049..0be3929 100644 --- a/backend/src/strata/models/sourcing_gap.py +++ b/backend/src/strata/models/sourcing_gap.py @@ -47,7 +47,12 @@ class SourcingGap(BaseModel): recorder collapses repeats onto, bumping ``hit_count`` instead of inserting a duplicate — so the queue counts demand. ``resource_target`` (e.g. ``bls:displacement-by-occupation``) names the connector/method that would fill - the gap, when known.""" + the gap, when known. + + A resolve-side and a verify-side phrasing of the same market can land two + rows (the key is the normalized query text and the two surfaces see + different strings) — acceptable: both still rank as demand for the same + missing number.""" model_config = ConfigDict(from_attributes=True) diff --git a/backend/tests/test_admin_feedback.py b/backend/tests/test_admin_feedback.py index 070d7b4..823d2f8 100644 --- a/backend/tests/test_admin_feedback.py +++ b/backend/tests/test_admin_feedback.py @@ -237,6 +237,245 @@ async def test_admin_list_feedback_sees_every_org(_clean: None) -> None: assert {r.org_id for r in a_rows} == {_TENANT_A} +@pytest.fixture(scope="module") +def reg() -> Any: + from pathlib import Path + + from strata.loader import load_data_dir + from strata.registry import Registry + + return Registry.from_loaded(load_data_dir(Path(__file__).resolve().parents[2] / "data")) + + +@_PG +@pytest.mark.asyncio +async def test_whoami_admin_sees_open_corrections( + reg: Any, _clean: None, monkeypatch: pytest.MonkeyPatch +) -> None: + """An admin caller's whoami carries the open primitive_correction backlog + count; a noise-quarantined or resolved flag does not inflate it.""" + import inspect + + from strata.db.engine import org_session_scope + from strata.mcp import context + from strata.mcp.tools import _h_whoami + + monkeypatch.setattr(settings, "admin_orgs", _ADMIN_ORG) + async with org_session_scope(_TENANT_A) as s: + await record_feedback( + s, + org_id=_TENANT_A, + user_id=None, + kind=FeedbackKind.PRIMITIVE_CORRECTION, + target_type=TargetType.PRIMITIVE, + body="headcount looks stale", + ) + + token = context.current_principal.set(Principal(org_id=_ADMIN_ORG, key_id="k", scopes=("*",))) + try: + res = _h_whoami(reg, {}) + assert inspect.isawaitable(res) + out = await res + finally: + context.current_principal.reset(token) + assert out["ok"] is True + assert out["open_corrections"] >= 1 + + +def test_whoami_non_admin_has_no_backlog(reg: Any, monkeypatch: pytest.MonkeyPatch) -> None: + """A non-admin (or stdio) caller keeps the synchronous, DB-free whoami — + the backlog count is an admin-only field.""" + from strata.mcp import context + from strata.mcp.tools import _h_whoami + + monkeypatch.setattr(settings, "admin_orgs", _ADMIN_ORG) + token = context.current_principal.set(Principal(org_id=_TENANT_A, key_id="k", scopes=("*",))) + try: + out = _h_whoami(reg, {}) + assert isinstance(out, dict) + assert "open_corrections" not in out + finally: + context.current_principal.reset(token) + # stdio: no principal at all → same sync path. + out = _h_whoami(reg, {}) + assert isinstance(out, dict) + assert "open_corrections" not in out + + +@_PG +@pytest.mark.asyncio +async def test_admin_list_feedback_correction_path( + _clean: None, monkeypatch: pytest.MonkeyPatch +) -> None: + """primitive_correction rows carry a correction_path: the apply→promote→CLI + chain when a promotable payload exists, a submit_feedback template when not. + Other kinds carry none.""" + from strata.db.engine import org_session_scope + from strata.mcp import context + from strata.mcp.admin_tools import admin_list_feedback as mcp_admin_list + + monkeypatch.setattr(settings, "admin_orgs", _ADMIN_ORG) + async with org_session_scope(_TENANT_A) as s: + with_payload = await record_feedback( + s, + org_id=_TENANT_A, + user_id=None, + kind=FeedbackKind.PRIMITIVE_CORRECTION, + target_type=TargetType.PRIMITIVE, + body="corrected headcount attached", + suggested_value=_sourced_primitive("adm-path-with-payload"), + ) + without_payload = await record_feedback( + s, + org_id=_TENANT_A, + user_id=None, + kind=FeedbackKind.PRIMITIVE_CORRECTION, + target_type=TargetType.PRIMITIVE, + target_id="some-flagged-slug", + body="this number looks wrong", + ) + commentary = await record_feedback( + s, + org_id=_TENANT_A, + user_id=None, + kind=FeedbackKind.BAD_ASSUMPTION, + target_type=TargetType.TAM, + body="adoption looks aggressive", + ) + + token = context.current_principal.set(Principal(org_id=_ADMIN_ORG, key_id="k", scopes=("*",))) + try: + out = await mcp_admin_list(None, org_id=_TENANT_A) # type: ignore[arg-type] + finally: + context.current_principal.reset(token) + by_id = {r["id"]: r for r in out["feedback"]} + + path = by_id[with_payload.id]["correction_path"] + assert [step.get("tool", step.get("cli")) for step in path] == [ + "apply_feedback", + "admin_promote_correction", + f"strata admin promote {with_payload.id}", + ] + assert path[0]["arguments"] == {"feedback_id": with_payload.id} + + template = by_id[without_payload.id]["correction_path"] + assert len(template) == 1 + assert template[0]["tool"] == "submit_feedback" + payload = template[0]["arguments"]["payload"] + assert payload["target_id"] == "some-flagged-slug" + assert "suggested_value" in payload # shows exactly what a promotable correction needs + + assert "correction_path" not in by_id[commentary.id] + + +# --------------------------------------------------------------------------- # +# `strata admin correct` — the operator's correction loop. # +# --------------------------------------------------------------------------- # + + +@_PG +@pytest.mark.asyncio +async def test_admin_correct_org_private_happy_path(_clean: None, tmp_path: Any) -> None: + """Default mode applies the correction ORG-PRIVATE via the existing apply + path: primitive created under the flag's own org, flag resolved + linked.""" + from strata.cli import _admin_correct_run + from strata.db.engine import org_session_scope + from strata.db.feedback import list_feedback + from strata.db.user_primitives import list_user_primitives + + async with org_session_scope(_TENANT_A) as s: + ev = await record_feedback( + s, + org_id=_TENANT_A, + user_id=None, + kind=FeedbackKind.PRIMITIVE_CORRECTION, + target_type=TargetType.PRIMITIVE, + body="corrected headcount attached", + suggested_value=_sourced_primitive("adm-correct-private"), + ) + fid = ev.id + + outcome = await _admin_correct_run( + fid, payload=None, shared=False, data_dir=tmp_path, assume_yes=True + ) + assert outcome["mode"] == "org-private" + assert outcome["org_id"] == _TENANT_A + assert outcome["slug"] == "adm-correct-private" + assert outcome["status"] == "resolved" + + async with org_session_scope(_TENANT_A) as s: + prims = await list_user_primitives(s, org_id=_TENANT_A) + assert any(p.slug == "adm-correct-private" and p.visibility == "private" for p in prims) + rows = await list_feedback(s, org_id=_TENANT_A) + assert rows[0].status is FeedbackStatus.RESOLVED + assert rows[0].resolved_primitive_id is not None + + +@_PG +@pytest.mark.asyncio +async def test_admin_correct_shared_happy_path(_clean: None, tmp_path: Any) -> None: + """--shared promotes through the existing fence and writes the committable + data/primitives/.yaml into the given data dir.""" + import yaml + + from strata.cli import _admin_correct_run + from strata.db.engine import org_session_scope + + async with org_session_scope(_TENANT_A) as s: + ev = await record_feedback( + s, + org_id=_TENANT_A, + user_id=None, + kind=FeedbackKind.PRIMITIVE_CORRECTION, + target_type=TargetType.PRIMITIVE, + body="corrected headcount attached", + suggested_value=_sourced_primitive("adm-correct-shared"), + ) + fid = ev.id + + outcome = await _admin_correct_run( + fid, payload=None, shared=True, data_dir=tmp_path, assume_yes=True + ) + assert outcome["mode"] == "shared" + written = tmp_path / "primitives" / "adm-correct-shared.yaml" + assert written.exists() + shared = yaml.safe_load(written.read_text(encoding="utf-8")) + assert shared["visibility"] == "public" + assert shared.get("org_id") is None + + +@_PG +@pytest.mark.asyncio +async def test_admin_correct_shared_refuses_modeled(_clean: None, tmp_path: Any) -> None: + """--shared on a modeled-tier correction is refused by the EXISTING shared + fence (Q20) — the flag stays open, the payload stays intact (never deleted), + and nothing lands in the shared data dir.""" + from strata.cli import _admin_correct_run + from strata.db.engine import org_session_scope + from strata.db.feedback import list_feedback + + async with org_session_scope(_TENANT_A) as s: + ev = await record_feedback( + s, + org_id=_TENANT_A, + user_id=None, + kind=FeedbackKind.PRIMITIVE_CORRECTION, + target_type=TargetType.PRIMITIVE, + body="proposed (modeled) correction", + suggested_value=_modeled_primitive("adm-correct-modeled"), + ) + fid = ev.id + + with pytest.raises(ModeledFenceError): + await _admin_correct_run(fid, payload=None, shared=True, data_dir=tmp_path, assume_yes=True) + + assert not (tmp_path / "primitives" / "adm-correct-modeled.yaml").exists() + async with org_session_scope(_TENANT_A) as s: + rows = await list_feedback(s, org_id=_TENANT_A) + assert rows[0].status is FeedbackStatus.OPEN + assert rows[0].suggested_value is not None # feedback is never deleted + + @_PG @pytest.mark.asyncio async def test_admin_promote_correction_returns_shared_yaml(_clean: None) -> None: diff --git a/backend/tests/test_feedback.py b/backend/tests/test_feedback.py index 8bae3aa..7fced38 100644 --- a/backend/tests/test_feedback.py +++ b/backend/tests/test_feedback.py @@ -107,6 +107,9 @@ async def _cleanup() -> None: has_prims = await conn.fetchval( "SELECT 1 FROM information_schema.tables WHERE table_name = 'user_primitives'" ) + has_gaps = await conn.fetchval( + "SELECT 1 FROM information_schema.tables WHERE table_name = 'sourcing_gaps'" + ) for org in (_ORG_A, _ORG_B): async with conn.transaction(): await conn.execute("SELECT set_config('app.org_id', $1, true)", org) @@ -114,6 +117,9 @@ async def _cleanup() -> None: if has_prims: # The apply tests create org-private primitives; clean them too. await conn.execute("DELETE FROM user_primitives WHERE org_id = $1", org) + if has_gaps: + # The resolve-miss tests now record sourcing gaps. + await conn.execute("DELETE FROM sourcing_gaps WHERE org_id = $1", org) finally: await conn.close() @@ -311,16 +317,19 @@ async def test_rest_bad_enum_is_422(client: AsyncClient, stub_clerk: None, _clea # --------------------------------------------------------------------------- # -# Failed-query auto-capture on /resolve. # +# Miss → sourcing gap (demand), exception → feedback (bug) on /resolve. # # --------------------------------------------------------------------------- # @_PG @pytest.mark.asyncio -async def test_failed_query_autocaptures_feedback( +async def test_resolve_miss_records_gap_not_feedback( client: AsyncClient, stub_clerk: None, _clean: None, monkeypatch: pytest.MonkeyPatch ) -> None: - """A resolve dead-end records a failed_query flag under the caller's org.""" + """A market-data miss is DEMAND for a missing number — it lands in the + de-duplicated sourcing-gap queue, never in the feedback triage queue.""" + from strata.models.sourcing_gap import GapSource + monkeypatch.setenv("STRATA_OFFLINE", "1") # A novel, unmatched query with no live/fallback path → terminal "no data". r = await client.post( @@ -333,12 +342,61 @@ async def test_failed_query_autocaptures_feedback( _ = r.text from strata.db.engine import org_session_scope + from strata.db.sourcing_gaps import list_gaps + + async with org_session_scope(_ORG_A) as session: + feedback_rows = await list_feedback(session, org_id=_ORG_A) + gap_rows = await list_gaps(session, org_id=_ORG_A) + assert feedback_rows == [], "a market-data miss must not create a feedback row" + assert len(gap_rows) == 1, "a market-data miss must record exactly one sourcing gap" + assert gap_rows[0].source is GapSource.RESOLVE_MISS + assert gap_rows[0].query_text == "wholly unrelated zylophone market 2099" + + # A repeat of the same miss bumps hit_count via the dedup key — no new row. + r2 = await client.post( + "/resolve", + headers=_BEARER_A, + json={"query": "Wholly Unrelated zylophone market 2099"}, + ) + assert r2.status_code == 200, r2.text + _ = r2.text + async with org_session_scope(_ORG_A) as session: + gap_rows = await list_gaps(session, org_id=_ORG_A) + assert len(gap_rows) == 1 + assert gap_rows[0].hit_count == 2 + + +@_PG +@pytest.mark.asyncio +async def test_internal_error_still_records_feedback( + client: AsyncClient, stub_clerk: None, _clean: None, monkeypatch: pytest.MonkeyPatch +) -> None: + """An exception in the pipeline is a BUG REPORT, not a demand signal — it + stays a failed_query feedback row (the gap queue is for missing numbers).""" + import strata.api.routes.resolve as resolve_mod + + def _boom(*_a: Any, **_k: Any) -> Any: + raise RuntimeError("kaboom") + + monkeypatch.setattr(resolve_mod, "run", _boom) + r = await client.post( + "/resolve", + headers=_BEARER_A, + json={"query": "wholly unrelated zylophone market 2099"}, + ) + assert r.status_code == 200, r.text + _ = r.text + + from strata.db.engine import org_session_scope + from strata.db.sourcing_gaps import list_gaps async with org_session_scope(_ORG_A) as session: rows = await list_feedback(session, org_id=_ORG_A) + gap_rows = await list_gaps(session, org_id=_ORG_A) assert any( r.kind is FeedbackKind.FAILED_QUERY and r.target_type is TargetType.QUERY for r in rows - ), "a failed resolve must auto-create a failed_query feedback row" + ), "an internal error must auto-create a failed_query feedback row" + assert gap_rows == [], "an internal error is not demand — no sourcing gap" # --------------------------------------------------------------------------- # @@ -624,3 +682,87 @@ def _novel_resolved_query() -> Any: from strata.resolver.slug import ResolvedQuery, ResolverIntent return ResolvedQuery(intent=ResolverIntent.NOVEL, confidence=0.0) + + +# --------------------------------------------------------------------------- # +# Noise quarantine at the feedback seam — wontfix + severity=noise, never drop. # +# --------------------------------------------------------------------------- # + + +@_PG +@pytest.mark.asyncio +async def test_noise_feedback_is_quarantined_not_deleted(_clean: None) -> None: + """Throwaway probes and the canned auto-miss body are INSERTED as + wontfix/severity=noise (auditable quarantine) — never dropped — while a real + flag stays open. The default open-queue view stays clean.""" + from strata.db.engine import org_session_scope + from strata.db.feedback import CANNED_MISS_BODY + + async with org_session_scope(_ORG_A) as session: + probe = await record_feedback( + session, + org_id=_ORG_A, + user_id=_USER_A, + kind=FeedbackKind.GENERAL, + target_type=TargetType.QUERY, + body="test", + ) + canned = await record_feedback( + session, + org_id=_ORG_A, + user_id=_USER_A, + kind=FeedbackKind.FAILED_QUERY, + target_type=TargetType.QUERY, + body=CANNED_MISS_BODY, + query_text="some genuinely missing market", + ) + short = await record_feedback( + session, + org_id=_ORG_A, + user_id=_USER_A, + kind=FeedbackKind.FAILED_QUERY, + target_type=TargetType.QUERY, + body="resolve came up empty", + query_text="ai", + ) + real = await record_feedback( + session, + org_id=_ORG_A, + user_id=_USER_A, + kind=FeedbackKind.BAD_ASSUMPTION, + target_type=TargetType.TAM, + body="adoption rate looks aggressive", + ) + + for noisy in (probe, canned, short): + assert noisy.status is FeedbackStatus.WONTFIX + assert noisy.severity == "noise" + assert real.status is FeedbackStatus.OPEN + assert real.severity != "noise" + + async with org_session_scope(_ORG_A) as session: + all_rows = await list_feedback(session, org_id=_ORG_A) + open_rows = await list_feedback(session, org_id=_ORG_A, status=FeedbackStatus.OPEN) + assert len(all_rows) == 4, "quarantined rows must persist — never auto-deleted" + assert {r.id for r in open_rows} == {real.id} + + +@_PG +@pytest.mark.asyncio +async def test_real_failed_query_is_not_noise(_clean: None) -> None: + """A failed_query flag carrying a real query stays open — only the probe + strings and the canned echo are quarantined.""" + from strata.db.engine import org_session_scope + + async with org_session_scope(_ORG_A) as session: + ev = await record_feedback( + session, + org_id=_ORG_A, + user_id=_USER_A, + kind=FeedbackKind.FAILED_QUERY, + target_type=TargetType.QUERY, + body="resolver gave a wrong framework for this", + query_text="US dental practice management software market", + ) + assert ev.status is FeedbackStatus.OPEN + assert ev.severity is None diff --git a/backend/tests/test_sourcing_gaps.py b/backend/tests/test_sourcing_gaps.py index 5690fe3..4fc4695 100644 --- a/backend/tests/test_sourcing_gaps.py +++ b/backend/tests/test_sourcing_gaps.py @@ -45,14 +45,19 @@ def test_dedup_key_prefers_resource_target() -> None: # --------------------------------------------------------------------------- # -def test_resolve_market_miss_is_flagged_unsourced() -> None: +@pytest.fixture(scope="module") +def reg() -> Any: from pathlib import Path from strata.loader import load_data_dir - from strata.mcp.tools import resolve_market from strata.registry import Registry - reg = Registry.from_loaded(load_data_dir(Path(__file__).resolve().parents[2] / "data")) + return Registry.from_loaded(load_data_dir(Path(__file__).resolve().parents[2] / "data")) + + +def test_resolve_market_miss_is_flagged_unsourced(reg: Any) -> None: + from strata.mcp.tools import resolve_market + out = resolve_market(reg, query="zzqx nonexistent market xyzzy qwerty") assert out["matched"] is False assert out["unsourced"] is True @@ -102,6 +107,36 @@ async def test_list_gaps_scope_all_is_admin_only(monkeypatch: pytest.MonkeyPatch context.current_principal.reset(token) +# --------------------------------------------------------------------------- # +# record_gap_safe — fire-and-forget gate (no DB needed for the drop paths). # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_record_gap_safe_drops_anonymous_and_missing_org() -> None: + from strata.db.sourcing_gaps import record_gap_safe + + assert await record_gap_safe(org_id=None, source=GapSource.RESOLVE_MISS, query_text="x") is None + assert ( + await record_gap_safe(org_id="public", source=GapSource.RESOLVE_MISS, query_text="x") + is None + ) + + +def test_mcp_resolve_miss_without_tenant_is_writeless(reg: Any) -> None: + """Over stdio (no principal) the miss path stays synchronous — no gap write + is even attempted, mirroring the no-tenant honesty of the write tools.""" + from mcp.types import CallToolResult + + from strata.mcp.tools import _h_resolve_market + + res = _h_resolve_market(reg, {"query": "zzqx nonexistent market xyzzy qwerty"}) + assert isinstance(res, CallToolResult) + assert res.structuredContent is not None + assert res.structuredContent["matched"] is False + assert "gap_recorded" not in res.structuredContent + + # --------------------------------------------------------------------------- # # DB-backed: de-dup, ranking, cross-org admin. # # --------------------------------------------------------------------------- # @@ -190,7 +225,7 @@ async def test_gaps_rank_by_demand_and_isolate(_clean: None) -> None: async with org_session_scope(_TENANT_A) as s: rows = await list_gaps(s, org_id=_TENANT_A) - assert [r.query_text for r in rows][0] == "hot market" # most-demanded first + assert next(r.query_text for r in rows) == "hot market" # most-demanded first assert all(r.org_id == _TENANT_A for r in rows) assert "B only" not in {r.query_text for r in rows} @@ -254,3 +289,68 @@ async def test_admin_sees_every_org_gap_and_can_mark_sourced(_clean: None) -> No async with org_session_scope(_TENANT_A) as s: open_rows = await list_gaps(s, org_id=_TENANT_A, status=GapStatus.OPEN) assert gid not in {r.id for r in open_rows} + + +# --------------------------------------------------------------------------- # +# MCP auto-record: a tenanted miss / unverifiable claim becomes ranked demand. # +# --------------------------------------------------------------------------- # + + +@_PG +@pytest.mark.asyncio +async def test_mcp_resolve_miss_autorecords_gap(reg: Any, _clean: None) -> None: + """A tenanted resolve_market miss records the gap automatically — and keeps + the report_sourcing_gap hint so the agent's follow-up enriches the SAME row.""" + import inspect + + from strata.db.engine import org_session_scope + from strata.db.sourcing_gaps import list_gaps + from strata.mcp import context + from strata.mcp.tools import _h_resolve_market + + token = context.current_principal.set(Principal(org_id=_TENANT_A, key_id="k", scopes=("*",))) + try: + res = _h_resolve_market(reg, {"query": "zzqx nonexistent market xyzzy qwerty"}) + assert inspect.isawaitable(res) + out = await res + finally: + context.current_principal.reset(token) + + assert out["matched"] is False and out["unsourced"] is True + assert out["next_action"]["tool"] == "report_sourcing_gap" + assert out["gap_recorded"] is True + + async with org_session_scope(_TENANT_A) as s: + rows = await list_gaps(s, org_id=_TENANT_A) + assert len(rows) == 1 + assert rows[0].source is GapSource.RESOLVE_MISS + assert rows[0].id == out["gap_id"] + + +@_PG +@pytest.mark.asyncio +async def test_mcp_unverifiable_claim_autorecords_gap(reg: Any, _clean: None) -> None: + """An unverifiable verify_market_claim is the same demand signal as a resolve + miss — recorded under its own source so the queue knows which surface saw it.""" + import inspect + + from strata.db.engine import org_session_scope + from strata.db.sourcing_gaps import list_gaps + from strata.mcp import context + from strata.mcp.tools import _h_verify_market_claim + + token = context.current_principal.set(Principal(org_id=_TENANT_A, key_id="k", scopes=("*",))) + try: + res = _h_verify_market_claim(reg, {"claim": "zzqx nonexistent market xyzzy qwerty"}) + assert inspect.isawaitable(res) + out = await res + finally: + context.current_principal.reset(token) + + assert out["verdict"] == "unverifiable" + assert out["gap_recorded"] is True + + async with org_session_scope(_TENANT_A) as s: + rows = await list_gaps(s, org_id=_TENANT_A) + assert len(rows) == 1 + assert rows[0].source is GapSource.UNVERIFIABLE_CLAIM