agnes-the-ai-analyst/connectors/llm/factory.py
Vojtech d6ad08f107
Flea-market upload guardrails + soft delete + JOIN-based admin queue (#233)
* feat(store): flea-market upload guardrails + soft delete + JOIN-based admin queue

Adds an end-to-end guardrails pipeline for store uploads (manifest +
static-security + LLM review), persists blocked bundles for forensics,
introduces soft-delete (Archive) semantics, consolidates the legacy
/store/{id} surface into /marketplace/flea/{id}, and reworks the admin
queue so lifecycle filters read live entity visibility via LEFT JOIN
rather than a denormalized submission column.

Schema v29 → v35:
  * v29 store_submissions table + store_entities.visibility_status
  * v30 file_size, bundle_sha256, bundle_purged_at on submissions
  * v31 reshape store_submissions (drop legacy unique on entity_id)
  * v32 store_entities.archived_at/by + 'archived' visibility value
  * v33 drop store_submissions.retry_count (unused)
  * v34 ensure idx_store_submissions_entity exists post column-drop
  * v35 broaden visibility_status enum + JOIN architecture cutover

Pipeline (src/store_guardrails/):
  * Inline checks: manifest_check, static_scan, quality_check
  * LLM review configurable haiku|sonnet|opus (default haiku)
  * BackgroundTasks-driven async path with structured-output JSON
  * Per-submitter daily quota (default 50)
  * 30-day TTL purge job (POST /api/admin/run-blocked-purge)
  * Bundle SHA256 + size persisted; sha256 survives purge for forensics

Visibility model:
  * pending | approved | hidden | archived
  * _enforce_visibility returns 404 (no leak) for non-owner non-admin
  * Owner sees own non-approved entries via include_owner_id widening
  * Install refused with 409 entity_not_approved when not approved

Soft-delete (DELETE /api/store/entities/{id}):
  * Default = soft (visibility_status='archived'); existing installs
    keep getting served the bundle so users don't lose the plugin
  * ?hard=true admin-only: drops bundle + cascades user_store_installs
  * Hard-delete preserves entity_id on submission as tombstone so
    audit_log linkage survives for the activity timeline

Admin queue lifecycle (the JOIN refactor):
  * Verdict (store_submissions.status) is immutable forensic record
  * Lifecycle (store_entities.visibility_status) is live state
  * /admin/store/submissions Archived chip translates to
    `e.visibility_status='archived'` via LEFT JOIN — any path that
    flips visibility surfaces in the queue immediately
  * Detail page renders Status (verdict) and Entity lifecycle side by
    side so admins see "approved at review, now archived" at a glance

URL consolidation:
  * /store/{id} deleted (no redirect, stale bookmarks 404)
  * /marketplace/flea/{id} is the canonical detail surface
  * Three in-tree callers (upload-success, my-stack card, store
    listing card) updated to point at the new URL
  * Quarantine banner extracted to _quarantine_banner.html partial,
    self-guarded, included from both flea detail templates
  * Banner JS auto-refreshes when the verdict lands by polling
    /api/marketplace/flea/{id}/detail (visibility_status +
    submission_status — the latter is needed because blocked_llm
    keeps the entity at visibility_status='pending')

Audit log resource format:
  * runner.py emits prefixed `store_submission:{id}` (post-fix)
  * Detail-page timeline query handles three patterns: prefixed
    submission, helper-emitted `store_entity:{sub_id}`, and bare-id
    legacy rows — all surface in the activity timeline

UX fixes:
  * Owner sees Under review / Quarantined / Hidden banner with status
  * Install button gray-disabled (not blue) when non-approved
  * Owner cannot delete quarantined entries (403); admin can
  * Admin queue: filter chips, sortable columns, paging, page-size
  * Auto-refresh queue every 5s while pending rows are visible
  * Store upload page file picker no longer opens twice (label →
    input default action collided with explicit JS handler)

Tests: 168 passed across the guardrails suites (admin submissions,
store API, inline / LLM / purge guardrails, store repositories,
marketplace filter, schema version). New regression coverage
includes: archive surfaces via JOIN even when API path is bypassed;
deleted submission renders activity timeline (tombstone); flea
detail surfaces submission_status only for owner/admin; detail page
renders Entity lifecycle row; audit log resource format covers both
helper and runner paths.

* fix(store-guardrails): PR #233 follow-up — prompt injection, atomic PUT, BG race, schema, reaper, sort whitelist

Addresses 9 of the 23 findings from the PR #233 review (spec at
docs/superpowers/specs/2026-05-09-pr233-guardrails-fixes-spec.md).
Merge-gate items #1-#6 plus high-value mediums #7, #9-#12, #23.
Architectural items (#8 enum split, #14 factory) and pure
maintainability (#15-#22) deferred to follow-ups.

Security:
* #1 prompt injection — SYSTEM_PROMPT now passed via the SDK's
  dedicated system= parameter; bundle wrapped in <bundle>...</bundle>
  sentinels declared data-only by the system prompt; literal
  sentinel strings in user content are escaped so an adversarial
  README can't forge a close tag.
* #6 static scan honesty — module docstring + admin copy + docs
  declare static scan as signal not gate; .md/.txt/.rst/.html/.json/
  .yaml/.yml/.toml skipped to avoid false positives on prose.
  AST mode for Python deferred (separate flag, FP comparison work).

Correctness:
* #2 PUT atomicity — bundles bake into plugin.staging-<rand>/
  alongside live, atomic-rename on success; failed checks leave
  live tree byte-for-byte intact.
* #3 BG-task race — set_visibility_if_pending guards verdict flips
  to the (pending, hidden) review window; admin archives during
  review survive; skipped flips audit-logged.
* #4 v35 NOT NULL/DEFAULT — schema v35→v36 re-applies them on
  store_entities.visibility_status. CHECK constraint enforced
  application-side (DuckDB ADD CHECK on existing column unsupported).
* #7 stuck-review reaper — reap_stuck_llm_reviews flips pending_llm
  rows older than guardrails.stuck_review_grace_seconds (default
  1800) to review_error. Scheduler runs every 15 min via new
  /api/admin/run-reap-stuck-reviews. Set knob to 0 to disable.
* #9 quota counter — count_blocked_for_submitter_since now counts
  blocked_inline + blocked_llm + review_error so a submitter
  triggering only LLM-blocked verdicts is bounded.
* #10 missing risk_level — surfaces as review_error with
  error='missing_risk_level' instead of silently defaulting to
  'medium' (which looked like a model-decided block).
* #11 archived_at clear — set_visibility nulls archived_at +
  archived_by when transitioning out of 'archived' so a future
  read doesn't show stale archive forensics on an approved row.

Maintainability:
* #12 FSM doc comment — accurate insert/transition/lifecycle
  description in src/db.py near store_submissions schema.
* #23 sort-key whitelist — admin queue rejects unknown sort keys
  with 400 invalid_sort_key; substring-replace footgun removed.

Deferred (separate PRs):
* #5 quota race — proper fix requires asyncio.Lock spanning the
  full pipeline; threading.Lock blocks event loop, DuckDB MVCC
  doesn't help. API-level slowapi bounds worst case for now.
* #6 part 3 (AST static scan), #8 (enum split), #13 (import
  bundle docs), #14 (factory consolidation), #15-#22 (maint).

Tests:
* New: tests/test_store_guardrails_prompt_injection.py (corpus +
  trust-boundary invariants), tests/test_store_put_atomic.py,
  tests/test_store_guardrails_reaper.py.
* Extended: test_store_guardrails_llm.py (system param, missing
  risk_level, BG race), test_admin_store_submissions.py (quota
  counter widening, sort whitelist 400), test_store_repositories.py
  (un-archive metadata clear), test_db_schema_version.py (v36).
* Full suite: 3738 passed; 17 pre-existing baseline failures
  unchanged (db migration tests, cli binary rename, catalog export,
  user mgmt v5 backfill — confirmed by stash + rerun on clean tree).
2026-05-09 17:32:53 +04:00

214 lines
7.5 KiB
Python

"""Factory for creating structured extractors from instance configuration.
Reads the ai: section from instance.yaml (already resolved by config/loader.py)
and creates the appropriate StructuredExtractor implementation.
"""
import logging
import os
from urllib.parse import urlparse
from .anthropic_provider import AnthropicExtractor
from .base import StructuredExtractor
from .openai_compat import OpenAICompatExtractor
logger = logging.getLogger(__name__)
# Default model when not specified in config
DEFAULT_MODEL = "claude-haiku-4-5-20251001"
# Default structured output strategy
DEFAULT_STRUCTURED_OUTPUT = "auto"
# Tier → concrete model ID. Used by guardrails (and any future feature)
# that wants to expose a "haiku|sonnet|opus" knob to operators without
# pinning them to a specific dated model. Update here when bumping the
# fleet to a newer model family — callers stay on the abstract tier.
MODEL_TIERS: dict[str, str] = {
"haiku": "claude-haiku-4-5-20251001",
"sonnet": "claude-sonnet-4-6",
"opus": "claude-opus-4-7",
}
def resolve_model_tier(tier: str) -> str:
"""Map an abstract tier ('haiku'|'sonnet'|'opus') to a concrete model ID.
Accepts the tier name OR a concrete model ID (passed through unchanged
so operators who already know the exact ID they want can hard-pin it
in instance.yaml). Unknown tier names raise ValueError so a typo in
config surfaces at startup, not at first review call.
"""
if not tier:
return DEFAULT_MODEL
tier = tier.strip()
if tier in MODEL_TIERS:
return MODEL_TIERS[tier]
if tier.startswith("claude-"):
return tier
raise ValueError(
f"Unknown model tier {tier!r}. Use one of "
f"{sorted(MODEL_TIERS)} or a concrete claude-* model ID."
)
def create_extractor(ai_config: dict) -> StructuredExtractor:
"""Create a structured extractor from the ai: config section.
Supports two configuration formats:
New format (explicit provider):
ai:
provider: anthropic | openai_compat
api_key: ${ANTHROPIC_API_KEY}
model: claude-haiku-4-5-20251001
base_url: https://api.example.com/v1 # required for openai_compat
structured_output: auto # strict | json | auto
Legacy format (backward compatible):
ai:
anthropic_api_key: ${ANTHROPIC_API_KEY}
Args:
ai_config: The ai: section dict from instance.yaml,
already resolved by config/loader.py.
Returns:
A StructuredExtractor instance.
Raises:
ValueError: If configuration is invalid or incomplete.
"""
if not ai_config or not isinstance(ai_config, dict):
raise ValueError(
"ai: section in instance.yaml must be a non-empty dict. "
"Example:\n ai:\n provider: anthropic\n api_key: ${ANTHROPIC_API_KEY}"
)
provider = ai_config.get("provider")
# Legacy format detection: anthropic_api_key present, no provider
if not provider and "anthropic_api_key" in ai_config:
api_key = ai_config["anthropic_api_key"]
_validate_api_key(api_key)
model = ai_config.get("model", DEFAULT_MODEL)
logger.info(
"Creating AnthropicExtractor (legacy config), model=%s", model
)
return AnthropicExtractor(api_key=api_key, model=model)
if not provider:
raise ValueError(
"ai.provider is required in instance.yaml. "
"Supported: 'anthropic', 'openai_compat'. "
"Hint: use ${ENV_VAR} syntax for secrets."
)
api_key = ai_config.get("api_key", "")
_validate_api_key(api_key)
model = ai_config.get("model", DEFAULT_MODEL)
if provider == "anthropic":
logger.info("Creating AnthropicExtractor, model=%s", model)
return AnthropicExtractor(api_key=api_key, model=model)
elif provider == "openai_compat":
base_url = ai_config.get("base_url", "")
if not base_url:
raise ValueError(
"ai.base_url is required when provider is 'openai_compat'. "
"Example: base_url: https://api.openai.com/v1"
)
structured_output = ai_config.get(
"structured_output", DEFAULT_STRUCTURED_OUTPUT,
)
if structured_output not in ("strict", "json", "auto"):
raise ValueError(
f"ai.structured_output must be 'strict', 'json', or 'auto', "
f"got '{structured_output}'"
)
verify_ssl = ai_config.get("verify_ssl", True)
safe_url = _sanitize_url(base_url)
logger.info(
"Creating OpenAICompatExtractor, url=%s, model=%s, "
"structured_output=%s, verify_ssl=%s",
safe_url, model, structured_output, verify_ssl,
)
return OpenAICompatExtractor(
api_key=api_key,
base_url=base_url,
model=model,
structured_output=structured_output,
verify_ssl=verify_ssl,
)
else:
raise ValueError(
f"Unknown ai.provider '{provider}'. "
f"Supported: 'anthropic', 'openai_compat'. "
f"Hint: use ${{ENV_VAR}} syntax for secrets."
)
def create_extractor_from_env_or_config(
ai_config: dict | None,
) -> StructuredExtractor:
"""Build an extractor from config, falling back to env vars.
Resolution order (#176):
1. ``ai_config`` is a non-empty dict → delegate to :func:`create_extractor`.
2. ``ANTHROPIC_API_KEY`` set → AnthropicExtractor with the default model.
3. ``LLM_API_KEY`` set without a base_url → AnthropicExtractor (the proxy
case typically also wires a base_url, in which case the operator should
use the explicit ai: block; this fallback is a best-effort convenience).
4. Otherwise raise ``ValueError`` with a clear actionable message — never
silently exit, never return ``None``. The previous "skip when ai: is
missing" behavior was the silent-failure root cause in #176.
"""
if ai_config:
return create_extractor(ai_config)
anthropic_key = os.environ.get("ANTHROPIC_API_KEY", "").strip()
llm_key = os.environ.get("LLM_API_KEY", "").strip()
if anthropic_key:
logger.info(
"No ai: block in instance.yaml; falling back to ANTHROPIC_API_KEY env var"
)
return AnthropicExtractor(api_key=anthropic_key, model=DEFAULT_MODEL)
if llm_key:
logger.info(
"No ai: block in instance.yaml; falling back to LLM_API_KEY env var"
)
return AnthropicExtractor(api_key=llm_key, model=DEFAULT_MODEL)
raise ValueError(
"LLM not configured. Add an ai: block to instance.yaml (see "
"config/instance.yaml.example) OR set ANTHROPIC_API_KEY / LLM_API_KEY "
"in the environment. The corporate-memory and verification-detector "
"services cannot run without one of these."
)
def _validate_api_key(api_key: str) -> None:
"""Validate that an API key is present and non-empty.
Raises:
ValueError: If api_key is empty or missing.
"""
if not api_key or not api_key.strip():
raise ValueError(
"ai.api_key (or ai.anthropic_api_key) must not be empty. "
"Check that the corresponding environment variable is set "
"and referenced with ${ENV_VAR} syntax in instance.yaml."
)
def _sanitize_url(url: str) -> str:
"""Extract scheme://host from a URL for safe logging."""
parsed = urlparse(url)
return f"{parsed.scheme}://{parsed.netloc}"