1283 lines
45 KiB
Python
1283 lines
45 KiB
Python
"""FastAPI app exposing DidiBrain's HTTP API contract.
|
|
|
|
Five routes, matching the existing web-gathering module 1:1 in request/response
|
|
shape so Didi's backend can swap brain <-> web transparently:
|
|
|
|
POST /v1/search — flat list of doc-level search results
|
|
POST /v1/fetch — look up atoms by URL, return extracted text
|
|
POST /v1/gather — full claim → ranked evidence pipeline
|
|
POST /v1/image-search — stub, always empty (we have no image corpus)
|
|
POST /v1/ingest — populate brain from web-module output (async extraction)
|
|
|
|
Plus GET /health for liveness probes.
|
|
|
|
No auth in v1 (bind to 127.0.0.1 only). Auth to be added when dockerized.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
import uuid
|
|
from contextlib import asynccontextmanager
|
|
from datetime import datetime
|
|
from typing import AsyncIterator
|
|
|
|
from dataclasses import asdict
|
|
|
|
from fastapi import BackgroundTasks, FastAPI, HTTPException, Query
|
|
|
|
from brain_api import deps
|
|
from brain_api.db import db as brain_db
|
|
from brain_api.schemas import (
|
|
AnalysisAtomData,
|
|
AnalysisAtomDetailResponse,
|
|
AnalysisAtomListItem,
|
|
AnalysisAtomListResponse,
|
|
AnalysisAtomLookupRequest,
|
|
AnalysisAtomLookupResponse,
|
|
AnalysisAtomPatchRequest,
|
|
AnalysisAtomStatsExtendedResponse,
|
|
AnalysisAtomStatsResponse,
|
|
AnalysisAtomWriteRequest,
|
|
AnalysisAtomWriteResponse,
|
|
AuditLogItem,
|
|
AuditLogResponse,
|
|
BrainMeta,
|
|
CacheInvalidateRequest,
|
|
CacheInvalidateResponse,
|
|
CanonicalizeRequest,
|
|
CanonicalizeResponse,
|
|
FactStatusItem,
|
|
FactStatusListResponse,
|
|
FactStatusPatchRequest,
|
|
FactStatusVersionItem,
|
|
FactStatusVersionsResponse,
|
|
FetchRequest,
|
|
FetchResponse,
|
|
GatherRequest,
|
|
GatherResponse,
|
|
GenericOkResponse,
|
|
ImageSearchRequest,
|
|
ImageSearchResponse,
|
|
IngestRequest,
|
|
IngestResponse,
|
|
SearchRequest,
|
|
SearchResponse,
|
|
TaxonomyInfoResponse,
|
|
TaxonomyReloadResponse,
|
|
VerificationCacheDetailResponse,
|
|
VerificationCacheListItem,
|
|
VerificationCacheListResponse,
|
|
VerificationCacheWriteRequest,
|
|
VerificationCacheWriteResponse,
|
|
)
|
|
from brain_api.services import admin as admin_svc
|
|
from brain_api.services import verification_cache as vcache
|
|
from brain_api.services import analysis_atom as atom_svc
|
|
from brain_api.services import fact_status as fact_svc
|
|
from brain_api.services.canonicalizer import canonicalize_claim_temporal
|
|
from brain_api.services.invalidation import (
|
|
InvalidateFilter,
|
|
invalidate_caches,
|
|
)
|
|
from brain_api.services.fetch import fetch as svc_fetch
|
|
from brain_api.services.gather import gather as svc_gather
|
|
from brain_api.services.ingest import ingest as svc_ingest
|
|
from brain_api.services.search import search as svc_search
|
|
from shared.atomic_api import AtomicClient
|
|
from shared.config import settings
|
|
from shared.embedding_client import EmbeddingClient
|
|
from shared.llm_client import LlmClient
|
|
from shared.logging import get_logger, setup_logging
|
|
from shared.taxonomy import TagResolver, build_path_map_from_tags
|
|
|
|
log = get_logger(__name__)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
|
|
setup_logging()
|
|
resolver = TagResolver()
|
|
|
|
atomic = AtomicClient()
|
|
embed = EmbeddingClient()
|
|
llm = LlmClient()
|
|
|
|
# Runtime config polling — pulls live overrides for atom_* + log_level
|
|
# from the AI platform dashboard. Pattern matches embeddings/rerank/catalog.
|
|
# Disabled when DASHBOARD_URL is not set (e.g. local dev runs).
|
|
from brain_api.runtime_config import RuntimeConfigClient, init_global_client
|
|
runtime_config = RuntimeConfigClient(
|
|
dashboard_url=settings.dashboard_url,
|
|
live_log_logger_name="brain_api",
|
|
live_log_key="brain.log.level",
|
|
)
|
|
init_global_client(runtime_config)
|
|
await runtime_config.start()
|
|
|
|
# Event sink — POSTs per-request events to dashboard /api/ingest/event so
|
|
# brain calls show up in the unified Insights/History view (module=brain).
|
|
# Same DASHBOARD_URL as runtime_config; disabled when unset.
|
|
from brain_api.events.sink import DashboardEventSink, init_global_sink
|
|
event_sink = DashboardEventSink(dashboard_url=settings.dashboard_url)
|
|
init_global_sink(event_sink)
|
|
await event_sink.start()
|
|
|
|
# Connect direct PG pool for brain-owned caches (verification, etc.). Same
|
|
# Postgres instance atomic-server uses — our tables are prefixed brain_ and
|
|
# live alongside atomic's own (no conflict). Failure here is soft: cache
|
|
# features degrade to miss but the rest of the service still starts.
|
|
try:
|
|
await brain_db.connect()
|
|
except Exception as e: # noqa: BLE001
|
|
log.error(
|
|
"brain_db_connect_failed",
|
|
error=f"{type(e).__name__}: {e}",
|
|
hint="verification_cache endpoints will return 503",
|
|
)
|
|
|
|
# Pull the current taxonomy from Atomic at startup so the container is
|
|
# self-sufficient and does not depend on a pre-generated _tag_ids.json.
|
|
# If Atomic is unreachable we fall back to whatever the resolver loaded
|
|
# from disk (may be empty); the service still starts so /health works
|
|
# for orchestration liveness probes.
|
|
try:
|
|
live_tags = await atomic.list_tags()
|
|
path_map = build_path_map_from_tags(live_tags)
|
|
if path_map:
|
|
resolver.load_from_mapping(path_map)
|
|
log.info("taxonomy_refreshed_from_atomic", count=len(path_map))
|
|
except Exception as e: # noqa: BLE001
|
|
log.warning(
|
|
"taxonomy_refresh_failed",
|
|
error=f"{type(e).__name__}: {e}",
|
|
hint="running with whatever is in _tag_ids.json or empty",
|
|
)
|
|
|
|
if not resolver.all:
|
|
log.warning(
|
|
"brain_api_starting_without_taxonomy",
|
|
hint="seed taxonomy by running scripts/04_seed_taxonomy.py on host",
|
|
)
|
|
|
|
deps.set_state(
|
|
deps.AppState(
|
|
atomic=atomic,
|
|
embed=embed,
|
|
llm=llm,
|
|
resolver=resolver,
|
|
)
|
|
)
|
|
log.info(
|
|
"brain_api_ready",
|
|
atomic_url=settings.atomic_url,
|
|
llm_router=settings.llm_router_url,
|
|
embed_url=settings.embedding_url,
|
|
reranker_url=settings.reranker_url,
|
|
taxonomy_size=len(resolver.all),
|
|
)
|
|
try:
|
|
yield
|
|
finally:
|
|
await runtime_config.stop()
|
|
await event_sink.stop()
|
|
await atomic.aclose()
|
|
await embed.aclose()
|
|
await llm.aclose()
|
|
await brain_db.close()
|
|
|
|
|
|
app = FastAPI(
|
|
title="DidiBrain API",
|
|
description=(
|
|
"HTTP interface that speaks the same contract as Didi's web-gathering "
|
|
"module, answering from pre-ingested knowledge atoms + semantic search "
|
|
"+ cross-encoder reranking. Drop-in cache/source for claim verification."
|
|
),
|
|
version="0.1.0",
|
|
lifespan=lifespan,
|
|
)
|
|
|
|
|
|
# ASGI middleware — emits a dashboard event for every non-meta request so brain
|
|
# calls show up in the unified Insights/History view alongside web-api activity.
|
|
from brain_api.events.middleware import event_emit_middleware # noqa: E402
|
|
|
|
app.middleware("http")(event_emit_middleware)
|
|
|
|
|
|
# ---------------------------------------------------------------- observability
|
|
# Prometheus /metrics endpoint
|
|
try:
|
|
from prometheus_fastapi_instrumentator import Instrumentator # type: ignore
|
|
|
|
Instrumentator(
|
|
should_group_status_codes=True,
|
|
should_ignore_untemplated=False,
|
|
).instrument(app).expose(app, endpoint="/metrics", include_in_schema=False)
|
|
except ImportError:
|
|
pass
|
|
|
|
# OpenTelemetry tracing (no-op if endpoint not set)
|
|
import os # noqa: E402
|
|
|
|
_otel_endpoint = os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT")
|
|
if _otel_endpoint:
|
|
try:
|
|
from opentelemetry import trace # type: ignore
|
|
from opentelemetry.sdk.resources import Resource # type: ignore
|
|
from opentelemetry.sdk.trace import TracerProvider # type: ignore
|
|
from opentelemetry.sdk.trace.export import BatchSpanProcessor # type: ignore
|
|
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import ( # type: ignore
|
|
OTLPSpanExporter,
|
|
)
|
|
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor # type: ignore
|
|
from opentelemetry.instrumentation.asyncpg import AsyncPGInstrumentor # type: ignore
|
|
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor # type: ignore
|
|
|
|
resource = Resource.create({
|
|
"service.name": os.environ.get("OTEL_SERVICE_NAME", "didibrain-api"),
|
|
"service.version": "0.1.0",
|
|
})
|
|
provider = TracerProvider(resource=resource)
|
|
provider.add_span_processor(
|
|
BatchSpanProcessor(OTLPSpanExporter(endpoint=_otel_endpoint, insecure=True))
|
|
)
|
|
trace.set_tracer_provider(provider)
|
|
FastAPIInstrumentor.instrument_app(app)
|
|
AsyncPGInstrumentor().instrument()
|
|
HTTPXClientInstrumentor().instrument()
|
|
print(f"[otel] didibrain-api instrumented, exporting to {_otel_endpoint}")
|
|
except ImportError as e:
|
|
print(f"[otel] init skipped (missing deps): {e}")
|
|
|
|
|
|
# --------------------------------------------------------------------- health
|
|
|
|
|
|
@app.get("/health", tags=["meta"])
|
|
async def health() -> dict:
|
|
return {
|
|
"status": "ok",
|
|
"service": "didibrain-api",
|
|
"version": app.version,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------- search
|
|
|
|
|
|
@app.post("/v1/search", response_model=SearchResponse, tags=["v1"])
|
|
async def post_search(req: SearchRequest) -> SearchResponse:
|
|
state = deps.get_state()
|
|
try:
|
|
return await svc_search(req, atomic=state.atomic)
|
|
except Exception as e: # noqa: BLE001
|
|
log.error("search_failed", error=str(e))
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
# ----------------------------------------------------------------------- fetch
|
|
|
|
|
|
@app.post("/v1/fetch", response_model=FetchResponse, tags=["v1"])
|
|
async def post_fetch(req: FetchRequest) -> FetchResponse:
|
|
state = deps.get_state()
|
|
try:
|
|
return await svc_fetch(req, atomic=state.atomic)
|
|
except Exception as e: # noqa: BLE001
|
|
log.error("fetch_failed", error=str(e))
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
# ----------------------------------------------------------------------- gather
|
|
|
|
|
|
@app.post("/v1/gather", response_model=GatherResponse, tags=["v1"])
|
|
async def post_gather(req: GatherRequest) -> GatherResponse:
|
|
state = deps.get_state()
|
|
try:
|
|
return await svc_gather(
|
|
req,
|
|
atomic=state.atomic,
|
|
embed=state.embed,
|
|
llm=state.llm,
|
|
resolver=state.resolver,
|
|
)
|
|
except Exception as e: # noqa: BLE001
|
|
log.exception("gather_failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
# ----------------------------------------------------------------- image-search
|
|
|
|
|
|
@app.post("/v1/image-search", response_model=ImageSearchResponse, tags=["v1"])
|
|
async def post_image_search(req: ImageSearchRequest) -> ImageSearchResponse:
|
|
"""Always returns an empty list — brain has no image corpus.
|
|
|
|
We still match the shape so Didi's backend can call us uniformly and know
|
|
to fall through to the real image-search service.
|
|
"""
|
|
t0 = time.perf_counter()
|
|
return ImageSearchResponse(
|
|
request_id=str(uuid.uuid4()),
|
|
results=[],
|
|
total_results=0,
|
|
execution_time_ms=round((time.perf_counter() - t0) * 1000, 1),
|
|
queries_processed=len(req.queries),
|
|
brain_meta=BrainMeta(
|
|
cache_status="MISS",
|
|
api_version="v1",
|
|
implementation="didibrain",
|
|
evidence_sources=0,
|
|
total_claim_atoms_matched=0,
|
|
),
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------- canonicalize
|
|
|
|
|
|
@app.post(
|
|
"/v1/canonicalize",
|
|
response_model=CanonicalizeResponse,
|
|
tags=["v1"],
|
|
)
|
|
async def post_canonicalize(req: CanonicalizeRequest) -> CanonicalizeResponse:
|
|
"""Resolve relative time markers + ambiguous entities in a claim (Pilon 7).
|
|
|
|
Caller (typically agent-v3 claims executor) sends raw user claim plus an
|
|
optional explicit current_date; brain returns the rewritten claim that
|
|
the caller should hash for cache lookups. Same surface text asked at
|
|
different times produces different canonical forms when temporal anchors
|
|
apply, so the same cache key cannot silently serve stale verdicts across
|
|
time horizons.
|
|
|
|
Failure-safe: if the LLM call fails, the response has canonical=original
|
|
and error populated. Caller should still proceed (no regression).
|
|
"""
|
|
state = deps.get_state()
|
|
|
|
# Optional explicit current_date — when agent-v3 wants to anchor a batch
|
|
# of claims to the same wall clock for determinism.
|
|
current_date: datetime | None = None
|
|
if req.current_date:
|
|
try:
|
|
current_date = datetime.fromisoformat(req.current_date)
|
|
except ValueError:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=f"current_date must be ISO format, got: {req.current_date}",
|
|
)
|
|
|
|
result = await canonicalize_claim_temporal(
|
|
state.llm,
|
|
claim=req.claim,
|
|
current_date=current_date,
|
|
)
|
|
return CanonicalizeResponse(
|
|
canonical=result.canonical,
|
|
original=result.original,
|
|
changed=result.changed,
|
|
anchors_added=result.anchors_added,
|
|
reasoning=result.reasoning,
|
|
error=result.error,
|
|
)
|
|
|
|
|
|
# ----------------------------------------------------------------- invalidate
|
|
|
|
|
|
@app.post(
|
|
"/v1/cache/invalidate",
|
|
response_model=CacheInvalidateResponse,
|
|
tags=["v1"],
|
|
)
|
|
async def post_cache_invalidate(
|
|
req: CacheInvalidateRequest,
|
|
) -> CacheInvalidateResponse:
|
|
"""Mass-invalidate analysis_atom + verification_cache rows by filter (Pilon 8).
|
|
|
|
Used by:
|
|
- admin dashboard "flush topic" button (manual)
|
|
- didibrain-breaking-watcher (real-time topic invalidation when
|
|
breaking news affects an entity or topic)
|
|
- cron jobs that flush evolved-news at end of day
|
|
|
|
Soft-deletes (sets ``expires_at = now()``) — rows stay for audit. Gold
|
|
atoms are spared by default; pass ``invalidate_gold=true`` to flush
|
|
those too (only in moderator-initiated flows).
|
|
|
|
Sub-endpoint of ``/v1/cache/...`` namespace (reserved for additional
|
|
cache-management endpoints in later phases — list-flushed, restore,
|
|
etc.).
|
|
"""
|
|
# Parse since if provided.
|
|
since: datetime | None = None
|
|
if req.since:
|
|
try:
|
|
since = datetime.fromisoformat(req.since)
|
|
except ValueError:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=f"since must be ISO datetime, got: {req.since}",
|
|
)
|
|
|
|
f = InvalidateFilter(
|
|
topic_codes=req.topic_codes,
|
|
entity_canonicals=req.entity_canonicals,
|
|
claim_pattern=req.claim_pattern,
|
|
since=since,
|
|
invalidate_gold=req.invalidate_gold,
|
|
dry_run=req.dry_run,
|
|
)
|
|
|
|
if f.is_empty():
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=(
|
|
"filter must include at least one of: topic_codes, "
|
|
"entity_canonicals, claim_pattern, since"
|
|
),
|
|
)
|
|
|
|
# Rate-limit: max 10 non-dry-run invalidations per actor per hour. Dry-runs are free.
|
|
if not f.dry_run:
|
|
actor = req.actor or "api"
|
|
try:
|
|
from brain_api.db import get_db
|
|
|
|
async with get_db().acquire() as conn:
|
|
count_q = await conn.fetchrow(
|
|
"""
|
|
SELECT COUNT(*) AS c
|
|
FROM brain_audit_log
|
|
WHERE action = 'invalidate'
|
|
AND actor = $1
|
|
AND created_at >= now() - interval '1 hour'
|
|
AND COALESCE((payload->>'dry_run')::boolean, false) = false
|
|
""",
|
|
actor,
|
|
)
|
|
recent_count = int(count_q["c"]) if count_q else 0
|
|
RATE_LIMIT_PER_HOUR = 10
|
|
if recent_count >= RATE_LIMIT_PER_HOUR:
|
|
raise HTTPException(
|
|
status_code=429,
|
|
detail=(
|
|
f"Rate limit exceeded: actor '{actor}' has "
|
|
f"{recent_count} invalidations in the last hour "
|
|
f"(limit: {RATE_LIMIT_PER_HOUR}). Try again later "
|
|
f"or use dry_run=true."
|
|
),
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e: # noqa: BLE001
|
|
log.warning("rate_limit_check_failed", error=str(e))
|
|
# Fail-open on rate-limit lookup error (don't block legitimate invalidations).
|
|
|
|
try:
|
|
result = await invalidate_caches(
|
|
f, actor=req.actor or "api", reason=req.reason
|
|
)
|
|
except RuntimeError as e:
|
|
log.error("invalidate_db_unavailable", error=str(e))
|
|
raise HTTPException(
|
|
status_code=503, detail="cache store unavailable"
|
|
)
|
|
except Exception as e: # noqa: BLE001
|
|
log.exception("invalidate_failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
return CacheInvalidateResponse(
|
|
invalidated_atoms=result.invalidated_atoms,
|
|
invalidated_vcache=result.invalidated_vcache,
|
|
dry_run=result.dry_run,
|
|
filters_applied=result.filters_applied,
|
|
executed_at=result.executed_at,
|
|
)
|
|
|
|
|
|
# ----------------------------------------------------------------------- ingest
|
|
|
|
|
|
@app.post("/v1/ingest", response_model=IngestResponse, tags=["v1"])
|
|
async def post_ingest(
|
|
req: IngestRequest, background: BackgroundTasks
|
|
) -> IngestResponse:
|
|
state = deps.get_state()
|
|
try:
|
|
return await svc_ingest(
|
|
req,
|
|
atomic=state.atomic,
|
|
resolver=state.resolver,
|
|
background=background,
|
|
llm=state.llm,
|
|
)
|
|
except Exception as e: # noqa: BLE001
|
|
log.exception("ingest_failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
# ----------------------------------------------------------- verification_cache
|
|
|
|
|
|
@app.post(
|
|
"/v1/verification_cache",
|
|
response_model=VerificationCacheWriteResponse,
|
|
tags=["v1"],
|
|
)
|
|
async def post_verification_cache(
|
|
req: VerificationCacheWriteRequest,
|
|
) -> VerificationCacheWriteResponse:
|
|
"""Store an LLM verification result for (claim, evidence_urls, tier).
|
|
|
|
Contract with didi-backend:
|
|
- backend runs its own LLM verification call (owns prompt + model)
|
|
- on success, fire-and-forget POSTs the result here
|
|
- next /v1/gather with include_verification=true for the same
|
|
(claim, evidence_urls, tier) returns the cached payload in brain_meta
|
|
|
|
Storage is last-wins on the unique key; payloads are stored verbatim as
|
|
opaque jsonb so backend can evolve the structure without brain changes.
|
|
"""
|
|
# Quick payload-size guard so we don't accept multi-MB junk by accident.
|
|
try:
|
|
import json as _json
|
|
|
|
body_bytes = len(_json.dumps(req.model_dump()).encode("utf-8"))
|
|
except Exception: # noqa: BLE001
|
|
body_bytes = 0
|
|
cap = settings.verification_cache_max_payload_kb * 1024
|
|
if body_bytes > cap:
|
|
raise HTTPException(
|
|
status_code=413,
|
|
detail=(
|
|
f"payload too large: {body_bytes} bytes > "
|
|
f"{cap} bytes cap (set via verification_cache_max_payload_kb)"
|
|
),
|
|
)
|
|
|
|
try:
|
|
state = deps.get_state()
|
|
entry = await vcache.upsert(
|
|
claim=req.claim,
|
|
evidence_urls=req.evidence_urls,
|
|
tier=req.tier,
|
|
prompt_hash=req.prompt_hash,
|
|
verification_processed=req.verification_processed,
|
|
verification_raw=req.verification_raw,
|
|
model=req.model,
|
|
framework_version=req.framework_version,
|
|
schema_name=req.schema_name,
|
|
llm=state.llm,
|
|
)
|
|
except RuntimeError as e:
|
|
# brain_db not connected — fail soft with 503 so backend logs + retries later
|
|
log.error("verification_cache_db_unavailable", error=str(e))
|
|
raise HTTPException(status_code=503, detail="verification cache unavailable")
|
|
except Exception as e: # noqa: BLE001
|
|
log.exception("verification_cache_write_failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
return VerificationCacheWriteResponse(
|
|
cached=True,
|
|
claim_hash=entry.claim_hash,
|
|
evidence_hash=entry.evidence_hash,
|
|
tier=entry.tier,
|
|
created_at=entry.created_at,
|
|
updated_at=entry.updated_at,
|
|
expires_at=entry.expires_at,
|
|
)
|
|
|
|
|
|
# ----------------------------------------------------------- analysis_atom
|
|
# Cache for full-component LLM results (techniques, ai_tampered).
|
|
# Lookup tier-agnostic, write only on tier=premium, gold tier from HIL moderation.
|
|
|
|
|
|
@app.post(
|
|
"/v1/analysis_atom/lookup",
|
|
response_model=AnalysisAtomLookupResponse,
|
|
tags=["v1"],
|
|
)
|
|
async def post_analysis_atom_lookup(
|
|
req: AnalysisAtomLookupRequest,
|
|
) -> AnalysisAtomLookupResponse:
|
|
"""Lookup an atom by (content_hash, component, prompt_hash).
|
|
|
|
Returns hit=True with the cached entry if a fresh gold or silver atom
|
|
matches. Bronze atoms are NEVER served. Gold atoms survive prompt change.
|
|
"""
|
|
try:
|
|
entry, staleness = await atom_svc.lookup(
|
|
content_hash=req.content_hash,
|
|
component=req.component,
|
|
prompt_hash=req.prompt_hash,
|
|
framework_version=req.framework_version,
|
|
)
|
|
except RuntimeError as e:
|
|
log.error("analysis_atom_db_unavailable", error=str(e))
|
|
raise HTTPException(status_code=503, detail="analysis atom store unavailable")
|
|
except Exception as e: # noqa: BLE001
|
|
log.exception("analysis_atom_lookup_failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
if entry is None or staleness == "miss":
|
|
return AnalysisAtomLookupResponse(hit=False, atom=None, staleness="miss")
|
|
|
|
# Only serve fresh entries — backend can choose to ignore stale
|
|
return AnalysisAtomLookupResponse(
|
|
hit=staleness == "fresh",
|
|
atom=AnalysisAtomData(
|
|
atom_id=entry.atom_id,
|
|
content_hash=entry.content_hash,
|
|
component=entry.component,
|
|
tier=entry.tier,
|
|
prompt_hash=entry.prompt_hash,
|
|
framework_version=entry.framework_version,
|
|
model_used=entry.model_used,
|
|
cache_tier=entry.cache_tier,
|
|
human_validated=entry.human_validated,
|
|
result_processed=entry.result_processed,
|
|
validator_user_id=entry.validator_user_id,
|
|
validated_at=entry.validated_at,
|
|
hit_count=entry.hit_count,
|
|
created_at=entry.created_at,
|
|
updated_at=entry.updated_at,
|
|
expires_at=entry.expires_at,
|
|
),
|
|
staleness=staleness,
|
|
match_type="exact",
|
|
)
|
|
|
|
|
|
@app.post(
|
|
"/v1/analysis_atom",
|
|
response_model=AnalysisAtomWriteResponse,
|
|
tags=["v1"],
|
|
)
|
|
async def post_analysis_atom_write(
|
|
req: AnalysisAtomWriteRequest,
|
|
) -> AnalysisAtomWriteResponse:
|
|
"""Write/upsert an atom. Rejects tier=free silently.
|
|
|
|
Runs the volatility classifier on ``content_preview`` (Pilon 1) before
|
|
the SQL upsert, deriving topic_codes / entity_bindings / TTL. Falls back
|
|
to legacy fixed-TTL behavior if the classifier is unavailable.
|
|
"""
|
|
try:
|
|
state = deps.get_state()
|
|
entry, skip_reason = await atom_svc.upsert(
|
|
content_hash=req.content_hash,
|
|
content_preview=req.content_preview,
|
|
component=req.component,
|
|
tier=req.tier,
|
|
prompt_hash=req.prompt_hash,
|
|
framework_version=req.framework_version,
|
|
model_used=req.model_used,
|
|
result_processed=req.result_processed,
|
|
result_raw=req.result_raw,
|
|
llm_confidence=req.llm_confidence,
|
|
cache_tier_override=req.cache_tier,
|
|
llm=state.llm,
|
|
)
|
|
except RuntimeError as e:
|
|
log.error("analysis_atom_db_unavailable", error=str(e))
|
|
raise HTTPException(status_code=503, detail="analysis atom store unavailable")
|
|
except Exception as e: # noqa: BLE001
|
|
log.exception("analysis_atom_write_failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
if entry is None:
|
|
return AnalysisAtomWriteResponse(cached=False, skipped_reason=skip_reason)
|
|
|
|
return AnalysisAtomWriteResponse(
|
|
cached=True,
|
|
atom_id=entry.atom_id,
|
|
cache_tier=entry.cache_tier,
|
|
)
|
|
|
|
|
|
@app.patch(
|
|
"/v1/analysis_atom/{atom_id}",
|
|
response_model=AnalysisAtomData,
|
|
tags=["v1"],
|
|
)
|
|
async def patch_analysis_atom(
|
|
atom_id: int,
|
|
req: AnalysisAtomPatchRequest,
|
|
) -> AnalysisAtomData:
|
|
"""Promote atom to gold (after moderator review)."""
|
|
try:
|
|
entry = await atom_svc.patch_to_gold(
|
|
atom_id=atom_id,
|
|
human_validated=req.human_validated,
|
|
human_corrections=req.human_corrections,
|
|
validator_user_id=req.validator_user_id,
|
|
result_processed=req.result_processed,
|
|
)
|
|
except RuntimeError as e:
|
|
log.error("analysis_atom_db_unavailable", error=str(e))
|
|
raise HTTPException(status_code=503, detail="analysis atom store unavailable")
|
|
except Exception as e: # noqa: BLE001
|
|
log.exception("analysis_atom_patch_failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
if entry is None:
|
|
raise HTTPException(status_code=404, detail=f"atom_id={atom_id} not found")
|
|
|
|
return AnalysisAtomData(
|
|
atom_id=entry.atom_id,
|
|
content_hash=entry.content_hash,
|
|
component=entry.component,
|
|
tier=entry.tier,
|
|
prompt_hash=entry.prompt_hash,
|
|
framework_version=entry.framework_version,
|
|
model_used=entry.model_used,
|
|
cache_tier=entry.cache_tier,
|
|
human_validated=entry.human_validated,
|
|
result_processed=entry.result_processed,
|
|
validator_user_id=entry.validator_user_id,
|
|
validated_at=entry.validated_at,
|
|
hit_count=entry.hit_count,
|
|
created_at=entry.created_at,
|
|
updated_at=entry.updated_at,
|
|
expires_at=entry.expires_at,
|
|
)
|
|
|
|
|
|
@app.get(
|
|
"/v1/analysis_atom/stats",
|
|
response_model=AnalysisAtomStatsResponse,
|
|
tags=["v1"],
|
|
)
|
|
async def get_analysis_atom_stats() -> AnalysisAtomStatsResponse:
|
|
"""Counts and hit rate for monitoring."""
|
|
try:
|
|
stats = await atom_svc.get_stats()
|
|
except RuntimeError as e:
|
|
log.error("analysis_atom_db_unavailable", error=str(e))
|
|
raise HTTPException(status_code=503, detail="analysis atom store unavailable")
|
|
except Exception as e: # noqa: BLE001
|
|
log.exception("analysis_atom_stats_failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
return AnalysisAtomStatsResponse(**stats)
|
|
|
|
|
|
# =========================================================================
|
|
# Admin endpoints (consumed by AI platform dashboard reskin)
|
|
# =========================================================================
|
|
# IMPORTANT: literal paths (/list, /stats/extended) MUST be declared before
|
|
# parameterized {atom_id} routes so FastAPI matches them first.
|
|
# =========================================================================
|
|
|
|
|
|
@app.get(
|
|
"/v1/analysis_atom/list",
|
|
response_model=AnalysisAtomListResponse,
|
|
tags=["admin"],
|
|
)
|
|
async def list_analysis_atoms(
|
|
component: str | None = Query(None, description="techniques | ai_tampered | claims | all"),
|
|
tier: str | None = Query(None, description="cache tier filter: gold | silver | bronze | all"),
|
|
freshness: str | None = Query(None, description="fresh | expiring | expired | all"),
|
|
q: str | None = Query(None, description="ILIKE search on content_preview / content_hash"),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(25, ge=1, le=100),
|
|
) -> AnalysisAtomListResponse:
|
|
"""Paginated atom browser for the dashboard."""
|
|
try:
|
|
result = await admin_svc.list_atoms(
|
|
component=component,
|
|
tier=tier,
|
|
freshness=freshness,
|
|
q=q,
|
|
page=page,
|
|
page_size=page_size,
|
|
)
|
|
except RuntimeError as e:
|
|
log.error("admin_atom_list_db_unavailable", error=str(e))
|
|
raise HTTPException(status_code=503, detail="atom store unavailable")
|
|
except Exception as e: # noqa: BLE001
|
|
log.exception("admin_atom_list_failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
return AnalysisAtomListResponse(
|
|
items=[AnalysisAtomListItem(**asdict(r)) for r in result.items],
|
|
total=result.total,
|
|
page=result.page,
|
|
page_size=result.page_size,
|
|
)
|
|
|
|
|
|
@app.get(
|
|
"/v1/analysis_atom/stats/extended",
|
|
response_model=AnalysisAtomStatsExtendedResponse,
|
|
tags=["admin"],
|
|
)
|
|
async def get_analysis_atom_stats_extended() -> AnalysisAtomStatsExtendedResponse:
|
|
"""Extended stats: per-tier hit counts, gold promotions, recent activity."""
|
|
try:
|
|
stats = await admin_svc.get_stats_extended()
|
|
except RuntimeError as e:
|
|
log.error("admin_stats_db_unavailable", error=str(e))
|
|
raise HTTPException(status_code=503, detail="atom store unavailable")
|
|
except Exception as e: # noqa: BLE001
|
|
log.exception("admin_stats_failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
return AnalysisAtomStatsExtendedResponse(**stats)
|
|
|
|
|
|
@app.get(
|
|
"/v1/analysis_atom/{atom_id}",
|
|
response_model=AnalysisAtomDetailResponse,
|
|
tags=["admin"],
|
|
)
|
|
async def get_analysis_atom(atom_id: int) -> AnalysisAtomDetailResponse:
|
|
"""Full atom row by ID (admin browser detail view)."""
|
|
try:
|
|
row = await admin_svc.get_atom(atom_id)
|
|
except RuntimeError as e:
|
|
log.error("admin_atom_get_db_unavailable", error=str(e))
|
|
raise HTTPException(status_code=503, detail="atom store unavailable")
|
|
except Exception as e: # noqa: BLE001
|
|
log.exception("admin_atom_get_failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail=f"atom_id={atom_id} not found")
|
|
|
|
return AnalysisAtomDetailResponse(**asdict(row))
|
|
|
|
|
|
@app.delete(
|
|
"/v1/analysis_atom/{atom_id}",
|
|
response_model=GenericOkResponse,
|
|
tags=["admin"],
|
|
)
|
|
async def delete_analysis_atom(atom_id: int) -> GenericOkResponse:
|
|
"""Mark an atom as expired (soft delete — keeps row for audit).
|
|
|
|
The row stays in PG with `expires_at = now()` so future lookups skip it.
|
|
A periodic TTL cleanup will hard-delete eventually.
|
|
"""
|
|
try:
|
|
ok = await admin_svc.expire_atom(atom_id)
|
|
except RuntimeError as e:
|
|
log.error("admin_atom_delete_db_unavailable", error=str(e))
|
|
raise HTTPException(status_code=503, detail="atom store unavailable")
|
|
except Exception as e: # noqa: BLE001
|
|
log.exception("admin_atom_delete_failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
if not ok:
|
|
raise HTTPException(status_code=404, detail=f"atom_id={atom_id} not found")
|
|
|
|
return GenericOkResponse(ok=True, message=f"atom_id={atom_id} marked expired")
|
|
|
|
|
|
# ---------------------------------------------------------------- verification
|
|
|
|
|
|
@app.get(
|
|
"/v1/verification_cache/list",
|
|
response_model=VerificationCacheListResponse,
|
|
tags=["admin"],
|
|
)
|
|
async def list_verifications(
|
|
tier: str | None = Query(None, description="free | premium | all"),
|
|
q: str | None = Query(None, description="ILIKE search on claim_hash / model"),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(25, ge=1, le=100),
|
|
) -> VerificationCacheListResponse:
|
|
"""Paginated browser of verification_cache entries."""
|
|
try:
|
|
result = await admin_svc.list_verifications(
|
|
tier=tier, q=q, page=page, page_size=page_size,
|
|
)
|
|
except RuntimeError as e:
|
|
log.error("admin_verif_list_db_unavailable", error=str(e))
|
|
raise HTTPException(status_code=503, detail="verification cache unavailable")
|
|
except Exception as e: # noqa: BLE001
|
|
log.exception("admin_verif_list_failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
return VerificationCacheListResponse(
|
|
items=[VerificationCacheListItem(**asdict(r)) for r in result.items],
|
|
total=result.total,
|
|
page=result.page,
|
|
page_size=result.page_size,
|
|
)
|
|
|
|
|
|
@app.get(
|
|
"/v1/verification_cache/{claim_hash}/{tier}",
|
|
response_model=VerificationCacheDetailResponse,
|
|
tags=["admin"],
|
|
)
|
|
async def get_verification(
|
|
claim_hash: str, tier: str
|
|
) -> VerificationCacheDetailResponse:
|
|
"""Full verification cache row by composite key."""
|
|
if tier not in ("free", "premium"):
|
|
raise HTTPException(status_code=422, detail="tier must be 'free' or 'premium'")
|
|
try:
|
|
row = await admin_svc.get_verification(claim_hash, tier) # type: ignore[arg-type]
|
|
except RuntimeError as e:
|
|
log.error("admin_verif_get_db_unavailable", error=str(e))
|
|
raise HTTPException(status_code=503, detail="verification cache unavailable")
|
|
except Exception as e: # noqa: BLE001
|
|
log.exception("admin_verif_get_failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="verification entry not found")
|
|
|
|
return VerificationCacheDetailResponse(**asdict(row))
|
|
|
|
|
|
@app.delete(
|
|
"/v1/verification_cache/{claim_hash}/{tier}",
|
|
response_model=GenericOkResponse,
|
|
tags=["admin"],
|
|
)
|
|
async def delete_verification(
|
|
claim_hash: str, tier: str
|
|
) -> GenericOkResponse:
|
|
"""Hard-delete a verification cache entry."""
|
|
if tier not in ("free", "premium"):
|
|
raise HTTPException(status_code=422, detail="tier must be 'free' or 'premium'")
|
|
try:
|
|
ok = await admin_svc.delete_verification(claim_hash, tier) # type: ignore[arg-type]
|
|
except RuntimeError as e:
|
|
log.error("admin_verif_delete_db_unavailable", error=str(e))
|
|
raise HTTPException(status_code=503, detail="verification cache unavailable")
|
|
except Exception as e: # noqa: BLE001
|
|
log.exception("admin_verif_delete_failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
if not ok:
|
|
raise HTTPException(status_code=404, detail="verification entry not found")
|
|
|
|
return GenericOkResponse(ok=True, message=f"deleted verification {claim_hash[:12]}.../{tier}")
|
|
|
|
|
|
# -------------------------------------------------------------------- taxonomy
|
|
|
|
|
|
@app.get(
|
|
"/v1/taxonomy",
|
|
response_model=TaxonomyInfoResponse,
|
|
tags=["admin"],
|
|
)
|
|
async def get_taxonomy() -> TaxonomyInfoResponse:
|
|
"""Snapshot of currently loaded taxonomy (in-memory resolver state)."""
|
|
state = deps.get_state()
|
|
info = await admin_svc.get_taxonomy_info(state.resolver)
|
|
return TaxonomyInfoResponse(**info)
|
|
|
|
|
|
@app.post(
|
|
"/v1/taxonomy/reload",
|
|
response_model=TaxonomyReloadResponse,
|
|
tags=["admin"],
|
|
)
|
|
async def reload_taxonomy() -> TaxonomyReloadResponse:
|
|
"""Re-fetch tags from atomic and replace the in-process resolver."""
|
|
state = deps.get_state()
|
|
result = await admin_svc.reload_taxonomy(state.resolver, state.atomic)
|
|
return TaxonomyReloadResponse(**result)
|
|
|
|
|
|
# =========================================================================
|
|
# Phase D2 — fact_status admin endpoints
|
|
# =========================================================================
|
|
# Read-only browsing + moderator override on (subject, predicate, object)
|
|
# triples extracted from claims. Used by the admin dashboard to inspect
|
|
# what brain knows about the world and to lock high-stakes facts so the
|
|
# auditor can't auto-flip them.
|
|
# =========================================================================
|
|
|
|
|
|
def _fact_to_item(fact: fact_svc.FactRecord) -> FactStatusItem:
|
|
"""Map FactRecord (dataclass) → FactStatusItem (pydantic) for response.
|
|
|
|
Maps the dataclass attribute ``obj`` to the schema field ``object`` so
|
|
the JSON shape matches the canonical (subject, predicate, object) form.
|
|
"""
|
|
return FactStatusItem(
|
|
fact_id=fact.fact_id,
|
|
subject=fact.subject,
|
|
predicate=fact.predicate,
|
|
object=fact.obj,
|
|
canonical_form=fact.canonical_form,
|
|
canonical_form_hash=fact.canonical_form_hash,
|
|
current_truth=fact.current_truth,
|
|
current_version_id=fact.current_version_id,
|
|
current_confidence=fact.current_confidence,
|
|
last_verified_at=fact.last_verified_at,
|
|
last_evidence_urls=fact.last_evidence_urls,
|
|
volatility=fact.volatility,
|
|
topic_codes=fact.topic_codes,
|
|
next_check_at=fact.next_check_at,
|
|
check_interval_hours=fact.check_interval_hours,
|
|
moderator_locked=fact.moderator_locked,
|
|
moderator_user_id=fact.moderator_user_id,
|
|
moderator_notes=fact.moderator_notes,
|
|
created_at=fact.created_at,
|
|
updated_at=fact.updated_at,
|
|
)
|
|
|
|
|
|
@app.get(
|
|
"/v1/fact_status/list",
|
|
response_model=FactStatusListResponse,
|
|
tags=["admin"],
|
|
)
|
|
async def list_fact_status(
|
|
entity: str | None = Query(None, description="ILIKE match on subject OR object"),
|
|
predicate: str | None = Query(None, description="Exact match on predicate"),
|
|
current_truth: bool | None = Query(None, description="Filter to TRUE/FALSE only"),
|
|
locked_only: bool = Query(False, description="Show only moderator-locked facts"),
|
|
topic: str | None = Query(None, description="Filter by topic_code in topic_codes"),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(25, ge=1, le=100),
|
|
) -> FactStatusListResponse:
|
|
"""Paginated browser of brain_fact_status."""
|
|
try:
|
|
result = await fact_svc.list_facts_admin(
|
|
entity=entity,
|
|
predicate=predicate,
|
|
current_truth=current_truth,
|
|
locked_only=locked_only,
|
|
topic=topic,
|
|
page=page,
|
|
page_size=page_size,
|
|
)
|
|
except RuntimeError as e:
|
|
log.error("fact_status_list_db_unavailable", error=str(e))
|
|
raise HTTPException(status_code=503, detail="fact store unavailable")
|
|
except Exception as e: # noqa: BLE001
|
|
log.exception("fact_status_list_failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
return FactStatusListResponse(
|
|
items=[_fact_to_item(f) for f in result.items],
|
|
total=result.total,
|
|
page=result.page,
|
|
page_size=result.page_size,
|
|
)
|
|
|
|
|
|
@app.get(
|
|
"/v1/fact_status/due_for_recheck",
|
|
tags=["admin"],
|
|
)
|
|
async def fact_status_due_for_recheck(
|
|
limit: int = Query(50, ge=1, le=500),
|
|
volatility: str | None = Query(None, description="Filter: volatile|evolving|stable"),
|
|
) -> dict:
|
|
"""Facts whose `next_check_at` has passed and need re-verification.
|
|
|
|
Excludes moderator-locked facts (those are managed by humans). Sorted by
|
|
next_check_at ASC (oldest first). Use this for proactive moderation —
|
|
show in dashboard so operators can prioritize re-checking stale facts.
|
|
"""
|
|
try:
|
|
items = await fact_svc.list_facts_due_for_check(
|
|
limit=limit, volatility=volatility
|
|
)
|
|
except RuntimeError as e:
|
|
log.error("fact_due_for_recheck_db_unavailable", error=str(e))
|
|
raise HTTPException(status_code=503, detail="fact store unavailable")
|
|
except Exception as e: # noqa: BLE001
|
|
log.exception("fact_due_for_recheck_failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
return {
|
|
"items": [_fact_to_item(f).model_dump(mode="json") for f in items],
|
|
"count": len(items),
|
|
"limit": limit,
|
|
"volatility_filter": volatility,
|
|
}
|
|
|
|
|
|
@app.get(
|
|
"/v1/fact_status/{fact_id}",
|
|
response_model=FactStatusItem,
|
|
tags=["admin"],
|
|
)
|
|
async def get_fact_status(fact_id: int) -> FactStatusItem:
|
|
"""Single fact by ID."""
|
|
try:
|
|
fact = await fact_svc.get_fact_by_id(fact_id)
|
|
except RuntimeError as e:
|
|
log.error("fact_status_get_db_unavailable", error=str(e))
|
|
raise HTTPException(status_code=503, detail="fact store unavailable")
|
|
if fact is None:
|
|
raise HTTPException(status_code=404, detail=f"fact_id={fact_id} not found")
|
|
return _fact_to_item(fact)
|
|
|
|
|
|
@app.get(
|
|
"/v1/fact_status/{fact_id}/versions",
|
|
response_model=FactStatusVersionsResponse,
|
|
tags=["admin"],
|
|
)
|
|
async def get_fact_versions(fact_id: int) -> FactStatusVersionsResponse:
|
|
"""Timeline of all versions for one fact, newest first."""
|
|
try:
|
|
versions = await fact_svc.list_versions(fact_id)
|
|
except RuntimeError as e:
|
|
log.error("fact_versions_db_unavailable", error=str(e))
|
|
raise HTTPException(status_code=503, detail="fact store unavailable")
|
|
|
|
return FactStatusVersionsResponse(
|
|
fact_id=fact_id,
|
|
total=len(versions),
|
|
versions=[
|
|
FactStatusVersionItem(
|
|
version_id=v.version_id,
|
|
fact_id=v.fact_id,
|
|
truth_value=v.truth_value,
|
|
confidence=v.confidence,
|
|
valid_from=v.valid_from,
|
|
valid_to=v.valid_to,
|
|
source_atom_ids=v.source_atom_ids,
|
|
evidence_urls=v.evidence_urls,
|
|
llm_reasoning=v.llm_reasoning,
|
|
created_by=v.created_by,
|
|
moderator_user_id=v.moderator_user_id,
|
|
notes=v.notes,
|
|
created_at=v.created_at,
|
|
)
|
|
for v in versions
|
|
],
|
|
)
|
|
|
|
|
|
@app.patch(
|
|
"/v1/fact_status/{fact_id}",
|
|
response_model=FactStatusItem,
|
|
tags=["admin"],
|
|
)
|
|
async def patch_fact_status(
|
|
fact_id: int, req: FactStatusPatchRequest
|
|
) -> FactStatusItem:
|
|
"""Moderator override: set truth, lock, or unlock a fact.
|
|
|
|
All three operations are independent — caller may set truth and lock
|
|
in the same call, or just toggle lock without changing truth.
|
|
|
|
Returns the updated FactRecord. Raises 404 if fact_id doesn't exist.
|
|
"""
|
|
# Look up the fact first so we can address it by canonical_form_hash
|
|
# in the service layer (which keys on hash, not id).
|
|
fact = await fact_svc.get_fact_by_id(fact_id)
|
|
if fact is None:
|
|
raise HTTPException(status_code=404, detail=f"fact_id={fact_id} not found")
|
|
|
|
# Step 1: optional set_truth.
|
|
if req.set_truth is not None:
|
|
try:
|
|
result = await fact_svc.assert_fact_truth(
|
|
canonical_form_hash=fact.canonical_form_hash,
|
|
truth_value=req.set_truth,
|
|
confidence=req.confidence,
|
|
evidence_urls=req.evidence_urls,
|
|
source_atom_ids=None,
|
|
llm_reasoning=None,
|
|
created_by="moderator",
|
|
moderator_user_id=req.moderator_user_id,
|
|
notes=req.notes,
|
|
)
|
|
except RuntimeError as e:
|
|
log.error("fact_assert_db_unavailable", error=str(e))
|
|
raise HTTPException(status_code=503, detail="fact store unavailable")
|
|
except Exception as e: # noqa: BLE001
|
|
log.exception("fact_assert_failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
if result is None:
|
|
# locked + non-moderator path is only triggered for created_by!=moderator
|
|
# so this branch shouldn't fire here, but stay defensive.
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail="fact is locked and cannot be changed by this caller",
|
|
)
|
|
|
|
# Step 2: optional lock/unlock.
|
|
if req.lock is True:
|
|
await fact_svc.lock_fact(
|
|
canonical_form_hash=fact.canonical_form_hash,
|
|
moderator_user_id=req.moderator_user_id,
|
|
moderator_notes=req.notes,
|
|
)
|
|
elif req.lock is False:
|
|
await fact_svc.unlock_fact(
|
|
canonical_form_hash=fact.canonical_form_hash,
|
|
moderator_user_id=req.moderator_user_id,
|
|
)
|
|
|
|
# Re-fetch and return the latest state.
|
|
refreshed = await fact_svc.get_fact_by_id(fact_id)
|
|
if refreshed is None:
|
|
raise HTTPException(status_code=404, detail=f"fact_id={fact_id} disappeared")
|
|
return _fact_to_item(refreshed)
|
|
|
|
|
|
# =========================================================================
|
|
# Phase D2 — audit log browser
|
|
# =========================================================================
|
|
|
|
|
|
@app.get(
|
|
"/v1/cache/audit_log",
|
|
response_model=AuditLogResponse,
|
|
tags=["admin"],
|
|
)
|
|
async def list_audit_log(
|
|
action: str | None = Query(None, description="Action prefix (ILIKE 'X%')"),
|
|
target_table: str | None = Query(None, description="Exact target_table match"),
|
|
actor: str | None = Query(None, description="ILIKE match on actor"),
|
|
since: str | None = Query(None, description="ISO datetime — entries at or after"),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(50, ge=1, le=200),
|
|
) -> AuditLogResponse:
|
|
"""Browse brain_audit_log with filters + pagination."""
|
|
parsed_since: datetime | None = None
|
|
if since:
|
|
try:
|
|
parsed_since = datetime.fromisoformat(since)
|
|
except ValueError:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=f"since must be ISO datetime, got: {since}",
|
|
)
|
|
|
|
try:
|
|
items, total = await fact_svc.list_audit_log(
|
|
action=action,
|
|
target_table=target_table,
|
|
actor=actor,
|
|
since=parsed_since,
|
|
page=page,
|
|
page_size=page_size,
|
|
)
|
|
except RuntimeError as e:
|
|
log.error("audit_log_db_unavailable", error=str(e))
|
|
raise HTTPException(status_code=503, detail="audit log unavailable")
|
|
except Exception as e: # noqa: BLE001
|
|
log.exception("audit_log_failed")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
return AuditLogResponse(
|
|
items=[AuditLogItem(**i) for i in items],
|
|
total=total,
|
|
page=page,
|
|
page_size=page_size,
|
|
)
|