soc_ai/main.py
Rama main · réplica de lectura del repositorio original.
"""FastAPI application entry point.
``uv run soc-ai`` boots the API via :func:`main`; ``uvicorn soc_ai.main:app``
boots it directly. The lifespan manager constructs every long-lived dependency
once and tears them down on shutdown.
"""
from __future__ import annotations
import asyncio
import contextlib
import ipaddress
import json
import logging
import os
import re
import time
from collections.abc import AsyncIterator, Mapping
from contextlib import asynccontextmanager
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any
from fastapi import FastAPI, Request
from fastapi.encoders import jsonable_encoder
from fastapi.exception_handlers import http_exception_handler
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.utils import is_body_allowed_for_status_code
from starlette.exceptions import HTTPException as StarletteHTTPException
from starlette.responses import JSONResponse, RedirectResponse, Response
from starlette.types import Scope
from soc_ai import __version__, notify
from soc_ai.agent.orchestrator import build_local_enrichment_context
from soc_ai.api.routes import router
from soc_ai.api.webui_api import open_router as api_v1_open_router
from soc_ai.api.webui_api import router as api_v1_router
from soc_ai.audit.logger import AuditLogger
from soc_ai.bootstrap_credential import bootstrap_credential_path
from soc_ai.config import SERVE_BIND_ENV, get_settings
from soc_ai.hunting.window import sweep_window
from soc_ai.so_client.auth import make_auth
from soc_ai.so_client.elastic import ElasticClient
from soc_ai.store import backtests as bt_svc
from soc_ai.store import chat as chat_svc
from soc_ai.store import general_chat as general_chat_svc
from soc_ai.store import hunt_templates as hunt_templates_svc
from soc_ai.store import hunts as hunt_svc
from soc_ai.store import investigations as inv_svc
from soc_ai.store import leads as leads_store
from soc_ai.store import user_tours
from soc_ai.store.auth import bootstrap_admin, purge_expired_sessions
from soc_ai.store.config_overrides import apply_to_settings, load_overrides
from soc_ai.store.db import describe_store, make_engine, make_sessionmaker, run_migrations
from soc_ai.store.secret_box import make_secret_box
from soc_ai.tools.enrichment import MispClient
_LOGGER = logging.getLogger(__name__)
# The built React SPA. Shipped to the deploy target alongside the package; absent
# in source checkouts until `npm run build` runs, so serving is best-effort.
FRONTEND_DIST = Path(__file__).resolve().parent.parent / "frontend" / "dist"
class SpaStaticFiles(StaticFiles):
"""StaticFiles that falls back to index.html on 404 so client-side
(BrowserRouter) deep links like /app/investigation/INV-1 resolve.
index.html is additionally served ``Cache-Control: no-cache`` (revalidate
every load — NOT "don't cache"): a deploy replaces the content-hashed
``/assets/*`` files, so a browser reusing a stale cached index.html points
at chunk filenames that no longer exist and the SPA dynamic-imports 404
until a hard refresh. The hashed assets themselves keep StaticFiles'
default ETag/Last-Modified caching.
"""
async def get_response(self, path: str, scope: Scope) -> Response:
try:
return await super().get_response(path, scope)
except StarletteHTTPException as exc:
if exc.status_code == 404:
return await super().get_response("index.html", scope)
raise
def file_response(
self,
full_path: str | os.PathLike[str],
stat_result: os.stat_result,
scope: Scope,
status_code: int = 200,
) -> Response:
# Every path that ends up serving index.html funnels through here: the
# direct file, the html-mode directory root (/app/), and the SPA
# fallback above. Setting the header on the returned response also
# covers the 304 NotModifiedResponse branch (revalidation replies must
# keep carrying the policy so clients don't regress to heuristics).
response = super().file_response(full_path, stat_result, scope, status_code)
if Path(full_path).name == "index.html":
response.headers["Cache-Control"] = "no-cache"
return response
def _console_not_found(request: Request, status_code: int) -> bool:
"""Whether a 404 is a browser asking for a page address the server lacks.
Such a request goes to the console, where the SPA shows its not-found
page. An API path, an ``/app`` path, a write, a caller that does not ask
for HTML, and a server with no console build keep the plain 404.
"""
if status_code != 404 or request.method not in ("GET", "HEAD"):
return False
path = request.url.path
if path == "/api" or path.startswith(("/api/", "/app")):
return False
if "text/html" not in request.headers.get("accept", ""):
return False
return bool(getattr(request.app.state, "spa_mounted", False))
async def _reaper_loop(db_sessionmaker: Any, settings: Any) -> None:
"""Periodically mark stale ``running`` investigations + ``pending`` chat
turns as ``error``.
Runs until cancelled at shutdown. A failed iteration is logged and the loop
continues — the reaper must never take the app down. Disabled (no-op loop)
when either investigation knob is <= 0.
The chat sweeps ride the same cadence: they mark ``pending`` assistant rows
older than ``chat_turn_timeout_s`` as ``error`` (a turn still inside its
timeout is legitimately in flight and is spared). They are a backstop for the
in-process timeout/cancel handlers — they catch a turn whose handlers never
ran (e.g. a wedged event loop) or a row a transient DB error left pending.
Investigation chat and dashboard (general) chat are separate tables with
separate reapers, swept in separate try blocks: neither store's sweep may be
skipped because the other one raised, since a table that goes unswept has no
symptom until an analyst is looking at a turn that will never finish.
"""
interval_min = settings.investigation_reaper_interval_minutes
age_min = settings.investigation_reaper_minutes
if interval_min <= 0 or age_min <= 0:
return
# The demo runs no work that can go stale: a replay ends its own row, and a
# demo chat stores nothing. The one row that stays in flight is the seeded
# hunt under the in-progress lead, and the reaper would end it.
if getattr(settings, "soc_ai_demo", False) is True:
return
# A turn that hits its timeout can still write a partial answer from its
# tool results, in at most PARTIAL_ANSWER_TIMEOUT_S more. The sweep waits
# for that too, so it never reaps a turn while the answer is written.
from soc_ai.webui.chat_turn import PARTIAL_ANSWER_TIMEOUT_S # noqa: PLC0415 - lazy
chat_age = timedelta(
seconds=max(int(getattr(settings, "chat_turn_timeout_s", 180)), 1)
+ PARTIAL_ANSWER_TIMEOUT_S
)
# Backtests and hunts legitimately run FAR longer than a single investigation,
# so reaping them at the investigation age (default 30 min) flips an in-flight
# job to 'error' mid-run. Derive each its own reap age from its real ceiling:
# * a backtest replays up to ``backtest_max_sample`` full investigations
# back-to-back, each bounded by ``investigation_run_timeout_s`` — worst-case
# budget is the product. Every backtest is clamped to that sample cap, so
# this age never reaps a legitimately-running one regardless of sample size.
# * a hunt runs to its ``hunt_run_timeout_s`` wall-clock backstop and then
# does partial synthesis — budget that plus a synthesis margin (one age_min).
# ``max(age_min, …)`` keeps each at least as generous as the investigation knob.
inv_run_min = max(int(getattr(settings, "investigation_run_timeout_s", 900)) // 60, 1)
hunt_age_min = max(age_min, int(getattr(settings, "hunt_run_timeout_s", 1800)) // 60 + age_min)
bt_age_min = max(
age_min, int(getattr(settings, "backtest_max_sample", 50)) * inv_run_min + age_min
)
while True:
await asyncio.sleep(interval_min * 60)
try:
async with db_sessionmaker() as db:
n = await inv_svc.reap_stale_running(db, older_than_minutes=age_min)
if n:
_LOGGER.info("reaper: marked %d stale 'running' investigation(s) as error", n)
async with db_sessionmaker() as db:
nh = await hunt_svc.reap_stale_running(db, older_than_minutes=hunt_age_min)
if nh:
_LOGGER.info("reaper: marked %d stale 'running' hunt(s) as error", nh)
async with db_sessionmaker() as db:
nb = await bt_svc.reap_stale_running(db, older_than_minutes=bt_age_min)
if nb:
_LOGGER.info("reaper: marked %d stale 'running' backtest(s) as error", nb)
except asyncio.CancelledError:
raise
except Exception:
_LOGGER.exception("investigation reaper iteration failed; continuing")
try:
async with db_sessionmaker() as db:
nc = await chat_svc.reap_stale_pending(db, older_than=chat_age)
if nc:
_LOGGER.info("reaper: marked %d stale 'pending' chat turn(s) as error", nc)
except asyncio.CancelledError:
raise
except Exception:
_LOGGER.exception("chat reaper iteration failed; continuing")
try:
async with db_sessionmaker() as db:
ng = await general_chat_svc.reap_stale_pending(db, older_than=chat_age)
if ng:
_LOGGER.info(
"reaper: marked %d stale 'pending' dashboard chat turn(s) as error", ng
)
except asyncio.CancelledError:
raise
except Exception:
_LOGGER.exception("dashboard chat reaper iteration failed; continuing")
def _is_loopback_host(host: str) -> bool:
if host.lower() == "localhost":
return True
try:
return ipaddress.ip_address(host).is_loopback
except ValueError:
return False
def auth_off_warning(settings: Any, environ: Mapping[str, str] | None = None) -> str | None:
"""The start line for ``API_AUTH_REQUIRED=false``, or None when authentication is on.
The line states what the app knows, and nothing more. ``soc-ai serve``
writes its bind into :data:`SERVE_BIND_ENV` before it starts uvicorn, so on
that path the line names the real bind. Under a bare uvicorn command (the
systemd unit, the container image) the bind is a uvicorn argument the app
cannot see. The old line read ``SOC_AI_HOST`` there, which defaults to
127.0.0.1, and said "loopback bind 127.0.0.1" while the unit passed
``--host 0.0.0.0`` and the port was open to the network (range dogfood,
2026-10-05). The network branch could never fire on the systemd path.
"""
if settings.api_auth_required:
return None
env = os.environ if environ is None else environ
bind = (env.get(SERVE_BIND_ENV) or "").strip()
head = (
"API_AUTH_REQUIRED=false. Authentication is off. A caller that reaches the port "
"can use the admin routes with no login."
)
if not bind:
return (
f"{head} The bind is the server's. soc-ai cannot see it. Run `soc-ai doctor` "
"to list the addresses that listen on the port. Set API_AUTH_REQUIRED=true "
"for a shared deployment."
)
host = bind.rsplit(":", 1)[0].strip("[]")
if _is_loopback_host(host):
return (
f"{head} The server binds to {bind}. Only this host can reach it. Set "
"API_AUTH_REQUIRED=true before you open the port."
)
return (
f"{head} The server binds to {bind}. Other hosts can reach that address. Set "
"API_AUTH_REQUIRED=true for a shared deployment."
)
def _discovery_due(last_scan_iso: str | None, interval_hours: int) -> bool:
"""True iff a scheduled discovery scan is due.
Due when there has been no scan this process (`last_scan_iso is None`) or at
least `interval_hours` have elapsed since the last completed scan. A bad/
unparseable timestamp is treated as 'due' (fail toward running the scan).
"""
if last_scan_iso is None:
return True
try:
last = datetime.fromisoformat(last_scan_iso)
except ValueError:
return True
return (datetime.now(UTC) - last) >= timedelta(hours=interval_hours)
def _dossier_due(last_run_iso: str | None, interval_hours: int) -> bool:
"""True iff a scheduled host-dossier sweep is due.
The elapsed-time maths is :func:`_discovery_due`'s; what differs is where
``last_run_iso`` comes from. It is the newest ``dossier_run.started_at`` —
a DURABLE stamp — not an ``app.state`` field. An in-memory stamp is ``None``
on every boot and ``None`` reads as due, so a crash-restart loop would
re-sweep the whole network (hundreds of hosts, several ES round trips each) on
each boot. Read from the run table, ``None`` means "no sweep has ever
started", which genuinely is due.
"""
return _discovery_due(last_run_iso, interval_hours)
def _audit_verify_due(last_run_iso: str | None, interval_hours: int) -> bool:
"""True iff a scheduled audit-chain verification is due.
:func:`_discovery_due`'s elapsed-time maths against an ``app.state`` stamp.
In-memory on purpose, unlike the dossier's durable stamp: a restart
re-verifying is a bounded read of one index, and the failure this schedule
exists to catch is precisely the kind that hides behind "we already
checked".
"""
return _discovery_due(last_run_iso, interval_hours)
def _eval_nightly_due(
now: datetime,
*,
hour_utc: int,
last_scheduled_date: str | None,
latest_snapshot_date: str | None,
last_attempt_date: str | None = None,
) -> bool:
"""True iff the in-app nightly quality eval should run at *now*.
Runs at most once per UTC day, at the first wake at/after ``hour_utc``.
Three once-per-day guards, and each covers a case the others do not:
* ``last_scheduled_date`` — in memory, claimed the instant this process
schedules a run, so two wakes a few minutes apart cannot both fire.
* ``latest_snapshot_date`` — durable, and covers a restart after a run
that WROTE a point, plus a host-cron run that already landed today's.
* ``last_attempt_date`` — durable, and covers the gap between those two:
an exit-2 or exit-5 night writes no snapshot, so before this table the
only guard on a night the nightly failed was the in-memory one, and a
restart loop could re-run it repeatedly on exactly the nights something
was already wrong.
"""
today = now.date().isoformat()
if last_scheduled_date == today:
return False
if last_attempt_date == today:
return False
if now.hour < hour_utc:
return False
return latest_snapshot_date != today
async def _hunt_spec_sweep_loop(app: FastAPI) -> None:
"""Run the declarative hunt catalog on a loop (proactive-hunting slice 3).
The seventh lifespan task, and the one with the least in it: a sweep costs
no model call at all. It compiles each spec to two bounded Elasticsearch
queries, drops every condition the fire-once gate has already handled, and
records what survives as a ``Hunt(kind="triggered")``. Nothing generative
runs, so a finding from this path cannot hallucinate.
Mirrors :func:`_hunt_schedule_loop`'s discipline — fixed 60s wake, settings
read live each wake so a config-console toggle applies without a restart,
a no-op unless ``hunt_spec_sweeps_enabled`` — and differs in three ways
that follow from costing nothing:
- the interval floor is 5 minutes rather than 60, because two aggregations
per spec is not an LLM hunt;
- the LOOK-BACK window is wider than the interval (24h by default), so a
condition that landed during an outage, a restart or an ingest lag is
still seen. That overlap is deliberate and is exactly what a live-tailing
rule engine cannot do — the fire-once gate is what stops it becoming
repeat findings;
- there is no per-spec schedule row, so the single-flight is a module-level
guard on the whole sweep rather than per-item state.
A spec that fails is reported inside :func:`sweep_catalog` and never stops
the rest of the catalog. A sweep that fails as a whole is logged and the
loop continues; the next wake re-reads a fresh window.
Note (workers>1): a second uvicorn worker would run its own copy and double
the query load. The gate's unique constraint means it could not double the
FINDINGS, which is the part that matters, but soc-ai runs a single worker
today and distributed coordination is Epoch 6.2, as for the loop above.
"""
from soc_ai.hunting.catalog_tiers import effective_catalog # noqa: PLC0415
from soc_ai.hunting.sweep import sweep_catalog # noqa: PLC0415
last_run: datetime | None = None
while True:
await asyncio.sleep(60)
try:
settings = app.state.settings
if not getattr(settings, "hunt_spec_sweeps_enabled", False):
continue
# The floor and the clamp live in soc_ai.hunting.window, shared
# with `soc-ai spec-sweep` and the catalog route, so a hand-run
# sweep covers exactly what this loop covers and the panel reports
# the same number the trail rows record.
window = sweep_window(settings)
now = datetime.now(UTC).replace(tzinfo=None)
if last_run is not None and (now - last_run) < timedelta(
minutes=window.interval_minutes
):
continue
# Said once per sweep, after the interval check, so a
# misconfiguration is visible without being repeated every wake.
window.say_if_widened(_LOGGER)
# The effective catalog, not the files alone: a retired analytic
# must stop running and a local one in shadow must start, and both
# facts live in the database. ``seed`` writes the shadow row of a
# shipped analytic that ships in shadow, so an approval has a row
# to move.
async with app.state.db_sessionmaker() as db:
tiers = await effective_catalog(db, seed=True)
catalog = tiers.specs
if not catalog:
continue
async with app.state.db_sessionmaker() as db:
result = await sweep_catalog(
catalog,
session=db,
elastic=app.state.elastic,
settings=settings,
since=window.since,
until="now",
now=now,
shadow_ids=tiers.shadow_ids,
)
await db.commit()
# Stamped only after a completed sweep, so a crash mid-sweep retries
# on the next wake rather than skipping a whole interval.
last_run = now
# The hits this sweep wrote are in the store now. The hold reads
# them against each live analytic's budget.
await _run_self_heal(app, settings)
if result.hunts or result.blind or result.errored or result.gaps_cleared:
_LOGGER.info(
"spec sweep: %d hunt(s) from %d fresh candidate(s); "
"%d already handled, %d blind, %d errored",
len(result.hunts),
result.fresh_candidates,
result.already_handled,
len(result.blind),
len(result.errored),
)
# Its own line, and its own condition rather than a clause on the
# summary above: a sweep whose only news is a recovery has no
# hunts, nothing blind and nothing errored, so before this the
# summary did not fire for it and an outage ENDING was the quietest
# event in the log. INFO, not WARNING — the transition the other
# way is reported at INFO too, and a "things are better now" line
# at WARNING is one anybody alerting on the level would misread.
if result.gaps_cleared:
_LOGGER.info(
"spec sweep: %d spec(s) can see their telemetry again; "
"the visibility gap recorded for them is closed",
result.gaps_cleared,
)
for spec_id, err in result.errored.items():
_LOGGER.warning("spec sweep: %s failed against the grid: %s", spec_id, err)
except asyncio.CancelledError:
raise
except Exception as exc:
_LOGGER.warning("spec sweep loop: %s: %s", type(exc).__name__, exc)
async def _run_self_heal(app: FastAPI, settings: Any) -> None:
"""Run the self-healing hold after a completed sweep. Never raises.
Both sweep loops call it, so a deployment with only one loop on still
checks every live analytic. The check reads the store and writes only a
demotion. One run at a time: the two loops share a lock on ``app.state``,
so they cannot both read an analytic as live and both demote it.
"""
if not getattr(settings, "analytic_self_heal_enabled", True):
return
from soc_ai.hunting.self_heal import run_self_heal # noqa: PLC0415
lock = getattr(app.state, "self_heal_lock", None)
if lock is None:
lock = asyncio.Lock()
app.state.self_heal_lock = lock
try:
from soc_ai.store.grid_gaps import settle_of # noqa: PLC0415
async with lock, app.state.db_sessionmaker() as db:
result = await run_self_heal(db, settle=settle_of(settings))
except asyncio.CancelledError:
raise
except Exception as exc:
_LOGGER.warning("self-healing hold: %s: %s", type(exc).__name__, exc)
return
for analytic_id, why in result.skipped.items():
_LOGGER.info("self-healing hold: %s left as it is: %s", analytic_id, why)
# The shortest interval the profile sweep will accept. A sweep makes no model
# call, so the floor is far below the hour an LLM schedule needs. It is not
# zero: each sweep is several real aggregations per dimension against the
# analyst's grid.
PRIOR_SWEEP_MIN_INTERVAL_MINUTES = 15
def _prior_sweep_interval_minutes(settings: Any) -> int:
"""Minutes between profile sweeps, with the floor applied.
The console bounds the setting, and this bounds it again. An override
written before the bound existed, or a settings object built in code,
reaches the loop without passing the console at all.
"""
raw = getattr(settings, "hunting_prior_sweep_interval_minutes", 60)
try:
minutes = int(raw or 0)
except (TypeError, ValueError):
minutes = 0
return max(PRIOR_SWEEP_MIN_INTERVAL_MINUTES, minutes)
def _grid_is_known_down(app: FastAPI) -> bool:
"""Has the health probe loop seen Elasticsearch fail and not recover?
Reads the map ``_note_dep_transitions`` keeps on ``app.state``, so the
answer costs no query. Unknown reads as up: a process that has not probed
yet must sweep rather than wait.
"""
down = getattr(app.state, "_dep_down_since", None) or {}
return "es" in down
def _note_prior_skip(app: FastAPI, settings: Any) -> None:
"""Record a due profile sweep that the loop skipped because the grid is down.
The loop recorded nothing on a skip, so four days of outage read as a
quiet sweep. The record counts one skip per interval, and only when a run
is due: a wake inside the interval had no run to skip.
"""
from soc_ai.hunting.loop_health import SKIP_GRID_DOWN, prior_health # noqa: PLC0415
now = datetime.now(UTC).replace(tzinfo=None)
every = timedelta(minutes=_prior_sweep_interval_minutes(settings))
last_run = getattr(app.state, "prior_sweep_last_run", None)
if last_run is not None and now - last_run < every:
return
prior_health(app.state).note_skip(now, SKIP_GRID_DOWN, every)
def _profile_stale_after(settings: Any) -> timedelta:
"""How old a baseline may be before the prior sweep rebuilds it.
The dossier interval, floored at two prior-sweep intervals. A rebuild on
every wake would double the query load for baselines that change on the
timescale of a provisioning ticket, not a shift.
"""
try:
hours = int(getattr(settings, "dossier_schedule_interval_hours", 24) or 24)
except (TypeError, ValueError):
hours = 24
floor = 2 * _prior_sweep_interval_minutes(settings)
return timedelta(minutes=max(hours * 60, floor))
def _profiles_are_stale(newest: datetime | None, now: datetime, threshold: timedelta) -> bool:
"""Never built, older than the threshold, or stamped in the future.
The future case is a row written in local time east of UTC by the release
that stamped ``datetime.now()``. Rebuilding it is what ends the skew; no
migration can, because no migration knows the host's offset.
"""
if newest is None:
return True
return newest > now or (now - newest) >= threshold
def _profile_age(newest: datetime | None, now: datetime) -> str:
if newest is None:
return "never built"
hours = max(0, int((now - newest).total_seconds() // 3600))
return f"{hours} h old"
def _unmeasurable_reason(unmeasurable: dict[str, str]) -> str | None:
joined = "; ".join(f"{d}: {r}" for d, r in sorted(unmeasurable.items()))
return joined[:255] or None
async def _refresh_profiles_if_stale(app: FastAPI, settings: Any, now: datetime) -> Any:
"""Rebuild the baselines the sweep is about to read, when they are stale.
Returns the ``ProfileState`` the sweep records, or ``None`` when profiles
are off. Runs whether or not the dossier schedule is on: the sweep is the
only consumer of the baselines, so it owns their freshness. The rebuild
takes the dossier's single-flight slot, because two network sweeps at once
is the connection-pool pressure that has frozen this app before; if the
slot is held the rebuild waits for the next wake.
Never raises. A rebuild that fails is logged and the sweep runs on what
exists, reporting it stale.
"""
from soc_ai.api.webui import _get_dossier_status # noqa: PLC0415
from soc_ai.dossier import profile_job # noqa: PLC0415
from soc_ai.hunting.prior_sweep import ProfileState # noqa: PLC0415
from soc_ai.oracle.identifiers import effective_internal_identifiers # noqa: PLC0415
if not getattr(settings, "entity_profiles_enabled", False):
return None
before = await profile_job.freshness(app.state.db_sessionmaker)
age = _profile_age(before.newest_built_at, now)
reason = _unmeasurable_reason(before.unmeasurable)
# A release that changes the stored profile shape makes a build due at
# once, whatever the age of the rows. Without this, a live analytic that
# reads the new shape stays blind on every host until the age rule fires,
# most of a day after the upgrade (production and the range, 2026-10-05).
reshape = profile_job.shape_due(before)
if reshape is None and not _profiles_are_stale(
before.newest_built_at, now, _profile_stale_after(settings)
):
_LOGGER.info("prior sweep: profiles %s", age)
return ProfileState(built_at=before.newest_built_at, stale=False, reason=reason)
if reshape:
_LOGGER.info("prior sweep: profiles %s; %s", age, reshape)
status = _get_dossier_status(app.state)
if status.running:
_LOGGER.info(
"prior sweep: profiles %s; a dossier sweep holds the slot, rebuild skipped this wake",
age,
)
return ProfileState(built_at=before.newest_built_at, stale=True, reason=reason)
status.running = True
try:
async with app.state.db_sessionmaker() as db:
cidrs = (await effective_internal_identifiers(db, settings)).cidrs
build = await profile_job.build_profiles(
app.state.elastic, app.state.db_sessionmaker, settings, cidrs
)
except Exception as exc:
_LOGGER.warning("prior sweep: profile rebuild failed: %s: %s", type(exc).__name__, exc)
return ProfileState(built_at=before.newest_built_at, stale=True, reason=reason)
finally:
status.running = False
for err in build.errors:
_LOGGER.warning("prior sweep: %s", err)
after = await profile_job.freshness(app.state.db_sessionmaker)
_LOGGER.info("prior sweep: profiles %s, rebuilt %d row(s)", age, build.written)
return ProfileState(
built_at=after.newest_built_at,
stale=False,
reason=_unmeasurable_reason(after.unmeasurable),
)
async def _prior_sweep_loop(app: FastAPI) -> None:
"""Run the profile sweep on a loop: the hunting layer's supply of observations.
This is the code path ``soc-ai priors --record`` runs. The test range runs
it from a host timer every hour. A container has no timer, so before this
loop a production deployment shipped the profile layer and never ran it:
no observation was recorded, no profile lead formed, and every surface
stayed green. A job nobody scheduled is the same false all-clear as a
surface that lies, arrived at from the supply side.
Mirrors :func:`_hunt_spec_sweep_loop`'s discipline — fixed 60 s wake,
settings read live each wake so a config-console toggle applies without a
restart, an interval floor, one line in the log per run — and differs in
three ways:
- it is ON by default. A catalog sweep writes findings, so it is opt-in. A
profile sweep writes observations and raises nothing, and a deployment
with no profile observation has no hunting layer;
- it skips a demo and a grid the health probe already knows is down. Both
write the same damage: a sweep that could not measure records that
nothing departed, which is a baseline learning from an outage;
- it owns the freshness of the baselines it reads. If profiles are on and
the newest is missing, older than the dossier interval (floored at two
sweep intervals) or stamped in the future, it rebuilds them first, under
the dossier's single-flight slot, whatever the dossier schedule says. A
rebuild that fails is logged and the sweep runs on what exists.
The sweep itself never raises: it returns its errors, and they are logged
beside the counts. A sweep that fails another way is logged and the loop
continues; the last-run stamp is written only after a sweep that returned,
so a failure retries on the next wake rather than costing a whole interval.
Note (workers>1): a second uvicorn worker would run its own copy and
double the query load. soc-ai runs a single worker today and distributed
coordination is Epoch 6.2, as for the loops above.
"""
from soc_ai.hunting.catalog_tiers import effective_catalog # noqa: PLC0415
from soc_ai.hunting.loop_health import OUTCOME_FAILED, prior_health # noqa: PLC0415
from soc_ai.hunting.prior_sweep import run_prior_sweep # noqa: PLC0415
from soc_ai.hunting.priors import COVERAGE_BLIND # noqa: PLC0415
from soc_ai.oracle.identifiers import effective_internal_identifiers # noqa: PLC0415
app.state.prior_sweep_last_run = None
while True:
await asyncio.sleep(60)
try:
settings = app.state.settings
if not getattr(settings, "hunting_prior_sweep_enabled", True):
continue
if getattr(settings, "soc_ai_demo", False):
continue
if _grid_is_known_down(app):
_note_prior_skip(app, settings)
continue
now = datetime.now(UTC).replace(tzinfo=None)
last_run = getattr(app.state, "prior_sweep_last_run", None)
if last_run is not None and (now - last_run) < timedelta(
minutes=_prior_sweep_interval_minutes(settings)
):
continue
# The baselines first. The sweep reads them, so their freshness
# is the sweep's job, not a second schedule's.
profiles = await _refresh_profiles_if_stale(app, settings, now)
async with app.state.db_sessionmaker() as db:
# The estate's own address space, so the sweep does not record
# observations about the internet and form leads out of them.
cidrs = (await effective_internal_identifiers(db, settings)).cidrs
# The effective catalog, for the same reason the spec sweep
# reads it: a retired prior must stop running and a local one
# in shadow must start.
tiers = await effective_catalog(db, seed=True)
sweep = await run_prior_sweep(
elastic=app.state.elastic,
settings=settings,
db=db,
record=True,
cidrs=cidrs,
catalog=tiers.specs,
shadow_ids=tiers.shadow_ids,
profiles=profiles,
)
await db.commit()
# Stamped only after a sweep that returned, so a failure retries on
# the next wake rather than skipping the whole interval.
app.state.prior_sweep_last_run = now
prior_health(app.state).note_sweep(now, sweep)
await _run_self_heal(app, settings)
observations = sum(len(r.departures) + len(r.hits) for r in sweep.results)
leads = len(sweep.leads.formed) if sweep.leads is not None else 0
blind = sweep.coverage_counts().get(COVERAGE_BLIND, 0)
# Said every run, not only when something departed. A quiet sweep
# is how an operator knows the loop is alive, and the blind count
# is what separates "nothing departed" from "nothing could be
# measured" — the same empty list, and completely different news.
_LOGGER.info(
"prior sweep: %d observation(s), %d lead(s), %d blind",
observations,
leads,
blind,
)
for err in sweep.errors:
_LOGGER.warning("prior sweep: %s", err)
except asyncio.CancelledError:
raise
except Exception as exc:
_LOGGER.warning("prior sweep loop: %s: %s", type(exc).__name__, exc)
prior_health(app.state).note(
datetime.now(UTC).replace(tzinfo=None),
OUTCOME_FAILED,
f"{type(exc).__name__}: {exc}",
)
async def _eval_nightly_loop(app: FastAPI, settings: Any) -> None:
"""Run the nightly quality micro-eval in-app when scheduled.
The nightly used to be host-cron-only; ``eval_nightly_enabled`` makes it
schedulable from the UI (Config → Quality). Models the discovery loop:
fixed wake cadence, live settings read each wake (console toggles apply
without a restart), and the run-now single-flight ``_QualityEvalStatus``
shared with POST /quality/eval/run so a scheduled run and a manual run
can never overlap. A failed iteration is logged and the loop continues.
"""
# Lazy import: reuse the run-now single-flight + worker (one direction).
from soc_ai.api.webui_api import ( # noqa: PLC0415
_get_quality_eval_status,
_quality_eval_worker,
)
wake_seconds = 300
while True:
await asyncio.sleep(wake_seconds)
try:
if not settings.eval_nightly_enabled:
continue
status = _get_quality_eval_status(app.state)
if status.running:
continue
now = datetime.now(UTC)
# Durable freshness check — fail-soft toward "no snapshot today"
# (running twice is cheaper than silently never running).
latest_snapshot_date: str | None = None
last_attempt_date: str | None = None
try:
from soc_ai.store import quality as quality_store # noqa: PLC0415
async with app.state.db_sessionmaker() as db:
rows = await quality_store.recent_snapshots(db, limit=1)
attempt = await quality_store.latest_attempt(db)
if rows:
latest_snapshot_date = rows[0].created_at.date().isoformat()
if attempt is not None:
last_attempt_date = attempt.attempted_at.date().isoformat()
except Exception:
_LOGGER.warning(
"eval-nightly: snapshot freshness check failed (continuing)", exc_info=True
)
if not _eval_nightly_due(
now,
hour_utc=settings.eval_nightly_hour_utc,
last_scheduled_date=status.last_scheduled_date,
latest_snapshot_date=latest_snapshot_date,
last_attempt_date=last_attempt_date,
):
# A point already landed today (host cron / pre-restart run):
# consume the day so later wakes skip the DB check too.
if latest_snapshot_date == now.date().isoformat():
status.last_scheduled_date = latest_snapshot_date
continue
status.last_scheduled_date = now.date().isoformat()
status.running = True # claim the single-flight slot before scheduling
status._task = asyncio.create_task(_quality_eval_worker(app.state, trigger="schedule"))
except asyncio.CancelledError:
raise
except Exception:
_LOGGER.exception("eval-nightly scheduler iteration failed; continuing")
async def _health_probe_loop(app: FastAPI, settings: Any) -> None:
"""Probe the upstreams on a schedule, so an outage is noticed with no tab open.
Every health surface in this app is pull-only behind a TTL cache, and
``_note_dep_transitions`` — the code that records when a dependency went
down, and the only thing that puts a dependency-down entry on the bell —
runs solely as a side effect of a client polling ``/api/v1/health``.
So an open browser tab was the scheduler. Elasticsearch could be down all
night, come back before anyone looked, and nothing anywhere would know it
had happened: no bell entry, no transition recorded, no log. That is the
same false-quiet this codebase keeps finding, arrived at from the other
direction — not a surface that lies, a surface nobody asked.
Cheap by construction. It calls the same cached probe the endpoint does, so
a wake inside the TTL costs nothing and a polling client and this loop share
one result rather than doubling the load. PCAP is deliberately not probed:
it is a heavy SSH round trip on a much longer TTL and nothing on the bell
keys on it.
"""
from soc_ai.api.webui.routes_meta import _cached_health_probes # noqa: PLC0415 - lazy
wake_seconds = 60
while True:
await asyncio.sleep(wake_seconds)
try:
await _cached_health_probes(app.state, settings)
except Exception:
# A probe that could not run is not a dependency being down, and
# must not be recorded as one. probe_* never raise; this guards the
# loop against anything else so a single bad wake cannot kill the
# only scheduled health check in the process.
_LOGGER.warning("scheduled health probe failed to run", exc_info=True)
async def _audit_verify_loop(app: FastAPI, settings: Any) -> None:
"""Periodically verify the tamper-evident audit chain, and alarm on a break.
A tamper-evident log nobody verifies is a log. The chain has been
checkable since v1 — ``soc-ai audit verify``, and the admin verify-chain
diagnostic — and in practice nothing ever ran either, so a live deployment
carried a broken current epoch for weeks with nothing saying so. This loop
is the thing that says so.
Models :func:`_discovery_scheduler_loop`: fixed wake, live settings read
each wake so a console toggle applies without a restart, and a failed
iteration logged and swallowed. The scan runs inline rather than in a
spawned task — it is a bounded ES read, not a model run — so the loop
cannot overlap itself and needs no single-flight slot.
Three outcomes, three different responses:
- broken — an audit record (kind ``audit_chain_verification``, into the
very chain it is reporting on) plus the opt-in notification webhook plus
a standing entry on the in-app bell. Same two channels the nightly
quality regression uses, and the bell besides, because notifications are
off by default and a tamper finding that reaches nobody on a default
install is the failure this loop exists to end.
- intact — nothing is sent and any standing alarm is cleared, so the bell
entry disappears on its own once the trail is sound again.
- could not run (unreachable or half-read index) — logged, never alarmed.
A verification that did not happen is not a tamper finding, and treating
it as one would teach the operator to ignore the alarm that matters.
Until a fix lands for whatever is breaking the chain, this reports the
break on every run. That is the honest state; it is not suppressed.
"""
from soc_ai import notify # noqa: PLC0415 - lazy
from soc_ai.api.webui.routes_config import _get_audit_verify_status # noqa: PLC0415
from soc_ai.audit.verify import verify_audit_chain # noqa: PLC0415
wake_seconds = 300
while True:
await asyncio.sleep(wake_seconds)
try:
# Re-read every wake → console toggle / interval edits apply live.
if not settings.audit_verify_schedule_enabled:
continue
status = _get_audit_verify_status(app.state)
if not _audit_verify_due(
status.last_run, settings.audit_verify_schedule_interval_hours
):
continue
elastic = getattr(app.state, "elastic", None)
if elastic is None:
continue
days = max(1, int(settings.audit_verify_days))
try:
result = await verify_audit_chain(elastic, settings.audit_index_alias, days=days)
except Exception:
# Could not run. NOT a tamper finding — see the docstring.
_LOGGER.warning(
"scheduled audit-chain verification could not run (the audit "
"index was unreachable or only partly readable); no verdict "
"either way",
exc_info=True,
)
status.last_run = datetime.now(UTC).isoformat()
status.last_outcome = "not_checked"
continue
status.last_run = datetime.now(UTC).isoformat()
status.last_outcome = "intact" if result.ok else "broken"
if result.ok:
status.alarm = None
_LOGGER.info(
"audit chain verified: %d records, %d epoch(s), last %dd",
result.records_verified,
result.epochs,
days,
)
continue
await _raise_audit_chain_alarm(app, settings, result, status, days, notify)
except asyncio.CancelledError:
raise
except Exception:
_LOGGER.exception("audit chain verification scheduler iteration failed; continuing")
async def _raise_audit_chain_alarm(
app: FastAPI,
settings: Any,
result: Any,
status: Any,
days: int,
notify: Any,
) -> None:
"""Put one chain-break finding in front of a human, three ways.
Split out of the loop so each channel's failure is contained: the bell
entry is set first and from memory, so it survives an audit index that
cannot take the record and a webhook that will not answer.
Every channel carries the blast radius, not one sequence number out of
however many broke, and the bell entry carries an identity for the FINDING
rather than for the moment it was noticed, so a standing alarm can be
dismissed without a new break inheriting the dismissal. See
:func:`~soc_ai.audit.verify.finding_key` for what that identity is made of
and :func:`~soc_ai.audit.verify.finding_is_dismissible` for the one kind of
finding that is never offered as dismissible.
"""
from soc_ai.audit.verify import ( # noqa: PLC0415 - lazy
describe_blast_radius,
finding_is_dismissible,
finding_key,
)
blast_radius = describe_blast_radius(result)
payload: dict[str, Any] = {
"window_days": days,
"records_verified": result.records_verified,
"epochs": result.epochs,
"epochs_broken": result.epochs_broken,
"latest_epoch_broken": result.latest_epoch_broken,
"capped": result.capped,
"break_kind": result.newest_break_kind,
"break_seq": result.first_broken_seq,
"break_detail": result.newest_break_detail,
"newest_broken_epoch_start": result.newest_broken_epoch_start,
"break_kinds": list(result.break_kinds),
"duplicate_seqs": result.duplicate_seqs,
"extra_records": result.extra_records,
"max_claimants": result.max_claimants,
"altered_records": result.altered_records,
"missing_seqs": result.missing_seqs,
"oldest_break_at": result.oldest_break_at,
"newest_break_at": result.newest_break_at,
"blast_radius": blast_radius,
}
now = datetime.now(UTC).isoformat()
key = finding_key(result)
previous = status.alarm if isinstance(status.alarm, dict) else None
# The clock the bell renders is when THIS finding was first seen, not when
# it was last looked at, so a standing scar does not read as "just now"
# every morning.
since = previous.get("alarm_since") if previous and previous.get("alarm_key") == key else None
status.alarm = dict(
payload,
detected_at=now,
alarm_key=key,
alarm_since=since or now,
dismissible=finding_is_dismissible(result),
)
_LOGGER.error(
"AUDIT CHAIN BROKEN: %s (%d of %d epochs, last %dd)",
blast_radius or result.newest_break_detail or "the chain does not verify",
result.epochs_broken,
result.epochs,
days,
)
audit = getattr(app.state, "audit", None)
if audit is not None:
try:
await audit.log_kind(
session_id="audit-chain-verify",
kind="audit_chain_verification",
payload=payload,
)
except Exception:
_LOGGER.warning("audit-chain-break record could not be written", exc_info=True)
event = notify.event_for_audit_chain_break(
epochs_broken=result.epochs_broken,
seq=result.first_broken_seq,
kind=result.newest_break_kind,
detail=result.newest_break_detail,
latest_epoch_broken=result.latest_epoch_broken,
settings=settings,
blast_radius=blast_radius,
)
if event is not None:
await notify.fire_safe(event, settings, audit)
@dataclass
class _TlsStatusSlot:
"""The standing TLS expiry alarm. In memory, like the audit-chain alarm."""
last_check: str | None = None
alarm: dict[str, Any] | None = None
def _get_tls_status(state: Any) -> _TlsStatusSlot:
slot = getattr(state, "_tls_status", None)
if not isinstance(slot, _TlsStatusSlot):
slot = _TlsStatusSlot()
state._tls_status = slot
return slot
def _utcnow_for_tls() -> datetime:
return datetime.now(UTC)
_TLS_STATE_FILE = "tls-expiry-state.json"
"""The last TLS expiry webhook, as (fingerprint, band). A restart must not repeat it."""
def _tls_state_path(settings: Any) -> Path | None:
data_dir = getattr(settings, "soc_ai_data_dir", None)
return Path(data_dir) / _TLS_STATE_FILE if data_dir else None
def _tls_last_notified(settings: Any) -> tuple[str, int] | None:
"""The (fingerprint, band) pair of the last webhook. None when nothing fired yet."""
path = _tls_state_path(settings)
if path is None:
return None
try:
data = json.loads(path.read_text(encoding="utf-8"))
return str(data["fingerprint"]), int(data["band"])
except (OSError, ValueError, KeyError, TypeError):
return None
def _tls_remember_notified(settings: Any, fingerprint: str, band: int) -> None:
path = _tls_state_path(settings)
if path is None:
return
try:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps({"fingerprint": fingerprint, "band": band}), encoding="utf-8")
except OSError:
_LOGGER.warning("Cannot write %s. A restart may repeat the TLS expiry webhook.", path)
async def _tls_expiry_check(app: Any, settings: Any) -> None:
"""One check of the certificate soc-ai serves. Fires once per band.
The served certificate is ``app.state.tls_at_start``. soc-ai loads the
files once, so a swap on disk changes nothing until a restart. The alarm
is keyed by the served fingerprint and the band. It notes whether the
disk differs, and only a restart with a new certificate clears it.
"""
from soc_ai.tls_status import TlsStatus, days_left_at, expiry_band, inspect_tls # noqa: PLC0415
slot = _get_tls_status(app.state)
now = _utcnow_for_tls()
cert_path = getattr(settings, "soc_ai_tls_cert", None)
key_path = getattr(settings, "soc_ai_tls_key", None)
served = getattr(app.state, "tls_at_start", None)
if not isinstance(served, TlsStatus):
served = inspect_tls(cert_path, key_path, now=now)
app.state.tls_at_start = served
on_disk = inspect_tls(cert_path, key_path, now=now)
slot.last_check = now.isoformat()
fingerprint = served.fingerprint_sha256
if served.mode == "off" or not served.not_after or not fingerprint:
slot.alarm = None
return
days_left = days_left_at(served.not_after, now)
band = expiry_band(days_left)
if band is None:
slot.alarm = None
return
previous = slot.alarm or {}
same = previous.get("band") == band and previous.get("fingerprint") == fingerprint
slot.alarm = {
"band": band,
"days_left": days_left,
"subject": served.subject or "",
"since": previous.get("since") if same else now.isoformat(),
"fingerprint": fingerprint,
"disk_differs": on_disk.fingerprint_sha256 != fingerprint,
}
if same or _tls_last_notified(settings) == (fingerprint, band):
return
event = notify.event_for_tls_expiry(
subject=served.subject or "",
days_left=days_left,
band=band,
settings=settings,
)
if event is not None:
await notify.fire_safe(event, settings)
_tls_remember_notified(settings, fingerprint, band)
async def _tls_expiry_loop(app: Any, settings: Any) -> None:
"""Check once at start, then once a day."""
wake_seconds = 3600
while True:
try:
slot = _get_tls_status(app.state)
if _discovery_due(slot.last_check, 24):
await _tls_expiry_check(app, settings)
except Exception:
_LOGGER.exception("tls expiry check failed")
await asyncio.sleep(wake_seconds)
async def _discovery_scheduler_loop(app: FastAPI, settings: Any) -> None:
"""Periodically run the internal-identifier discovery scan when scheduled.
Runs until cancelled at shutdown. Models `_reaper_loop`: wake on a fixed
cadence, read the live settings each wake (so a config-console toggle takes
effect without a restart), and no-op unless scheduling is enabled. Shares the
scan-now single-flight `_DiscoveryStatus` on `app.state`, so a scheduled run
and a manual 'Scan now' can never overlap. A failed iteration is logged and
the loop continues — the scheduler must never take the app down.
"""
# Lazy import: reuse the scan-now single-flight + worker (one direction).
from soc_ai.api.webui_api import ( # noqa: PLC0415
_get_discovery_status,
_run_discovery_task,
)
# Fixed wake cadence (minutes). Decoupled from the hours-granularity
# interval so a freshly-toggled-on schedule starts within a wake, and the
# interval check itself is what enforces the spacing. 5 min is cheap (the
# body is a cheap settings read + timestamp compare unless a scan is due).
wake_seconds = 300
while True:
await asyncio.sleep(wake_seconds)
try:
# Re-read every wake → GUI toggle / interval edits apply live.
if not settings.discovery_schedule_enabled or not settings.discovery_enabled:
continue
status = _get_discovery_status(app.state)
if status.running:
continue # a scan (manual or scheduled) is already in flight
if not _discovery_due(status.last_scan, settings.discovery_schedule_interval_hours):
continue
status.running = True # claim the single-flight slot before scheduling
status._task = asyncio.create_task(_run_discovery_task(app.state))
except asyncio.CancelledError:
raise
except Exception:
_LOGGER.exception("discovery scheduler iteration failed; continuing")
async def _dossier_scheduler_loop(app: FastAPI, settings: Any) -> None:
"""Periodically refresh the host dossier (the network's asset context).
Models :func:`_discovery_scheduler_loop` in every respect — fixed wake
cadence, live settings read each wake (a config-console toggle applies
without a restart), the shared "Rebuild now" single-flight slot so a manual
and a scheduled sweep can never overlap, and a logged-and-swallowed failed
iteration because a scheduler must never take the app down.
The one deliberate difference is the due-check's input. Discovery compares
against an ``app.state`` timestamp; this loop reads the newest
``dossier_run`` row, because an in-memory stamp is ``None`` after every
restart and ``None`` reads as due — a restart loop would then re-sweep the
entire network on each boot. The durable stamp is read once and cached onto
the status object, so every later wake stays a settings read and a timestamp
compare.
A failing durable read degrades toward sweeping rather than toward never
sweeping again: the sweep itself opens a ``dossier_run`` row against the same
database, so a genuinely broken store returns immediately, and the worker
stamps ``last_run`` in its ``finally`` regardless — there is no retry storm
behind the fail-soft.
"""
# Lazy import: reuse the rebuild-now single-flight + worker (one direction).
from soc_ai.api.webui import _get_dossier_status, _run_dossier_task # noqa: PLC0415
from soc_ai.enrichment.host_dossier import latest_run_started_at # noqa: PLC0415
wake_seconds = 300
while True:
await asyncio.sleep(wake_seconds)
try:
# Re-read every wake → GUI toggle / interval edits apply live.
if not settings.dossier_enabled or not settings.dossier_schedule_enabled:
continue
status = _get_dossier_status(app.state)
if status.running:
continue # a sweep (manual or scheduled) is already in flight
if status.last_run is None:
try:
stamp = await latest_run_started_at(app.state.db_sessionmaker)
except Exception:
_LOGGER.warning(
"host dossier: durable last-run read failed; treating the sweep "
"as due (continuing)",
exc_info=True,
)
stamp = None
if stamp is not None:
status.last_run = stamp.isoformat()
# RE-READ the shared slot: the check above happened before the
# await, and a "Rebuild now" landing inside that window would
# otherwise get a second network sweep started on top of it —
# hundreds of hosts times several ES round trips, twice, which
# is the connection-pool pressure that has frozen this app
# before. (Read through the accessor, not the narrowed local:
# the await is exactly where the value can change.)
if _get_dossier_status(app.state).running:
continue
if not _dossier_due(status.last_run, settings.dossier_schedule_interval_hours):
continue
status.running = True # claim the single-flight slot before scheduling
status._task = asyncio.create_task(_run_dossier_task(app.state, trigger="schedule"))
except asyncio.CancelledError:
raise
except Exception:
_LOGGER.exception("host dossier scheduler iteration failed; continuing")
async def _auto_triage_scheduler_loop(app: FastAPI, settings: Any) -> None:
"""Continuously drain the untriaged backlog when scheduled auto-triage is on.
Mirrors :func:`_discovery_scheduler_loop`: a fixed wake cadence (derived from
``auto_triage_schedule_interval_minutes``), live settings read each wake (so a
config-console toggle applies without a restart), and a no-op unless
``auto_triage_schedule_enabled``. Single-flight via the shared
``AutoTriageStatus.active`` slot, so a scheduled sweep and a manual ⚡ press
can never overlap. A failed iteration is logged and the loop continues — the
scheduler must never take the app down.
"""
from soc_ai.webui import autotriage as at # noqa: PLC0415
# Fixed short wake cadence + an internal "is due" check (mirrors the discovery
# scheduler). Sleeping the whole interval up front meant toggling the schedule
# ON only took effect up to interval-length later; a 60s wake makes a fresh
# enable fire on the next wake. ``_last_sweep`` is None until the first sweep so
# the first enabled wake always fires — a 0.0 sentinel collided with a small
# ``time.monotonic()`` on a freshly-booted host (monotonic's epoch is arbitrary,
# near-zero right after boot), which wrongly read as "just swept" and skipped the
# first sweep for up to one interval.
_last_sweep: float | None = None
while True:
await asyncio.sleep(60)
try:
if not settings.auto_triage_schedule_enabled:
continue
interval_min = int(getattr(settings, "auto_triage_schedule_interval_minutes", 5))
now = time.monotonic()
if _last_sweep is not None and now - _last_sweep < max(60, interval_min * 60):
continue
if at.get_status(app.state).active:
continue # a sweep (manual or scheduled) is already in flight
n = await at.start_config_sweep(app.state, started_by="auto-triage:scheduler")
# Record the sweep time whether or not it launched targets. The ES
# planning pass (one grouped aggregation per severity + per-group
# fetches) is exactly what the interval throttles; on a drained
# backlog every sweep plans 0 targets, so gating the timestamp on
# ``n`` left ``_last_sweep`` unset and re-ran full planning every 60s,
# ignoring ``auto_triage_schedule_interval_minutes``.
_last_sweep = time.monotonic()
if n:
_LOGGER.info("auto-triage scheduler: launched a sweep of %d target(s)", n)
except asyncio.CancelledError:
raise
except Exception:
_LOGGER.exception("auto-triage scheduler iteration failed; continuing")
async def _hunt_schedule_loop(app: FastAPI) -> None:
"""Fire recurring (scheduled) hunts when they come due (E3.1).
Mirrors :func:`_auto_triage_scheduler_loop`: a fixed 60s wake, live settings
read each wake (a config-console toggle applies without a restart), a no-op
unless ``hunt_schedules_enabled``. Each wake fetches the DUE schedules
(:func:`soc_ai.store.hunt_schedules.due_schedules`) and spawns a normal hunt
per schedule via the shared ``HuntConsoleManager`` (tagged ``kind="scheduled"``),
then stamps ``last_run_at`` — but only after the spawn actually succeeds. A
spawn rejected at the shared concurrency ceiling stays unstamped so the
schedule remains due and retries on the next wake (no lost cycle).
**Single-flight** is per-SCHEDULE, not global: distinct schedules run
concurrently (the manager keys tasks by hunt_id, so they never collide), but a
schedule can't re-fire while its own hunt is still running because
:func:`mark_ran` resets its interval clock the instant a spawn succeeds — so the same
schedule is no longer "due" next wake until the interval elapses again. This
relies on the interval being ≥ the hunt runtime (enforced as a 60-min floor at
the store). A per-schedule failure is logged and skipped so one bad schedule
can never take out the others or the loop.
Note (workers>1): a second uvicorn worker would run its own copy of this loop
and double-fire every schedule — soc-ai runs a SINGLE worker today; distributed
scheduler coordination is Epoch 6.2, deliberately not built here.
Each hunt also reads the findings of the last complete runs of its objective
(:func:`soc_ai.hunting.earlier.earlier_for_objective`). The objective does not
change, so the run keeps its link to the earlier runs and its diff.
"""
from soc_ai.hunting import earlier as earlier_mod # noqa: PLC0415
from soc_ai.store import hunt_schedules as hs_svc # noqa: PLC0415
from soc_ai.webui import hunt_console_manager as hcm # noqa: PLC0415
while True:
await asyncio.sleep(60)
try:
settings = app.state.settings
if not getattr(settings, "hunt_schedules_enabled", False):
continue
now = datetime.now(UTC).replace(tzinfo=None) # naive UTC, matches the store
async with app.state.db_sessionmaker() as db:
due = await hs_svc.due_schedules(db, now)
if not due:
continue
manager = hcm.get_manager(app.state)
launched = 0
for sched in due:
try:
# Opens its own session and never raises, so a failed read
# cannot reach the except branch below that stamps mark_ran.
earlier = await earlier_mod.earlier_for_objective(
app.state.db_sessionmaker, sched.objective, now=now
)
hunt_id = await manager.start(
app.state,
objective=sched.objective,
started_by="scheduler",
kind="scheduled",
starter="schedule",
earlier=earlier,
)
if hunt_id is None:
# Rejected: the shared concurrency ceiling is full (or the
# generator died before hunt_created). Leave last_run_at
# UNSTAMPED so the schedule stays "due" and the next wake
# retries it — stamping here would silently drop a whole
# interval and show a "last ran" for a run that never began.
_LOGGER.warning(
"hunt scheduler: schedule id=%s could not start "
"(at capacity?) — left due to retry next wake",
sched.id,
)
continue
# Stamp the interval clock only AFTER a real spawn: this is the
# per-schedule single-flight guard and records the run that
# actually started. The loop is sequential, so the next wake
# can't fire mid-pass — a slow start() can't double-fire it.
async with app.state.db_sessionmaker() as db:
await hs_svc.mark_ran(db, sched.id, now)
launched += 1
except asyncio.CancelledError:
raise
except Exception:
_LOGGER.exception(
"hunt scheduler: schedule id=%s failed to fire; skipping", sched.id
)
# A schedule whose spawn RAISED (a broken objective, a bug in
# start()) is stamped so it can't retry-storm every wake — a
# permanently-bad schedule fires at most once per interval, not
# once per wake. This is distinct from the capacity rejection
# above (hunt_id is None), which is transient and SHOULD retry.
try:
async with app.state.db_sessionmaker() as db:
await hs_svc.mark_ran(db, sched.id, now)
except Exception:
_LOGGER.exception(
"hunt scheduler: schedule id=%s could not be stamped after "
"a failed spawn",
sched.id,
)
if launched:
_LOGGER.info("hunt scheduler: launched %d scheduled hunt(s)", launched)
except asyncio.CancelledError:
raise
except Exception:
_LOGGER.exception("hunt scheduler iteration failed; continuing")
# How many leads one wake reads at a time. Wider than any sensible concurrency
# cap, because a lead that cites no documents is skipped and must not hold the
# slot of a lead behind it that does.
_LEAD_AUTO_HUNT_BATCH = 50
# How many such pages one wake reads before it gives up looking. A skipped lead
# stays open and stays where it is, so the skipped leads pile up at the head of
# the queue, and a wake that read one page found nothing but skips once a page
# of them was older than every lead with evidence. The wake reads on past them.
# The bound keeps a table of thousands of skipped leads from pinning one wake.
_LEAD_AUTO_HUNT_PAGES = 20
# How often the skip of one lead is said. The loop wakes every 60 seconds, so
# an unthrottled line would write sixty an hour about one lead that is not
# moving.
_LEAD_AUTO_HUNT_QUIET = timedelta(hours=1)
async def _leads_the_loop_may_hunt(db: Any, *, need: int) -> tuple[list[int], list[int]]:
"""The waiting leads, split by whether their observations cite a document.
The second list is skipped rather than hunted: a hunt of a lead that cites
nothing has no evidence to read first, so it searches the grid from
scratch. The split is re-read on every wake, because the sweep can record
documents for a lead that had none.
``need`` is how many leads the wake can start. The pages are read in order
until a page has brought the first list to that many, until a page comes
back short, or until the page bound is spent. Every skipped lead read on
the way is in the second list, so the log still names each one.
"""
from soc_ai.hunting import lead_hunt # noqa: PLC0415
from soc_ai.store import leads as leads_store # noqa: PLC0415
cited: list[int] = []
bare: list[int] = []
after: tuple[datetime, int] | None = None
for _ in range(_LEAD_AUTO_HUNT_PAGES):
page = await lead_hunt.leads_awaiting_a_hunt(db, limit=_LEAD_AUTO_HUNT_BATCH, after=after)
for lead in page:
rows = await leads_store.timeline(db, int(lead.id))
(cited if leads_store.cites_documents(rows) else bare).append(int(lead.id))
if len(cited) >= need or len(page) < _LEAD_AUTO_HUNT_BATCH:
break
after = (page[-1].formed_at, int(page[-1].id))
return cited, bare
def _say_the_skipped_leads(
said: dict[int, datetime], no_documents: list[int], now: datetime
) -> dict[int, datetime]:
"""Log each skipped lead once an hour. Returns the times to keep.
The returned map holds only the leads still being skipped, so a process
that runs for months does not accumulate an entry per lead ever formed.
"""
kept = {lead_id: at for lead_id, at in said.items() if lead_id in no_documents}
for lead_id in no_documents:
last = kept.get(lead_id)
if last is None or (now - last) >= _LEAD_AUTO_HUNT_QUIET:
kept[lead_id] = now
_LOGGER.info(
"lead auto-hunt: lead %s cites no documents, so it is not hunted. "
"Run the sweep again to record them.",
lead_id,
)
return kept
async def _lead_auto_hunt_loop(app: FastAPI) -> None:
"""Start a hunt for every lead that has never had one (D1).
Mirrors :func:`_hunt_schedule_loop`'s discipline: a fixed 60s wake, live
settings read each wake so a config-console toggle applies without a
restart, and a no-op unless ``lead_auto_hunt``. It differs in what it
reads: not a schedule table but the leads themselves.
The loop starts the hunt rather than the sweep that forms the lead,
because leads also form in the timer process that runs the priors, and
that process cannot run an agent.
The rule, in one sentence: an open lead with no hunt, never dismissed,
with no hunt row that names it, oldest first, until the number of the
loop's own lead hunts in flight reaches ``lead_auto_hunt_concurrency``.
The selection lives in :func:`soc_ai.hunting.lead_hunt.leads_awaiting_a_hunt`
and the start in :func:`soc_ai.hunting.lead_hunt.start_lead_hunt`, which
the Hunt button on the lead calls as well.
Before it selects, each wake settles every hunting lead whose hunt has
finished, through :func:`soc_ai.store.leads.settle_finished_hunts`. The
startup reaper runs the same call, so an install that upgrades with leads
stuck in ``hunting`` needs no migration of data by hand.
A lead whose observations cite no documents is skipped, not started: its
hunt would have no evidence to read first and would search the grid from
scratch. The skip is re-read on every wake, because the sweep can record
documents later, but the log line about it is said once an hour per lead.
One lead that fails is logged and the tick continues, so a single bad lead
can never take out the others or the loop.
Note (workers>1): a second uvicorn worker would run its own copy and could
start a second hunt for the same lead between the console call and the
mark. soc-ai runs a single worker today; distributed scheduler
coordination is Epoch 6.2, as for the loops above.
"""
from soc_ai.hunting import lead_hunt # noqa: PLC0415
# The demo starts no hunt. Its model gateway is a mock, so a lead hunt
# could only fail, and each failed hunt would grow the store of a public
# site. The seeded leads already show every state.
if getattr(app.state.settings, "soc_ai_demo", False) is True:
return
# Per-process, per-lead: when the skip was last said. Rebuilt each wake
# from the leads still being skipped, so it cannot grow without bound.
said_no_documents: dict[int, datetime] = {}
while True:
await asyncio.sleep(60)
try:
settings = app.state.settings
if not getattr(settings, "lead_auto_hunt", False):
continue
cap = max(1, int(getattr(settings, "lead_auto_hunt_concurrency", 2) or 1))
async with app.state.db_sessionmaker() as db:
# Bookkeeping before selection. A lead whose hunt finished
# while this process was down, or before the settle rule
# existed, moves now, and an errored one is back to open in
# time for the selection below to retry it.
settled = await leads_store.settle_finished_hunts(db)
if settled:
_LOGGER.info(
"lead auto-hunt: settled %d lead(s) whose hunt had finished", settled
)
running = await lead_hunt.running_auto_hunts(db)
if running >= cap:
continue
eligible, no_documents = await _leads_the_loop_may_hunt(db, need=cap - running)
said_no_documents = _say_the_skipped_leads(
said_no_documents, no_documents, datetime.now(UTC).replace(tzinfo=None)
)
started = 0
for lead_id in eligible:
if running + started >= cap:
break
try:
out = await lead_hunt.start_lead_hunt(
app.state, lead_id=lead_id, started_by=lead_hunt.AUTO_HUNT_ACTOR
)
except asyncio.CancelledError:
raise
except lead_hunt.LeadHuntRefused as exc:
# The lead closed or the console is full. Both are normal.
# The lead stays open and the next wake reads it again.
_LOGGER.info(
"lead auto-hunt: lead %s did not take a hunt (%s)", lead_id, exc.reason
)
continue
except Exception:
_LOGGER.exception(
"lead auto-hunt: lead %s failed to start a hunt; skipping", lead_id
)
continue
if out.existing:
continue
started += 1
_LOGGER.info("lead auto-hunt: lead %s started hunt %s", lead_id, out.hunt_id)
except asyncio.CancelledError:
raise
except Exception:
_LOGGER.exception("lead auto-hunt iteration failed; continuing")
def _persist_bootstrap_credential(settings: Any, created_pw: str) -> None:
"""Write the one-shot bootstrap admin password to a locked-down sidecar
file instead of the shared log stream, and log only a pointer to it.
setup.sh pre-generates BOOTSTRAP_ADMIN_PASSWORD so this path is off the
happy path. journald/container logs are often readable by the same
audience (other analysts, integrations) this credential must stay secret
from. Mirrors the chmod(0o600) treatment backup.py gives the Ed25519
signing key sidecar.
"""
cred_path = bootstrap_credential_path(settings)
try:
# Born 0600 rather than created under the umask and locked down after:
# a write-then-chmod leaves a window where any local user can read the
# password. Unlink first because O_TRUNC on a sidecar left over from an
# earlier run would keep that file's old mode. The chmod afterwards is
# belt-and-braces, mirroring the signing-key sidecar.
cred_path.unlink(missing_ok=True)
fd = os.open(cred_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
with os.fdopen(fd, "w") as fh:
fh.write(created_pw + "\n")
cred_path.chmod(0o600)
except OSError:
# Data dir not writable for some reason — fall back to the log line
# rather than leaving the operator with no way to reach the account.
# Whatever did land on disk must not stay there next to the log line,
# or the credential is exposed in two places instead of none.
with contextlib.suppress(OSError):
cred_path.unlink()
_LOGGER.warning(
"BOOTSTRAP CREDENTIAL (change at first login, then scrub this log line): "
"initial admin user 'admin' password=%s",
created_pw,
)
else:
_LOGGER.warning(
"BOOTSTRAP CREDENTIAL written to %s (mode 0600) — log in as 'admin', "
"change the password, then delete that file.",
cred_path,
)
def _require_prompt_assets() -> None:
"""Refuse to serve when a declared prompt asset is not on disk.
The prompt assets (``soc_ai.agent.prompts.PROMPT_ASSETS``) are markdown
files that live beside the package rather than inside it, so a build that
forgets to copy them produces an app that starts, answers, and is wrong:
the primer degrades to a stub that tells the model the query language is
unavailable, and the verdicts that come back still look like verdicts.
This is the one place the product prefers a crash. The standing rule is
that a false all-clear outranks a crash, and it holds for upstreams: a
half-answering grid is reported, not fatal, because the honest degraded
answer exists and we can hand it to the analyst. Here there is no honest
degraded answer to hand over. The absence is invisible in the output, it
is knowable with two stat calls before the first request, and it is a
packaging fault an operator fixes by redeploying, so we fail at the point
where somebody is already watching, rather than at the point where an
analyst is trusting the answer. The doctor carries the same finding
(``check_prompt_assets``) for the installs that reach it another way.
"""
from soc_ai.agent.prompts import missing_prompt_assets # noqa: PLC0415 - startup-only
missing = missing_prompt_assets()
if not missing:
return
detail = "; ".join(f"{asset.name} ({asset.path}): {asset.cost}" for asset in missing)
raise RuntimeError(
f"soc-ai does not start. {len(missing)} prompt assets are missing: {detail}. "
"The deployment is incomplete. Redeploy from an image or an install that "
"ships the docs/ directory. Then run `soc-ai doctor` to confirm."
)
async def _reap_orphans_at_startup(db_sessionmaker: Any) -> None:
"""Resolve every row the previous process left mid-flight.
Anything still ``running``/``pending`` when we boot can never finish — the
background task that owned it died with the old process — so each store is
swept here, with the terminal status that fits its semantics: ``interrupted``
where the work is re-runnable, ``error`` where it is a one-shot measurement
or an answer nobody will ever receive.
Split out of :func:`_init_store` because it is ONE concern spread over five
stores, and the list only grows: each new background-backed table needs a
line here, and a missing one is invisible until a restart happens to land
mid-run.
"""
# Investigations: mark 'interrupted' (NOT 'error') — a clean restart cut them
# off; they didn't fail. 'interrupted' is re-huntable, so continuous
# auto-triage (or a manual re-hunt) picks them back up, and the UI shows a
# benign state instead of a scary "error" in a healthy environment.
async with db_sessionmaker() as db:
orphaned = await inv_svc.reap_stale_running(
db, older_than_minutes=None, status="interrupted"
)
if orphaned:
_LOGGER.info("reaped %d orphaned 'running' investigation(s) at startup", orphaned)
# Hunts mirror the investigation reaper: re-runnable, so 'interrupted'.
async with db_sessionmaker() as db:
orphaned_hunts = await hunt_svc.reap_stale_running(
db, older_than_minutes=None, status="interrupted"
)
if orphaned_hunts:
_LOGGER.info("reaped %d orphaned 'running' hunt(s) at startup", orphaned_hunts)
# A hunt that investigated its findings when the process stopped. Its
# report stands and its investigations were reaped above, so it completes.
async with db_sessionmaker() as db:
investigating = await hunt_svc.reap_investigating(db)
if investigating:
_LOGGER.info("completed %d 'investigating' hunt(s) at startup", investigating)
# Leads: a lead whose hunt finished with no process there to settle it,
# and every lead an older release left in 'hunting'. Runs after the hunt
# reaper, so an orphaned hunt is terminal by the time the rule reads it.
async with db_sessionmaker() as db:
settled = await leads_store.settle_finished_hunts(db)
if settled:
_LOGGER.info("settled %d lead(s) whose hunt had finished, at startup", settled)
# Backtests: mark 'error' — a backtest is a one-shot measurement whose replay
# task died, not a re-huntable target.
async with db_sessionmaker() as db:
orphaned_bt = await bt_svc.reap_stale_running(db, older_than_minutes=None, status="error")
if orphaned_bt:
_LOGGER.info("reaped %d orphaned 'running' backtest(s) at startup", orphaned_bt)
# Investigation chat: a 'pending' assistant row would otherwise stay pending —
# empty — forever, because the task that was writing the answer is gone.
async with db_sessionmaker() as db:
orphaned_chat = await chat_svc.reap_stale_pending(db, older_than=None)
if orphaned_chat:
_LOGGER.info("reaped %d orphaned 'pending' chat turn(s) at startup", orphaned_chat)
# …and the dashboard's general chat, a SEPARATE table with its own reaper
# (chat_svc only scans chat_messages). Skipping it leaves an empty answer
# bubble on the landing screen — the first thing every analyst sees — with no
# path back to 'done'.
async with db_sessionmaker() as db:
orphaned_general = await general_chat_svc.reap_stale_pending(db, older_than=None)
if orphaned_general:
_LOGGER.info(
"reaped %d orphaned 'pending' dashboard chat turn(s) at startup", orphaned_general
)
# …and the hunt follow-up chat, whose turns live in hunt_events rather than
# in either chat table. Neither reaper above reaches them, and a pending
# turn left there keeps the hunt's chat spinning and refuses every later
# question on that hunt as busy. Startup only: the rows carry no timestamp
# for the periodic sweep to age them by.
async with db_sessionmaker() as db:
orphaned_hunt_chat = await hunt_svc.reap_stale_pending_chat(db)
if orphaned_hunt_chat:
_LOGGER.info(
"reaped %d orphaned 'pending' hunt chat turn(s) at startup", orphaned_hunt_chat
)
# Expired sessions: get_session_user rejects them but leaves the row, so an
# abandoned cookie's row lingers forever. Sweep them here (storage hygiene) —
# the lookup is indexed so this is not a hot-path cost, just unbounded growth.
async with db_sessionmaker() as db:
purged_sessions = await purge_expired_sessions(db)
if purged_sessions:
_LOGGER.info("purged %d expired session(s) at startup", purged_sessions)
async def _mark_backtest_leads(db_sessionmaker: Any) -> None:
"""Mark the leads that formed from backtest replays. A one-time repair.
Before 2026-10-06 a backtest replay wrote an alert observation like a live
run, so a replay could form a lead. The store writes none now. This marks
the leads that formed before that. Idempotent: a marked lead is not marked
again, and no new replay forms a lead, so a later start finds nothing to
do. Fail-soft: the mark is a note on a lead and must not block serving.
"""
try:
async with db_sessionmaker() as db:
marked = await leads_store.mark_backtest_leads(db)
if marked:
_LOGGER.info("marked %d lead(s) that formed from backtest replays", marked)
except Exception:
_LOGGER.warning("backtest lead repair failed; continuing", exc_info=True)
def _apply_demo_autotriage(state: Any, batch: Any) -> None:
"""Set the last Auto-Investigate batch the demo shows.
The panel reads process memory, and a demo never runs a batch, so the
fixture file carries one. Unknown keys are ignored: the fixture cannot
reach the private fields of the status object.
"""
if not isinstance(batch, dict):
return
from soc_ai.webui import autotriage # noqa: PLC0415
status = autotriage.get_status(state)
for key in (
"total",
"hunted",
"skipped",
"failed",
"finished_at",
"tool_calls",
"inherited_acked",
"inherited_acked_total",
"prior_covered",
):
if key in batch:
setattr(status, key, batch[key])
if isinstance(batch.get("skipped_reasons"), dict):
status.skipped_reasons = {str(k): int(v) for k, v in batch["skipped_reasons"].items()}
if isinstance(batch.get("severities"), list):
status.severities = tuple(str(v) for v in batch["severities"])
status.active = False
async def _init_store(db_engine: Any, settings: Any, secret_box: Any = None) -> Any:
"""Migrate the store, bootstrap the admin, apply config overrides, reap orphans.
Returns the sessionmaker. Migration failure disposes the engine and re-raises
(the app must not serve against a corrupt/newer schema).
"""
try:
await run_migrations(db_engine)
except Exception:
_LOGGER.exception(
"store migration failed — DB at %s is corrupt or newer than this build; "
"back up/remove the file or upgrade soc-ai",
describe_store(settings),
)
await db_engine.dispose()
raise
db_sessionmaker = make_sessionmaker(db_engine)
async with db_sessionmaker() as db:
created_pw = await bootstrap_admin(db, settings.bootstrap_admin_password)
if created_pw is not None:
_persist_bootstrap_credential(settings, created_pw)
# Re-apply persisted admin config overrides onto the live settings singleton
# so operator choices (e.g. Oracle on/off) survive a restart.
async with db_sessionmaker() as db:
overrides = await load_overrides(db)
apply_to_settings(settings, overrides, secret_box=secret_box)
if overrides:
_LOGGER.info("applied %d persisted config override(s)", len(overrides))
# Seed the builtin hunt templates (E3.2) — idempotent upsert-by-name, so it's
# safe on every startup (never duplicates a builtin, never touches a custom
# template). Fail-soft: a seed failure must not block serving; the picker just
# falls back to whatever templates already exist (or the frontend's static
# pills if the store is empty).
try:
async with db_sessionmaker() as db:
seeded = await hunt_templates_svc.seed_builtins(db)
if seeded:
_LOGGER.info("seeded/refreshed %d builtin hunt template(s)", seeded)
except Exception:
_LOGGER.warning("builtin hunt-template seed failed; continuing", exc_info=True)
# A shipped analytic that ships in shadow gets its shadow row on the first
# catalog load. The catalog reads it as shadow without the row, so a seed
# failure costs the approval route its row and never runs the analytic
# live. Fail-soft for that reason.
try:
from soc_ai.hunting.catalog_tiers import effective_catalog # noqa: PLC0415
async with db_sessionmaker() as db:
await effective_catalog(db, seed=True)
except Exception:
_LOGGER.warning("shipped shadow analytic seed failed; continuing", exc_info=True)
await _mark_backtest_leads(db_sessionmaker)
await _reap_orphans_at_startup(db_sessionmaker)
# Demo mode: seed the fixture set so every console page has content. The
# seed runs AFTER the reaper on purpose: the demo shows a hunt in flight
# and a lead that waits on it, and the reaper would end both. Idempotent
# per row (restart-safe) and fail-soft: a missing or invalid fixtures.json
# must never block serving; the demo starts with whatever the store holds.
if settings.soc_ai_demo:
try:
from soc_ai.demo.fixtures import ( # noqa: PLC0415
DEFAULT_FIXTURES,
load_fixtures,
seed_fixtures,
)
added = await seed_fixtures(db_sessionmaker, load_fixtures(DEFAULT_FIXTURES))
if added:
_LOGGER.info("demo mode: seeded %d fixture row(s)", added)
except Exception:
_LOGGER.warning("demo fixture seed failed; continuing with empty store", exc_info=True)
# The hunt catalog's sweep trail is generated from the shipped catalog,
# not read from the fixture file (see soc_ai/demo/catalog_trail.py), so
# it has its own fail-soft step: a fixture problem must not cost the
# Status page its analytic sweep panel, and the reverse holds too.
try:
from soc_ai.demo.catalog_trail import seed_catalog_trail # noqa: PLC0415
if await seed_catalog_trail(db_sessionmaker):
_LOGGER.info("demo mode: seeded a week of hunt catalog sweeps")
except Exception:
_LOGGER.warning("demo catalog trail seed failed; continuing without it", exc_info=True)
return db_sessionmaker
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]: # noqa: PLR0915 — linear app setup
"""Wire up app-scoped clients, tear them down on shutdown."""
settings = get_settings()
logging.basicConfig(
level=settings.log_level,
format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
# Before anything is built: the prompt assets are a packaging fact, and a
# deployment missing one answers wrongly rather than not at all.
_require_prompt_assets()
# DB + persisted overrides FIRST, before any client is built. Connection /
# secret Danger-Zone overrides feed the SO/ES/LiteLLM clients, so they must
# land on `settings` before make_auth/ElasticClient/etc. read it. The secret
# box decrypts at-rest secret overrides (None when CONFIG_SECRET_KEY is unset
# → secret overrides are skipped, env values stand).
secret_box = make_secret_box(settings)
db_engine = make_engine(settings)
# The values before any console override. GET /config names them as the
# .env or default origin of a setting, and "Use the default" restores them
# (dogfood 2026-10-19, R5-D7, R5-D8).
settings_baseline = settings.model_copy(deep=True)
db_sessionmaker = await _init_store(db_engine, settings, secret_box)
_LOGGER.info(
"soc-ai starting; auth=%s, host=%s",
"connect" if settings.use_connect_api else "kratos",
settings.so_host,
)
from soc_ai.tls_status import TlsStatus, describe, inspect_tls # noqa: PLC0415 - lazy
try:
tls_at_start = inspect_tls(settings.soc_ai_tls_cert, settings.soc_ai_tls_key)
except Exception as exc: # a surprise in the inspector must not stop the start
_LOGGER.exception("The TLS inspection failed at start. soc-ai starts anyway.")
tls_at_start = TlsStatus(mode="direct", errors=[f"Cannot inspect the certificate: {exc}."])
app.state.tls_at_start = tls_at_start
app.state.tls_loaded_at = datetime.now(UTC).isoformat()
# A self-signed certificate alone is the stock install. It reads at INFO
# here and in the doctor. An expiry band or a broken chain reads at WARNING.
serious = [
w for w in tls_at_start.warnings if not w.startswith("The certificate is self-signed")
]
if tls_at_start.errors:
_LOGGER.error(
"TLS: %s soc-ai starts anyway. Fix the files and restart.", describe(tls_at_start)
)
elif serious:
_LOGGER.warning("TLS: %s %s", describe(tls_at_start), " ".join(tls_at_start.warnings))
else:
_LOGGER.info("TLS: %s %s", describe(tls_at_start), " ".join(tls_at_start.warnings))
# Loud warning when API auth is disabled — with auth off, require_admin_api is
# a no-op, so secret mutation, user creation, and token minting are open to any
# caller that can reach the port. See auth_off_warning for what the line may
# claim about the bind.
auth_line = auth_off_warning(settings)
if auth_line is not None:
_LOGGER.warning("%s", auth_line)
auth = make_auth(settings)
elastic = ElasticClient(settings)
misp = MispClient(settings) if settings.misp_url is not None else None
audit = AuditLogger(settings, elastic)
enrichment = build_local_enrichment_context(settings)
# Ed25519 signer for decision-record exports (load-or-generate the key).
# Best-effort: a signing failure must not block startup — exports then carry
# the sha256 checksum only.
from soc_ai.store.signing import DecisionSigner # noqa: PLC0415
try:
decision_signer: Any = DecisionSigner.load_or_create(settings.soc_ai_data_dir)
except Exception:
_LOGGER.warning("decision-record signer unavailable; exports use checksum only")
decision_signer = None
app.state.settings = settings
app.state.settings_baseline = settings_baseline
app.state.secret_box = secret_box
app.state.auth = auth
app.state.elastic = elastic
app.state.misp = misp
app.state.audit = audit
app.state.enrichment = enrichment
app.state.decision_signer = decision_signer
app.state.db_engine = db_engine
app.state.db_sessionmaker = db_sessionmaker
# Demo mode: cache the parsed fixture document ONCE for the replay runner —
# the two allowlisted POSTs (soc_ai/demo/replay.py) look up replays[] per
# request without re-reading the file. Fail-soft like the seed hook in
# _init_store: a missing/invalid fixtures.json leaves the cache None, and
# replays then report unknown alerts the same way the live pipeline does.
app.state.demo_fixtures = None
if settings.soc_ai_demo:
try:
from soc_ai.demo.fixtures import ( # noqa: PLC0415
DEFAULT_FIXTURES,
load_fixtures,
rebase_document,
)
app.state.demo_fixtures = rebase_document(load_fixtures(DEFAULT_FIXTURES))
_apply_demo_autotriage(app.state, app.state.demo_fixtures.get("autotriage"))
except Exception:
_LOGGER.warning(
"demo replay fixtures unavailable; replay lookups will find nothing",
exc_info=True,
)
reaper_task = asyncio.create_task(_reaper_loop(db_sessionmaker, settings))
discovery_task = asyncio.create_task(_discovery_scheduler_loop(app, settings))
dossier_task = asyncio.create_task(_dossier_scheduler_loop(app, settings))
autotriage_task = asyncio.create_task(_auto_triage_scheduler_loop(app, settings))
hunt_schedule_task = asyncio.create_task(_hunt_schedule_loop(app))
lead_auto_hunt_task = asyncio.create_task(_lead_auto_hunt_loop(app))
eval_nightly_task = asyncio.create_task(_eval_nightly_loop(app, settings))
spec_sweep_task = asyncio.create_task(_hunt_spec_sweep_loop(app))
prior_sweep_task = asyncio.create_task(_prior_sweep_loop(app))
# The estate model: a daily fit beside the profile build, off by default.
from soc_ai.hunting.estate_model.schedule import estate_model_loop # noqa: PLC0415
estate_model_task = asyncio.create_task(estate_model_loop(app))
audit_verify_task = asyncio.create_task(_audit_verify_loop(app, settings))
tls_expiry_task = asyncio.create_task(_tls_expiry_loop(app, settings))
health_probe_task = asyncio.create_task(_health_probe_loop(app, settings))
try:
yield
finally:
_LOGGER.info("soc-ai shutting down; releasing clients")
reaper_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await reaper_task
discovery_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await discovery_task
dossier_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await dossier_task
autotriage_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await autotriage_task
hunt_schedule_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await hunt_schedule_task
lead_auto_hunt_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await lead_auto_hunt_task
eval_nightly_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await eval_nightly_task
spec_sweep_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await spec_sweep_task
prior_sweep_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await prior_sweep_task
estate_model_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await estate_model_task
health_probe_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await health_probe_task
audit_verify_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await audit_verify_task
tls_expiry_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await tls_expiry_task
# An in-flight quality-eval worker (scheduled or run-now) holds its own
# engine/ES clients — cancel + drain like the discovery worker below.
from soc_ai.api.webui_api import _get_quality_eval_status # noqa: PLC0415
_qs = _get_quality_eval_status(app.state)
if _qs._task is not None and not _qs._task.done():
_qs._task.cancel()
with contextlib.suppress(BaseException):
await _qs._task
# A scheduled (or manual "Scan now") discovery worker may be mid-scan,
# tracked on the shared single-flight status object (the same one the
# scan-now endpoint uses). Cancel + drain it BEFORE the ES/DB clients it
# holds are torn down, so a shutdown racing an in-flight scan doesn't log
# a spurious "scan failed".
from soc_ai.api.webui_api import _get_discovery_status # noqa: PLC0415
_st = _get_discovery_status(app.state)
if _st._task is not None and not _st._task.done():
_st._task.cancel()
with contextlib.suppress(BaseException):
await _st._task
# Same for an in-flight host-dossier sweep (scheduled or "Rebuild now"):
# it holds the ES + DB clients for minutes at a time, so it must be
# cancelled and drained BEFORE they are torn down or it lands a
# use-after-close ES search partway through the network.
from soc_ai.api.webui import _get_dossier_status # noqa: PLC0415
_ds = _get_dossier_status(app.state)
if _ds._task is not None and not _ds._task.done():
_ds._task.cancel()
with contextlib.suppress(BaseException):
await _ds._task
# The scheduler LOOPS are cancelled above, but the WORKER tasks they
# spawn (auto-triage drain, backtest replay) and manually-started hunt
# console drains hold references to the ES + DB clients. Cancel + drain
# them here too, before those clients are torn down, so an in-flight
# worker can't do a use-after-close ES search / DB write on shutdown.
from soc_ai.webui import autotriage as _at # noqa: PLC0415
from soc_ai.webui import backtest as _bac # noqa: PLC0415
from soc_ai.webui import chat_manager as _cm # noqa: PLC0415
from soc_ai.webui import hunt_console_manager as _hcm # noqa: PLC0415
from soc_ai.webui import hunt_manager as _hm # noqa: PLC0415
_worker_tasks: list[asyncio.Task[Any]] = []
_at_task = _at.get_status(app.state)._task
if _at_task is not None:
_worker_tasks.append(_at_task)
_bac_task = _bac.get_status(app.state)._task
if _bac_task is not None:
_worker_tasks.append(_bac_task)
_worker_tasks.extend(_hcm.get_manager(app.state)._tasks.values())
# HuntManager (per-alert manual Investigate via POST /hunt) and ChatManager
# (chat turns) hold the same ES/DB references via app.state — drain them too,
# else an in-flight Investigate or chat turn can use-after-close on shutdown.
_worker_tasks.extend(_hm.get_manager(app.state)._tasks.values())
_worker_tasks.extend(_cm.get_manager(app.state)._tasks.values())
# The "Chat about this hunt" follow-up turns (HuntChatManager, stored as
# hunt_events) are a THIRD chat drainer. Without this, a pending hunt-chat
# turn is abandoned mid-flight at shutdown — never cancelled, so its
# done-callback backstop never fires and the row stays 'pending' forever
# (the hunt's chat then spins and every later POST .../chat 409s).
_worker_tasks.extend(_hcm.get_chat_manager(app.state)._tasks.values())
# Demo replay drains (soc_ai/demo/replay.py start_background_replay) hold
# the DB sessionmaker via the recorder — drain them too, else an in-flight
# replay can write after db_engine.dispose(). Empty set outside demo.
_worker_tasks.extend(getattr(app.state, "demo_replay_tasks", set()))
for _t in _worker_tasks:
if not _t.done():
_t.cancel()
for _t in _worker_tasks:
with contextlib.suppress(BaseException):
await _t
await auth.aclose()
await elastic.aclose()
if misp is not None:
await misp.aclose()
await db_engine.dispose()
def _resolve_cors_origins(cors_setting: str, so_host: str) -> list[str]:
"""Resolve the CORS allow-origins list from config (pure, testable).
Precedence: an explicit ``"*"`` (opt-in wildcard) > a comma-separated
``CORS_ALLOW_ORIGINS`` list > the SO host. With none of those configured we
**fail closed** to ``[]`` (no cross-origin callers) rather than ``["*"]`` —
the React app is same-origin, and a wildcard would let any site read
responses on behalf of a bearer-token caller.
"""
cors = cors_setting.strip()
if cors == "*":
return ["*"]
if cors:
return [o.strip() for o in cors.split(",") if o.strip()]
if so_host:
return [so_host]
return []
# Demo mode (SOC_AI_DEMO): the only mutating routes left open — the replay
# triggers (Task 6 wires them to recorded fixtures). Mounted paths verified
# against the live route table: the soc_ai.api.routes router is included with
# NO prefix, so investigate lives at /investigate (not /api/v1/investigate);
# the webui hunts router is under /api/v1.
#
# The Dashboard general chat is here too, and it is the one entry that is not a
# replay: on a demo BOTH its verbs resolve inside soc_ai/api/webui/routes_chat.py
# without touching the store (see the demo section there), so allow-listing them
# hands a visitor no write. Both verbs are needed and neither is surplus:
# POST is the reply itself — this box sits on the LANDING SCREEN, so refusing it
# means the first thing a visitor touches errors; DELETE backs the panel's
# "Clear conversation" control, which appears as soon as a reply does, and
# refusing it would put an error under the landing screen instead. PUT/PATCH are
# not routes here and stay refused.
_DEMO_WRITE_ALLOW: set[tuple[str, str]] = {
("POST", "/investigate"),
("POST", "/api/v1/hunt"),
("POST", "/api/v1/chat"),
("DELETE", "/api/v1/chat"),
}
# Chat routes carry a variable id in the path, so they can't live in the
# exact-match set above — match them by (method, pattern). In demo mode these
# POSTs are turned into canned, ZERO-EGRESS replies by the demo branches in the
# chat/hunt route handlers and managers (soc_ai/api/webui/routes_chat.py,
# soc_ai/webui/chat_manager.py, host_chat_manager.py, lead_chat_manager.py,
# hunt_console_manager.py);
# ``/api/v1/hunts/chat`` (hunt-start) is wired to a fixture replay in a later
# task. The host chat's DELETE is the one non-POST: like the general chat's
# DELETE it backs the panel's "Clear conversation" control and, on a demo,
# deletes nothing (routes_chat serves it as a no-op) — refusing it would put an
# error under the first reply a visitor gets.
# Anchored ``^...$`` so a pattern can only allow the exact chat routes — never a
# broader mutating route that merely contains the substring.
_DEMO_WRITE_ALLOW_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
("POST", re.compile(r"^/api/v1/investigations/[^/]+/chat$")),
("POST", re.compile(r"^/api/v1/hunts/[^/]+/chat$")),
("POST", re.compile(r"^/api/v1/hunts/chat$")), # hunt-start → seeded hunt id (ephemeral)
("POST", re.compile(r"^/api/v1/dossiers/[^/]+/chat$")),
("DELETE", re.compile(r"^/api/v1/dossiers/[^/]+/chat$")),
# The lead chat: the host chat's two verbs, for the host chat's reasons.
("POST", re.compile(r"^/api/v1/hunts/leads/[^/]+/chat$")),
("DELETE", re.compile(r"^/api/v1/hunts/leads/[^/]+/chat$")),
)
# A field whose name matches this is reported WITHOUT its value. See
# `_validation_error_without_input` for why: the rejected value of a credential
# field is the plaintext secret, and a 4xx body is copied into proxy logs.
_SECRET_FIELD_RE = re.compile(r"pass|secret|token|key|credential|otp|pin", re.IGNORECASE)
# The request parts pydantic names in `loc`. They are not field names, so they
# are dropped before the field is composed.
_LOC_PARTS = frozenset({"body", "query", "path", "header", "cookie"})
# How much of a rejected value the hint quotes. Long enough to recognise a
# typo, short enough that a pasted document does not become the error message.
_HINT_VALUE_CHARS = 60
def _bound(value: Any) -> str:
return str(int(value)) if isinstance(value, float) and value.is_integer() else str(value)
def _bounded_sentence(field: str, kind: str, ctx: dict[str, Any]) -> str | None:
"""The sentence for a value outside a length, size, number or choice bound."""
if kind == "string_too_long" and "max_length" in ctx:
return f"{field} must be at most {ctx['max_length']} characters."
if kind == "string_too_short" and "min_length" in ctx:
if ctx["min_length"] == 1:
return f"{field} is empty. Send a value."
return f"{field} must be at least {ctx['min_length']} characters."
if kind == "too_long" and "max_length" in ctx:
return f"{field} must hold at most {ctx['max_length']} items."
if kind == "too_short" and "min_length" in ctx:
if ctx["min_length"] == 1:
return f"{field} is empty. Send at least one item."
return f"{field} must hold at least {ctx['min_length']} items."
if kind == "greater_than_equal" and "ge" in ctx:
return f"{field} must be {_bound(ctx['ge'])} or more."
if kind == "less_than_equal" and "le" in ctx:
return f"{field} must be {_bound(ctx['le'])} or less."
if kind == "greater_than" and "gt" in ctx:
return f"{field} must be more than {_bound(ctx['gt'])}."
if kind == "less_than" and "lt" in ctx:
return f"{field} must be less than {_bound(ctx['lt'])}."
if kind in {"literal_error", "enum"} and "expected" in ctx:
return f"{field} accepts one of these values: {ctx['expected']}."
return None
def _validation_hint(err: dict[str, Any]) -> str:
"""One STE sentence for one rejected field: what it is and what it must be.
Built from the pydantic error ``type``. The raw ``msg`` read as noise in an
operator's toast: "0 jSON decode error; got '{}'" for a body that was not
JSON, "username field required; got '{}'" for a missing field.
"""
parts = [str(p) for p in (err.get("loc") or ()) if str(p) not in _LOC_PARTS]
kind = str(err.get("type") or "")
ctx = err.get("ctx") or {}
if kind == "json_invalid":
return "The body is not valid JSON."
# A JSON body that is a list or a string where an object belongs.
if kind in {"model_attributes_type", "dict_type", "model_type"} and not parts:
return "The body must be a JSON object."
field = ".".join(p for p in parts if not p.isdigit()) or ".".join(parts)
if not field:
if kind == "missing":
return "The request body is required. Send a JSON object."
field = "The request"
if kind == "missing":
return f"{field} is required."
if kind == "extra_forbidden":
return f"{field} is not a known field. Remove it."
if kind in {"int_parsing", "int_type", "int_from_float"}:
return f"{field} must be a whole number."
if kind in {"float_parsing", "float_type", "decimal_parsing"}:
return f"{field} must be a number."
if kind in {"bool_parsing", "bool_type"}:
return f"{field} must be true or false."
if kind in {"string_type"}:
return f"{field} must be a string."
if kind in {"list_type"}:
return f"{field} must be a list."
sentence = _bounded_sentence(field, kind, ctx)
if sentence is None:
message = str(err.get("msg") or "").strip().rstrip(".")
if message.lower().startswith("value error, "):
message = message[len("value error, ") :]
sentence = f"{field} is not valid."
if message:
sentence += f" {message[0].upper()}{message[1:]}."
# The value is quoted so the analyst sees what arrived, with one exception:
# the rejected value of a credential field is the plaintext secret.
if _SECRET_FIELD_RE.search(field) or "input" not in err:
return sentence
value = str(err.get("input"))
if len(value) > _HINT_VALUE_CHARS:
# A long value is named by its start and its length, with no ellipsis.
return (
f"{sentence} soc-ai got a value of {len(value)} characters that starts "
f"with '{value[:_HINT_VALUE_CHARS]}'."
)
return f"{sentence} soc-ai got '{value}'."
# The largest request body soc-ai reads. No route takes an upload: the largest
# body is a runbook, 64 Ki characters, well under this cap. A 20 MB JSON body
# on /hunts/bulk-delete was parsed to the end while login refused at 8 KiB
# (fleet 2026-10-01, RA19).
_MAX_BODY_BYTES = 1024 * 1024
# Tighter caps for named routes. Login is unauthenticated and sits in front of
# the throttle, so it gets the smallest one.
_ROUTE_BODY_CAPS: dict[str, int] = {
"/api/v1/login": 8 * 1024,
# The tour state is a small object (soc_ai.store.user_tours).
"/api/v1/me/tours": user_tours.MAX_BYTES,
}
def _too_large(limit: int) -> JSONResponse:
size = f"{limit // 1024} KiB" if limit < 1024 * 1024 else f"{limit // (1024 * 1024)} MiB"
return JSONResponse(
status_code=413,
content={
"detail": {
"reason": "payload_too_large",
"hint": f"The request body is larger than {size}. soc-ai did not read it.",
}
},
)
class _BodySizeCap:
"""Refuse a request body over the cap, counted in bytes as it arrives.
A declared Content-Length over the cap is refused before any byte is read.
A body with no declared length is read up to the cap and then refused, so
the cap holds whatever framing a proxy in front chose. A body under the cap
is handed to the app unchanged.
"""
def __init__(self, app: Any, max_bytes: int = _MAX_BODY_BYTES) -> None:
self.app = app
self.max_bytes = max_bytes
async def __call__(self, scope: Scope, receive: Any, send: Any) -> None:
if scope.get("type") != "http":
await self.app(scope, receive, send)
return
limit = _ROUTE_BODY_CAPS.get(str(scope.get("path", "")), self.max_bytes)
headers = dict(scope.get("headers") or [])
declared = headers.get(b"content-length")
if declared is not None:
if declared.isdigit() and int(declared) > limit:
await _too_large(limit)(scope, receive, send)
return
await self.app(scope, receive, send)
return
if b"transfer-encoding" not in headers or scope.get("path") == "/api/v1/login":
# No body, or login: its own guard answers 411 to a body with no
# declared length without reading a byte of it.
await self.app(scope, receive, send)
return
chunks: list[bytes] = []
total = 0
while True:
message = await receive()
if message.get("type") != "http.request":
# A disconnect before the body ended: nothing to hand on.
return
body = message.get("body", b"")
total += len(body)
if total > limit:
await _too_large(limit)(scope, receive, send)
return
chunks.append(body)
if not message.get("more_body"):
break
buffered = b"".join(chunks)
replayed = False
async def replay() -> Any:
nonlocal replayed
if not replayed:
replayed = True
return {"type": "http.request", "body": buffered, "more_body": False}
return await receive()
await self.app(scope, replay, send)
# The API prefix every refusal shape rule below applies to. Routes outside it
# (/healthz, /investigate, the SPA under /app) keep their own answers.
_API_V1_PREFIX = "/api/v1/"
# The default hint for a refusal that names a reason and no hint. A bare
# {"reason": "not_found"} on a red toast told the operator nothing, and the
# 2026-10-01 fleet found more than a dozen of them. A route that knows more
# writes its own hint; this table is the floor under the rest.
_REASON_HINTS: dict[str, str] = {
"admin_required": "This action needs the admin role. Ask an admin to do it.",
"already_applied": "This proposal is already applied. Reload the page.",
"bad_token": "The confirmation token does not match. Reload the page and try again.",
"chat_busy": "The chat is answering an earlier message. Wait for the answer, then send again.",
"empty_message": "The message is empty. Type a message and send it again.",
"finding_not_found": "The hunt has no finding with this id. Reload the hunt.",
"hunt_not_found": "soc-ai has no hunt with this id. Check the hunt id.",
"malformed_proposal": "The proposal is not complete. Ask the chat for a new proposal.",
"no_session": "Sign in first. This route needs a session.",
"no_such_action": "The investigation has no action at this index. Reload the investigation.",
"not_applyable": "This proposal cannot be applied. Ask the chat for a new proposal.",
"planning_failed": "soc-ai could not plan the sweep. Check the server log and try again.",
"proposal_not_found": "soc-ai has no proposal with this id. Reload the chat.",
"still_running": "The run is still in progress. Wait for it to finish, then try again.",
}
# The noun a 404 names, read off the first path segment after /api/v1/.
_PATH_NOUNS: dict[str, str] = {
"investigations": "investigation",
"hunts": "hunt",
"analyst": "investigation",
"backtest": "backtest run",
"runbooks": "runbook",
"leads": "lead",
"analytics": "analytic",
"detections": "detection",
"identifiers": "identifier",
"users": "user",
"tokens": "token",
"notifications": "notification",
"dossiers": "dossier",
"events": "event",
}
# The reason a refusal gets when the route raised a bare string, by status.
_STATUS_REASONS: dict[int, str] = {
400: "bad_request",
401: "unauthorized",
403: "forbidden",
404: "not_found",
405: "method_not_allowed",
409: "conflict",
411: "length_required",
413: "payload_too_large",
422: "bad_request",
429: "rate_limited",
500: "server_error",
502: "upstream_error",
503: "unavailable",
504: "upstream_timeout",
}
def _default_hint(status: int, reason: str, path: str) -> str:
"""One STE sentence for a refusal that came with no hint of its own."""
if reason in _REASON_HINTS:
return _REASON_HINTS[reason]
if status == 404:
segment = path[len(_API_V1_PREFIX) :].split("/", 1)[0]
noun = _PATH_NOUNS.get(segment)
if noun:
return f"soc-ai has no {noun} with this id. Check the id in the address."
return "soc-ai has no record at this address. Check the id in the address."
if status == 401:
return "Sign in and try again."
if status == 403:
return "Your role does not allow this action."
if status == 409:
return "The record changed state. Reload it and try again."
if status >= 500:
return "soc-ai could not complete the request. Check the server log and try again."
return "The request is not valid. Check the values and try again."
def _sentence(text: str) -> str:
"""A bare-string detail as one sentence: capital first letter, final period."""
text = text.strip()
if not text:
return text
text = text[0].upper() + text[1:]
return text if text.endswith((".", "!", "?")) else text + "."
def _api_refusal_detail(status: int, detail: Any, path: str, method: str) -> Any:
"""The ``{"reason", "hint"}`` shape for any refusal under /api/v1.
Three shapes reached clients before this: a dict with a hint (the good
one), a dict with a reason and no hint, and a bare string. The router's own
404 for an unknown path was the bare string "Not Found".
"""
if isinstance(detail, dict):
reason = detail.get("reason")
if isinstance(reason, str) and not detail.get("hint"):
return {**detail, "hint": _default_hint(status, reason, path)}
return detail
if status == 404 and detail == "Not Found":
return {
"reason": "unknown_route",
"hint": "soc-ai has no API route at this address. Check the path.",
}
if status == 405 and detail == "Method Not Allowed":
return {
"reason": "method_not_allowed",
"hint": f"This route does not accept the {method} method.",
}
reason = _STATUS_REASONS.get(status, "error")
hint = _sentence(detail) if isinstance(detail, str) and detail.strip() else ""
return {"reason": reason, "hint": hint or _default_hint(status, reason, path)}
def create_app() -> FastAPI: # noqa: PLR0915 - app factory wires many middlewares + routers
"""Application factory."""
# Gate the interactive docs + raw schema behind a setting (off in prod) so a
# security product doesn't publish its full admin API surface unauthenticated.
try:
_expose_docs = get_settings().expose_api_docs
except Exception:
_expose_docs = False
app = FastAPI(
title="soc-ai",
description="Open, self-hosted LLM-powered triage assistant for Security Onion.",
# Single source of truth: the installed package version (pyproject).
# Was hardcoded "0.1.0", which drifted from the `__version__` the
# /healthz and /metrics routes already report.
version=__version__,
lifespan=lifespan,
docs_url="/docs" if _expose_docs else None,
redoc_url="/redoc" if _expose_docs else None,
openapi_url="/openapi.json" if _expose_docs else None,
)
@app.exception_handler(RequestValidationError)
async def _validation_error_without_input(
request: Request, exc: RequestValidationError
) -> JSONResponse:
"""422 with the same ``reason`` + ``hint`` shape every other 4xx uses.
FastAPI's default handler answers with a list of error dicts. The SPA
reads ``detail.reason`` and ``detail.hint`` everywhere else, so a
validation failure was the one refusal it could not render, and the
dogfood saw a bare "422" toast. The hint names the field, says what is
wrong with it and quotes the value.
The value is quoted with one exception, and it is the reason this
handler exists. For every credential field in this API — ``/login``'s
password, create-user's password, ``/me/password``'s current AND new
password — the rejected value IS the plaintext secret, and a 422 is
exactly what an over-long one produces. The echoed plaintext then lands
wherever 4xx bodies land: reverse-proxy capture logs, frontend error
reporters, browser devtools history. A field whose name looks like a
credential is reported without its value.
"""
hints = list(dict.fromkeys(_validation_hint(err) for err in exc.errors()))
hint = " ".join(hints[:3]) if hints else "The request is not valid."
if len(hints) > 3:
hint = f"{hint} {len(hints) - 3} more fields are not valid."
return JSONResponse(
status_code=422,
content={"detail": jsonable_encoder({"reason": "bad_request", "hint": hint})},
)
@app.exception_handler(StarletteHTTPException)
async def _api_refusal_shape(request: Request, exc: StarletteHTTPException) -> Response:
"""Every refusal under /api/v1 answers ``{"detail": {"reason", "hint"}}``.
Routes outside the API prefix keep FastAPI's own answer. One case
differs: a browser that asks for an unknown page address goes to the
console under ``/app``, and the SPA shows its not-found page in the
shell. ``/nope`` answered a browser with raw ``{"detail":"Not Found"}``
(dogfood 2026-10-19, R1-D19).
"""
path = request.url.path
if _console_not_found(request, exc.status_code):
query = request.url.query
return RedirectResponse(f"/app{path}" + (f"?{query}" if query else ""), 307)
if not path.startswith(_API_V1_PREFIX) or not is_body_allowed_for_status_code(
exc.status_code
):
return await http_exception_handler(request, exc)
detail = _api_refusal_detail(exc.status_code, exc.detail, path, request.method)
return JSONResponse(
status_code=exc.status_code,
content={"detail": jsonable_encoder(detail)},
headers=getattr(exc, "headers", None),
)
try:
_demo = get_settings().soc_ai_demo
except Exception:
_demo = False
if _demo:
# Demo read-only lock: refuse every mutating request with the structured
# 403 shape the SPA already renders (detail.reason + detail.hint), except
# the replay triggers in _DEMO_WRITE_ALLOW, which replay recorded runs
# instead of doing real work (their handlers hit the egress guards if
# they try). GET/HEAD/OPTIONS pass untouched — /healthz (docker
# healthcheck) and all reads are unaffected. SOC_AI_DEMO is env-only
# (not a UI-editable override), so gating registration at create time is
# safe and keeps the non-demo request path completely untouched.
# Registered FIRST → innermost middleware: on a public demo the refusal
# is the most-served mutating response, so it must flow back out through
# _security_headers (and CORSMiddleware), which are registered after it
# and therefore wrap outside it.
@app.middleware("http")
async def _demo_readonly(request: Any, call_next: Any) -> Response:
_path = request.url.path
_allowed = (request.method, _path) in _DEMO_WRITE_ALLOW or any(
request.method == m and p.match(_path) for m, p in _DEMO_WRITE_ALLOW_PATTERNS
)
if request.method in {"POST", "PUT", "PATCH", "DELETE"} and not _allowed:
return JSONResponse(
status_code=403,
content={
"detail": {
"reason": "demo_mode",
"hint": (
"The demo is read-only. It replays stored runs. This action is off."
),
}
},
)
response: Response = await call_next(request)
return response
# Cross-origin API clients (automation / integrations hosted on another
# origin) fetch soc-ai cross-origin (the React /app is same-origin and needs
# no CORS). Scope to CORS_ALLOW_ORIGINS if set, else the SO host; "*" only as
# a last resort (with a warning). allow_credentials stays False — the
# cross-origin caller authenticates with a bearer token, not a cookie.
# get_settings() may raise at import-time construction if no .env is present
# (e.g. CI just importing the module); fall back to a warned "*" then.
try:
_settings = get_settings()
_cors = _settings.cors_allow_origins.strip()
_so_host = str(_settings.so_host).rstrip("/") if _settings.so_host else ""
except Exception:
_cors, _so_host = "", ""
cors_origins = _resolve_cors_origins(_cors, _so_host)
if cors_origins == ["*"]:
_LOGGER.warning("CORS allow_origins='*' — set CORS_ALLOW_ORIGINS (or SO_HOST) to scope it")
elif not cors_origins:
_LOGGER.warning(
"CORS allow_origins empty — no cross-origin callers permitted; "
"set CORS_ALLOW_ORIGINS (or SO_HOST) to enable cross-origin API clients"
)
app.add_middleware(
CORSMiddleware,
allow_origins=cors_origins,
allow_credentials=False,
allow_methods=["*"],
allow_headers=["*"],
expose_headers=["*"],
)
try:
_csp = get_settings().content_security_policy.strip()
except Exception:
_csp = ""
from soc_ai.api.webui_api import _request_is_https # noqa: PLC0415
@app.middleware("http")
async def _security_headers(request: Any, call_next: Any) -> Response:
"""Set conservative security response headers on every response.
CSP (``content_security_policy``, default tuned for the bundled Vite SPA)
plus ``frame-ancestors 'none'`` / ``X-Frame-Options: DENY`` block
clickjacking; ``Cross-Origin-Opener-Policy`` isolates the browsing
context. HSTS is only sent over HTTPS so a plain-HTTP dev run is
unaffected.
"""
response: Response = await call_next(request)
response.headers.setdefault("X-Content-Type-Options", "nosniff")
response.headers.setdefault("X-Frame-Options", "DENY")
response.headers.setdefault("Referrer-Policy", "no-referrer")
response.headers.setdefault(
"Permissions-Policy", "geolocation=(), microphone=(), camera=()"
)
response.headers.setdefault("Cross-Origin-Opener-Policy", "same-origin")
if _csp:
response.headers.setdefault("Content-Security-Policy", _csp)
# Emit HSTS when the browser reached us over HTTPS — directly, or via a
# TLS-terminating reverse proxy that forwards plain HTTP with
# X-Forwarded-Proto: https. The same trust rule as the Secure cookie
# flag decides: the forwarded header counts only from a peer in
# proxy_trusted_ips, so an untrusted client cannot switch HSTS on.
# The resolved settings sit on app.state (set at startup); the helper
# tolerates a missing state.
if _request_is_https(request, getattr(request.app.state, "settings", None)):
response.headers.setdefault(
"Strict-Transport-Security", "max-age=63072000; includeSubDomains"
)
return response
# /api/v1/login is unauthenticated by design ("this IS the gate") and runs
# before the login throttle's counters are touched, so it's the one route an
# anonymous caller can flood with an arbitrarily large request body — the
# deployed stack terminates TLS in uvicorn directly with nothing in front to
# impose a size cap. Reject early on a declared Content-Length so the body is
# never buffered. A body with NO declared length (chunked transfer) gets no
# such early answer: Starlette reads it to the end before LoginIn's
# Field(max_length=...) ever runs, so the declared length is REQUIRED here
# (411). Browsers, fetch() with a string body, and httpx all send
# Content-Length for a JSON login, so no real client is affected. A request
# that carries Content-Length AND Transfer-Encoding is framed as chunked by
# the HTTP parser, so its length header cannot be trusted either.
_LOGIN_MAX_BODY_BYTES = 8 * 1024 # ample for a username+password JSON body
@app.middleware("http")
async def _login_body_size_guard(request: Any, call_next: Any) -> Response:
if request.url.path == "/api/v1/login" and request.method == "POST":
content_length = request.headers.get("content-length")
if (
request.headers.get("transfer-encoding")
or content_length is None
or not content_length.isdigit()
):
return JSONResponse(
status_code=411,
content={
"detail": {
"reason": "length_required",
"hint": "Login requires a Content-Length header.",
}
},
)
if int(content_length) > _LOGIN_MAX_BODY_BYTES:
return JSONResponse(
status_code=413,
content={
"detail": {
"reason": "payload_too_large",
"hint": "The login body is larger than 8 KiB. soc-ai did not read it.",
}
},
)
response: Response = await call_next(request)
return response
try:
_rl_limit = get_settings().api_rate_limit_per_min
except Exception:
_rl_limit = 0
if _rl_limit > 0:
# ip -> [window (monotonic minute), count]. Per-app-instance, bounded.
app.state.rate_buckets = {}
from soc_ai.api.webui_api import client_ip as _client_ip # noqa: PLC0415
@app.middleware("http")
async def _rate_limit(request: Any, call_next: Any) -> Response:
# The limiter counts API calls, not page loads. One load of the
# SPA pulls a dozen JavaScript chunks and a document, and counting
# those spent the minute's budget on the app's own assets. The
# browser then rendered a blank screen with 429s in the console.
# Health checks are exempt for the same reason they always were.
path = request.url.path
if path in {"/healthz", "/app"} or path.startswith("/app/"):
exempt: Response = await call_next(request)
return exempt
# Proxy-aware: attribute to the real client, not a shared proxy IP,
# when proxy_trusted_ips is configured (else the socket peer). Read the
# resolved settings off app.state (set at startup) rather than a
# per-request get_settings(); client_ip tolerates a missing state.
ip = _client_ip(request, getattr(request.app.state, "settings", None))
window = int(time.monotonic()) // 60
buckets: dict[str, list[int]] = app.state.rate_buckets
entry = buckets.get(ip)
if entry is None or entry[0] != window:
if len(buckets) > 8192: # crude bound against unique-IP floods
buckets.clear()
buckets[ip] = [window, 1]
else:
entry[1] += 1
if entry[1] > _rl_limit:
return JSONResponse(
status_code=429,
headers={"Retry-After": "60"},
content={
"detail": {
"reason": "rate_limited",
"hint": (
f"This address sent more than {_rl_limit} requests in one "
"minute. Wait 60 s and try again."
),
}
},
)
response: Response = await call_next(request)
return response
# Added last, so it is the outermost layer and runs before any other
# middleware reads the body.
app.add_middleware(_BodySizeCap)
app.include_router(router)
# Open (pre-auth) endpoints first so FastAPI resolves /api/v1/login before
# the auth-gated router's blanket dependency can reject the request.
app.include_router(api_v1_open_router, prefix="/api/v1")
app.include_router(api_v1_router, prefix="/api/v1")
# The React SPA at /app. Only mounted when a build is present so source-only
# checkouts still boot.
app.state.spa_mounted = FRONTEND_DIST.is_dir()
if app.state.spa_mounted:
app.mount("/app", SpaStaticFiles(directory=FRONTEND_DIST, html=True), name="app")
else:
_LOGGER.info("frontend build not found at %s — /app not served", FRONTEND_DIST)
# Bare `/` → the React app front door. The SPA is always built in deployment;
# if `spa_mounted` is False, /app isn't served (startup logged a warning) and
# this redirect 404s — which is the correct signal that the build is missing.
# The SPA root picks the page: the Dashboard, or sign-in. `/` sent the
# reader to Alerts while `/app` opened the Dashboard, so the console had
# two home pages (dogfood 2026-10-19, R1-D19).
@app.get("/", include_in_schema=False)
async def _root() -> RedirectResponse:
return RedirectResponse("/app/", 307)
return app
app = create_app()
# CLI entry now lives in soc_ai.cli; pyproject's [project.scripts] points
# `soc-ai` there. The systemd unit still uses `uvicorn soc_ai.main:app`
# directly, so this module just defines the FastAPI `app`.