soc_ai/agent/orchestrator.py
Rama main · réplica de lectura del repositorio original.
"""Investigation orchestrator: the synth-first pipeline entry point.
The API layer calls :func:`investigate` directly, passing an
:class:`~soc_ai.agent.context.InvestigationContext` assembled per request.
Agents are built lazily inside the pipeline — there is no lifecycle-managed
``build_agent`` construction at the API layer.
The pipeline (:func:`_run_synth_first_pipeline`) runs:
1. **Phase A** — :func:`get_enriched_alert_context` pivots + locally enriches
the alert into an :class:`EnrichedAlertContext`.
2. **Phase B** — :func:`match_decision_template` produces an optional
*candidate verdict* anchor.
3. **Round-1 skip checks, before Phase C.** Two of them, each emitting
``synth_round1_skipped`` and routing straight to the investigation loop.
``reason="definitely_investigate"``: the alert carries malware/exploit
signal or a threat-context flag. ``reason="cannot_settle"``: the round-1
verdict could not end the run anyway (:func:`_round1_can_settle`), so the
loop would overwrite it. Round 1 settles a case only when a DISPOSITIVE
decision template already cleared the alert ``false_positive``. Set
``synth_round1_always`` to run the call on every alert again.
4. **Phase C round 1** — the heavy synthesizer model reads materialized
evidence + candidate and emits a :class:`TriageReport`. It has no tools;
it may name one gap via ``gap_for_investigator``.
5. **Phase D** — if a gap was named and the loop was not entered,
:mod:`soc_ai.agent.targeted_investigator` dispatches that single tool
deterministically (no LLM), then round-2 synthesis re-reads the result.
6. **Investigation loop** — full tool-equipped agent, entered when
``_should_investigate`` returns true, the definitely-investigate flag was
set, OR ``fast_triage_enabled=false`` forces it for every alert
(``loop_reason="fast_triage_disabled"``); synthesizer runs over the
transcript to produce the final report.
7. **Post-synth validators** — :func:`~soc_ai.agent.gates._synth_first_post_validate`
applies citation validation, confidence caps, verdict floor rewrites, and
targeted downgrades deterministically.
8. **Oracle escalation** — optional second-opinion raw HTTP call (opt-in;
not a pydantic-ai agent).
Write tools surface via :attr:`TriageReport.recommended_actions` only — the
pipeline never pauses waiting for a human mid-run. The analyst executes
recommended actions on demand through the actions API
(:mod:`soc_ai.api.webui.routes_actions`), which runs them through the single
audited :func:`~soc_ai.tools.write_exec.execute_write_tool` path.
"""
from __future__ import annotations
import asyncio
import dataclasses
import logging
import re
import uuid
from collections import Counter
from collections.abc import AsyncGenerator, AsyncIterator, Sequence
from datetime import UTC, datetime
from typing import Any, cast
from pydantic_ai import Agent, capture_run_messages
from pydantic_ai.exceptions import UsageLimitExceeded
from pydantic_ai.models import Model
from pydantic_ai.usage import RequestUsage, RunUsage, UsageLimits
from soc_ai import metrics
from soc_ai.agent import budget, context_budget
from soc_ai.agent._gateway_retry import capture_backend_attribution
from soc_ai.agent._partial_replay import (
repair_dangling_tool_calls,
replay_reasoning_context,
)
from soc_ai.agent.classifier import AlertClass, classify_alert, normalize_classtype
# Backward-compat re-exports — _DedupTracker, InvestigationContext, StepEvent now
# live in soc_ai.agent.context; tests and callers reach these via orchestrator.
from soc_ai.agent.context import (
HuntSubject,
InvestigationContext,
StepEvent,
_DedupTracker,
)
from soc_ai.agent.egress_guard import EgressGuard, EgressResidueError
# Backward-compat re-exports — evidence helpers now live in soc_ai.agent.evidence;
# tests and callers reach these via orchestrator.
from soc_ai.agent.evidence import ( # noqa: F401
_NON_EVIDENCE_RESULT_KEYS,
_PIVOT_DECISIVE_ATTRS,
_beacon_profile_bullet,
_bundle_dump_text,
_classify_citation,
_dns_tunnel_profile_bullet,
_loop_evidence_marker,
_materialize_prefetch_evidence,
_num,
_path_exists_in_alert,
_pivot_decisive_evidence,
_targeted_result_has_data,
_tool_was_invoked,
count_successful_tool_calls,
)
# Backward-compat re-exports — the deterministic verdict gates (citation
# validation + the post-synthesis downgrade stack) now live in
# soc_ai.agent.gates; tests and callers reach these via orchestrator.
from soc_ai.agent.gates import ( # noqa: F401
_CITATION_STOP_WORDS,
_ESCALATION_CONF_FLOOR,
_FUZZY_TOKEN_RE,
_GROUNDED_EVIDENCE_TOKENS,
_PIVOT_ATTRS,
_apply_targeted_downgrades,
_citation_confidence_cap,
_downgrade_unevidenced_verdict,
_downgrade_ungrounded_host_anchored_tp,
_has_ioc_hit,
_is_solicited_internal_icmp_echo,
_is_strong_grounded_template,
_no_semantic_evidence,
_pivot_evidence_tokens,
_refuse_benign_verdict_without_baseline,
_resolve_citations,
_synth_first_post_validate,
_validate_citations,
_verdict_cites_decisive_pivot_value,
_verdict_grounded_in_pivot,
enforce_session_verdict_consistency,
)
from soc_ai.agent.loop_result import SINGLE, SWARM_STRATEGIES, LoopResult, strategy_of
from soc_ai.agent.models import (
build_investigator_model,
build_model,
build_synthesizer_model,
)
from soc_ai.agent.narrative_grounding import coverage_subject, ground_host_coverage_claims
from soc_ai.agent.prompts import (
BUDGET_PARTIAL_SYNTH_PROMPT,
INVESTIGATOR_PROMPT,
SYNTHESIZER_PROMPT,
FocusOrigin,
_format_investigator_prompt,
_format_transcript_for_synthesizer,
build_investigator_prompt,
case_conditions,
format_endpoint_coverage_block,
format_host_coverage_block,
format_more_tools_block,
format_prefetched_web_search_block,
hunt_subject_rules,
rule_body_in_alert,
unresolved_external_indicator,
)
from soc_ai.agent.reasoning import extract_reasoning_trace
# The unified read-tool surface (single registration source for all three
# agents) lives in soc_ai.agent.toolset. The private helpers are re-imported
# here because existing tests and callers reach them via this module
# (`from soc_ai.agent.orchestrator import _clamp_tool_result`, ...).
from soc_ai.agent.toolset import ( # noqa: F401
_TOOL_RESULT_BUDGET_BYTES,
_clamp_tool_result,
_dedup_result,
_tool_error,
prefetched_community_ids,
prime_playbook_presence,
register_read_tools,
)
from soc_ai.agent.triage import (
InvestigationTranscript,
RecommendedAction,
TargetedGap,
TriageReport,
)
from soc_ai.config import Settings
from soc_ai.dossier.prompt import MAX_PROMPT_HOSTS, host_dossier_prompt_block
from soc_ai.errors import OqlValidationError, SoApiError
# Module import (not `from ... import adjudicate`) so tests patching
# `soc_ai.oracle.client.adjudicate` keep intercepting the escalation call.
from soc_ai.oracle import breaker as _oracle_breaker
from soc_ai.oracle import client as _oracle_client
from soc_ai.oracle.identifiers import EffectiveIdentifiers, effective_internal_identifiers
from soc_ai.so_client.inventory import inventory_prompt_block
from soc_ai.so_client.models import SoAlert
from soc_ai.tools.enrichment import build_local_enrichment_context
from soc_ai.tools.get_alert_context import ENDPOINT_COVERAGE_GAP_KEY
from soc_ai.tools.write_exec import execute_write_tool
from soc_ai.triage_models import is_pipeline_fallback
_LOGGER = logging.getLogger(__name__)
# =====================================================================
# Agent factory
# =====================================================================
# The model + provider builders — ``build_investigator_model``,
# ``build_synthesizer_model``, ``build_model`` and their ``_build_provider`` /
# ``_nemotron_profile`` helpers — now live in :mod:`soc_ai.agent.models`. They
# are re-imported at the top of this module so every existing call site, the
# ``agent`` package re-exports, and the tests that patch
# ``soc_ai.agent.orchestrator.build_*_model`` keep working unchanged.
# `build_local_enrichment_context` now lives in `soc_ai.tools.enrichment`
# (so the MCP server can build the same local sources without importing this
# heavy module); it is re-imported above and re-exported below for the
# existing `soc_ai.agent.orchestrator.build_local_enrichment_context` callers.
def build_investigator(
model: Model,
ctx: InvestigationContext,
*,
system_prompt: str | None = None,
emits_report: bool = False,
output_type: Any = None,
) -> Agent[None, Any]:
"""Investigator agent: fast model + read tools + InvestigationTranscript output.
The read-tool surface comes from
:func:`soc_ai.agent.toolset.register_read_tools` (role ``investigator``);
closures there capture ``ctx`` so the LLM-facing tool signatures stay
semantic-only (no auth/elastic/etc. parameters in the schema).
``system_prompt`` overrides the default :data:`INVESTIGATOR_PROMPT`.
``output_type`` overrides the output type. A swarm worker of the swarm
triage experiment (:mod:`soc_ai.agent.swarm`) passes its brief model, and
passes the system prompt that asks for it. None keeps the two types below.
``emits_report`` (W3, ``Settings.investigator_emits_report``) swaps the
output type to :class:`~soc_ai.triage_models.TriageReport`, so the loop's
own final output IS the report and the round-2 synthesis call does not run.
The tool surface, the retry budget and the system prompt are unchanged; the
user message gains
:data:`~soc_ai.agent.prompts.INVESTIGATOR_EMITS_REPORT_BLOCK`. Kept as a
plain ``TriageReport`` rather than the synthesizer's
``synthesizer_output_mode`` wrappers: this agent interleaves real tool
calls, and tool mode is what that path is validated on.
The coverage gate previously lived here as an
``output_validator`` raising ``ModelRetry``. Smoke testing surfaced a
pathological interaction: PydanticAI's ``retries`` budget is shared
between schema-validation retries (Nemotron-30B routinely needs 2-3
attempts to land a schema-valid InvestigationTranscript) AND
output_validator retries. The combined retry budget exhausted before
the model could produce a schema-valid transcript. The coverage gate
was removed; the synthesis-floor rewrite handles missing enrichment by
downing confidence below the floor when semantic citation coverage is
absent. ``retries=5`` gives schema validation room.
"""
agent: Agent[None, Any] = Agent(
model,
output_type=(
output_type
if output_type is not None
else (TriageReport if emits_report else InvestigationTranscript)
),
# The report-writing loop reads a system prompt that says it writes
# the report and carries the verdict policy; the transcript loop keeps
# the prompt the synthesizer expects it to have followed.
system_prompt=system_prompt
or (build_investigator_prompt(emits_report=True) if emits_report else INVESTIGATOR_PROMPT),
# The default of 10 retries is generous on a per-output basis but
# Nemotron-30B's schema-format wobble is genuinely stochastic (some
# runs land in 2 attempts, others need 8+); stronger models may need
# far less — tune via the investigator_retries setting. The
# per-investigation request_limit still bounds the worst case, and a
# failed alert burns ~10 quick retries (each emitting almost no
# output) which is cheaper than an unrecoverable run.
retries=ctx.settings.investigator_retries,
)
# Note: no `t_get_alert_context` tool is registered for the investigator
# since the orchestrator pre-fetches the alert context and embeds it in
# the user prompt. The fast 30B was unable to consistently honor a
# "do not call this tool" rubric — removing the tool entirely is the
# only reliable way to enforce the contract. If a future iteration
# needs secondary-alert context, expose it through a renamed tool
# (`t_get_other_alert_context(alert_id)`) so the model can't
# accidentally re-fetch the alert under triage.
register_read_tools(agent, ctx, role="investigator")
return agent
def _synth_output_type(output_mode: str) -> Any:
"""Map :attr:`Settings.synthesizer_output_mode` to a pydantic-ai output spec.
``tool`` returns the bare TriageReport (today's synthetic final_result tool
call). ``native`` wraps it in :class:`~pydantic_ai.output.NativeOutput`
(response_format json_schema — server-side guided decoding, no tool-call
parser in the path). ``prompted`` wraps it in
:class:`~pydantic_ai.output.PromptedOutput` (schema in prompt, JSON parsed
from text). See the Settings docstring for when each applies; validate a
candidate mode with ``soc-ai model-probe`` before flipping in prod.
"""
if output_mode == "tool":
return TriageReport
if output_mode == "native":
from pydantic_ai.output import NativeOutput # noqa: PLC0415
return NativeOutput(TriageReport)
if output_mode == "prompted":
from pydantic_ai.output import PromptedOutput # noqa: PLC0415
return PromptedOutput(TriageReport)
raise ValueError(f"unknown synthesizer output_mode: {output_mode!r}")
def build_synthesizer(model: Model, *, output_mode: str = "tool") -> Agent[None, TriageReport]:
"""Synthesizer agent: heavy model, no tools, TriageReport output.
The synthesizer reads the investigator's transcript (passed as the user
message) and emits a TriageReport. It has no tools — synthesis happens
entirely from the gathered evidence.
``retries=3`` matches :func:`build_synth_first_agent` and
:func:`build_partial_triage_synthesizer`. This builder previously took no
``retries`` at all, inheriting pydantic-ai's default of **1** — a single
schema wobble ended the run. It is live on the round-2 loop-synthesis path
(the ``investigation_loop_synth`` phase), which accounted for 7 of 11
pipeline fallbacks in the 14-day prod window ending 2026-08-03. Schema
wobble is stochastic on lower-tier models, so one attempt is not enough.
"""
return Agent(
model,
output_type=_synth_output_type(output_mode),
system_prompt=SYNTHESIZER_PROMPT,
retries=3,
)
def build_partial_triage_synthesizer(
model: Model, *, output_mode: str = "tool"
) -> Agent[None, TriageReport]:
"""No-tools synthesizer that concludes a budget-cut investigation loop.
Mirrors the hunt runner's partial-report path (:func:`soc_ai.api.
hunt_runner._synthesize_partial_hunt`): replays the loop's gathered
(repaired) message history and forces a TriageReport from ONLY that
evidence. ``retries=3`` for reasoning-model schema-wobble parity with
:func:`build_synth_first_agent`.
The prompt goes in ``instructions=``, never ``system_prompt=``. This agent
always runs with a NON-empty replayed ``message_history``, and pydantic-ai
emits an agent's ``system_prompt`` only when the history is empty. A
``system_prompt`` here was dropped on every run: the partial verdict was
written under the replayed investigator prompt with none of the
anti-over-claim rules. ``instructions`` are applied on every request, the
same fix :func:`soc_ai.agent.hunt.build_hunt_synthesizer` carries.
"""
return Agent(
model,
output_type=_synth_output_type(output_mode),
instructions=BUDGET_PARTIAL_SYNTH_PROMPT,
retries=3,
)
# Error-shaped on purpose: ``count_successful_tool_calls`` must never count a
# synthetic closure as gathered evidence (the hunt path's plain-string closure
# predates that gate and stays as-is for prompt-compat).
_PARTIAL_CLOSURE_CONTENT: dict[str, Any] = {
"error": True,
"message": "not executed — tool-call budget exhausted",
}
async def _synthesize_partial_triage(
settings: Settings,
guard: EgressGuard | None,
gathered: list[Any],
*,
result_out: list[Any] | None = None,
) -> tuple[Any, list[Any]]:
"""Force a TriageReport from a budget-cut investigation loop's history.
Returns ``(report, repaired_history)``. The repaired history doubles as
``loop_messages`` for the downstream evidence/citation gates, so the
partial verdict earns the loop evidence exemption ONLY from tool results
that actually landed. Raises on any failure — the caller lands the honest
pipeline-fallback as the last resort.
``result_out`` receives the agent run result, so the caller can record the
model usage of this synthesis with the run's other usage.
"""
repaired = repair_dangling_tool_calls(gathered, closure_content=_PARTIAL_CLOSURE_CONTENT)
user_msg = replay_reasoning_context(repaired) + (
"Write the TriageReport now from the evidence already gathered above."
)
if guard is not None:
# The replayed history is already label space (the loop conversed over
# sanitized inputs); this sweep covers the lifted reasoning block.
user_msg = guard.sanitize_text(user_msg)
user_msg = _guard_egress(guard, user_msg, settings)
agent = build_partial_triage_synthesizer(
build_synthesizer_model(settings, temperature=settings.synthesizer_temperature),
output_mode=settings.synthesizer_output_mode,
)
async with asyncio.timeout(settings.investigation_turn_timeout_s):
result = await agent.run(
user_msg,
message_history=repaired,
usage_limits=UsageLimits(request_limit=3, tool_calls_limit=0),
)
if result_out is not None:
result_out.append(result)
return result.output, repaired
def build_synth_first_agent(
model: Model, *, output_mode: str = "tool"
) -> Agent[None, TriageReport]:
"""Build the synth Agent for the synth-first pipeline (no tools).
Identical to build_synthesizer except it uses the synth-first system
prompt that includes the gap_for_investigator + decision-template
rules.
``retries=3`` (vs pydantic_ai's default of 1) gives some reasoning
models multiple chances to emit valid
TriageReport JSON. Repeated batches showed a recurring fraction of
synth alerts failing with ``UnexpectedModelBehavior: Exceeded maximum
retries (1) for output validation`` — the same scenarios across runs,
so the schema-validation retry budget was the bottleneck, not a
transient model fault.
"""
from soc_ai.agent.prompts import SYNTH_FIRST_SYSTEM_PROMPT # noqa: PLC0415
return Agent(
model=model,
system_prompt=SYNTH_FIRST_SYSTEM_PROMPT,
output_type=_synth_output_type(output_mode),
retries=3,
)
_CODE_TOOL_CALL_ID = "pipeline-prefetch"
def _code_tool_exchange(tool_name: str, args: dict[str, Any], result: Any) -> list[Any]:
"""A tool call the pipeline made in code, as a call and a return part.
The gates read what a run gathered from its message history: the evidence
gate counts the returns, a tool citation resolves against the calls, the
support gate reads the results. A call the pipeline makes in place of the
loop's own turn joins that record in the shape the loop's call had. It is
never sent to the model as history.
"""
from pydantic_ai.messages import ( # noqa: PLC0415 - local, one caller
ModelRequest,
ModelResponse,
ToolCallPart,
ToolReturnPart,
)
return [
ModelResponse(
parts=[ToolCallPart(tool_name=tool_name, args=args, tool_call_id=_CODE_TOOL_CALL_ID)]
),
ModelRequest(
parts=[
ToolReturnPart(tool_name=tool_name, content=result, tool_call_id=_CODE_TOOL_CALL_ID)
]
),
]
def _guard_egress(guard: EgressGuard | None, text: str, settings: Settings) -> str:
"""Fail-closed residue check on a FINAL composed outbound analyst message.
Returns *text* unchanged when there is nothing to enforce — either no guard
(``analyst_cloud_redaction`` off ⇒ local model, byte-identical to the
pre-feature path) or fail-closed mode off (best-effort: a sanitize miss is
logged elsewhere but still egresses). When the guard is present AND
``settings.analyst_redaction_fail_closed`` is on, run the INDEPENDENT residue
sweep on the (already-sanitized) composed string and raise
:class:`~soc_ai.agent.egress_guard.EgressResidueError` if an internal
identifier survived — the caller catches it, does NOT call the model, and
lands a pipeline error.
Wire it as ``msg = _guard_egress(guard, guard.sanitize_text(msg), settings)``
so the check runs on the FINAL string that is about to hit the model, not on
any single fragment composed into it.
"""
if guard is None:
return text
guard.check_or_raise(text, fail_closed=settings.analyst_redaction_fail_closed)
return text
def _synth_failure_fallback_report(
alert_id: str,
phase: str,
exc: BaseException,
*,
retry_causes: list[str] | None = None,
served_backend: dict[str, Any] | None = None,
) -> Any:
"""Build a fallback TriageReport when the synth-first model raises.
When the synth fails schema-validation
retries (UnexpectedModelBehavior) or any other exception, the
pipeline previously emitted an ``error`` event and returned without
a ``triage_report``. That produced ``verdict=None`` rows in
``index.jsonl`` that were unscoreable. Now we synthesize a low-
confidence ``needs_more_info`` report from the failure so the row is
structured and the downstream post-validators + audit run uniformly.
The fallback report:
- ``verdict='needs_more_info'`` (correct: we genuinely don't know)
- ``confidence=0.3`` (visibly low — analyst sees it's a fallback)
- ``summary`` names the failure phase + exception type
- ``citations=['synth_first_failure']`` (single audit-trail marker)
- ``gap_for_investigator=None`` (don't recurse into Phase D)
- ``resolution={'provenance': 'pipeline_fallback', ...}`` — the marker that
makes this failure-driven needs_more_info render distinctly from a genuine
one, filterable, and excluded from the Needs-info KPI (E1.2). The verdict
stays needs_more_info by design; only the presentation differs. This is a
DIFFERENT key from the manual/chat override's ``resolved_via`` (added
post-hoc by the resolve endpoint), so the two never conflate — see
:func:`soc_ai.triage_models.is_pipeline_fallback`.
"""
from soc_ai.agent.triage import ( # noqa: PLC0415
PIPELINE_FALLBACK_PROVENANCE,
TriageReport,
)
return TriageReport(
verdict="needs_more_info",
confidence=0.3,
summary=(
f"Synth-first pipeline fallback: {phase} raised "
f"{type(exc).__name__}. The alert is recorded as "
f"needs_more_info pending investigator-path retry. "
f"Underlying error: {exc}"
),
citations=["synth_first_failure"],
recommended_actions=[],
gap_for_investigator=None,
resolution={
"provenance": PIPELINE_FALLBACK_PROVENANCE,
"phase": phase,
"error_type": type(exc).__name__,
"hint": _hint_for(exc, phase=phase),
# Only when captured (a schema-retry exhaustion): WHY each attempt
# failed, so the pipeline-error drilldown is actionable. Old rows /
# other failure classes keep their exact shape.
**({"retry_causes": retry_causes} if retry_causes else {}),
# WHICH deployment produced the failure, from LiteLLM's own response
# headers. The alias soc-ai asks for hides both aliasing and fallback
# routing, so without this a failure can't be attributed to a backend.
# Absent when it couldn't be determined.
**({"served_backend": served_backend} if served_backend else {}),
},
)
def _round2_failure_fallback(
alert_id: str,
round1: Any,
exc: BaseException,
retry_causes: list[str] | None = None,
served_backend: dict[str, Any] | None = None,
) -> Any:
"""Fallback verdict when the round-2 investigation loop / synth crashes.
The agent already reached a round-1 verdict before the (expensive, flakier)
round-2 path ran — don't discard it on a round-2 crash. If round-1 settled on
a confident verdict, land THAT (annotated in the summary so the operator sees
round-2 didn't finish) rather than erroring the whole run with no verdict. If
round-1 was itself inconclusive (skipped / untriaged), fall back to the
needs_more_info synth-failure report.
``round1`` is None when the round-1 call never ran. The pipeline skips that
call whenever the loop will overwrite its verdict (:func:`_round1_can_settle`),
so most runs now arrive here with nothing to preserve. The caller replays the
gathered history through the partial synthesizer BEFORE it comes here; this
function is the last resort, and it must stay honest about an empty hand.
"""
settled = {"true_positive", "false_positive"}
verdict = getattr(round1, "verdict", None)
if round1 is not None and verdict in settled:
note = (
f" (Round-2 deep investigation did not complete: {type(exc).__name__}; "
"showing the round-1 verdict.)"
)
try:
return round1.model_copy(
update={"summary": (getattr(round1, "summary", "") or "") + note}
)
except Exception:
return round1
return _synth_failure_fallback_report(
alert_id,
"investigation_loop_synth",
exc,
retry_causes=retry_causes,
served_backend=served_backend,
)
# Backwards-compat shim — pre-split callers used `build_agent(model, ctx)`
# and assumed Agent[None, TriageReport]. Their tests build a single agent
# manually; route to the synthesizer (no tools) since that produces the
# TriageReport. Tests that need tool-calling now build the investigator
# directly.
def build_agent( # pragma: no cover - thin alias
model: Model, ctx: InvestigationContext
) -> Agent[None, TriageReport]:
"""Deprecated: use build_investigator + build_synthesizer."""
return build_synthesizer(model)
# =====================================================================
# Investigation runner
# =====================================================================
_GATEWAY_BACKEND_HINT = (
"the LLM gateway could not reach its model backend. The engine behind the "
"LiteLLM route is down, is restarting or returns 5xx. Elasticsearch is not "
"the cause. Check that the serving engine is up and that the api_base of the "
"route points at it. Then retry."
)
_GATEWAY_UNREACHABLE_HINT = (
"the LLM gateway did not answer. soc-ai could not open a connection to it. "
"This phase makes no Elasticsearch call, so the grid is not the cause. "
"Check that the gateway runs and that litellm_base_url points at it. Then "
"retry."
)
_GRID_UNREACHABLE_HINT = (
"Elasticsearch and Security Onion are unreachable. Check that the SO grid is "
"online and that ES_HOSTS in the soc-ai .env points at the right node."
)
_GRID_SLOW_HINT = (
"the Security Onion grid did not answer in time. Check that the cluster is "
"online and not saturated. Then retry. If the cluster is healthy and slow, "
"raise es_request_timeout_s."
)
# Which dependency a phase can possibly have failed against.
#
# Working this out from the exception string is archaeology, and it got the
# answer wrong: a refused gateway connection raises a bare "Connection error."
# with no LiteLLM marker in it, so the ambiguous connect arm below fell through
# to the grid branch and the recorded pipeline error sent the operator off to
# check Elasticsearch during an LLM outage. The call site already knows.
#
# The synthesizer runs (round-1, round-2, the loop synth and its partial-report
# variant) are model-only: no tool is registered on them. The investigation
# loop does hold the grid tools, but every tool boundary catches its own
# exception and hands the model a structured result
# (:func:`soc_ai.agent.toolset._tool_error`), so no Elasticsearch exception can
# leave that phase either. Phase-A prefetch is the mirror image: it only ever
# reads the grid. Anything not listed here stays on the old string matching,
# which is right for phases that genuinely could be either.
_GATEWAY = "gateway"
_GRID = "grid"
_PHASE_DEPENDENCY: dict[str, str] = {
"synth_first_round1": _GATEWAY,
"synth_first_round2": _GATEWAY,
"investigation_loop": _GATEWAY,
"investigation_loop_synth": _GATEWAY,
"investigation_loop_partial_synth": _GATEWAY,
"prefetch": _GRID,
}
def _hint_for(exc: BaseException, *, phase: str | None = None) -> str | None:
"""Return a short, actionable hint string for the analyst, or None.
``phase`` is the pipeline phase that raised. When it names a dependency in
:data:`_PHASE_DEPENDENCY` it decides the ambiguous transport arms (connect
failure, timeout) instead of the exception text, because the call site is
evidence and the text is a guess. Omitted, every arm behaves exactly as it
did before.
"""
dependency = _PHASE_DEPENDENCY.get(phase or "")
if isinstance(exc, EgressResidueError):
# Name only the COUNT/CLASS — never the leaked values (that would defect
# on the fail-closed block). This hint lands in the report summary.
return (
f"redacted egress blocked: {len(exc.leaked)} internal identifiers "
"survived sanitization on the outbound analyst payload. soc-ai did not "
"call the model, because analyst_redaction_fail_closed is on. Investigate "
"locally, or use a local analyst model."
)
if isinstance(exc, OqlValidationError):
frag = getattr(exc, "fragment", None)
base = "The OQL validator rejected the query."
if frag:
return (
f"{base} The fragment it rejected is {frag!r}. Use the full ECS "
"field name, for example 'destination.ip'. A short name such as "
"'dest.ip' fails."
)
return f"{base} Check the field names against the OQL primer."
if isinstance(exc, SoApiError):
return "The alert id can be wrong. Check that the alert is in the grid."
msg = str(exc).lower()
# Pattern-match on the LiteLLM/PydanticAI error strings.
if "exceeded maximum output retries" in msg or (
"exceeded maximum retries" in msg and "output validation" in msg
):
return (
"the model output failed the TriageReport schema validation on each "
"attempt until the retry budget ran out. The error event and the "
"resolution of this run record the validation error of each attempt as "
"retry_causes. Read them. A schema failure that repeats usually means "
"that the analyst model or its gateway route changed its reasoning or "
"tool-call format. A transient fault is a rare cause."
)
if "token limit" in msg and "before any response" in msg:
return (
"the model hit its response-token cap while it was still reasoning, so "
"it produced no structured output. A reasoning model can spend the whole "
"budget on reasoning. Raise synthesizer_max_response_tokens in the config "
"console under Agent, or use a less verbose analyst model."
)
if "contextwindowexceeded" in msg or "context length" in msg:
return (
"The context window is exceeded. The transcript or the prompt is too "
"large. The alert context or the evidence can exceed the model window."
)
# MUST precede the generic timeout arm: a partial-read error's message reads
# "the search timed out before all shards answered", which that arm would
# blame on the model gateway — sending the operator to debug LiteLLM while a
# shard is down. Same misdirection class as the connect-error split below,
# which accounted for 8 of 15 recorded prod error events (2026-08-03).
if "partial search results" in msg:
return (
"the grid answered and did not read the whole index. Shards failed or "
"were unassigned, or a search stopped part-way. The results were "
"incomplete, so the run stopped. The run does not reason from a partial "
"view of the network. Check the shard health in Elasticsearch. Then retry."
)
if "timed out" in msg or "timeout" in msg:
# Same misdirection, mirrored: an ES ConnectionTimeout raised inside
# Phase-A prefetch renders as "Connection timed out" and was reported
# as a slow LiteLLM gateway, which prefetch never calls.
if dependency == _GRID:
return _GRID_SLOW_HINT
return "The LiteLLM gateway is slow or unreachable. Retry."
# Markers LiteLLM stamps into its OWN error strings. Their presence means the
# failure was raised by the gateway about a model call, never by the ES client.
is_gateway = any(
m in msg for m in ("litellm", "model_name:", "vllmexception", "openaiexception")
)
# Transport-layer "can't reach the host" is ambiguous BY TEXT: it fires for
# BOTH a dead Elasticsearch/SO grid and a dead model backend behind LiteLLM.
# Getting this backwards sent operators to debug Security Onion for a dead
# vLLM engine on 8 of 15 recorded prod error events (2026-08-03). It is not
# ambiguous by CALL SITE, so the phase answers first where it can, and only
# a phase that could genuinely be either falls back to the markers.
if "cannot connect to host" in msg or "connection error" in msg or "connection refused" in msg:
if dependency == _GATEWAY:
# A gateway that never accepted the connection is a different fault
# from one that answered and could not reach its own backend, and
# the two need different remedies. The marker still separates them.
return _GATEWAY_BACKEND_HINT if is_gateway else _GATEWAY_UNREACHABLE_HINT
if dependency == _GRID:
return _GRID_UNREACHABLE_HINT
if is_gateway:
return _GATEWAY_BACKEND_HINT
return _GRID_UNREACHABLE_HINT
# A bare gateway 5xx with no connection text ("status_code: 502, model_name:
# …, body:" — LiteLLM truncates the body on a proxy-level failure). These
# previously produced NO hint at all, leaving 5 of 15 prod error events with
# zero guidance during an outage. The status code alone is enough: a 5xx from
# the gateway is always a serving-side failure, never an analyst mistake.
if is_gateway and any(code in msg for code in ("status_code: 5", "status code: 5")):
return _GATEWAY_BACKEND_HINT
return None
def _retry_causes_from_messages(messages: Any) -> list[str]:
"""Lift the per-attempt validation errors out of a captured message history.
When a synth run dies with ``UnexpectedModelBehavior: Exceeded maximum
output retries``, the actual reasons live in the ``RetryPromptPart``s
pydantic-ai fed back to the model — and are otherwise discarded with the
run. Called on a ``capture_run_messages()`` capture from the failing run;
returns compact, whitespace-normalized strings (most recent last, at most
4, each capped at 400 chars) for the error event payload and the fallback
report's ``resolution`` marker. Defensive: a surprise message/part shape is
skipped, never raised — this runs inside an error handler.
"""
causes: list[str] = []
try:
for msg in messages or []:
for part in getattr(msg, "parts", None) or []:
if type(part).__name__ != "RetryPromptPart":
continue
try:
rendered = part.model_response()
except Exception:
rendered = str(getattr(part, "content", "") or "")
compact = " ".join(str(rendered).split())
if compact:
causes.append(compact[:400])
except Exception:
return causes[-4:]
return causes[-4:]
def _served_backend(attribution: Any) -> dict[str, Any] | None:
"""Normalize a gateway attribution sink for an error payload, or None.
Takes the dict filled by
:func:`soc_ai.agent._gateway_retry.capture_backend_attribution` — LiteLLM's
own response headers naming the deployment that actually served the call.
Deliberately NOT derived from ``ModelResponse.model_name``: that echoes the
ALIAS soc-ai requested, so on a route like ``qwen3.6-35b-instruct`` (aliased
to the deepseek-v4-flash engine) it reports the wrong model with total
confidence. ``api_base`` identifies the engine; ``attempted_fallbacks``
reports whether the gateway demoted the request to another deployment.
Defensive: runs inside error handlers, so a surprise shape yields None.
"""
try:
if not isinstance(attribution, dict) or not attribution:
return None
return {k: v for k, v in attribution.items() if v not in (None, "")}
except Exception:
return None
def _usage_event_payload(
round_num: int, u: Any, served_backend: dict[str, Any] | None = None
) -> dict[str, Any]:
"""Usage-event payload; ``served_backend`` when gateway attribution landed.
Error events have carried ``served_backend`` since the diagnosability fix;
without it on SUCCESS too, the one question an operator asks after a
fallback window — "which verdicts did the fallback backend produce?" — is
unanswerable. Key absent (not null) when attribution is unknown, so old
rows and no-attribution runs keep their exact shape.
"""
payload: dict[str, Any] = {
"phase": "synthesizer",
"round": round_num,
"tool_calls": u.tool_calls,
"requests": u.requests,
"input_tokens": u.input_tokens,
"output_tokens": u.output_tokens,
"total_tokens": u.total_tokens,
}
if served_backend:
payload["served_backend"] = served_backend
return payload
def _run_usage(run_result: Any) -> RunUsage | None:
"""The ``RunUsage`` of a pydantic-ai run result, or None for a test double.
A property since pydantic-ai 1.107; an older release or a stub exposes a
method. Anything that is not a ``RunUsage`` reads as None.
"""
try:
u = run_result.usage
if not isinstance(u, RunUsage) and callable(u):
u = u()
except Exception:
return None
return u if isinstance(u, RunUsage) else None
# The note a verdict carries when the loop stopped at its budget and the
# partial synthesis wrote it from the evidence gathered before the cutoff.
_BUDGET_CUT_NOTE = (
" (Investigation stopped at the tool-call budget; "
"verdict synthesized from the evidence gathered before "
"the cutoff.)"
)
def _cut_short_report(report: Any, note: str, guard: EgressGuard | None) -> Any:
"""A partial synthesis report, made fit to land as a cut-short verdict.
One rule for the primary loop and for a vote sample cut short. A cut-short
investigation must not assert high confidence (mirrors the hunt humility
clamp), never recurses into Phase D, and does not set the failure marker.
"""
out = report.model_copy(
update={
"confidence": min(report.confidence, 0.6),
"summary": (report.summary or "") + note,
# Never recurse into Phase D off a cut-short synthesis.
"gap_for_investigator": None,
# Model output: the failure marker is not its to set.
"resolution": None,
}
)
if guard is not None:
# Assignment-source restore — the gates below compare this report's
# text against RAW enriched/pivot values.
out = _desanitize_report(out, guard)
return out
def _usage_from_messages(messages: Sequence[Any]) -> RunUsage:
"""The usage of a run that raised, read off the messages it exchanged.
A run cut short by its budget has no result object to read. Each model
response carries the usage of its request, and each tool return is a tool
call the run executed. The output tool is not a tool the run executed.
"""
usage = RunUsage()
for message in messages:
kind = getattr(message, "kind", None)
if kind == "response":
usage.requests += 1
request_usage = getattr(message, "usage", None)
if isinstance(request_usage, RequestUsage):
usage.incr(request_usage)
elif kind == "request":
for part in getattr(message, "parts", []) or []:
if (
getattr(part, "part_kind", None) == "tool-return"
and getattr(part, "tool_name", None) != "final_result"
):
usage.tool_calls += 1
return usage
def _error_payload(
exc: BaseException,
*,
phase: str,
round_num: int,
retry_causes: list[str] | None = None,
served_backend: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Typed error event payload with phase/round/type/message + optional hint."""
payload: dict[str, Any] = {
"phase": phase,
"round": round_num,
"type": type(exc).__name__,
"message": str(exc),
}
hint = _hint_for(exc, phase=phase)
if hint:
payload["hint"] = hint
if retry_causes:
payload["retry_causes"] = retry_causes
if served_backend:
payload["served_backend"] = served_backend
return payload
def _is_high_stakes_alert(alert: SoAlert) -> bool:
"""Whether an alert is too high-stakes to auto-ack, even on a confident FP.
Reuses the existing deterministic rule-class signals (no new classifier):
- :func:`classify_alert` lands the alert in EXPLOIT_ATTEMPT / POST_COMPROMISE
when its Suricata ``classtype`` or ``signature_severity`` declares an
exploit / attack / malware / C2 family.
- :func:`_alert_signals_malware` (from :mod:`decision_templates`) catches the
malware/exploit token case where ``classtype`` is absent but the rule name
or ``rule_metadata.metadata_tags`` carry a malware-family signal (the
BPFDoor-style ET MALWARE label).
- :func:`_alert_signals_decoy` (same module) catches a deception sensor,
which is the one detection class that carries none of the above.
- SO's own severity: ``severity_label`` of critical/high, or
``severity_score`` >= 3 (SO buckets 3=high, 4=critical).
Any one of these makes the alert high-stakes. The verdict still stands —
we just refuse to *auto-write* an ack on it.
What makes a rule high-stakes is the classification its author declared,
read through the one classification table this codebase has
(:mod:`soc_ai.agent.classifier`). The token list is the fallback for alerts
that carry no classification at all, and it is deliberately not where new
signals get added: it also drives benign-template routing across the whole
decision layer, so a token added here to close an auto-ack hole changes
verdicts for every rule whose name happens to contain it, and it would still
leave the next unclassified classtype open. "GPL SHELLCODE x86 setgid 0"
got past all four arms for exactly that reason: its classification,
system-call-detect, had no routing decision. The fix was to make the
routing map total over the classification table, not to add "shellcode".
The decoy arm is the exception to that paragraph, and it is there because
the other four are all readings of fields a honeypot document does not
have. Measured on the deployed grid on 2026-09-06: OpenCanary writes no
``event.severity_label``, no ``event.severity`` and no ``rule.name``, and it
is not a Suricata signature so it has no ``classtype`` either. Arms one and
two read the classtype, arm three reads the rule name, arm four reads the
severity — every one of them is looking at an absent field, so all four
return False and the alert is treated as ordinary. Elastic Defend is only
covered because it happens to write a 99 into ``event.severity``; that is an
accident of one shipper, not a guard. The decoy arm reads
``event.dataset``/``event.module`` instead, which is the field the honeypot
does write, and which the deception-sensor predicate already keys on for the
benign-verdict gate (:func:`~soc_ai.agent.gates._refuse_benign_verdict_without_baseline`).
This is deliberately NOT a restatement of that gate. That one coerces a
false_positive verdict on the alert being investigated, so on the direct
path it already stops the ack by making the verdict ineligible. The write
volume is not on the direct path: 110,693 of the recorded grid writes came
out of :func:`soc_ai.webui.autotriage._ack_inherited_fps`, which acks an
alert on a verdict produced for a DIFFERENT alert in the same cluster. That
alert's own verdict gate never ran, so this guard is the only thing between
a honeypot interaction and an unattended acknowledgement — and an
acknowledged honeypot hit is a silenced intrusion, since a decoy has no
benign population to be a false positive from.
"""
from soc_ai.agent.decision_templates import ( # noqa: PLC0415 — circular
_ATTACK_CLASSTYPES,
_alert_signals_decoy,
_alert_signals_malware,
)
if classify_alert(alert) in (AlertClass.EXPLOIT_ATTEMPT, AlertClass.POST_COMPROMISE):
return True
# Attack-class Suricata classtypes (denial-of-service / exploit-kit / …) that
# the Oracle-escalation guard treats as attack-signalling but classify_alert's
# _CLASSTYPE_MAP doesn't map — mirror _rule_signals_attack so the auto-ack cap
# and the escalation guard never disagree on what's an attack.
if normalize_classtype(alert.classtype) in _ATTACK_CLASSTYPES:
return True
if _alert_signals_malware(alert):
return True
# A deception sensor. Nothing has a legitimate reason to touch a decoy, so
# there is no benign population for an ack to be drawn from — and the
# document carries none of the fields the four arms above read.
if _alert_signals_decoy(alert):
return True
sev_label = (alert.severity_label or "").strip().lower()
if sev_label in ("critical", "high"):
return True
return alert.severity_score is not None and alert.severity_score >= 3
async def _standing_analyst_verdict(ctx: InvestigationContext, es_id: str) -> tuple[Any, bool]:
"""``(verdict, read_ok)``: the analyst verdict that stands on the alert.
``verdict`` is an :class:`~soc_ai.store.investigations.AnalystVerdict` or
None. ``read_ok`` is False when the store did not answer. No store on the
context (CLI, eval, a direct caller) has no analyst verdict to read, so
that is ``(None, True)``.
"""
maker = ctx.db_sessionmaker
if maker is None or not es_id:
return None, True
from soc_ai.store.investigations import analyst_verdicts_for_alerts # noqa: PLC0415
try:
async with maker() as db:
found = await analyst_verdicts_for_alerts(db, [es_id])
except Exception as exc:
_LOGGER.warning("analyst verdict lookup failed for alert %s: %s", es_id, exc)
return None, False
return found.get(es_id), True
async def _analyst_verdict_holds_ack(
report: TriageReport,
es_id: str,
*,
ctx: InvestigationContext,
emit_ev: Any,
) -> StepEvent | None:
"""The ``auto_ack_skipped`` event when an analyst verdict forbids the ack.
An unattended write never acts against a person. The analyst set the
verdict of this alert, on this run or on any earlier run of it, and it is
not false positive: the ack waits. The dogfood of 2026-10-19 (R2-D1) found
the path. The analyst overrode a false positive to true positive, pressed
Re-run, and the new run reached false positive at 0.80 and acknowledged the
alert in Security Onion. Nothing on that path read the analyst verdict.
FAILS CLOSED. A store that does not answer is not permission to write to
the analyst's grid.
"""
analyst, read_ok = await _standing_analyst_verdict(ctx, es_id)
if read_ok and (analyst is None or analyst.verdict == "false_positive"):
return None
payload: dict[str, Any] = {
"es_id": es_id,
"reason": "analyst_verdict" if read_ok else "analyst_verdict_unknown",
"confidence": report.confidence,
"threshold": ctx.settings.auto_ack_fp_threshold,
}
if analyst is not None:
payload["analyst_verdict"] = analyst.verdict
payload["analyst_investigation_id"] = analyst.investigation_id
payload["analyst"] = analyst.resolved_by
_LOGGER.info(
"auto-ack suppressed for alert %s: %s",
es_id,
f"an analyst set {analyst.verdict}" if analyst is not None else "analyst verdict unknown",
)
held: StepEvent = emit_ev("auto_ack_skipped", payload)
return held
async def maybe_auto_ack_fp(
report: TriageReport,
es_id: str,
*,
alert: SoAlert,
ctx: InvestigationContext,
emit_ev: Any,
audit_ev: Any,
investigated: bool,
citation_coverage: float | None = None,
oracle_dissent: bool = False,
) -> StepEvent | None:
"""Auto-acknowledge a high-confidence FP alert in Security Onion.
Called from both the synth-first and legacy finalization paths after the
final verdict and confidence are settled (including Oracle adjudication).
Gating (all must be true):
- ``settings.auto_ack_fp_enabled`` is True
- ``report.verdict == "false_positive"``
- ``report.confidence >= settings.auto_ack_fp_threshold``
- ``investigated`` — the run retrieved something
- the report SUPPORTS ITSELF — it cites something, and what it cites
resolved (see ``citation_coverage`` below)
- the alert is NOT high-stakes (see :func:`_is_high_stakes_alert`)
- no analyst verdict other than false positive stands on the alert, on
this run or on any earlier run of it (see :func:`_analyst_verdict_holds_ack`)
``citation_coverage`` is the resolved share of the report's citations, as
:func:`soc_ai.agent.gates._resolve_citations` measured them, or ``None``
when it does not describe THIS report (the Oracle rewrote the verdict after
the local resolution ran, so the local coverage is about different
citations). None means the empty-citation check still applies and the
zero-coverage one does not.
The citation state is a GATE here, not only an input to confidence. Nothing
else in the chain refuses an uncited verdict on the way to the grid: the
coverage cap deliberately stands down when there are no citations to
measure (see :func:`~soc_ai.agent.gates._citation_confidence_cap`), so an
uncited report keeps its confidence and sails past the threshold. On the
deployed instance 1,315 of 3,696 completed runs shipped a report that cited
nothing and 834 of them were acknowledged in Security Onion, at a mean
confidence of 0.77. A coverage of exactly zero over a non-empty citation
list is refused for the stronger reason: the report named its grounds and
not one of them could be resolved to anything the run retrieved.
``investigated`` is True when the run made at least one successful tool
call, dispatched a Phase-D targeted tool, or the Oracle's own loop called a
tool. Confidence was previously the only quantitative condition, and a
decision template supplies confidence without supplying evidence: on the
production instance 13 alerts were acknowledged in Security Onion at 0.85 to
0.90 with nothing looked up about them, nine of them the same exploitation-
attempt signature, repeating daily. An unattended write to the analyst's
grid is the last thing that should rest on a verdict the run did no work
for. The verdict itself is unaffected: a template-settled false positive
still reads as a confident false positive on the console, it just waits for
a person to press the button.
The high-stakes guard is a blast-radius cap: a prompt-injected confident
``false_positive`` must never auto-ack a critical/high-severity or
malware/exploit/attack-class alert. We skip the auto-write (the verdict is
unchanged) and leave the ack to a human.
Best-effort: any write error is logged as a warning and does NOT propagate.
The investigation is never failed by auto-ack.
The ack is written DIRECTLY (via :func:`execute_write_tool`) — it does NOT
wait for an analyst action. That is the whole point of the opt-in:
``auto_ack_fp_enabled`` is an explicit operator decision to let confident,
low-stakes FP acks write to SO unattended. The audit logger is routed
through so the unattended write is always recorded (fail-closed on intent).
Coupling: this only runs from investigation finalisation, so an alert is
auto-acked ONLY if it is investigated while the toggle is on — there is no
retroactive backlog sweep. See ``Settings.auto_ack_fp_enabled`` for how this
interacts with the auto-triage floor.
Returns the ``auto_ack`` StepEvent (for the caller to yield into the stream)
when the write was attempted; an ``auto_ack_skipped`` StepEvent (with
``reason`` = ``analyst_verdict`` | ``analyst_verdict_unknown`` |
``below_threshold`` | ``no_investigation`` | ``uncited`` |
``high_stakes`` | ``oracle_dissent``)
when auto-ack was armed for this FP but a guard held it back — recorded so
the drawer can explain why the pending ack needs a human; or ``None`` when
auto-ack simply doesn't apply (disabled, or a non-FP verdict).
"""
settings = ctx.settings
if not settings.auto_ack_fp_enabled:
return None
if report.verdict != "false_positive":
return None
analyst_ev = await _analyst_verdict_holds_ack(report, es_id, ctx=ctx, emit_ev=emit_ev)
if analyst_ev is not None:
return analyst_ev
if oracle_dissent:
# The Oracle answered another class with no evidence that resolves.
# Its answer is an opinion on the run and the local verdict stands,
# but a verdict the second opinion disputes is not written back to
# the grid unattended (design 2026-10-04, "No override without
# evidence").
dissent_ev: StepEvent = emit_ev(
"auto_ack_skipped",
{
"es_id": es_id,
"reason": "oracle_dissent",
"confidence": report.confidence,
"threshold": settings.auto_ack_fp_threshold,
},
)
return dissent_ev
if (report.confidence or 0.0) < settings.auto_ack_fp_threshold:
# Record WHY this FP wasn't auto-acked so the drawer can explain the
# pending ack instead of leaving the analyst to guess (two identical-
# looking ack cards behaved differently in dogfood 2026-07-15).
skipped_ev: StepEvent = emit_ev(
"auto_ack_skipped",
{
"es_id": es_id,
"reason": "below_threshold",
"confidence": report.confidence,
"threshold": settings.auto_ack_fp_threshold,
},
)
return skipped_ev
if not investigated:
# Nothing was retrieved this run. The verdict may well be right, and it
# stands, but it is not something to write back to the analyst's grid
# unattended.
_LOGGER.info(
"auto-ack suppressed for alert %s: no tool call, targeted dispatch "
"or Oracle retrieval behind the verdict (conf=%.2f)",
es_id,
report.confidence or 0.0,
)
uninvestigated_ev: StepEvent = emit_ev(
"auto_ack_skipped",
{
"es_id": es_id,
"reason": "no_investigation",
"confidence": report.confidence,
"threshold": settings.auto_ack_fp_threshold,
},
)
return uninvestigated_ev
n_citations = len(report.citations or [])
uncited_reason = None
if n_citations == 0:
uncited_reason = "the report cited nothing"
elif citation_coverage is not None and citation_coverage <= 0.0:
uncited_reason = "none of the report's citations resolved"
if uncited_reason is not None:
# Retrieval and support are different questions and this one has never
# been asked on the way to an unattended write. A run can call tools and
# still hand the analyst a verdict with nothing under it; on the
# deployed instance 714 of the 834 acknowledged uncited verdicts had
# retrieved something, so the retrieval bar alone lets almost all of
# them through. The verdict stands and still reads as a confident false
# positive — the write waits for a person.
_LOGGER.info(
"auto-ack suppressed for alert %s: %s (conf=%.2f, citations=%d)",
es_id,
uncited_reason,
report.confidence or 0.0,
n_citations,
)
uncited_ev: StepEvent = emit_ev(
"auto_ack_skipped",
{
"es_id": es_id,
"reason": "uncited",
"confidence": report.confidence,
"threshold": settings.auto_ack_fp_threshold,
"citations": n_citations,
"coverage_ratio": citation_coverage,
},
)
return uncited_ev
if _is_high_stakes_alert(alert):
# Blast-radius cap: never auto-write an ack on a high-stakes alert, even
# on a confident FP. The verdict stands; a human must ack it.
_LOGGER.info(
"auto-ack suppressed for high-stakes alert %s "
"(class/severity gate) despite verdict=false_positive conf=%.2f",
es_id,
report.confidence or 0.0,
)
high_stakes_ev: StepEvent = emit_ev(
"auto_ack_skipped",
{
"es_id": es_id,
"reason": "high_stakes",
"confidence": report.confidence,
"threshold": settings.auto_ack_fp_threshold,
},
)
return high_stakes_ev
_LOGGER.info(
"auto-acking FP alert %s (confidence=%.2f >= threshold=%.2f)",
es_id,
report.confidence or 0.0,
settings.auto_ack_fp_threshold,
)
success: bool
try:
# Auto-ack is an unattended write. Route the audit logger through so the
# ack is recorded as a mutating tool_call intent BEFORE the SO write
# (fail-closed under audit_fail_closed) and a result record after — an
# analyst-review-free write must always land in the audit trail.
_result, error = await execute_write_tool(
"ack_alert",
{"alert_id": es_id},
auth=ctx.auth,
settings=settings,
audit=ctx.audit,
session_id=f"auto-ack:{es_id}",
user="auto-ack",
)
if error:
_LOGGER.warning("auto-ack write failed for alert %s: %s", es_id, error)
success = False
else:
success = True
except Exception as exc:
_LOGGER.warning("auto-ack unexpected error for alert %s: %s", es_id, exc)
success = False
ack_ev: StepEvent = emit_ev(
"auto_ack",
{
"es_id": es_id,
"confidence": report.confidence,
"threshold": settings.auto_ack_fp_threshold,
"success": success,
},
)
try:
await audit_ev(ack_ev)
except Exception as exc:
_LOGGER.warning("auto-ack audit log failed for alert %s: %s", es_id, exc)
# Yield is caller's responsibility — we return the event for the caller to yield.
# (Generators can't be called from non-generator helpers in Python.)
return ack_ev
async def _maybe_auto_ack_fp_gated(
report: TriageReport,
es_id: str,
*,
alert: SoAlert,
ctx: InvestigationContext,
emit_ev: Any,
audit_ev: Any,
investigated: bool = False,
citation_coverage: float | None = None,
allow_so_writes: bool = True,
oracle_dissent: bool = False,
) -> StepEvent | None:
"""Call-site guard in front of :func:`maybe_auto_ack_fp` (Task 6, finding
promotion): a promoted hunt finding's ``alert_es_id`` is a cited telemetry
document, not an SO alert — there is nothing in Security Onion to ack, ever,
regardless of verdict/confidence/threshold. When ``allow_so_writes`` is
False, the write is skipped BEFORE ``maybe_auto_ack_fp`` (and therefore
``execute_write_tool``) is ever reached, and an ``auto_ack_skipped`` event
with reason ``promoted_finding`` records why — mirroring the
``below_threshold`` / ``no_investigation`` / ``uncited`` / ``high_stakes``
skip events maybe_auto_ack_fp itself emits for its own held-back cases.
``investigated`` defaults False so a caller that has not worked out whether
the run retrieved anything cannot write to Security Onion by omission.
``citation_coverage`` defaults None, which is the safe default for the other
reason: the empty-citation refusal applies regardless, and only the
stricter zero-coverage refusal needs a number to stand on.
"""
if not allow_so_writes:
# emit_ev is Any-typed at this seam; annotate so --strict sees StepEvent.
skipped: StepEvent = emit_ev(
"auto_ack_skipped", {"es_id": es_id, "reason": "promoted_finding"}
)
return skipped
return await maybe_auto_ack_fp(
report,
es_id,
alert=alert,
ctx=ctx,
emit_ev=emit_ev,
audit_ev=audit_ev,
investigated=investigated,
citation_coverage=citation_coverage,
oracle_dissent=oracle_dissent,
)
async def investigate(
alert_id: str,
*,
ctx: InvestigationContext,
focus_hint: str | None = None,
deep: bool = False,
allow_so_writes: bool = True,
focus_origin: FocusOrigin = "rerun",
subject: HuntSubject | None = None,
) -> AsyncIterator[StepEvent]:
"""Public entry point for the synth-first triage pipeline.
Async-yields :class:`StepEvent` items throughout the run.
``subject`` (optional, D2): a HUNT subject. The run reads the hunt: its
objective, its findings and every document they cite. ``alert_id`` stays
the anchor document the prefetch reads, so the tools keep their time
anchor, but the subject block replaces the alert block in the investigator
prompt. For a finding subject the verdict judges the claim of the promoted
finding, and the rest of the hunt is context. For a lead subject the
verdict answers the hunt's objective (change E, 2026-10-30). ``None`` is
the ordinary alert subject and nothing changes.
``deep`` (optional): force the full tool-driven investigation loop for THIS
run regardless of ``fast_triage_enabled`` — the analyst's "deep re-run" of
a heuristic verdict must go deeper than the heuristic it double-checks.
``focus_hint`` (optional): when this run was launched to close a prior
``needs_more_info`` verdict (the "request more info" action), the prior
open questions are passed here and woven into the seed prompt so the fresh
run TARGETS those gaps. ``None`` ⇒ normal cold run.
``allow_so_writes`` (default True): a promoted hunt finding's anchor is
cited telemetry, not an SO alert — there is nothing to ack. When False
(finding-promotion, Task 6), the opt-in auto-ack step is skipped entirely
regardless of verdict/confidence.
``focus_origin`` (default ``"rerun"``): which header ``focus_hint`` gets
in the seed prompt — see :func:`soc_ai.agent.prompts.format_focus_hint_block`.
``"hunt_finding"`` for a promoted finding's framing; the default assumes
``focus_hint`` (when present) is a prior run's open questions.
**Pipeline stages** (all executed by :func:`_run_synth_first_pipeline`):
1. **Prefetch / enrichment** — alert fields extracted; IOC enrichment tools
run; enrichment results written to the cross-alert cache.
2. **Decision-template candidate** — deterministic template match attempted;
strong grounded templates may produce a verdict without any LLM call.
3. **Tool-less synthesis round 1** — analyst model synthesizes over the
prefetch payload; emits a :class:`TriageReport` candidate.
4. **Phase D or investigation loop** (optional) — if round-1 leaves a
``gap_for_investigator``, either a single targeted tool dispatch (Phase D)
or a full tool-equipped investigation loop gathers additional evidence.
5. **Synthesis over result** — analyst model re-synthesizes over the
combined evidence to produce the final :class:`TriageReport`.
6. **Deterministic gates / downgrades** — evidence floor rewrite, malware
payload gate (GATE A), distinctive-token citation gate (GATE C), targeted
verdict downgrades.
7. **Oracle escalation** (optional) — a second analyst-model opinion when the
Oracle feature is enabled and the confidence warrants it.
8. **Final TriageReport** — recommended actions persisted (analyst-executed
via the actions API); optional auto-ack applied.
Per-phase ``usage`` SSE events expose real token / tool-call counts so we
can right-size limits and the confidence floor with audit data.
The pipeline constructs its own agents internally; callers supply only the
alert, the context, and an optional focus hint.
"""
# Single pipeline: synth-first (A→B→C→optional D→C round 2). The legacy
# investigator→synthesizer loop was deleted 2026-07 after synth-first had
# been the production default since 2026-05-29 (0a02f9c).
async for ev in _run_synth_first_pipeline(
alert_id=alert_id,
ctx=ctx,
focus_hint=focus_hint,
deep=deep,
# One source of truth for the unattended write on a hunt subject:
# nothing in Security Onion holds a hunt, so there is never anything to
# ack. Forced here rather than trusted from the caller, the same rule
# the hunt manager applies to ``kind='hunt'``.
allow_so_writes=allow_so_writes and subject is None,
focus_origin=focus_origin,
subject=subject,
):
yield ev
# Citation-path prefixes that point at REAL gathered evidence (tool returns,
# enrichment results, or pivot events) rather than the alert's own fields.
# A verdict cited only against `alert.*` paths is self-referential — it
# restates the alert rather than investigating it. The QVOD beacon false-FP
# cited 5 `alert.*` paths (rule_name, payload_printable, classtype,
# rule_metadata.*) and called a Cobalt Strike beacon benign on that basis.
#
# The document lists come first and ``_GATHERED_DOCUMENT_ATTRS`` names them, so
# a list added here is read for ids too.
_GATHERED_DOCUMENT_ATTRS: tuple[str, ...] = (
"community_id_events",
"host_events",
"user_events",
"process_events",
"file_events",
# The documents a hunt subject cites (D2). Empty on every alert run.
"subject_documents",
)
_EVIDENCE_PATH_PREFIXES: tuple[str, ...] = (
*_GATHERED_DOCUMENT_ATTRS,
"enrichments",
"typed_zeek",
)
def _pivot_event_ids(alert_ctx: Any) -> set[str]:
"""Collect the ES ``_id`` of every prefetched pivot event.
An ``id``-shaped citation only counts as real evidence when it matches a
pivot event the orchestrator actually pulled — otherwise the model could
fabricate a long-alphanumeric string and have it trusted by
:func:`_classify_citation`'s id branch.
"""
ids: set[str] = set()
for pivot_attr in _GATHERED_DOCUMENT_ATTRS:
for ev in getattr(alert_ctx, pivot_attr, None) or []:
ev_id = getattr(ev, "id", None)
if isinstance(ev_id, str) and ev_id:
ids.add(ev_id)
return ids
def _is_evidence_backed(report: Any, enriched: Any, *, messages: list[Any] | None = None) -> bool:
"""True only when the verdict rests on REAL gathered evidence.
Theme-1 Task 1. "Real evidence" means at least one citation resolves to
an actual tool/enrichment result or a prefetched pivot event — NOT merely
to a self-referential field on the alert under triage
(``alert.rule_name``, ``alert.payload_printable``, ``alert.classtype``,
``alert.rule_metadata.*``, …). A citation qualifies when:
- it names a tool (``(tool t_query_zeek_logs)`` / bare ``t_…``) that was
actually invoked in the loop's message history, OR
- it is a path into a pivot list / enrichment / typed-Zeek block
(``community_id_events.0.…``, ``enrichments.1.2.3.4.…``, …) that
resolves against the bundle, OR
- it is an id that matches a prefetched pivot event's ``_id``.
Pure ``alert.*`` paths (and bare ``alert.*`` field names) are
self-referential and never count. An empty citation list is, by
definition, not evidence-backed.
``messages`` is the loop's ``all_messages()`` history when available
(lets tool citations resolve against real ``ToolCallPart`` events). At
round-1 there is no message history, so tool citations can't be proven —
which is correct: a zero-tool round-1 guess naming a tool it never
called is exactly what this gate exists to catch.
"""
citations = list(getattr(report, "citations", None) or [])
if not citations:
return False
pivot_ids = _pivot_event_ids(enriched)
for c in citations:
kind, target = _classify_citation(c)
if kind == "tool":
if target and _tool_was_invoked([], target, messages=messages):
return True
elif kind == "path":
if not target:
continue
head = target.split(".", 1)[0]
# `alert.*` is self-referential; only non-alert evidence paths count.
# Fix A: path citations into pivot/enrichment lists only count when
# messages is not None — i.e. a real investigation loop ran. At
# round 1 (messages=None) these paths come from _materialize_prefetch_evidence;
# citing them is restating the prefetch, not investigation.
if (
head in _EVIDENCE_PATH_PREFIXES
and messages is not None
and _path_exists_in_alert(enriched, target)
):
return True
elif (
kind == "id"
and target
and target in pivot_ids
# Fix A: id citations matching prefetched pivot events only count when
# messages is not None. At round 1 the synth was given these ids via
# _materialize_prefetch_evidence; a zero-tool citation of a prefetched
# id is not evidence of investigation.
and messages is not None
):
return True
return False
# The trigger behind a ``definitely_investigate`` run, recorded on the
# ``investigation_loop_entered`` event. The timeline named one cause, an
# exploit or malware signal, for all three triggers. An NTLMv1 alert matched
# by a provisional template read as malware (dogfood 2026-10-19, V1-A1).
DI_TRIGGER_MALWARE_RULE = "malware_rule"
DI_TRIGGER_HOST_THREAT = "host_concurrent_threat"
DI_TRIGGER_PROVISIONAL_TEMPLATE = "provisional_template"
def _definitely_investigate(enriched: Any, candidate: Any) -> bool:
"""Whether a report-INDEPENDENT trigger fires. See :func:`_definitely_investigate_trigger`."""
return _definitely_investigate_trigger(enriched, candidate) is not None
def _definitely_investigate_trigger(enriched: Any, candidate: Any) -> str | None:
"""Report-INDEPENDENT investigate triggers: the one that fired, or None.
A trigger fires when the case will run the investigation loop REGARDLESS of
the round-1 verdict: a malware/exploit-signalled rule (the QVOD/beacon/BPFDoor
failure mode: a zero-tool synth citing prefetched pivots is not evidence of
benignness), a threat alert on the same host, or a PROVISIONAL benign
decision template. The checks run in that order, and the first that holds
names the trigger.
A provisional template proposes a benign verdict it is not entitled to
settle: its grounds are properties of the endpoints or the absence of a
reputation hit, never what the rule detected. The case therefore needs
evidence before it can close, and the cheapest way to get it is to skip
round 1 and go straight to the loop. This subsumes the older
EXTERNAL_REPUTATION_TEMPLATES trigger (pushplanet settled FP on an unknown
external host with zero tools) and adds ``clean_internal_traffic``, which
settled nine ET HUNTING OGNL exploitation-attempt alerts the same way.
Because these don't depend on the round-1 report, the pipeline pre-checks
this BEFORE Phase C and skips the ~10-15s round-1 synth call when True — that
verdict would be discarded the moment the loop runs.
"""
from soc_ai.agent.decision_templates import ( # noqa: PLC0415
_host_has_concurrent_threat,
_rule_signals_malware,
)
if _rule_signals_malware(enriched):
return DI_TRIGGER_MALWARE_RULE
# Host-context trigger: the focus alert may look benign
# (internal east-west, INFO) while its host is concurrently beaconing to a
# C2 — the "context not being considered" failure. A threat-signalling pivot
# alert on the same host/flow forces a real investigation of this leg.
if _host_has_concurrent_threat(enriched):
return DI_TRIGGER_HOST_THREAT
if (
candidate is not None
and getattr(candidate, "verdict", None) == "false_positive"
and getattr(candidate, "authority", "provisional") != "dispositive"
):
return DI_TRIGGER_PROVISIONAL_TEMPLATE
return None
def _round1_can_settle(enriched: Any, candidate: Any) -> bool:
"""True when a round-1 verdict is allowed to end the run.
W2. The pipeline reads this BEFORE the round-1 call and
:func:`_should_investigate` reads it after, so the two cannot disagree.
Round 1 has no tools and no message history, so
:func:`_is_evidence_backed` rejects every round-1 report: a zero-tool
citation proves nothing. One case is left. A DISPOSITIVE decision
template already cleared the alert ``false_positive`` on grounds that
read what the rule detected, and round 1 only has to agree with it.
Every other alert runs the loop, and the loop overwrites the round-1
verdict, so the call before it is pure cost: 18 s on production and 67 s
on the range, for 83% of the runs that made it.
This predicate is report-independent. The caller adds the one part that
needs the report itself: round 1 must also say ``false_positive``.
"""
# A malware or exploit signal, a concurrent threat on the host, or a
# PROVISIONAL benign template all force the loop whatever round 1 says.
if _definitely_investigate(enriched, candidate):
return False
return (
candidate is not None
and getattr(candidate, "verdict", None) == "false_positive"
and getattr(candidate, "authority", "provisional") == "dispositive"
)
def _should_investigate(report: Any, enriched: Any, candidate: Any) -> bool:
"""Decide whether to run the real investigation loop after round 1.
Theme-1 Task 1. True when ALL hold:
- ``investigate_when_unsure`` is on (settings flag is read by the
caller, passed positionally via ``report``'s pipeline — see below),
- the round-1 verdict is NOT evidence-backed
(:func:`_is_evidence_backed`), AND
- the alert is non-trivial — i.e. NOT a routine benign that a DISPOSITIVE
decision template already cleared without any malware signal.
"Trivially benign" = a non-malware-signalling alert whose decision
template landed a benign verdict AND was entitled to settle it
(``needs_more_info`` is treated as non-benign; only ``false_positive``
from a dispositive template on a non-malware rule short-circuits). Such
alerts keep the fast zero-tool path; everything else that lacks evidence
gets the loop.
Note: the ``investigate_when_unsure`` flag check lives at the call site
(it needs ``ctx.settings``); this helper assumes it has already passed
and concerns itself only with the evidence + triviality gates.
"""
# Report-INDEPENDENT triggers (malware/exploit signal, external-reputation
# template). Extracted to _definitely_investigate so the pipeline can
# pre-check them BEFORE Phase C and skip the wasted round-1 synth. Checked
# before _is_evidence_backed because a template's own cited evidence (or a
# round-1 FP citing a prefetched pivot) would otherwise read as "backed".
if _definitely_investigate(enriched, candidate):
return True
if _is_evidence_backed(report, enriched):
return False
# Routine benign: a DISPOSITIVE decision template cleared it false_positive
# on a rule with no malware signal → keep the fast path. Everything else
# that lacks evidence gets the loop. _round1_can_settle holds that test and
# the pipeline reads the SAME function before the round-1 call, so the rule
# that skips the call and the rule that accepts its answer are one rule.
return not (
_round1_can_settle(enriched, candidate)
and getattr(report, "verdict", None) == "false_positive"
)
def _synth_reasoning_payload(run_result: Any) -> dict[str, Any] | None:
"""Project a synthesizer run's thinking into a ``model_response`` payload.
The synth-first synthesizer runs (round-1 / loop-synth / Phase-D round-2)
complete via ``agent.run`` — nothing walks their messages, so their
ThinkingParts (deepseek ``reasoning_content`` bound via the model profile)
were silently dropped and a no-loop investigation surfaced NO "Model
reasoning" panel. Same part projection as :func:`_walk_message` (named
ThinkingPart content, plus inline ``<think>`` blocks in TextParts), but
collapsed to ONE payload for the whole run since a no-tools synthesis is a
single model turn. Returns None when there is no non-empty trace (emit
nothing) or on any message-shape surprise (defensive: never fail a verdict
over explainability bookkeeping).
"""
try:
messages = run_result.all_messages()
except Exception:
return None
traces: list[str] = []
texts: list[str] = []
try:
for msg in messages or []:
for part in getattr(msg, "parts", []) or []:
ptype = type(part).__name__
if ptype == "RetryPromptPart":
# pydantic-ai rejected the PRECEDING attempt (a
# schema-validation retry). Everything accumulated so far
# belongs to a superseded attempt — drop it so the panel
# reflects only the attempt that produced the accepted report.
traces.clear()
texts.clear()
elif ptype == "ThinkingPart":
content = getattr(part, "content", "") or ""
if content.strip():
traces.append(content.strip())
elif ptype == "TextPart":
content = getattr(part, "content", "") or ""
trace, cleaned = extract_reasoning_trace(content)
if trace and trace.strip():
traces.append(trace.strip())
if cleaned.strip():
texts.append(cleaned.strip())
except Exception:
return None
if not traces:
return None
return {"content": "\n\n".join(texts), "reasoning_trace": "\n\n".join(traces)}
async def _walk_message(
msg: Any,
ev_factory: Any,
*,
phase: str | None = None,
round_num: int | None = None,
) -> AsyncIterator[StepEvent]:
"""Yield StepEvent records for every interesting part of a PydanticAI message.
PydanticAI's message objects are a structured (model_request, model_response,
tool_call, tool_return) sequence; we project only what the SSE consumer cares
about and capture the ``<think>`` trace separately for audit. ``phase`` /
``round_num`` are stamped onto every emitted payload so consumers can
distinguish investigator-vs-synthesizer events and round 1 vs round 2.
"""
def _stamp(payload: dict[str, Any]) -> dict[str, Any]:
if phase is not None:
payload["phase"] = phase
if round_num is not None:
payload["round"] = round_num
return payload
parts = getattr(msg, "parts", []) or []
# Track the most-recent ThinkingPart so we can attach it to the next
# TextPart (or emit it standalone if no TextPart follows in this message).
pending_trace: str | None = None
for part in parts:
ptype = type(part).__name__
if ptype == "ThinkingPart":
content = getattr(part, "content", "") or ""
if content:
pending_trace = (pending_trace + "\n\n" + content) if pending_trace else content
continue
if ptype == "TextPart":
content = getattr(part, "content", "") or ""
trace, cleaned = extract_reasoning_trace(content)
payload: dict[str, Any] = {"content": cleaned}
# Prefer a same-message ThinkingPart trace; fall back to inline
# <think>...</think> if the model embedded the trace in text.
if pending_trace:
payload["reasoning_trace"] = pending_trace
pending_trace = None
elif trace:
payload["reasoning_trace"] = trace
yield ev_factory("model_response", _stamp(payload))
# The ``search_tools`` call that loads a deferred tool (the standard
# class) is a tool call too: it costs a turn, and the trail shows it.
elif ptype in ("ToolCallPart", "ToolSearchCallPart"):
yield ev_factory(
"tool_call",
_stamp(
{
"tool_name": getattr(part, "tool_name", ""),
"args": getattr(part, "args", {}),
"tool_call_id": getattr(part, "tool_call_id", ""),
}
),
)
elif ptype in ("ToolReturnPart", "ToolSearchReturnPart"):
yield ev_factory(
"tool_result",
_stamp(
{
"tool_name": getattr(part, "tool_name", ""),
"result": getattr(part, "content", None),
"tool_call_id": getattr(part, "tool_call_id", ""),
}
),
)
# Trace without a follow-up TextPart in the same message — emit it as a
# standalone reasoning-only model_response so it isn't lost.
if pending_trace:
yield ev_factory(
"model_response",
_stamp({"content": "", "reasoning_trace": pending_trace}),
)
# The Oracle escalation reasons (stage 1, item 6). Each names the uncertainty
# that sent the case: the egress audit row (``oracle_escalation``) carries it.
ORACLE_REASON_BAND = "confidence_in_band"
ORACLE_REASON_SPLIT = "template_split"
ORACLE_REASON_DEEP_NMI = "deep_needs_more_info"
def oracle_escalation_reason(
report: TriageReport,
enriched: Any,
settings: Settings,
*,
candidate: Any = None,
ran_loop: bool = False,
run_class: str = budget.STANDARD,
) -> str | None:
"""Why the local verdict goes to the Oracle, or None when it stays local.
The Oracle escalates on UNCERTAINTY, never on the verdict class alone
(docs/dev/specs/2026-10-04-four-tier-detection-methodology.md, "The unified
investigator"). Production escalated 59 times in 30 days, 48 of them a
malware rule cleared false positive. The Oracle answered 17 of those 48 in
the window and confirmed the false positive in all 17 (Oracle failure
review, 2026-10-04). Three triggers, first match wins:
1. ``confidence_in_band``: the confidence sits in the gate band
(:func:`soc_ai.agent.gates.gate_band`, ``[0.4, 0.7)``). The low edge is
where a coercing gate parks a verdict it could not ground; the high edge
is the confidence the gates let stand.
2. ``template_split``: the decision template and the model disagree on the
verdict.
3. ``deep_needs_more_info``: a deep run, the analyst's deep re-run, still
ended needs_more_info.
Never: a verdict the template and the model agree on with confidence at or
above the band; a pipeline fallback (a mechanical failure, nothing to
adjudicate); a cheap or rule-prior run (soc_ai.agent.budget).
The per-verdict opt-ins narrow the triggers. They never add one:
- a needs_more_info verdict needs ``oracle_escalate_needs_more_info``;
- on a malware or attack rule, a true positive stays local, a non-TP needs
``oracle_escalate_malware_non_tp``, and a loop verdict at or above
``oracle_skip_after_confident_loop`` stays local;
- an in-band escalation needs confidence below
``oracle_escalate_below_confidence``.
"""
if not settings.oracle_enabled:
return None
if run_class not in budget.ORACLE_CLASSES:
return None
# A pipeline-fallback placeholder is a MECHANICAL failure (model truncation,
# gateway 5xx), not a model opinion — there is nothing for the Oracle to
# adjudicate. Re-running is the fix (dogfood 2026-07-15).
if is_pipeline_fallback({"resolution": report.resolution}):
return None
from soc_ai.agent.gates import gate_band # noqa: PLC0415 - re-exported module
low, high = gate_band()
confidence = float(report.confidence or 0.0)
template_verdict = getattr(candidate, "verdict", None) if candidate is not None else None
agrees = template_verdict is not None and template_verdict == report.verdict
if agrees and confidence >= high:
return None
if low <= confidence < high:
reason = ORACLE_REASON_BAND
elif template_verdict is not None and not agrees:
reason = ORACLE_REASON_SPLIT
elif run_class == budget.DEEP and report.verdict == "needs_more_info":
reason = ORACLE_REASON_DEEP_NMI
else:
return None
# ----- The opt-ins narrow -----
if report.verdict == "needs_more_info" and not settings.oracle_escalate_needs_more_info:
return None
from soc_ai.agent.decision_templates import ( # noqa: PLC0415
_rule_signals_attack,
_rule_signals_malware,
)
if _rule_signals_malware(enriched) or _rule_signals_attack(enriched):
# A flagged-malicious verdict on a malware or attack rule is the right
# call whatever its confidence (the CryptoWall citation-cap case).
if report.verdict == "true_positive":
return None
if not settings.oracle_escalate_malware_non_tp:
return None
if ran_loop and confidence >= settings.oracle_skip_after_confident_loop:
return None
if reason == ORACLE_REASON_BAND and confidence >= settings.oracle_escalate_below_confidence:
return None
return reason
# The classic rule's reasons: the audit codes the verdict-class rule wrote
# before stage 1.
ORACLE_REASON_CLASSIC_NMI = "needs_more_info"
ORACLE_REASON_CLASSIC_MALWARE = "malware_non_tp"
ORACLE_REASON_CLASSIC_FLOOR = "below_confidence"
# The confidence floor of the classic rule. ``oracle_escalate_below_confidence``
# held it before stage 1 with a default of 0.6. Stage 1 gave that setting the
# band ceiling to hold and moved its default to 0.7, so the classic rule keeps
# 0.6 and takes the setting only when the operator set it lower.
ORACLE_CLASSIC_BELOW_CONFIDENCE = 0.6
def classic_oracle_escalation_reason(
report: TriageReport,
enriched: Any,
settings: Settings,
*,
ran_loop: bool = False,
run_class: str = budget.STANDARD,
) -> str | None:
"""The verdict-class rule production ran before stage 1, or None.
Kept so ``oracle_rule_mode`` can run it (``classic``) or run it and record
what the uncertainty rule would do (``shadow``). First match wins:
1. ``needs_more_info``: the opt-in is on and the verdict is needs_more_info.
2. A malware or attack rule with a true positive stays local.
3. ``malware_non_tp``: the opt-in is on, the rule signals malware or an
attack, and the verdict is not a true positive. A loop that ran and
reached ``oracle_skip_after_confident_loop`` keeps it local.
4. ``below_confidence``: the confidence is below
:data:`ORACLE_CLASSIC_BELOW_CONFIDENCE`, or below
``oracle_escalate_below_confidence`` when that is lower.
"""
if not settings.oracle_enabled:
return None
if run_class not in budget.ORACLE_CLASSES:
return None
if is_pipeline_fallback({"resolution": report.resolution}):
return None
from soc_ai.agent.decision_templates import ( # noqa: PLC0415
_rule_signals_attack,
_rule_signals_malware,
)
if settings.oracle_escalate_needs_more_info and report.verdict == "needs_more_info":
return ORACLE_REASON_CLASSIC_NMI
malware_or_attack = _rule_signals_malware(enriched) or _rule_signals_attack(enriched)
if malware_or_attack and report.verdict == "true_positive":
return None
confidence = float(report.confidence or 0.0)
resolved_by_confident_loop = (
ran_loop and confidence >= settings.oracle_skip_after_confident_loop
)
if (
settings.oracle_escalate_malware_non_tp
and malware_or_attack
and not resolved_by_confident_loop
):
return ORACLE_REASON_CLASSIC_MALWARE
floor = min(settings.oracle_escalate_below_confidence, ORACLE_CLASSIC_BELOW_CONFIDENCE)
if confidence < floor:
return ORACLE_REASON_CLASSIC_FLOOR
return None
@dataclasses.dataclass(frozen=True)
class OracleRuleDecision:
"""What the Oracle rule mode decided for one local verdict.
``reason`` is the deciding rule's reason, None when the verdict stays
local. ``classic_reason`` and ``uncertainty_reason`` are each rule's own
answer; a rule the mode does not run answers None.
"""
mode: str
reason: str | None
classic_reason: str | None
uncertainty_reason: str | None
def oracle_rule_decision(
report: TriageReport,
enriched: Any,
settings: Settings,
*,
candidate: Any = None,
ran_loop: bool = False,
run_class: str = budget.STANDARD,
) -> OracleRuleDecision:
"""Apply ``oracle_rule_mode``: classic, shadow or uncertainty.
Classic and shadow let the classic rule decide. Shadow also runs the
uncertainty rule, and the pipeline records its answer on an
``oracle_shadow`` event. Uncertainty lets the uncertainty rule decide.
"""
mode = str(getattr(settings, "oracle_rule_mode", "shadow") or "shadow")
classic: str | None = None
uncertainty: str | None = None
if mode in ("classic", "shadow"):
classic = classic_oracle_escalation_reason(
report, enriched, settings, ran_loop=ran_loop, run_class=run_class
)
if mode in ("uncertainty", "shadow"):
uncertainty = oracle_escalation_reason(
report,
enriched,
settings,
candidate=candidate,
ran_loop=ran_loop,
run_class=run_class,
)
reason = uncertainty if mode == "uncertainty" else classic
return OracleRuleDecision(
mode=mode, reason=reason, classic_reason=classic, uncertainty_reason=uncertainty
)
def _should_escalate_to_oracle(
report: TriageReport,
enriched: Any,
settings: Settings,
*,
candidate: Any = None,
ran_loop: bool = False,
run_class: str = budget.STANDARD,
) -> bool:
"""True when the rule that ``oracle_rule_mode`` selects names a reason."""
return (
oracle_rule_decision(
report,
enriched,
settings,
candidate=candidate,
ran_loop=ran_loop,
run_class=run_class,
).reason
is not None
)
async def _resolve_effective_identifiers(
ctx: InvestigationContext,
) -> EffectiveIdentifiers | None:
"""Resolve the full effective internal-identifier set ONCE per investigation.
Opens a one-off session from ``ctx.db_sessionmaker`` and computes the merged
*effective* set (env-config union active detected/manual identifiers, minus
muted) via
:func:`~soc_ai.oracle.identifiers.effective_internal_identifiers`. The
returned :class:`EffectiveIdentifiers` carries ``.suffixes``/``.hosts`` (for
the Oracle egress sanitizer) and ``.cidrs`` (for internal-IP classification).
Returns ``None`` when:
* no ``db_sessionmaker`` is on ``ctx`` (CLI / eval / direct callers), or
* resolution raised (DB error, missing table).
A ``None`` return is the BACKWARD-COMPAT escape hatch: callers fall back to
the raw ``settings`` values (``oracle_internal_suffixes`` / ``oracle_extra_hosts``
for redaction, ``internal_cidrs`` for classification), so a db-less path — or
any failure — leaves both redaction and classification behavior unchanged.
SECURITY (redaction): threading this can never under-redact relative to
today's settings-only behavior. The effective suffix/host set is
``(settings/reserved, always) + (active detected/manual) - (muted)``, and the
sanitizer always re-adds ``settings.oracle_internal_suffixes`` (plus the
reserved ``.lan/.local/.internal/.corp`` floor), so reserved/env defaults
cannot be muted away — relative to raw settings this only ever *adds*.
CLASSIFICATION (cidrs): with NO active ``cidr`` rows the effective cidrs ==
``settings.internal_cidrs`` (a muted detected CIDR is suppressed, an active
one is added) — so classification is byte-identical to today until an
operator un-mutes a suggested subnet. Detected CIDRs are always muted, so
discovery alone never reclassifies a host.
"""
maker = ctx.db_sessionmaker
if maker is None:
return None
try:
async with maker() as db:
return await effective_internal_identifiers(db, ctx.settings)
except Exception: # pragma: no cover - defensive; never block egress on a DB hiccup
_LOGGER.warning(
"orchestrator: failed to resolve effective internal-identifier set; "
"falling back to settings (oracle suffixes/hosts + internal_cidrs)",
exc_info=True,
)
return None
def _classification_cidrs(
ctx: InvestigationContext, effective: EffectiveIdentifiers | None
) -> Sequence[Any]:
"""The internal CIDR set internal-IP classification should use.
``effective.cidrs`` when the effective set resolved (env ``internal_cidrs``
union active ``cidr`` rows minus muted), else ``settings.internal_cidrs``
(db-less path / resolution failure). With no active ``cidr`` rows the two are
identical, so classification is unchanged until an operator un-mutes a
suggested subnet.
"""
if effective is not None:
return effective.cidrs
return ctx.settings.internal_cidrs
async def _resolve_oracle_identifiers(
ctx: InvestigationContext,
) -> tuple[tuple[str, ...], tuple[str, ...]] | None:
"""Resolve the effective (suffixes, hosts) for the Oracle egress sanitizer.
Thin wrapper over :func:`_resolve_effective_identifiers` preserving the
historical ``(suffixes, hosts)`` shape the Oracle client consumes. ``None``
⇒ fall back to the raw settings tuples (redaction unchanged). See
:func:`_resolve_effective_identifiers` for the full contract.
"""
effective = await _resolve_effective_identifiers(ctx)
if effective is None:
return None
return effective.suffixes, effective.hosts
def _ground_host_claims(report: TriageReport, enriched: Any) -> tuple[TriageReport, int]:
"""Rewrite the report's "no host telemetry" claims from the prefetch coverage."""
subjects = [
coverage_subject(entry.ip, entry.coverage)
for entry in (getattr(enriched, "host_coverage", None) or [])
]
grounded, rewritten = ground_host_coverage_claims(report, subjects)
return cast("TriageReport", grounded), rewritten
def _strip_model_resolution(report: TriageReport) -> TriageReport:
"""Drop a model-authored ``resolution`` from a *report*.
``resolution`` is the pipeline-FAILURE marker. It suppresses Oracle
escalation, drops the run out of the Needs-info KPI, excludes the row from
the same-session consistency constraint and prior-outcome memory, and
renders it as an infrastructure error — and only the orchestrator's own
fallback builder (``_synth_failure_fallback_report``) may set it (see the
field's note in triage_models). The field is part of the model's output
schema, though, so any synthesizer or tool-using investigator reading
attacker-influenceable alert text can emit it next to a real verdict.
Every assignment from a model result goes through this, so a settled
verdict never stores as a pipeline failure on the model's say-so.
"""
if report.resolution is None:
return report
return report.model_copy(update={"resolution": None})
def _desanitize_report(report: TriageReport, guard: EgressGuard) -> TriageReport:
"""Restore real identifiers in every string field of a *report*.
When the cloud-egress guard is active the synth/loop models emit their
TriageReports in LABEL space (``HOST_01`` cited in summaries, citations,
recommended-action args, and ``gap_for_investigator.tool_args``). This
round-trips the whole report through ``guard.desanitize_obj`` —
``model_dump`` → recursive label restore → ``model_validate`` — so
everything downstream (the deterministic gates comparing against real
enriched values, Phase-D dispatch args hitting Elasticsearch, the stored
``triage_report`` event, the analyst-facing UI) sees real values.
Defensive: a desanitize/validation surprise must never cost the verdict —
on any failure the labeled report is returned unchanged (labels in the UI
beat a crashed investigation).
"""
try:
restored = guard.desanitize_obj(report.model_dump(mode="json"))
return TriageReport.model_validate(restored)
except Exception:
_LOGGER.warning(
"egress guard: TriageReport desanitize failed; keeping labeled report",
exc_info=True,
)
return report
def _round1_skipped_report(alert_id: str) -> TriageReport:
"""Placeholder round-1 verdict for cases that skip the round-1 synth and
route straight to the investigation loop. Always overwritten by the loop's
synthesizer output — it only serves as the ``triage_final`` default."""
return TriageReport(
verdict="needs_more_info",
confidence=0.0,
summary="Round-1 synth skipped — routed directly to the investigation loop.",
citations=[],
)
def _self_consistency_vote(reports: list[Any]) -> tuple[str, float, str]:
"""Majority-vote a verdict across N independent final-synthesis samples.
Pure helper for the flag-gated self-consistency vote
(``settings.verdict_consistency_samples > 1``). Given the TriageReports
from N runs of the SAME final synthesis call, returns
``(verdict, confidence, note)``:
- STRICT majority (count > N/2): that verdict wins; confidence is the mean
of the confidences of the samples that voted for it (3 dp); note =
``"self-consistency K/N agreed on <verdict>"``.
- No strict majority (tie or 3-way split): ``("inconclusive", mean
confidence of ALL samples capped at 0.5, "self-consistency split M ways:
<tally> — inconclusive")``.
- Single report (defensive — the caller guards N>1): passthrough of that
report's verdict/confidence unchanged.
"""
if len(reports) == 1:
r = reports[0]
return r.verdict, r.confidence, "self-consistency: single sample — no vote"
n = len(reports)
tally = Counter(r.verdict for r in reports)
top_verdict, top_count = tally.most_common(1)[0]
if top_count > n / 2:
confs = [r.confidence for r in reports if r.verdict == top_verdict]
mean_conf = round(sum(confs) / len(confs), 3)
return (
top_verdict,
mean_conf,
f"self-consistency {top_count}/{n} agreed on {top_verdict}",
)
conf = round(min(sum(r.confidence for r in reports) / n, 0.5), 3)
tally_str = ", ".join(
f"{v}={c}" for v, c in sorted(tally.items(), key=lambda kv: (-kv[1], kv[0]))
)
return (
"inconclusive",
conf,
f"self-consistency split {len(tally)} ways: {tally_str} — inconclusive",
)
# How many same-session verdicts reach the prompt. One is nearly always all
# there is, and the gate only ever reads the true positive out of the list.
_SESSION_MAX_ITEMS = 3
# Header framing the same-session constraint. Deliberately the opposite framing
# to the prior-outcomes header below: that block is a RESEMBLANCE the current
# evidence may overrule, and this one is the SAME conversation, already read.
# Two alerts on one session that settle opposite ways are not two opinions, they
# are one contradiction, and the range produced exactly that.
_SESSION_PRIOR_HEADER = (
"## Same network session, already investigated (BINDING — read before you decide)"
)
_SESSION_PRIOR_INSTRUCTION = (
"This is the SAME conversation as the alert above: same source, same "
"destination, same ports, same protocol (the community id matches), inside "
"the session window. It is not a similar alert. You may not settle this "
"alert `false_positive` while a `true_positive` verdict stands on the same "
"session. If the evidence in front of you genuinely contradicts that "
"verdict, say so in the summary and emit `needs_more_info` naming the "
"disagreement — do not close it quietly."
)
def _format_session_prior_block(digests: list[dict[str, Any]]) -> str:
"""Render same-session verdicts into the binding constraint block.
Same line shape and the same fail-soft age phrasing as
:func:`_format_prior_outcomes_block`, and composed at the same point in the
message so it rides the caller's sanitize sweep and ``_guard_egress``: a
prior rationale is redacted on the cloud-analyst path like everything else.
"""
from soc_ai.store.auth import utcnow # noqa: PLC0415 - lazy: store dep only when a DB exists
now = utcnow()
lines = [_SESSION_PRIOR_HEADER, ""]
for d in digests:
conf = d.get("confidence")
conf_part = f" ({conf:.2f})" if isinstance(conf, int | float) else ""
digest = d.get("rationale_digest") or "(no rationale recorded)"
lines.append(
f"- {_prior_age_phrase(d.get('created_at'), now)} · investigation "
f"{d.get('id')} · {d.get('verdict')}{conf_part} — {digest}"
)
lines.extend(["", _SESSION_PRIOR_INSTRUCTION])
return "\n".join(lines)
_ANALYST_VERDICT_HEADER = "## Analyst verdict on this alert"
_ANALYST_VERDICT_INSTRUCTION = (
"A person read this alert and set this verdict. The verdict is not evidence. "
"Do not cite it. If your evidence disagrees with it, state the disagreement "
"in the summary."
)
def _analyst_note(rationale: str | None) -> str | None:
"""The analyst's own words from an override headline, without the name.
A manual override stores ``Verdict set by <user>: <reason>.``. The prompt
carries the reason and leaves the user name out: a name is not a fact
about the alert.
"""
text = (rationale or "").strip()
if text.startswith("Verdict set by "):
_, sep, rest = text.partition(": ")
text = rest.strip() if sep else ""
return text or None
def _format_analyst_verdict_block(analyst: Any) -> str:
"""Render the standing analyst verdict as one fact for every prompt site.
R2-D1 (dogfood 2026-10-19): a re-run after an analyst override read
nothing of the override and landed the opposite verdict. The block rides
the same injection sites and the same sanitize sweep as the same-session
constraint.
"""
from soc_ai.store.investigations import _digest_rationale # noqa: PLC0415
age = "earlier"
resolved_at = getattr(analyst, "resolved_at", None)
if resolved_at:
from soc_ai.store.auth import utcnow # noqa: PLC0415
try:
at = datetime.fromisoformat(str(resolved_at))
if at.tzinfo is not None:
at = at.astimezone(UTC).replace(tzinfo=None)
age = _prior_age_phrase(at, utcnow())
except ValueError:
age = "earlier"
line = (
f"- {age}: on investigation {analyst.investigation_id}, an analyst set the "
f"verdict to {analyst.verdict}."
)
note = _digest_rationale(_analyst_note(analyst.rationale))
if note:
line += f" The analyst wrote: {note}"
return "\n".join([_ANALYST_VERDICT_HEADER, "", line, "", _ANALYST_VERDICT_INSTRUCTION])
# Header framing the E4.2 prior-outcome memory block. The anti-anchoring
# instruction lives HERE (read before any verdict line), because the whole
# point of the default-off flag is that prior verdicts can bias the model —
# the block must present as fallible context, never as evidence to cite.
_PRIOR_OUTCOMES_HEADER = (
"## Prior outcomes for similar alerts (CONTEXT ONLY — NOT evidence. "
"Do not cite these; they may be wrong; weigh the current evidence on its "
"own merits.)"
)
def _prior_age_phrase(created_at: datetime | None, now: datetime) -> str:
"""Compact "how long ago" phrase for one prior-outcome line ("3d ago").
Both datetimes are naive UTC (the store's convention — ``created_at`` is
stamped by SQLite ``func.now()`` and compared against
:func:`soc_ai.store.auth.utcnow` everywhere else). Sub-hour ages render
"<1h ago". A missing value or an aware/naive mismatch renders "recently"
instead of raising — the memory block is advisory context and must never
kill an investigation over a timestamp quirk (fail-soft, like the fetch).
"""
if created_at is None:
return "recently"
try:
delta_s = max((now - created_at).total_seconds(), 0.0)
except TypeError: # aware vs naive mismatch from an exotic backend
return "recently"
days = int(delta_s // 86400)
if days >= 1:
return f"{days}d ago"
hours = int(delta_s // 3600)
return f"{hours}h ago" if hours >= 1 else "<1h ago"
def _format_prior_outcomes_block(digests: list[dict[str, Any]]) -> str:
"""Render prior-outcome digests into the round-1 memory block (E4.2).
One compact line per digest — age, match tier, verdict (+confidence), and
the word-boundary-truncated rationale — under the anti-anchoring header.
The returned block is passed into :func:`build_synth_first_user_message`
BEFORE the composed message's final sanitize sweep + ``_guard_egress``, so
internal identifiers inside a prior rationale are redacted on the
cloud-analyst path exactly like the rest of the prompt.
"""
from soc_ai.store.auth import utcnow # noqa: PLC0415 - lazy: store dep only when memory is on
now = utcnow()
lines = [_PRIOR_OUTCOMES_HEADER, ""]
for d in digests:
conf = d.get("confidence")
conf_part = f" ({conf:.2f})" if isinstance(conf, int | float) else ""
digest = d.get("rationale_digest") or "(no rationale recorded)"
lines.append(
f"- {_prior_age_phrase(d.get('created_at'), now)} · {d.get('matched_on')} · "
f"{d.get('verdict')}{conf_part} — {digest}"
)
return "\n".join(lines)
# Header framing the chat-transcript memory block. Like the priors header the
# anti-grounding instruction lives HERE (read before any excerpt line) — and it
# is stronger, per the operator's hard rule: a transcript's USER turns are an
# analyst's unverified opinion (the user is NOT always right), and its
# ASSISTANT turns reasoned about DIFFERENT alerts. Context only, never evidence.
_CHAT_MEMORY_HEADER = (
"## Prior discussion excerpts (CONTEXT ONLY — NOT evidence. These are past "
"chat messages; USER statements are unverified operator opinions and may be "
"wrong; ASSISTANT statements were about different alerts. Do not cite; "
"weigh current evidence on its own merits.)"
)
def _chat_memory_query_terms(
rule_name: str | None, src_ip: str | None, dest_ip: str | None
) -> list[str]:
"""Alert-derived FTS query terms for the chat-transcript recall.
Rule-name WORDS (split here, so each word matches independently — the
store would treat a multi-word term as an exact phrase) plus the endpoint
IPs as whole terms (the store turns each into an FTS phrase, keeping an IP
selective). Everything is deterministic alert-row data — no model in the
loop, mirroring the E4.2 (rule, src, dest) keying.
"""
terms: list[str] = re.findall(r"[A-Za-z0-9]+", rule_name or "")
terms.extend(ip for ip in (src_ip, dest_ip) if ip)
return terms
def _format_chat_memory_block(digests: list[dict[str, Any]]) -> str:
"""Render chat-snippet digests into the round-1 "prior discussion" block.
One compact line per snippet — ``[age · source · ROLE] "snippet"`` — under
the context-NEVER-evidence header. The ROLE is uppercased so a USER line is
visibly labeled as operator opinion right where it's read, not only in the
header. Same egress contract as the priors block: the returned text is
composed BEFORE the final sanitize sweep + ``_guard_egress``, so internal
identifiers inside a snippet are redacted on the cloud-analyst path.
"""
from soc_ai.store.auth import utcnow # noqa: PLC0415 - lazy: store dep only when memory is on
now = utcnow()
lines = [_CHAT_MEMORY_HEADER, ""]
for d in digests:
role = str(d.get("role") or "").upper() or "UNKNOWN"
lines.append(
f"- [{_prior_age_phrase(d.get('created_at'), now)} · {d.get('source')} · "
f'{role}] "{d.get("snippet")}"'
)
return "\n".join(lines)
# ---------------------------------------------------------------------------
# Host dossier — the ambient "what IS this host?" block
# ---------------------------------------------------------------------------
# Headings the dossier block is woven in AHEAD of, first match wins. The model
# should learn what the hosts ARE before it reads what they DID, and the current
# evidence must still arrive last. Two anchors because the messages come from
# two different formatters: the synth-first ones carry the enriched context, the
# post-loop synthesizer message carries only the transcript. A message with
# NEITHER is appended to rather than skipped — silently dropping the block is
# the exact failure this feature exists to fix, and it would be invisible.
#
# Woven here rather than threaded as a `host_dossier_block=` kwarg through each
# formatter in prompts.py: the ordering that actually matters is "composed
# before this site's sanitize_text", and keeping composition in the orchestrator
# puts that ordering on ONE screen beside the sweep it must precede. A kwarg
# passed at a call site three hundred lines from its guard is a correct-looking
# line that cannot be read for the property it has to hold.
_DOSSIER_ANCHORS = ("## Enriched alert context", "## Investigation transcript")
# Floor under the enriched context's token budget once the dossier's cost has
# been taken out of it. `trim_enriched_for_budget` grinds every pivot down to
# two events and STILL returns an over-budget string, so an unclamped
# subtraction cannot buy the dossier its space — it can only destroy the
# correlation signal on the way to failing anyway.
_MIN_ENRICHED_BUDGET = 2000
def dossier_hosts_for_alert(context: Any, settings: Any) -> dict[str, str]:
"""Every host this alert puts in play, as ``{ip: role-in-alert}``, source first.
The two endpoints come first — order is the caller's contract with the
renderer, so the source host reads first — followed by every distinct
INTERNAL address the alert's prefetched group events name. A host that
appears twice is described once: two identical entries would spend the
block's budget saying the same thing twice.
Widened past the endpoints because the endpoints are not the cast. The
motivating incident's real actor was a VM one pivot away from the address
the alert named, and it was already sitting in ``community_id_events`` —
prefetched, in memory, and never described. Reading the pivots costs no I/O
(:func:`~soc_ai.tools.get_alert_context.get_alert_context` already fetched
them) and the returned set is bounded downstream by
:data:`~soc_ai.dossier.prompt.MAX_PROMPT_HOSTS`, which also states how many
it left out.
The internal filter applies to the WIDENED hosts only. An external
destination on the alert itself is still named, because "the network sweep
has no record of this address" is something the block owes the model about
the host under discussion; an external peer three pivots deep is just noise
that no dossier will ever describe.
The address is passed through EXACTLY as the alert spells it. The egress
guard allocates one label per spelling, and the whole mechanic — "IP_01 is a
hypervisor whose policy forbids interactive SSH" — depends on the dossier's
address collapsing onto the same ``IP_01`` the alert already carries.
"""
alert = getattr(context, "alert", context)
hosts: dict[str, str] = {}
for ip, label in ((alert.source_ip, "source"), (alert.destination_ip, "destination")):
if isinstance(ip, str) and ip.strip() and ip not in hosts:
hosts[ip] = label
for event in _group_events(context):
for ip in (event.source_ip, event.destination_ip):
if not isinstance(ip, str) or not ip.strip() or ip in hosts:
continue
if settings.network_is_internal(ip):
hosts[ip] = "related event"
return hosts
def _group_events(context: Any) -> list[Any]:
"""The alert's five prefetched pivots, flattened in the order they are read.
``getattr`` per list rather than attribute access: the block is also built
for a bare :class:`~soc_ai.so_client.models.SoAlert` (the chat seeds from a
stored investigation whose context fetch may have failed), and asking an
alert for its pivots must degrade to "no group" rather than raise.
The field list is ``context_budget.PIVOT_FIELDS`` rather than a local copy:
the two must not drift, or a sixth pivot gets trimmed for budget but never
read for identity and nothing fails. Its order is the order
``get_alert_context`` fans the pivots out in, so the hosts nearest the
alert's own flow (its community_id) are the ones that survive the cap.
"""
events: list[Any] = []
for name in context_budget.PIVOT_FIELDS:
events.extend(getattr(context, name, None) or [])
return events
def _inject_dossier_block(message: str, block: str) -> str:
"""Weave the ``\\n\\n``-prefixed dossier *block* into a composed *message*.
Called at every site whose message reaches the analyst model, and always
BEFORE that site's ``guard.sanitize_text`` — appending after the sweep leaks
the dossier's internal hostnames and addresses to a cloud model in the
clear, and appending between the sweep and the fail-closed residue gate
kills the investigation with an ``egress_blocked`` verdict instead.
"""
if not block:
return message
for anchor in _DOSSIER_ANCHORS:
at = message.find(anchor)
if at >= 0:
return f"{message[:at]}{block.strip()}\n\n{message[at:]}"
return f"{message}{block}"
def _inject_session_constraint(message: str, block: str) -> str:
"""Weave the same-session constraint into a composed *message*.
Same anchoring and the same before-the-sanitize-sweep rule as
:func:`_inject_dossier_block`, and injected at every site the dossier is,
including the loop and round-2 prompts the memory blocks deliberately skip.
The memory blocks stay out of those prompts because a resemblance would
compete with the evidence the loop just gathered. This is not a resemblance:
it is the same session, and a verdict that contradicts it has to say so
rather than arrive by not being told.
Runs after the dossier at each site, so the constraint ends up between the
dossier and the enriched context: the last thing read before the evidence.
"""
return _inject_dossier_block(message, block)
def _inject_subject_block(message: str, block: str) -> str:
"""Weave the hunt-subject block into a composed *message*.
D2. The investigator prompt renders the subject itself; this is for the
message that has no room for it, the loop synthesis over the transcript.
That call writes the verdict and it sees only the transcript, so without
the block it would score the hunt as an alert. Same anchoring and the same
before-the-sanitize-sweep rule as :func:`_inject_dossier_block`. Empty
block, no change, which is every alert run.
"""
return _inject_dossier_block(message, block)
async def _run_synth_first_pipeline( # noqa: PLR0912, PLR0915 - multi-phase pipeline is inherently long
*,
alert_id: str,
ctx: InvestigationContext,
focus_hint: str | None = None,
deep: bool = False,
allow_so_writes: bool = True,
focus_origin: FocusOrigin = "rerun",
subject: HuntSubject | None = None,
) -> AsyncGenerator[StepEvent, None]:
"""Phase A → B → C → optional D → C round 2 → done.
The synth-first pipeline, and the SOLE production triage path: ``investigate()``
calls only this, having deleted the legacy investigator→synthesizer loop in
2026-07 after synth-first had been the production default since 2026-05-29.
Any change to a phase here changes the one path every alert takes.
``focus_hint`` (optional): prior open questions from a re-launched
``needs_more_info`` investigation, woven into the round-1 seed + the
investigation-loop investigator prompt so this run targets those gaps.
``allow_so_writes`` / ``focus_origin`` / ``subject``: see :func:`investigate`.
A HUNT subject changes five things and nothing else. The cited documents
join the prefetch bundle, so every id resolver reads them. The decision
templates do not run: they match alert rule classes and a hunt has none.
The same-session prior does not run: a hunt has no community id of its own.
Round 1 does not run: it is a tool-less synthesis over one alert, and the
hunt has to be read with the tools. The host history gate also reads the
documents that the promoted finding cites. Every gate still runs.
"""
from soc_ai.agent._prefetch_retry import retry_prefetch # noqa: PLC0415
from soc_ai.agent.decision_templates import match_decision_template # noqa: PLC0415
from soc_ai.agent.prompts import ( # noqa: PLC0415
build_synth_first_round2_user_message,
build_synth_first_user_message,
)
from soc_ai.agent.targeted_investigator import ( # noqa: PLC0415
run_targeted_investigation,
)
from soc_ai.tools.enrichment import EnrichmentContext # noqa: PLC0415
from soc_ai.tools.get_alert_context import get_enriched_alert_context # noqa: PLC0415
session_id = uuid.uuid4().hex
sequence_counter = [0]
def _ev(kind: str, payload: dict[str, Any]) -> StepEvent:
sequence_counter[0] += 1
return StepEvent(
kind=kind, session_id=session_id, sequence=sequence_counter[0], payload=payload
)
async def _audit(ev: StepEvent) -> None:
# Audit must never crash the in-flight investigation. The audit logger
# already swallows ES errors, but a Pydantic ValidationError on
# AuditKind would propagate before the ES call - catch it here too.
# Also feed the per-process Prometheus counters so /metrics reflects this
# run — the synth-first pipeline is the DEFAULT path, and without this the
# /metrics counters stay frozen at 0 in production (mirrors the legacy
# investigate() _audit).
try:
await metrics.get_metrics().record_event(ev.kind, ev.payload)
except Exception as e:
_LOGGER.warning("metrics record failed (kind=%s): %s", ev.kind, e)
if ctx.audit is None:
return
try:
await ctx.audit.log_kind(session_id, ev.kind, ev.payload)
except Exception as e: # audit must never crash the investigation
_LOGGER.warning("audit log_kind failed: %s", e)
def _usage_ev(
round_num: int, run_result: Any, served_backend: dict[str, Any] | None = None
) -> StepEvent | None:
"""Build a ``usage`` event from a pydantic_ai result.
The synth-first pipeline previously emitted NO usage events (only
the legacy investigate() path did), so the UI's token KPI /
sparkline / context meter stayed dead at 0. Mirror the legacy
`_build_usage_event` shape so the panel populates.
``served_backend``: gateway attribution for the calls that produced
this usage — see :func:`_usage_event_payload`.
"""
try:
# A property since pydantic-ai 1.107; the call form warns today and
# will go, and a caught TypeError would silently empty the token
# panel. A test double or an older release still exposes a method.
u = run_result.usage
if not isinstance(u, RunUsage) and callable(u):
u = u()
except Exception:
return None
return _ev("usage", _usage_event_payload(round_num, u, served_backend))
def _reasoning_ev(round_num: int, run_result: Any) -> StepEvent | None:
"""``model_response`` event carrying the synthesizer's reasoning trace.
The loop's investigator turns already stream model_response via
``_walk_message``; the synthesizer ``agent.run`` calls did not, so a
round-1/round-2-settled investigation had an empty "Model reasoning"
panel. None when the run produced no trace — no event is emitted.
Called ONLY for the primary/final synthesis runs, never for the
self-consistency extra samples.
"""
payload = _synth_reasoning_payload(run_result)
if payload is None:
return None
if guard is not None:
# The synthesizer thought in LABEL space (its inputs were
# sanitized) — restore real identifiers before the trace is
# stored/displayed so the reasoning panel reads like the rest of
# the investigation. Local storage only, never egress.
payload = guard.desanitize_obj(payload)
return _ev("model_response", {**payload, "phase": "synthesizer", "round": round_num})
async def _emit_egress_blocked(
phase: str, exc: EgressResidueError
) -> AsyncGenerator[StepEvent, None]:
"""Audit + stream a fail-closed egress block into the timeline.
Yields (and audits) an ``egress_blocked`` event carrying ONLY the
leaked-identifier COUNT + the call site — NEVER the leaked values
(logging them would defect on the whole point of the block) — then the
paired ``error`` event, so the block renders in the timeline exactly like
every other synth-failure path. Best-effort: a failed audit index must
never turn a blocked egress into an actual egress. Drive it with
``async for ev in _emit_egress_blocked(...): yield ev``.
"""
block_ev = _ev("egress_blocked", {"phase": phase, "leaked_count": len(exc.leaked)})
await _audit(block_ev)
yield block_ev
err_ev = _ev("error", _error_payload(exc, phase=phase, round_num=0))
await _audit(err_ev)
yield err_ev
# D2. `alert_id` is the anchor document, which a hunt-subject run needs for
# its time windows. `subject` says what the run is about, so a reader of
# the trail cannot take the anchor for the subject.
# The budget class the caller asked for. None is the scheduler's request:
# the rungs below decide (soc_ai.agent.budget).
_asked = getattr(ctx, "requested_run_class", None)
requested_class: str | None = (
budget.DEEP if deep else (_asked if _asked in (budget.STANDARD, budget.DEEP) else None)
)
start_payload: dict[str, Any] = {"alert_id": alert_id, "pipeline": "synth_first"}
if requested_class is not None:
# An explicit request is the class the run will run in. A run the rungs
# decide states its class when they have (the loop entry, the report).
start_payload["run_class"] = requested_class
if subject is not None:
start_payload["subject"] = "hunt"
yield _ev("session_start", start_payload)
# Resolve the effective internal-identifier set ONCE per investigation
# (env-config union active detected/manual identifiers, minus muted). Used
# for BOTH internal-IP classification (``.cidrs`` → the targeted downgrades /
# post-validator below) and the Oracle egress sanitizer (``.suffixes`` /
# ``.hosts`` at the adjudication call). ``None`` ⇒ no DB on ctx (CLI / eval /
# tests) or a resolution failure → classification falls back to
# ``settings.internal_cidrs`` and redaction to the raw settings tuples
# (behavior unchanged). With no active ``cidr`` rows the effective cidrs ==
# ``settings.internal_cidrs``, so classification is byte-identical to today.
effective_idents = await _resolve_effective_identifiers(ctx)
classification_cidrs = _classification_cidrs(ctx, effective_idents)
# Pre-seed the effective suffix/host sets the ONLINE egress tool guards
# (web_search / crawl_page) consult, reusing the set already resolved above so
# those tools no longer read env-only identifiers (a DB-discovered internal
# name on an empty .env otherwise fanned out to public search). Marking it
# resolved even when None (no DB) keeps the tool closures from re-resolving —
# they fall back to the raw settings tuples, unchanged.
ctx.effective_internal_suffixes = (
effective_idents.suffixes if effective_idents is not None else None
)
ctx.effective_internal_hosts = effective_idents.hosts if effective_idents is not None else None
ctx._egress_idents_resolved = True
# ----- Cloud-egress guard for the ANALYST model path (opt-in) -----
# When analyst_cloud_redaction is on, everything sent to the analyst model
# below (enriched context, prompts, tool results — the toolset wraps tools
# via _guarded at registration) is redacted with ONE per-run label mapping,
# and every model output is label-restored where it lands. `is True` (not
# truthiness) so a non-Settings test double can never flip redaction on.
if ctx.settings.analyst_cloud_redaction is True and ctx.egress_guard is None:
if effective_idents is not None:
# Reuse the effective identifier set already resolved above —
# avoids a second DB round-trip and guarantees the guard redacts
# with exactly the same host/suffix set the classifier uses.
ctx.egress_guard = EgressGuard(
extra_hosts=effective_idents.hosts,
extra_suffixes=effective_idents.suffixes,
)
else:
# No DB on ctx (CLI / eval / tests) or resolution failed —
# env-only identifiers, same fallback as the Oracle path.
ctx.egress_guard = await EgressGuard.for_settings(ctx.settings)
guard = ctx.egress_guard
if guard is not None and focus_hint:
# focus_hint carries prior open questions (real host/IP text from a
# stored report) and is woven into every seed prompt below.
focus_hint = guard.sanitize_text(focus_hint)
# ----- Phase A: rich precompute -----
enrichment_ctx = EnrichmentContext(
blocklist=ctx.blocklist,
maxmind=ctx.maxmind,
cloud=ctx.cloud,
)
try:
enriched = await retry_prefetch(
lambda: get_enriched_alert_context(
alert_id,
elastic=ctx.elastic,
settings=ctx.settings,
enrichment=enrichment_ctx,
misp=ctx.misp,
include_synth=ctx.include_synth,
# Thread the effective CIDR set (settings.internal_cidrs union
# active 'cidr' rows minus muted, resolved once above) into
# Phase-A enrichment so an activated CIDR marks hosts internal
# here too — consistent with the ICMP-downgrade classification
# path. No active cidr rows / no DB ⇒ classification_cidrs ==
# settings.internal_cidrs (unchanged).
internal_cidrs=classification_cidrs,
),
max_retries=ctx.settings.prefetch_max_retries,
base_delay_s=ctx.settings.prefetch_retry_base_delay_s,
)
except Exception as e:
err_ev = _ev("error", _error_payload(e, phase="prefetch", round_num=0))
await _audit(err_ev)
yield err_ev
return
# ----- The hunt subject joins the prefetch -----
# D2. The subject's cited documents ARE prefetched evidence: the promotion
# fetched them by id before this run started. They go into the bundle here,
# BEFORE the enriched event and before the budget trim, so the stored
# record shows what the run held and every id resolver
# (gates._retrieved_evidence_tokens, _pivot_event_ids) reads them beside
# the pivots. Empty on every alert run.
subject_block = ""
# Change E. A subject with a promoted finding judges the claim of that
# finding, and the prompt says so. A lead promotion judges the hunt.
subject_finding = subject is not None and subject.promoted_finding is not None
if subject is not None:
enriched.subject_documents = list(subject.documents)
# Change A. The documents that the promoted finding cites. Only the
# host history gate reads them, and the field stays out of every dump.
enriched.subject_focus_document_ids = subject.promoted_document_ids()
subject_block = subject.render_block()
enriched_payload = enriched.model_dump(mode="json")
if subject is not None:
# What this step loaded. The timeline titles the step from these two
# keys plus the document list, so a hunt subject does not read
# "Loaded alert context + enrichments".
enriched_payload["subject"] = "hunt"
enriched_payload["subject_findings"] = len(subject.findings)
enriched_ev = _ev("enriched_alert_context", enriched_payload)
await _audit(enriched_ev)
yield enriched_ev
# Anchor every time-windowed tool on THIS alert's @timestamp immediately
# after prefetch. The fast / Phase-D path dispatches the query tools and
# t_get_pcap WITHOUT entering the investigation-loop branch below (which is
# where the anchor used to be set), so without this a round-1 gap dispatch
# (or a t_get_pcap capture) fell back to a now-relative window; and on a ctx
# reused across an auto-triage / backtest sweep it could inherit a PREVIOUS
# alert's timestamp. Set unconditionally here for every run so both the
# Phase-D and loop paths center on this alert (matches the documented
# InvestigationContext.default_time_anchor contract). The loop branch below
# re-sets it (with dedup / prefetched_community_ids) harmlessly.
ctx.default_time_anchor = enriched.alert.timestamp
# ----- Host dossier: what the hosts in this alert ARE -----
# Built ONCE here, from THIS alert's hosts (nothing is cached on ctx —
# an InvestigationContext is reused across auto-triage / backtest sweeps,
# the bug class that already bit default_time_anchor above), and woven into
# every message that reaches the analyst model below. Without it the model
# only ever learns what a host DID; the motivating failure was attributing
# SSH probing to an address without knowing it was the hypervisor running
# the SOC's own stack, whose policy forbids interactive SSH.
#
# RAW at this point, deliberately: every injection site below composes the
# block into its message BEFORE that site's sanitize sweep, so the dossier's
# addresses collapse onto the same labels the alert already carries.
# `host_dossier_prompt_block` never raises and returns "" on every off
# switch, so this costs a disabled or unswept deployment nothing.
dossier_hosts = dossier_hosts_for_alert(enriched, ctx.settings)
dossier_block = await host_dossier_prompt_block(dossier_hosts, ctx=ctx)
if dossier_block:
# Timeline transparency, emitted ONLY when a non-empty block is actually
# injected: WHAT the model was told about these hosts, with real values.
# Local storage, never egress — the same rule the enriched_alert_context
# event follows.
#
# `hosts` is capped to the same ceiling the block is, with the remainder
# as its own count. The full map here beside a block describing eight
# would be an audit record naming hosts the model was never told about,
# and the analyst reading it cannot tell the two apart.
dossier_ev = _ev(
"host_dossier",
{
"hosts": dict(list(dossier_hosts.items())[:MAX_PROMPT_HOSTS]),
"hosts_omitted": max(0, len(dossier_hosts) - MAX_PROMPT_HOSTS),
"block": dossier_block.strip(),
},
)
await _audit(dossier_ev)
yield dossier_ev
# ----- Phase B: decision template -----
# A hunt subject runs no template. Every template matches an alert RULE
# class (a classtype, a signature severity, a rule name) and a hunt has no
# rule. A template that fired here would settle a hunt on the rule class of
# whichever document the prefetch happened to anchor on.
candidate = None if subject is not None else match_decision_template(enriched)
template_ev = _ev(
"decision_template_match",
{
"matched": candidate is not None,
"skipped": "hunt_subject" if subject is not None else None,
"template_id": candidate.template_id if candidate else None,
"verdict": candidate.verdict if candidate else None,
"confidence": candidate.confidence if candidate else None,
"rationale": candidate.rationale if candidate else None,
},
)
await _audit(template_ev)
yield template_ev
# ----- Phase C: synth round 1 -----
# Speed: when the case will investigate REGARDLESS of the round-1 verdict
# (malware/exploit signal or external-reputation template), skip the round-1
# synth entirely — it's a ~10-15s HEAVY call whose verdict the loop discards.
#
# Context budgeting: the enriched context feeds EVERY model call below
# (round-1 synth, the investigation loop, round-2 synth). If it exceeds the
# analyst model's input budget (window discovered from the gateway, or the
# model_context_window_tokens override), drop the oldest pivot events now —
# a ContextWindowExceeded later burns the whole run into a fallback verdict.
_ctx_window = await context_budget.resolve_model_window(ctx.settings)
_enriched_budget = context_budget.input_budget_tokens(_ctx_window)
if _enriched_budget is not None and dossier_block:
# The dossier is paid for OUT OF this budget, not on top of it: it rides
# the same messages the enriched context does, so charging it nowhere
# would let a fat dossier push real evidence past the window. Clamped at
# _MIN_ENRICHED_BUDGET — see that constant.
_enriched_budget = max(
_MIN_ENRICHED_BUDGET,
_enriched_budget - context_budget.estimate_tokens(dossier_block),
)
enriched_json, _trim_note = context_budget.trim_enriched_for_budget(enriched, _enriched_budget)
if guard is not None:
# EGRESS BOUNDARY: enriched_json feeds every analyst-model call below
# (round-1 synth, the investigation-loop prompt, round-2 synth) —
# sanitize it ONCE here, after trimming, so all consumers share the
# same labeled text. The raw `enriched_alert_context` event above is
# deliberately untouched: it is local storage, never egress.
enriched_json = guard.sanitize_text(enriched_json)
if _trim_note is not None:
trim_ev = _ev("context_trimmed", _trim_note)
await _audit(trim_ev)
yield trim_ev
# ----- Same-session verdicts (BEFORE the template fast path) -----------
# Has this exact conversation already been read? Keyed on the community id,
# which is the hashed five-tuple, so unlike the rule-keyed memory below it
# relates two DIFFERENT rules firing on one session, and does not relate two
# different sessions between the same pair. Resolved here, ahead of Phase C,
# because a benign decision template can close a case with no tools at all
# and a constraint arriving after that has arrived too late: on the range,
# the second alert's own prefetch matched a benign template and shut the
# case while the first alert's true positive was already on the board.
#
# Deliberately NOT behind ``memory_enabled``. That flag governs whether the
# model is shown resemblances it may weigh; this is a contradiction inside
# one session, and a product that ships with it switched off ships the
# defect. Fail-soft everywhere else: no DB (CLI / eval / tests), no session
# on the alert, or a store error all leave the constraint absent.
#
# A hunt subject skips it. A hunt has no session of its own: the community
# id here belongs to whichever document the prefetch anchored on, and a
# verdict about a hunt is not a verdict about that one conversation.
session_digests: list[dict[str, Any]] = []
session_window_minutes = 0
session_community_id = (
None if subject is not None else getattr(enriched.alert, "network_community_id", None)
)
if ctx.db_sessionmaker is not None and session_community_id:
from soc_ai.store.investigations import ( # noqa: PLC0415
SESSION_WINDOW_MINUTES,
session_verdicts,
)
try:
async with ctx.db_sessionmaker() as sess_db:
session_digests = await session_verdicts(
sess_db,
community_id=str(session_community_id),
# This run's own row is still ``running`` and the lookup is
# complete-only, so it cannot self-match — the same reason
# prior_outcomes passes no id from inside the pipeline.
exclude_id=None,
# An earlier run of THIS alert is not a second alert on
# the session. A re-run read its own earlier verdict as
# one (R2-D9).
exclude_alert_es_id=alert_id,
limit=_SESSION_MAX_ITEMS,
)
session_window_minutes = SESSION_WINDOW_MINUTES
except Exception as e:
_LOGGER.warning("same-session verdict lookup failed (skipping constraint): %s", e)
session_digests = []
session_block = _format_session_prior_block(session_digests) if session_digests else ""
# A true positive on this session is the case that must not be closed
# quietly, so it also forces the investigation loop: a fast-path close is
# exactly the outcome being prevented.
session_conflict_possible = any(d.get("verdict") == "true_positive" for d in session_digests)
if session_digests:
# The analyst is told the two are related whether or not the verdicts
# end up disagreeing. Light payload, same rule as the memory events:
# ids and verdicts, never rationale text.
session_ev = _ev(
"session_prior",
{
"count": len(session_digests),
"window_minutes": session_window_minutes,
"forces_investigation": session_conflict_possible,
"items": [
{"id": d.get("id"), "verdict": d.get("verdict")} for d in session_digests
],
},
)
await _audit(session_ev)
yield session_ev
# ----- The analyst verdict on this alert (R2-D1) -----
# A re-run after an analyst override read nothing of the override. The
# standing analyst verdict reaches every prompt site as one fact, on the
# same injection path as the session constraint. It is context: the run
# may still disagree, and the auto-acknowledge reads the analyst verdict
# again on its own (:func:`_analyst_verdict_holds_ack`). A hunt subject
# skips it, for the reason it skips the session lookup above. Fail-soft:
# a store error leaves the fact out, and the ack guard fails closed.
if subject is None and ctx.db_sessionmaker is not None:
standing_analyst, _ = await _standing_analyst_verdict(ctx, alert_id)
if standing_analyst is not None:
analyst_block = _format_analyst_verdict_block(standing_analyst)
session_block = (
f"{session_block}\n\n{analyst_block}" if session_block else analyst_block
)
analyst_ev = _ev(
"analyst_verdict",
{
"id": standing_analyst.investigation_id,
"verdict": standing_analyst.verdict,
"analyst": standing_analyst.resolved_by,
"resolved_via": standing_analyst.resolved_via,
"resolved_at": standing_analyst.resolved_at,
},
)
await _audit(analyst_ev)
yield analyst_ev
di_trigger = (
_definitely_investigate_trigger(enriched, candidate)
if ctx.settings.investigate_when_unsure
else None
)
definitely_investigate = di_trigger is not None
# ----- Budget class (stage 1, soc_ai.agent.budget) -----
# The rungs above the model plan the class: an explicit request from the
# caller, the hunt subject, the session constraint, the decision template.
# Planned cheap is the round-1 path. An explicit standard or deep request
# runs the loop even where a template could settle the alert, because a
# person asked for an investigation.
class_plan = budget.plan_run_class(
requested=requested_class,
subject_is_hunt=subject is not None,
round1_can_settle=subject is None and _round1_can_settle(enriched, candidate),
session_true_positive=session_conflict_possible,
fast_triage_enabled=bool(ctx.settings.fast_triage_enabled),
)
# fast_triage_enabled=False forces the tool-driven loop regardless of how
# confident round-1 was ("agent does agent things"): deeper but slower.
# `deep` is the same override scoped to THIS run (the analyst's deep re-run),
# and an analyst's standard request is the same override at the standard
# budget.
force_investigate = deep or not ctx.settings.fast_triage_enabled or class_plan.forces_loop
# W2. Round 1 settles a case only when _round1_can_settle holds; on every
# other alert the loop runs and overwrites the verdict, so the call is pure
# cost (18 s and ~8K tokens on production, discarded on 83% of runs). Skip
# it. The second conjunct is the safety rail: with the loop switched off,
# round 1 is the only verdict there is, so it always runs.
loop_runs_anyway = (
force_investigate or session_conflict_possible or ctx.settings.investigate_when_unsure
)
# A hunt subject never settles at round 1. Round 1 is a tool-less synthesis
# over one alert's prefetch, and a hunt is read with the tools over many
# documents. Skipping it also keeps Phase D out of the run: the skipped
# report names no gap, so the loop is the only path to a verdict.
round1_cannot_settle = subject is not None or (
not definitely_investigate
and loop_runs_anyway
and not getattr(ctx.settings, "synth_round1_always", False)
and not _round1_can_settle(enriched, candidate)
)
# A run whose class forces the loop (an analyst's standard or deep request)
# never lets round 1 settle, so the call before the loop is pure cost.
round1_class_skip = (
class_plan.forces_loop
and not definitely_investigate
and not round1_cannot_settle
and not getattr(ctx.settings, "synth_round1_always", False)
)
round1_ok = False
# (agent, user_message, usage_limits|None) describing how to RE-RUN the
# final verdict synthesis — consumed by the flag-gated self-consistency
# vote below. Set at each site whose output becomes ``triage_final``;
# cleared (None) on every fallback path so a vote never re-runs a call
# that just failed.
final_synth_rerun: tuple[Any, str, Any] | None = None
# True when ``final_synth_rerun`` re-runs the investigation loop itself,
# tools and all (the W3 report path). A sample of the loop that stops at
# its budget then lands a partial verdict and votes, as the loop does.
rerun_is_loop = False
# True when the primary loop stopped at its budget and its vote is the
# partial verdict the partial synthesis wrote.
primary_partial = False
if definitely_investigate or round1_cannot_settle or round1_class_skip:
triage_round1 = _round1_skipped_report(alert_id)
if subject is not None:
skip_reason = "hunt_subject"
elif definitely_investigate:
skip_reason = "definitely_investigate"
elif round1_class_skip:
skip_reason = "class_runs_loop"
else:
skip_reason = "cannot_settle"
skip_payload: dict[str, Any] = {"reason": skip_reason}
if skip_reason == "definitely_investigate":
skip_payload["trigger"] = di_trigger
skip_ev = _ev("synth_round1_skipped", skip_payload)
await _audit(skip_ev)
yield skip_ev
else:
materialized = _materialize_prefetch_evidence(enriched)
if guard is not None:
# list[str] of evidence bullets built from RAW enriched fields —
# redact before it joins the synth prompt (labels stay stable
# because the guard's mapping is shared with enriched_json above).
materialized = guard.sanitize_obj(materialized)
# ----- E4.2: deterministic prior-outcome memory (round-1 ONLY) -----
# Flag-gated context block: the most relevant PRIOR verdicts for this
# alert's (rule, src, dest), fetched with one indexed SQL query — no
# embeddings, no extra model calls. Fail-SOFT end to end: no DB on ctx
# (CLI / eval / tests), no rule name, a store error, or zero matches
# all just skip the block — memory must never kill an investigation.
# ``is True`` (not truthiness) so a non-Settings test double can never
# flip it on. The loop / round-2 / retask prompts are deliberately
# untouched: they reason over GATHERED evidence, where a prior verdict
# would compete with real tool results.
prior_outcomes_block: str | None = None
prior_digests: list[dict[str, Any]] = []
if (
getattr(ctx.settings, "memory_enabled", False) is True
and ctx.db_sessionmaker is not None
and enriched.alert.rule_name
):
from soc_ai.store import investigations as investigations_store # noqa: PLC0415
try:
async with ctx.db_sessionmaker() as mem_db:
prior_digests = await investigations_store.prior_outcomes(
mem_db,
rule_name=enriched.alert.rule_name,
src_ip=enriched.alert.source_ip,
dest_ip=enriched.alert.destination_ip,
# This run's own row is still status='running' (finalize
# lands after the stream ends) and prior_outcomes is
# complete-only, so it can never self-match — there is
# no row id to exclude from inside the pipeline.
exclude_id=None,
window_days=ctx.settings.memory_window_days,
limit=ctx.settings.memory_max_items,
)
except Exception as e:
_LOGGER.warning("prior-outcome memory lookup failed (skipping block): %s", e)
prior_digests = []
if prior_digests:
prior_outcomes_block = _format_prior_outcomes_block(prior_digests)
# Timeline transparency: record WHAT was recalled — ids/verdicts/
# tier only, never rationale text (the digests live in the prompt;
# the event payload stays light). Emitted ONLY when a non-empty
# block is actually injected.
mem_ev = _ev(
"prior_outcomes",
{
"count": len(prior_digests),
"window_days": ctx.settings.memory_window_days,
"items": [
{
"id": d.get("id"),
"verdict": d.get("verdict"),
"matched_on": d.get("matched_on"),
}
for d in prior_digests
],
},
)
await _audit(mem_ev)
yield mem_ev
# ----- Chat-transcript memory (round-1 ONLY, sub-switch of E4.2) -----
# Operator intent, verbatim rule: past chat transcripts are CONTEXT,
# never evidence — the user is not always right. Same fail-soft +
# same window/limit knobs as the priors; gated on memory_enabled AND
# memory_include_chat (both `is True` so a test double can't flip it).
chat_memory_block: str | None = None
chat_digests: list[dict[str, Any]] = []
if (
getattr(ctx.settings, "memory_enabled", False) is True
and getattr(ctx.settings, "memory_include_chat", False) is True
and ctx.db_sessionmaker is not None
):
from soc_ai.store import chat_memory as chat_memory_store # noqa: PLC0415
chat_terms = _chat_memory_query_terms(
enriched.alert.rule_name,
enriched.alert.source_ip,
enriched.alert.destination_ip,
)
if chat_terms:
try:
async with ctx.db_sessionmaker() as mem_db:
chat_digests = await chat_memory_store.relevant_chat_snippets(
mem_db,
query_terms=chat_terms,
# This run has no chat thread yet (chats attach to
# COMPLETED investigations), so there is no own
# thread to exclude from inside the pipeline —
# mirrors the prior_outcomes exclude_id=None note.
exclude_thread=None,
window_days=ctx.settings.memory_window_days,
limit=ctx.settings.memory_max_items,
)
except Exception as e:
_LOGGER.warning("chat-transcript memory lookup failed (skipping): %s", e)
chat_digests = []
if chat_digests:
chat_memory_block = _format_chat_memory_block(chat_digests)
# Sibling of the prior_outcomes event, same light-payload rule:
# source/thread/role only — snippet text lives in the prompt,
# never in the timeline payload. Emitted ONLY on injection.
chat_ev = _ev(
"chat_memory",
{
"count": len(chat_digests),
"window_days": ctx.settings.memory_window_days,
"items": [
{
"source": d.get("source"),
"thread_id": d.get("thread_id"),
"role": d.get("role"),
}
for d in chat_digests
],
},
)
await _audit(chat_ev)
yield chat_ev
user_msg_round1 = build_synth_first_user_message(
alert_id=alert_id,
enriched_ctx_json=enriched_json,
materialized_evidence=materialized,
candidate=candidate,
focus_hint=focus_hint,
focus_origin=focus_origin,
# Included BEFORE the final sanitize sweep + _guard_egress below,
# so prior rationale text is redacted on the cloud-analyst path.
prior_outcomes_block=prior_outcomes_block,
# Same egress contract as the priors block (composed pre-sweep).
chat_memory_block=chat_memory_block,
)
# Injection 1 of 4. Rendered ahead of the enriched context so the model
# reads what the hosts ARE before what they DID, and so the current
# evidence still arrives last. Composed here, one statement before the
# sanitize sweep below — never after it.
user_msg_round1 = _inject_dossier_block(user_msg_round1, dossier_block)
user_msg_round1 = _inject_session_constraint(user_msg_round1, session_block)
if guard is not None:
# Final sweep over the COMPOSED message — catches the decision-
# template candidate block (rationale/cited_evidence carry real
# values) without mutating the candidate object itself, which the
# post-synth gates must later compare against RAW enriched values.
# Re-sanitizing the already-labeled parts is a no-op (labels don't
# match any redaction pattern).
user_msg_round1 = guard.sanitize_text(user_msg_round1)
synth_agent = build_synth_first_agent(
build_synthesizer_model(ctx.settings, temperature=ctx.settings.synthesizer_temperature),
output_mode=ctx.settings.synthesizer_output_mode,
)
# Captures the run's message history even when the run raises — on a
# schema-retry exhaustion the RetryPromptParts in here are the only
# record of WHY each attempt failed (the prod 2026-07-17/18 fallbacks
# recorded just "Exceeded maximum output retries (3)" with hint=null).
r1_captured: list[Any] = []
r1_attr: dict[str, Any] = {}
try:
# Fail-closed residue sweep on the FINAL composed outbound message
# (after every sanitize_text above): if fail-closed is on and an
# internal identifier survived, this raises BEFORE the model call so
# the payload never egresses. No-op when off / no guard.
user_msg_round1 = _guard_egress(guard, user_msg_round1, ctx.settings)
with (
capture_run_messages() as r1_captured,
capture_backend_attribution() as r1_attr,
):
async with asyncio.timeout(ctx.settings.investigation_turn_timeout_s):
synth_result_round1 = await synth_agent.run(user_msg_round1)
except EgressResidueError as e:
# Blocked egress: the model was NOT called. Audit the block (count
# only) + emit the paired error event, then land the SAME honest
# pipeline-fallback (E1.2) a synth crash would — the run couldn't
# safely proceed, so it IS a pipeline error.
async for ev in _emit_egress_blocked("synth_first_round1", e):
yield ev
triage_round1 = _synth_failure_fallback_report(alert_id, "egress_blocked", e)
await metrics.get_metrics().record_event("fallback_verdict", {})
except Exception as e:
# Emit error event for the audit trail, then
# fall through with a fallback NMI TriageReport so the row in
# index.jsonl is structured (not verdict=None). The post-validators
# + triage_report emission below run uniformly on the fallback.
r1_causes = _retry_causes_from_messages(r1_captured)
r1_served = _served_backend(r1_attr)
err_ev = _ev(
"error",
_error_payload(
e,
phase="synth_first_round1",
round_num=1,
retry_causes=r1_causes,
served_backend=r1_served,
),
)
await _audit(err_ev)
yield err_ev
triage_round1 = _synth_failure_fallback_report(
alert_id, "synth_first_round1", e, retry_causes=r1_causes, served_backend=r1_served
)
await metrics.get_metrics().record_event("fallback_verdict", {})
else:
triage_round1 = _strip_model_resolution(synth_result_round1.output)
if guard is not None:
# Restore real identifiers AT THE ASSIGNMENT SOURCE (not at a
# later convergence point): the desanitized report drives
# _should_investigate's evidence checks AND Phase-D dispatch
# (gap_for_investigator.tool_args must hit Elasticsearch with
# real values, not labels). Egress re-sanitizes downstream.
triage_round1 = _desanitize_report(triage_round1, guard)
round1_ok = True
# Round-1 IS the final synthesis when neither the loop nor Phase D
# supersedes it below (each overwrites/clears this on its own path).
final_synth_rerun = (synth_agent, user_msg_round1, None)
# Surface the synthesizer's thinking — without this a round-1-settled
# run stored NO model_response event and the reasoning panel was empty.
r1_reasoning_ev = _reasoning_ev(1, synth_result_round1)
if r1_reasoning_ev is not None:
await _audit(r1_reasoning_ev)
yield r1_reasoning_ev
usage_ev = _usage_ev(1, synth_result_round1, _served_backend(r1_attr))
if usage_ev is not None:
await _audit(usage_ev)
yield usage_ev
triage_final = triage_round1
# ----- Cheap class: a verdict gate escalates the run -----
# A cheap run settles on the round-1 false positive. The deterministic
# gates run after the verdict, so ask them now, on the same inputs they
# will see: when one of them would change the round-1 verdict, the cheap
# verdict cannot stand and the run escalates to the standard loop. The
# gates are pure, so the real pass below repeats this exactly.
gate_escalation = False
if (
class_plan.run_class == budget.CHEAP
and round1_ok
and getattr(triage_round1, "verdict", None) == "false_positive"
):
try:
gated_round1, _gate_audit = _synth_first_post_validate(
triage_round1,
enriched,
candidate,
synthesis_confidence_floor=ctx.settings.synthesis_confidence_floor,
blocklist=ctx.blocklist,
internal_cidrs=classification_cidrs,
)
gate_escalation = gated_round1.verdict != triage_round1.verdict
except Exception:
# A gate fault is no reason to trust the cheap verdict.
_LOGGER.warning("cheap-class gate pre-check failed; escalating", exc_info=True)
gate_escalation = True
# ----- Bounded investigation loop (Theme-1 Task 1) -----
# The Phase C synth is a NO-tools structured-output guess that
# rationalizes the prefetch. When its verdict isn't evidence-backed and
# the alert isn't trivially benign, run a REAL investigation loop: the
# tool-bound investigator (on the HEAVY model) chooses which read tools
# to call, then the synthesizer concludes from the gathered transcript.
# This replaces the zero-tool synthesis that scored 1-4/9 on synth-TP
# (confidently clearing a Cobalt Strike beacon). Reversible via the
# investigate_when_unsure flag.
ran_investigation_loop = False
# The tool exchange the pipeline ran in code before the loop (the web
# search). Joined to the loop's message history for the gates below.
prefetch_tool_messages: list[Any] = []
# Why the run ended in its class: the loop reason when the loop ran, else
# the plan's reason (a cheap run settled at round 1).
run_class_reason = (
class_plan.reason if class_plan.run_class == budget.CHEAP else "loop_not_entered"
)
loop_messages: list[Any] | None = None
# The investigator's evidence bullets (claim → supporting-id index), kept
# for the Oracle escalation payload. Set only on the completed-loop path;
# the budget-cut partial path has no InvestigationTranscript to read.
loop_evidence_bullets: list[str] | None = None
# The same bullets in REAL-VALUE space, for the report's citations. Separate
# from the list above because that one is kept in whatever space the loop ran
# in (label space under an egress guard) for the Oracle.
loop_evidence_for_report: list[str] | None = None
# W3 A/B marker: which call wrote the report the run lands. ``investigator``
# = the loop's own output (investigator_emits_report ON); ``synth_round2`` =
# the round-2 synthesis over the transcript. Stays None on every other path
# (a fallback, a budget-cut partial, a round-1-settled or Phase-D run), so
# the marker never claims a path that did not run. Read back off the stored
# ``triage_report`` event to compare the two arms.
report_path: str | None = None
# The swarm triage experiment (soc_ai.agent.loop_result): the strategy of
# the investigation loop, read once so the loop and the report agree even
# if an admin switches it mid-run. The report names it unless it is
# ``single``, and carries the swarm counters when a swarm ran.
loop_strategy = strategy_of(ctx.settings)
swarm_counters: dict[str, Any] | None = None
# A true positive already stands on this session, so a fast-path close is
# the exact outcome being prevented and the loop runs whatever the flags
# say. Deliberately outside the ``investigate_when_unsure`` conjunct: the
# contradiction does not stop being one because that flag is off.
# ``round1_cannot_settle`` sits at the top level for the same reason it was
# safe to skip the call: the alert has no settling verdict of any kind, so
# the loop is the only path to one.
if (
force_investigate
or session_conflict_possible
or round1_cannot_settle
or (
ctx.settings.investigate_when_unsure
and (
definitely_investigate
or gate_escalation
or (round1_ok and _should_investigate(triage_round1, enriched, candidate))
)
)
):
ran_investigation_loop = True
if definitely_investigate:
loop_reason = "definitely_investigate"
elif deep:
loop_reason = "deep_rerun"
elif class_plan.forces_loop:
loop_reason = class_plan.reason
elif force_investigate:
loop_reason = "fast_triage_disabled"
elif session_conflict_possible:
loop_reason = "session_true_positive_stands"
elif round1_cannot_settle:
loop_reason = "round1_cannot_settle"
elif gate_escalation:
loop_reason = "gate_would_change"
else:
loop_reason = "verdict_not_evidence_backed"
run_class_reason = loop_reason
loop_ev = _ev(
"investigation_loop_entered",
{
"reason": loop_reason,
# Round 1 did not run on the skip paths, so there is no verdict
# to report. ``round1_ok`` is the one honest test for that.
"round1_verdict": triage_round1.verdict if round1_ok else None,
"round1_confidence": triage_round1.confidence if round1_ok else None,
# The class the loop runs in. A planned cheap run that gets
# here escalated to standard.
"run_class": budget.ran_class(class_plan, ran_loop=True),
# Which of the three triggers sent the alert here, when the
# reason is definitely_investigate. The timeline names it.
**({"trigger": di_trigger} if loop_reason == "definitely_investigate" else {}),
},
)
await _audit(loop_ev)
yield loop_ev
# Reset per-investigation tool state so the investigator's tools
# anchor on THIS alert (mirrors the legacy investigate() prefetch
# block): time anchor, dedup tracker, prefetched community_ids.
ctx.default_time_anchor = enriched.alert.timestamp
ctx.dedup = _DedupTracker()
ctx.prefetched_community_ids = prefetched_community_ids(enriched)
# The alert message carries the rule body on some sensors. The loop then
# reads the rule in its prompt instead of spending a turn on it.
ctx.rule_body_in_prompt = rule_body_in_alert(enriched.alert)
loop_usage_limits = UsageLimits(
request_limit=ctx.settings.agent_request_limit,
tool_calls_limit=ctx.settings.agent_tool_calls_limit,
)
# HEAVY model (build_synthesizer_model), NOT the fast investigator
# model — the loop must reason on the strong model. The Nemotron
# profile on the heavy builder already carries the tool_choice
# workaround so tool-calling works. Moderate temperature: keep some
# pivot exploration while staying broadly reproducible.
#
# W3: with investigator_emits_report ON the loop writes the TriageReport
# itself and the round-2 synthesis below never runs. Read once here so
# the agent, the prompt and the convergence all see one value even if an
# admin hot-applies the setting mid-run.
emits_report = bool(getattr(ctx.settings, "investigator_emits_report", False))
# W1: ask the grid once an hour whether it holds any playbook. It
# answers which tools get registered below: a deployment with no
# playbook must not be offered t_get_playbooks, because the model spends
# a turn on it and reads back `[]`.
have_playbooks = await prime_playbook_presence(ctx)
# Which conditional tool can still change THIS verdict. A tool whose
# condition is not met is not named, so the loop is not invited to
# spend a turn on it.
conditions = case_conditions(
enriched,
playbooks_available=have_playbooks,
web_search_available=bool(ctx.settings.web_search_enabled),
)
# ----- Web search from code (stage 1, item 4) -----
# The loop spent its one tool turn on this search in 87% of production
# runs, on a condition the code already computes, at about 31K input
# tokens a turn. The pipeline makes the call here, before the loop, the
# same way Phase D dispatches a named tool. The result rides into the
# loop's message, the condition line leaves it, and the dedup tracker
# turns a repeat of the same query into a stub. A search that failed
# leaves the condition in place, so the loop can still try.
web_search_block = ""
web_indicator = (
unresolved_external_indicator(enriched)
if conditions.web_search and subject is None
else None
)
if web_indicator is not None:
search_args = {"query": web_indicator}
search_gap = TargetedGap(
question=f"What is the external indicator {web_indicator}?",
tool_name="t_web_search",
tool_args=search_args,
why_this_matters=(
"The prefetch enrichment left this external indicator unanswered."
),
)
search_dispatch_ev = _ev(
"targeted_dispatch",
{
"question": search_gap.question,
"tool_name": search_gap.tool_name,
"tool_args": search_args,
"why_this_matters": search_gap.why_this_matters,
"phase": "prefetch",
},
)
await _audit(search_dispatch_ev)
yield search_dispatch_ev
search_result = await run_targeted_investigation(search_gap, ctx=ctx)
search_result_ev = _ev(
"targeted_tool_result",
{"tool_name": "t_web_search", "result": search_result, "phase": "prefetch"},
)
await _audit(search_result_ev)
yield search_result_ev
if isinstance(search_result, dict) and search_result.get("ok") is True:
# Registered as seen: the loop's identical call short-circuits.
ctx.dedup.is_duplicate("t_web_search", search_args)
conditions = dataclasses.replace(conditions, web_search=False)
web_search_block = format_prefetched_web_search_block(web_indicator, search_result)
# The gates read what the run gathered from the loop's message
# history. This call replaced the loop's own turn, so it joins
# that record exactly as the loop's call would have.
prefetch_tool_messages = _code_tool_exchange(
"t_web_search", search_args, search_result
)
# ----- Prompt and tool schema trim per class (stage 1, item 5) -----
# The standard class sends the prompt sections and the tool schemas the
# alert's planes make useful. Every other investigator tool stays
# registered with deferred loading, one ``search_tools`` call away. The
# deep class, and a hunt subject, keep the whole prompt and every schema.
trim_planes = (
budget.alert_planes(enriched)
if subject is None and budget.ran_class(class_plan, ran_loop=True) == budget.STANDARD
else None
)
ctx.loop_visible_tools = (
budget.standard_visible_tools(trim_planes) if trim_planes is not None else None
)
ctx.deferred_tool_names = []
investigator_kwargs: dict[str, Any] = {}
if trim_planes is not None:
investigator_kwargs["system_prompt"] = build_investigator_prompt(
emits_report=emits_report,
planes=trim_planes,
visible_tools=ctx.loop_visible_tools,
)
investigator = build_investigator(
build_synthesizer_model(
ctx.settings, temperature=ctx.settings.investigator_temperature
),
ctx,
emits_report=emits_report,
**investigator_kwargs,
)
more_tools_block = format_more_tools_block(ctx.deferred_tool_names)
# Injection 2 of 4, beside the grid inventory: the two blocks are the
# same class of ambient ground truth, and rubric step 5 already tells the
# model to weigh what the host IS — this supplies the missing input.
#
# The endpoint-coverage block renders the prefetch's
# ``prefetch_gaps["endpoint.coverage"]`` verdict DIRECTLY AFTER the
# inventory, because the inventory's (true, grid-wide) "endpoint data
# exists — never conclude absence without querying" is exactly what
# sent budget-exhausted runs probing endpoint datasets for hosts with
# no endpoint agent. Empty string when covered/unknown.
inv_tail = (
await inventory_prompt_block(ctx.elastic, ctx.settings)
+ format_endpoint_coverage_block(enriched.prefetch_gaps.get(ENDPOINT_COVERAGE_GAP_KEY))
+ format_host_coverage_block(enriched.host_coverage, enriched.host_pivot_note)
+ dossier_block
)
inv_user_msg = (
_format_investigator_prompt(
alert_id,
enriched_json,
focus_hint=focus_hint,
focus_origin=focus_origin,
conditions=conditions,
emits_report=emits_report,
# D2. Non-empty only for a hunt subject, where it replaces the
# alert block: the hunt, its findings and its cited documents
# are the subject, so the enriched alert JSON is not rendered.
subject_block=subject_block or None,
subject_finding=subject_finding,
)
+ web_search_block
+ more_tools_block
+ inv_tail
)
# The investigator chooses what to retrieve, so it is the one place the
# constraint can change what gets looked at rather than only how the
# result is read.
inv_user_msg = _inject_session_constraint(inv_user_msg, session_block)
if guard is not None:
# enriched_json/focus_hint are already labeled; this sweep covers
# the dataset-inventory and host-dossier blocks (grid host/dataset
# names, dossier hostnames and addresses) so the WHOLE investigator
# prompt crosses the egress boundary sanitized.
inv_user_msg = guard.sanitize_text(inv_user_msg)
# The loop's outputs, whichever strategy runs it (soc_ai.agent.loop_result).
# ``loop.gathered`` holds the labeled node messages streamed so far: the
# budget-partial synthesis replays them when the loop is cut short
# (mirrors the hunt runner's `gathered`). ``loop.egress_blocked`` is set
# when the fail-closed residue sweep blocks the loop's first model call:
# it routes the budget-boundary convergence below to the HONEST
# pipeline-fallback (E1.2) instead of preserving a round-1 verdict, so a
# blocked egress renders as a pipeline error.
loop = LoopResult(
strategy=loop_strategy,
# The judge of a swarm writes a TriageReport whatever the W3 flag
# says, so the convergence takes the W3 branch for it.
emits_report=emits_report or loop_strategy in SWARM_STRATEGIES,
)
async def _land_cut_short_verdict(
exc: BaseException,
*,
note: str,
retry_causes: list[str] | None = None,
served_backend: dict[str, Any] | None = None,
) -> AsyncGenerator[StepEvent, None]:
"""Land a verdict when the loop stopped before it could conclude.
Audit R10: a fallback must never state a verdict that no evidence
supports. The order is fixed. A round-1 verdict stands only when
round 1 RAN and settled. Otherwise the gathered history is replayed
through the partial synthesizer, which reads the tool results that
did land. With no history at all, the honest failure report is the
answer, and the run stays retryable.
The budget cut and the loop-synth crash both come here, so the two
cannot drift apart. Sets ``triage_final`` and, on a partial
synthesis, ``loop_messages`` (the repaired history the citation
gates read).
"""
nonlocal triage_final, loop_messages, final_synth_rerun
nonlocal rerun_is_loop, primary_partial
partial_report: Any = None
repaired_history: list[Any] = []
# ``round1_ok`` is the honest test: the skip paths leave a
# placeholder report behind, and a crashed round 1 leaves a
# fallback one. Neither is a verdict anybody reached.
settled_r1 = round1_ok and getattr(triage_round1, "verdict", None) in (
"true_positive",
"false_positive",
)
partial_runs: list[Any] = []
if not settled_r1 and loop.gathered:
try:
partial_report, repaired_history = await _synthesize_partial_triage(
ctx.settings, guard, loop.gathered, result_out=partial_runs
)
except asyncio.CancelledError:
raise # cooperative cancel — propagate, never swallow
except EgressResidueError as e:
async for ev in _emit_egress_blocked("investigation_loop_partial_synth", e):
yield ev
except BaseException as e:
err_ev = _ev(
"error",
_error_payload(e, phase="investigation_loop_partial_synth", round_num=1),
)
await _audit(err_ev)
yield err_ev
final_synth_rerun = None # a cut-short verdict — never vote on it
for partial_run in partial_runs:
# The partial synthesis is a model run too. Its usage joins the
# run counters with the rest (soc_ai.run_meter).
partial_usage_ev = _usage_ev(2, partial_run)
if partial_usage_ev is not None:
await _audit(partial_usage_ev)
yield partial_usage_ev
if partial_report is not None:
triage_final = _cut_short_report(partial_report, note, guard)
# The repaired history feeds the downstream evidence/citation
# gates: the partial verdict earns the loop exemption only from
# tool results that actually landed (synthetic closures are
# error-shaped and never counted).
loop_messages = repaired_history
if loop.rerun is not None:
# The loop wrote the report (W3) and stopped at its budget.
# With the vote on, its partial verdict is its vote and the
# samples re-run the loop: an ensemble whose first run hit
# the budget still votes. With the vote off, nothing reads
# the rerun.
final_synth_rerun = loop.rerun
rerun_is_loop = True
primary_partial = True
return
triage_final = _round2_failure_fallback(
alert_id,
triage_round1 if round1_ok else None,
exc,
retry_causes,
served_backend,
)
await metrics.get_metrics().record_event("fallback_verdict", {})
async def _run_single_loop() -> AsyncGenerator[StepEvent, None]: # noqa: PLR0915 - the loop body moved whole
"""The one investigator: run the loop and fill ``loop``.
The ``single``, ``spend`` and ``ensemble`` strategies run this. It
streams every node as it lands, and it leaves the run's outputs on
``loop`` for the convergence below. The ``usage`` event of a run
that ended with a result is the last event it yields, at the point
in the stream where the convergence used to emit it.
"""
# Gateway attribution for the multi-turn loop's model calls.
# Mutable-dict sink (see capture_backend_attribution): filled per
# response, read after the loop for the usage event, valid even if a
# later turn raised.
loop_attr: dict[str, Any] = {}
inv_result: Any = None
try:
# Fail-closed residue sweep on the FINAL composed investigator prompt
# BEFORE the loop's first model call — if an internal identifier
# survived, this raises so the payload never reaches the model.
guarded_msg = _guard_egress(guard, inv_user_msg, ctx.settings)
if loop.emits_report:
# W3: the loop writes the report, so the vote re-runs the
# loop itself, tools and all. Set before the run, so a run
# cut short at its budget can still vote.
loop.rerun = (investigator, guarded_msg, loop_usage_limits)
# Stream the run NODE-BY-NODE via agent.iter() so each tool_call /
# tool_result / model_response lands in the timeline THE MOMENT it
# happens (the recorder persists every event immediately, FLUSH_EVERY=1)
# instead of replaying the whole investigation in one burst at the end.
# CallToolsNode carries the model's response (text + tool requests);
# the following ModelRequestNode carries that step's tool results.
node_msg: Any = None
with capture_backend_attribution() as loop_attr:
async with investigator.iter(
guarded_msg, usage_limits=loop_usage_limits
) as inv_run:
# Advance NODE-BY-NODE with a PER-TURN wall-clock timeout on each
# iterator step (the model-run await that produces the next node),
# NOT one timeout spanning the whole multi-turn loop. Event
# projection + yielding stays outside the timeout so streaming is
# unchanged. A per-turn TimeoutError propagates to the same
# except-handlers below (error event + honest stop) as any other
# investigator failure.
node_iter = inv_run.__aiter__()
while True:
async with asyncio.timeout(ctx.settings.investigation_turn_timeout_s):
try:
node = await node_iter.__anext__()
except StopAsyncIteration:
break
# CallToolsNode carries the model's response (text + tool
# requests); the following ModelRequestNode carries that
# step's tool results. Detect by attribute so the message is
# projected the moment its node arrives.
node_msg = getattr(node, "model_response", None)
if node_msg is None:
node_msg = getattr(node, "request", None)
if node_msg is not None:
loop.gathered.append(node_msg)
async for ev in _walk_message(
node_msg, _ev, phase="investigation_loop", round_num=1
):
# The loop converses in label space; restore real
# values in the streamed timeline events (tool
# calls/results, reasoning traces) — they are
# local storage/display, never egress.
out_ev = (
ev
if guard is None
else ev.model_copy(
update={"payload": guard.desanitize_obj(ev.payload)}
)
)
await _audit(out_ev)
yield out_ev
inv_result = inv_run.result
except asyncio.CancelledError:
raise # cooperative cancel — propagate, never swallow
except EgressResidueError as e:
# Fail-closed block on the investigator prompt — the model was NEVER
# called. Audit the block (count only) + emit the paired error event,
# then route the convergence below to the HONEST pipeline-fallback.
async for ev in _emit_egress_blocked("investigation_loop", e):
yield ev
loop.egress_blocked = e
return
except UsageLimitExceeded as e:
# Budget exhaustion is an EXPECTED outcome of a thorough investigation,
# NOT an infrastructure failure. The tool calls + model responses
# already streamed live above, so don't discard them with status=error
# and no verdict: land the round-1 verdict via the same fallback used
# when the loop-synth crashes. (This is the "conclude gracefully at the
# budget boundary" behaviour, vs. the old "error out" one.)
_LOGGER.warning("investigation loop hit budget limit: %s", e)
err_ev = _ev(
"error", _error_payload(e, phase="investigation_loop_budget", round_num=1)
)
await _audit(err_ev)
yield err_ev
loop.budget_exc = e
return
except TimeoutError as e:
# A per-turn wall-clock timeout (investigation_turn_timeout_s) fired
# while advancing the investigator on a slow stack. The evidence
# gathered before the hung turn already streamed live, so conclude
# GRACEFULLY with the round-1 verdict via the same budget-boundary
# fallback as UsageLimitExceeded — do NOT discard the run with
# status=error. (Mirrors the hunt runner's timeout→partial-report
# behaviour; a slow stack is the expected trigger, not an infra fault.)
_LOGGER.warning("investigation loop turn timed out: %s", e)
err_ev = _ev(
"error", _error_payload(e, phase="investigation_loop_timeout", round_num=1)
)
await _audit(err_ev)
yield err_ev
loop.budget_exc = e
return
except BaseException as e:
# The investigator could not gather evidence — most often a transient
# LLM-gateway connection drop (the openai client has already retried
# litellm_max_retries times). Do NOT fabricate a needs_more_info
# verdict from an empty transcript: that reads as if the agent
# investigated and was unsure. Surface an honest error and stop — the
# recorder marks the run 'error' (retryable), not a fake verdict.
_LOGGER.exception("investigation loop investigator run failed")
err_ev = _ev("error", _error_payload(e, phase="investigation_loop", round_num=1))
await _audit(err_ev)
yield err_ev
loop.stopped = True
return
if inv_result is None:
# No End node and no exception: the convergence below lands the
# error. ``loop.output`` stays None.
return
loop.output = inv_result.output
loop.messages = inv_result.all_messages()
loop.served_backend = _served_backend(loop_attr)
loop.usage = _run_usage(inv_result)
if loop.emits_report:
loop.report_path = "investigator"
inv_usage_ev = _usage_ev(1, inv_result, loop.served_backend)
if inv_usage_ev is not None:
await _audit(inv_usage_ev)
yield inv_usage_ev
if loop.strategy in SWARM_STRATEGIES:
# ----- The swarm strategies (arms D and E) -----
# soc_ai.agent.swarm fills the same LoopResult. Every step before
# this line and after the convergence below is the same for it.
from soc_ai.agent.swarm import ( # noqa: PLC0415 - the swarm imports this module
SwarmInputs,
run_strategy,
)
def _swarm_msg(more_tools: str) -> str:
# The loop's message with no report block: a worker writes a
# brief. Composed and sanitized as the loop's message is.
msg = (
_format_investigator_prompt(
alert_id,
enriched_json,
focus_hint=focus_hint,
focus_origin=focus_origin,
conditions=conditions,
emits_report=False,
subject_block=subject_block or None,
subject_finding=subject_finding,
)
+ web_search_block
+ more_tools
+ inv_tail
)
msg = _inject_session_constraint(msg, session_block)
return guard.sanitize_text(msg) if guard is not None else msg
swarm_inputs = SwarmInputs(
alert_id=alert_id,
ctx=ctx,
enriched=enriched,
worker_msg=_swarm_msg(more_tools_block),
# A plane worker and the judge cannot load a deferred tool.
context_msg=_swarm_msg(""),
trim_planes=trim_planes,
guard=guard,
make_ev=_ev,
)
async for ev in run_strategy(loop, swarm_inputs):
await _audit(ev)
yield ev
swarm_counters = dict(loop.counters)
else:
async for ev in _run_single_loop():
yield ev
if loop.stopped:
return
if loop.egress_blocked is not None:
# A blocked egress is a pipeline error (the run couldn't safely
# proceed) — land the HONEST pipeline-fallback (E1.2), NOT the
# round-1-preserving budget fallback. loop_messages stays None.
triage_final = _synth_failure_fallback_report(
alert_id, "egress_blocked", loop.egress_blocked
)
final_synth_rerun = None # fallback verdict — never vote on it
await metrics.get_metrics().record_event("fallback_verdict", {})
elif loop.budget_exc is not None:
# A settled round-1 verdict stands (evidence gathered pre-budget is
# preserved in the streamed timeline). With NO settled round-1
# (the round-1 call was skipped, or round-1 was itself NMI) the old
# path discarded every gathered tool result into a generic 0.3
# fallback — the 2026-07-18 prod BPFDoor run burned 25 tool calls
# and landed nothing. Now: synthesize a PARTIAL verdict from the
# gathered history (mirrors the hunt runner's budget path); the
# honest fallback remains the last resort.
async for ev in _land_cut_short_verdict(loop.budget_exc, note=_BUDGET_CUT_NOTE):
yield ev
elif loop.output is None:
# The agent run ended without a final result (no End node reached, and
# no exception raised). Emit an honest error instead of crashing with an
# UnboundLocalError on loop_transcript below.
err_ev = _ev(
"error",
_error_payload(
RuntimeError("investigation loop returned no result"),
phase="investigation_loop",
round_num=1,
),
)
await _audit(err_ev)
yield err_ev
return
elif loop.emits_report:
# ----- W3: the loop wrote the report -----
# The investigator's output type IS TriageReport here, so the run
# that gathered the evidence is the run that concludes on it. No
# second model call: the 2026-09-19 turn audit measured round 2
# adding no new evidence in 40 of 41 runs and reversing two correct
# range verdicts. Everything BELOW this block is shared with the
# round-2 path — the self-consistency vote, every post-synth
# validator, the Oracle, the session gate and the auto-ack all read
# ``triage_final`` / ``loop_messages`` and cannot tell the two apart.
# The loop's usage event already streamed with the loop.
report_path = loop.report_path
loop_messages = loop.messages
# This path hands the privileged failure marker to a tool-using
# model reading attacker-influenceable text; the strip is the same
# one every model-authored report gets.
loop_report = _strip_model_resolution(loop.output)
# The Oracle payload stays in the space the loop ran in (label space
# under a guard), exactly like the transcript bullets it replaces.
loop_evidence_bullets = [
str(c).strip() for c in (loop_report.citations or []) if str(c).strip()
]
triage_final = loop_report
if guard is not None:
# Assignment-source restore — the gates below compare this
# report's text against RAW enriched/pivot values.
triage_final = _desanitize_report(triage_final, guard)
# No separate transcript exists to lend citations to an uncited
# report: this report IS the loop's own words, and an empty citation
# list on it is a real finding for the A/B, not a handoff loss.
loop_evidence_for_report = None
# The self-consistency vote (OFF by default) re-runs the SAME call
# that wrote the report. Here that call is the loop itself, tools
# and all.
final_synth_rerun = loop.rerun
rerun_is_loop = loop.rerun is not None
# The loop replaces Phase D — strip any gap so we don't also
# dispatch a single-tool targeted round on top of it.
if triage_final.gap_for_investigator is not None:
triage_final = triage_final.model_copy(update={"gap_for_investigator": None})
else:
# Events and the loop's usage already streamed live above; land the
# transcript, and keep the full message history for the downstream
# citation/evidence check (loop_messages feeds _is_evidence_backed /
# targeted-cite check).
loop_transcript = loop.output
loop_messages = loop.messages
# Kept in whatever value-space the loop ran in (label space under a
# guard) — consistent with transcript_text at the Oracle call below.
loop_evidence_bullets = list(loop_transcript.evidence)
transcript_payload = {
"round": 1,
"phase": "investigation_loop",
**loop_transcript.model_dump(mode="json"),
}
if guard is not None:
# Stored/displayed copy gets real values; loop_transcript
# itself stays in label space — it feeds the loop-synth
# message below, which crosses the egress boundary.
transcript_payload = guard.desanitize_obj(transcript_payload)
# The REAL-VALUE copy of the same bullets. The citation gates below
# compare against raw enriched/pivot values and the analyst reads
# what they resolve to, so a label-space bullet would neither
# resolve nor be readable. Taken from the desanitized payload rather
# than desanitized a second time, so the stored transcript and the
# citations carried out of it can never disagree.
loop_evidence_for_report = [
str(e) for e in (transcript_payload.get("evidence") or []) if str(e).strip()
]
transcript_ev = _ev("investigation_transcript", transcript_payload)
await _audit(transcript_ev)
yield transcript_ev
# Synthesize over the gathered evidence — REUSE the legacy
# synthesizer-over-transcript (build_synthesizer + the transcript
# user-message formatter). HEAVY model, no tools.
loop_synth = build_synthesizer(
build_synthesizer_model(
ctx.settings, temperature=ctx.settings.synthesizer_temperature
),
output_mode=ctx.settings.synthesizer_output_mode,
)
# Injection 3 of 4, and the one that matters most: this is the
# verdict writer, and it sees ONLY the transcript plus the candidate
# — no enriched JSON, no inventory, no memory. Without the block, a
# loop-settled investigation reaches its verdict knowing nothing
# about what the hosts are.
loop_synth_msg = _inject_dossier_block(
_format_transcript_for_synthesizer(
alert_id, [loop_transcript], candidate=candidate
),
dossier_block,
)
loop_synth_msg = _inject_session_constraint(loop_synth_msg, session_block)
# D2. The verdict writer has to know the subject is a hunt, or it
# scores the transcript as an alert and the summary answers the
# wrong question. Empty on every alert run. Change E: on a finding
# run it reads the rules for one finding.
loop_synth_msg = _inject_subject_block(
loop_synth_msg,
f"{subject_block}\n\n{hunt_subject_rules(finding=subject_finding)}"
if subject_block
else "",
)
if guard is not None:
# The transcript is already in label space (the loop ran over
# sanitized inputs/tool results); this sweep covers the RAW
# candidate block woven into the synthesizer message.
loop_synth_msg = guard.sanitize_text(loop_synth_msg)
# Same capture as round-1: on schema-retry exhaustion the causes
# live only in the run's RetryPromptParts.
ls_captured: list[Any] = []
ls_attr: dict[str, Any] = {}
try:
# Fail-closed residue sweep on the FINAL composed loop-synth
# message before the model call — raises BEFORE egress on residue.
loop_synth_msg = _guard_egress(guard, loop_synth_msg, ctx.settings)
with (
capture_run_messages() as ls_captured,
capture_backend_attribution() as ls_attr,
):
async with asyncio.timeout(ctx.settings.investigation_turn_timeout_s):
loop_synth_result = await loop_synth.run(
loop_synth_msg,
usage_limits=loop_usage_limits,
)
except asyncio.CancelledError:
raise # cooperative cancel — propagate, never swallow
except EgressResidueError as e:
# Blocked egress: the model was NOT called. Audit the block (count
# only) + emit the paired error event, then land the HONEST
# pipeline-fallback (E1.2) — a blocked run IS a pipeline error.
async for ev in _emit_egress_blocked("investigation_loop_synth", e):
yield ev
triage_final = _synth_failure_fallback_report(alert_id, "egress_blocked", e)
final_synth_rerun = None # fallback verdict — never vote on it
await metrics.get_metrics().record_event("fallback_verdict", {})
except BaseException as e:
# A BaseException here (e.g. a gateway timeout/cancel that escaped
# the client retries) previously propagated past the recorder,
# landing status=error with NO verdict + no recorded error event.
# Catch it, record the error, and DON'T discard the evidence: a
# settled round-1 verdict stands, and with none the gathered
# history is replayed (same rule as the budget cut).
ls_causes = _retry_causes_from_messages(ls_captured)
ls_served = _served_backend(ls_attr)
err_ev = _ev(
"error",
_error_payload(
e,
phase="investigation_loop_synth",
round_num=2,
retry_causes=ls_causes,
served_backend=ls_served,
),
)
await _audit(err_ev)
yield err_ev
async for ev in _land_cut_short_verdict(
e,
note=(
" (The concluding synthesis did not complete; verdict "
"synthesized from the evidence gathered before it failed.)"
),
retry_causes=ls_causes,
served_backend=ls_served,
):
yield ev
else:
report_path = "synth_round2"
triage_final = _strip_model_resolution(loop_synth_result.output)
if guard is not None:
# Assignment-source restore — the gates below compare this
# report's text against RAW enriched/pivot values.
triage_final = _desanitize_report(triage_final, guard)
# The loop synth is now the final synthesis (supersedes round-1).
final_synth_rerun = (loop_synth, loop_synth_msg, loop_usage_limits)
# The loop's investigator turns streamed their own reasoning via
# _walk_message above; this covers only the concluding synth run.
loop_reasoning_ev = _reasoning_ev(2, loop_synth_result)
if loop_reasoning_ev is not None:
await _audit(loop_reasoning_ev)
yield loop_reasoning_ev
loop_synth_usage_ev = _usage_ev(2, loop_synth_result, _served_backend(ls_attr))
if loop_synth_usage_ev is not None:
await _audit(loop_synth_usage_ev)
yield loop_synth_usage_ev
# The loop replaces Phase D — strip any gap so we don't also
# dispatch a single-tool targeted round on top of it.
if triage_final.gap_for_investigator is not None:
triage_final = triage_final.model_copy(update={"gap_for_investigator": None})
# ----- Phase D (optional): targeted investigator -----
# Skipped when the investigation loop ran — the loop already gathered
# evidence agentically (it supersedes the deterministic targeted
# dispatch). Bounded loop: up to ctx.settings.phase_d_max_rounds
# gap→dispatch→re-synthesize rounds (default 1 = the original single
# dispatch). A non-final round's message allows the synth to chain ONE
# more gap (e.g. t_get_event_raw -> t_decode_payload).
targeted_result: dict[str, Any] | str | None = None
# EVERY Phase-D dispatch result, in order — the targeted path threads no
# message history into the post-validators, so the decisive-value support
# gate needs the raw results to know what the dispatches retrieved.
targeted_results_all: list[dict[str, Any] | str] = []
# Tool name of the most recent dispatch whose result carried discriminating
# data — feeds the post-validators' evidence exemption below.
targeted_tool_with_data: str | None = None
phase_d_synth_ok = False
if not ran_investigation_loop and triage_round1.gap_for_investigator is not None:
gap = triage_round1.gap_for_investigator
current_report = triage_round1
max_rounds = ctx.settings.phase_d_max_rounds
for dispatch_round in range(1, max_rounds + 1):
rounds_left = max_rounds - dispatch_round
# Co-emit `retask` so eval/batch.py:read_retask_count
# picks up Phase D dispatches. Without this the metric is
# mathematically guaranteed 0 even when Phase D fires every alert.
# retask precedes targeted_dispatch — semantic ordering: "agent asked
# for more" then "here's the specific call".
retask_ev = _ev(
"retask",
{
"reason": "phase_d_targeted_dispatch",
"tool_name": gap.tool_name,
"gap_question": gap.question,
"gap_why_this_matters": gap.why_this_matters,
"confidence": current_report.confidence,
},
)
await _audit(retask_ev)
yield retask_ev
dispatch_ev = _ev(
"targeted_dispatch",
{
"question": gap.question,
"tool_name": gap.tool_name,
"tool_args": gap.tool_args,
"why_this_matters": gap.why_this_matters,
},
)
await _audit(dispatch_ev)
yield dispatch_ev
targeted_result = await run_targeted_investigation(gap, ctx=ctx)
targeted_results_all.append(targeted_result)
targeted_result_ev = _ev(
"targeted_tool_result",
{"tool_name": gap.tool_name, "result": targeted_result},
)
await _audit(targeted_result_ev)
yield targeted_result_ev
# Only a SUCCESSFUL dispatch that returned DISCRIMINATING DATA
# counts as evidence for the post-validators. An errored dispatch
# (error string / tool error dict) OR an empty-but-non-error result
# (zero OQL hits, internal IP with no blocklist/MISP hit) must NOT
# exempt the hard evidence gate.
if _targeted_result_has_data(targeted_result):
targeted_tool_with_data = gap.tool_name
# Re-synthesize with the targeted result. On a non-final round the
# message permits ONE more gap; the final round demands closure.
user_msg_round2 = build_synth_first_round2_user_message(
alert_id=alert_id,
enriched_ctx_json=enriched_json,
materialized_evidence=materialized,
candidate=candidate,
round1_gap=gap,
targeted_tool_result=targeted_result,
focus_hint=focus_hint,
allow_further_gap=rounds_left > 0,
focus_origin=focus_origin,
)
# Injection 4 of 4. Unlike the memory blocks — deliberately round-1
# only, because a prior verdict competes with gathered evidence —
# asset facts describe the hosts the targeted result is about, and
# this round is a verdict site.
user_msg_round2 = _inject_dossier_block(user_msg_round2, dossier_block)
user_msg_round2 = _inject_session_constraint(user_msg_round2, session_block)
if guard is not None:
# The Phase-D dispatch ran with REAL args (the round-1 report
# was desanitized at its assignment source) and returned a RAW
# targeted_result — correct for the local targeted_tool_result
# event above, but it and the gap text must be re-labeled
# before this composed message crosses the egress boundary.
user_msg_round2 = guard.sanitize_text(user_msg_round2)
r2_captured: list[Any] = []
r2_attr: dict[str, Any] = {}
try:
# Fail-closed residue sweep on the FINAL composed round-2 message
# before the model call — raises BEFORE egress on residue.
user_msg_round2 = _guard_egress(guard, user_msg_round2, ctx.settings)
with (
capture_run_messages() as r2_captured,
capture_backend_attribution() as r2_attr,
):
async with asyncio.timeout(ctx.settings.investigation_turn_timeout_s):
synth_result_round2 = await synth_agent.run(user_msg_round2)
except asyncio.CancelledError:
raise # cooperative cancel — propagate, never swallow
except EgressResidueError as e:
# Blocked egress: the model was NOT called. Audit the block (count
# only) + emit the paired error event, then land the HONEST
# pipeline-fallback (E1.2) and stop dispatching — a blocked run IS
# a pipeline error.
async for ev in _emit_egress_blocked("synth_first_round2", e):
yield ev
triage_final = _synth_failure_fallback_report(alert_id, "egress_blocked", e)
final_synth_rerun = None # fallback verdict — never vote on it
await metrics.get_metrics().record_event("fallback_verdict", {})
phase_d_synth_ok = False
break
except BaseException as e:
# Emit a recorded error, then fall back to the round-1 verdict (or a
# scoreable NMI) so the row is never a silent status=error. Lift the
# per-attempt schema-validation causes off the captured run so a
# round-2 retry-exhaustion isn't the hint=null blind spot the
# retry_causes feature closed for round-1/loop-synth.
r2_causes = _retry_causes_from_messages(r2_captured)
r2_served = _served_backend(r2_attr)
err_ev = _ev(
"error",
_error_payload(
e,
phase="synth_first_round2",
round_num=dispatch_round + 1,
retry_causes=r2_causes,
served_backend=r2_served,
),
)
await _audit(err_ev)
yield err_ev
triage_final = _round2_failure_fallback(
alert_id, triage_round1, e, r2_causes, r2_served
)
final_synth_rerun = None # fallback verdict — never vote on it
await metrics.get_metrics().record_event("fallback_verdict", {})
phase_d_synth_ok = False
break
else:
phase_d_synth_ok = True
triage_final = _strip_model_resolution(synth_result_round2.output)
if guard is not None:
# Assignment-source restore — a chained next-round gap's
# tool_args must dispatch with real values, and the gates
# compare against raw enriched values.
triage_final = _desanitize_report(triage_final, guard)
# This synthesis is now the final one (supersedes the previous).
final_synth_rerun = (synth_agent, user_msg_round2, None)
r2_reasoning_ev = _reasoning_ev(dispatch_round + 1, synth_result_round2)
if r2_reasoning_ev is not None:
await _audit(r2_reasoning_ev)
yield r2_reasoning_ev
usage2_ev = _usage_ev(
dispatch_round + 1, synth_result_round2, _served_backend(r2_attr)
)
if usage2_ev is not None:
await _audit(usage2_ev)
yield usage2_ev
if triage_final.gap_for_investigator is None:
break # verdict settled — no further dispatch wanted
current_report = triage_final
gap = triage_final.gap_for_investigator
# Defensive: enforce the dispatch budget — the FINAL round's synth must
# NOT emit a gap. Fires only when the model ignored the final-round
# closure instruction (never on the failure fallback above, whose
# round-1-derived gap is left as-is, matching the pre-loop behavior).
if phase_d_synth_ok and triage_final.gap_for_investigator is not None:
triage_final = triage_final.model_copy(update={"gap_for_investigator": None})
# The web search the pipeline ran for the loop is part of what the loop
# gathered. With no loop history (the run fell back), there is nothing to
# join, exactly as a loop call would have been lost with it.
if prefetch_tool_messages and loop_messages is not None:
loop_messages = [*prefetch_tool_messages, *loop_messages]
# ----- Self-consistency vote (flag-gated; OFF by default) -----
# verdict_consistency_samples=1 (the default) skips this entirely: single
# synthesis call, no vote, `inconclusive` never produced — byte-identical
# to the pre-vote pipeline. When >1, re-run the SAME final synthesis
# (same agent + prompt) N-1 more times and majority-vote the verdict
# BEFORE the deterministic post-validators (the evidence/citation guards
# below still apply to the voted report). Fewer than 2 voting samples ⇒ no
# vote (the single available report stands). Fallback-produced verdicts
# never vote (rerun is None).
#
# When the rerun is the loop itself (W3), a run that stops at its tool
# budget votes with its partial verdict: the primary loop through
# _land_cut_short_verdict, a sample here through the same partial
# synthesis (one more call, no tools). Only a sample that raises, or that
# ends with no verdict, is dropped. Arm C of the swarm triage experiment is
# three runs at 8 tool calls each, and a vote that dropped every run cut
# at the budget collapsed it into single runs.
vote_samples = int(getattr(ctx.settings, "verdict_consistency_samples", 1) or 1)
if vote_samples > 1 and final_synth_rerun is not None:
sample_agent, sample_msg, sample_limits = final_synth_rerun
sample_reports: list[TriageReport] = [triage_final]
partial_samples = 1 if primary_partial else 0
dropped_samples = 0
sample_round = 1 if rerun_is_loop else 2
for _ in range(vote_samples - 1):
if rerun_is_loop:
# The sample re-runs the loop, tools and all, against the same
# ctx the primary run used. The dedup tracker is per-run state:
# inherited, every pivot the primary already made comes back as
# a ``duplicate_call`` stub and the sample votes blind. Nothing
# after the vote reads the tracker, so a fresh one per sample
# is safe (time anchor / prefetched ids stay as set).
ctx.dedup = _DedupTracker()
sample_number = len(sample_reports)
sample_captured: list[Any] = []
sample_partial = False
try:
with capture_run_messages() as sample_captured:
async with asyncio.timeout(ctx.settings.investigation_turn_timeout_s):
if sample_limits is not None:
extra_result = await sample_agent.run(
sample_msg, usage_limits=sample_limits
)
else:
extra_result = await sample_agent.run(sample_msg)
except asyncio.CancelledError:
raise # cooperative cancel — propagate, never swallow
except (UsageLimitExceeded, TimeoutError) as e:
if not (rerun_is_loop and sample_captured):
_LOGGER.warning("self-consistency sample failed (dropped): %s", e)
dropped_samples += 1
continue
# The sample ran the loop to its budget. Its usage so far is a
# cost of the run, and the partial synthesis is one more call.
cut_usage_ev = _ev(
"usage",
{
**_usage_event_payload(sample_round, _usage_from_messages(sample_captured)),
"self_consistency_sample": sample_number,
},
)
await _audit(cut_usage_ev)
yield cut_usage_ev
sample_partial_runs: list[Any] = []
sample_partial_report: Any = None
try:
sample_partial_report, _repaired = await _synthesize_partial_triage(
ctx.settings, guard, list(sample_captured), result_out=sample_partial_runs
)
except asyncio.CancelledError:
raise # cooperative cancel — propagate, never swallow
except BaseException as partial_exc:
_LOGGER.warning(
"self-consistency sample cut at its budget wrote no verdict (dropped): %s",
partial_exc,
)
for partial_run in sample_partial_runs:
partial_usage = _run_usage(partial_run)
if partial_usage is not None:
partial_usage_ev = _ev(
"usage",
{
**_usage_event_payload(2, partial_usage),
"self_consistency_sample": sample_number,
},
)
await _audit(partial_usage_ev)
yield partial_usage_ev
if sample_partial_report is None:
dropped_samples += 1
continue
extra_report = _cut_short_report(sample_partial_report, _BUDGET_CUT_NOTE, guard)
sample_partial = True
except BaseException as e:
_LOGGER.warning("self-consistency sample failed (dropped): %s", e)
dropped_samples += 1
continue
else:
# A sample is a model run too. Its usage joins the run counters
# and the arms report, which sums the ``usage`` events: without
# it an ensemble arm reads at the cost of its first sample.
sample_usage = _run_usage(extra_result)
if sample_usage is not None:
sample_usage_ev = _ev(
"usage",
{
**_usage_event_payload(sample_round, sample_usage),
"self_consistency_sample": sample_number,
},
)
await _audit(sample_usage_ev)
yield sample_usage_ev
if getattr(extra_result.output, "verdict", None) is None:
_LOGGER.warning("self-consistency sample returned no verdict (dropped)")
dropped_samples += 1
continue
# Same privileged-field strip as every primary report: a winning
# sample's ``resolution`` would be the one the run stores.
extra_report = _strip_model_resolution(extra_result.output)
if guard is not None:
# Samples re-ran the SAME already-sanitized message, so their
# outputs are labeled too; restore before the vote so a winning
# sample's summary/citations land in real-value space.
extra_report = _desanitize_report(extra_report, guard)
# Samples never trigger Phase D — strip any gap defensively.
if extra_report.gap_for_investigator is not None:
extra_report = extra_report.model_copy(update={"gap_for_investigator": None})
sample_reports.append(extra_report)
partial_samples += int(sample_partial)
if len(sample_reports) >= 2:
voted_verdict, voted_conf, vote_note = _self_consistency_vote(sample_reports)
vote_tally = dict(Counter(r.verdict for r in sample_reports))
# Representative sample: first report matching the winning verdict;
# for an inconclusive split, the highest-confidence sample (its
# summary/citations/actions are kept — only verdict+confidence are
# overwritten by the vote, and the note is appended).
if voted_verdict == "inconclusive":
representative = max(sample_reports, key=lambda r: r.confidence)
else:
representative = next(r for r in sample_reports if r.verdict == voted_verdict)
triage_final = representative.model_copy(
update={
"verdict": voted_verdict,
"confidence": voted_conf,
"summary": f"{representative.summary}\n\n[{vote_note}]",
}
)
vote_ev = _ev(
"self_consistency_vote",
{
"samples": len(sample_reports),
"tally": vote_tally,
"chosen_verdict": voted_verdict,
"note": vote_note,
# What the arms report states for each vote: the runs asked
# for, the runs that voted, the votes a partial synthesis
# wrote after a budget cut, and the runs dropped.
"requested": vote_samples,
"voted": len(sample_reports),
"partial": partial_samples,
"dropped": dropped_samples,
},
)
await _audit(vote_ev)
yield vote_ev
# ----- Post-synth validators -----
# Mirror the legacy post-synth validator chain: citation validation,
# citation cap, and verdict floor rewrite. Coverage cap is NOT
# applied — no investigator ran, so there's no tool-call ledger.
# When the investigation loop ran, it supersedes Phase D: thread the
# loop's real message history into citation resolution (so tool/pivot
# citations resolve against actual ToolCallParts) and treat it like an
# investigator round — gathered evidence legitimately grounds the
# verdict, same as a Phase-D round-2. Otherwise keep the existing
# synth-first behavior.
targeted_tool: str | None = None
targeted_messages: list[Any] | None = None
if ran_investigation_loop:
targeted_messages = loop_messages
# Mark "an investigation gathered evidence" — which exempts the verdict
# from the hard evidence gate + GATE A — ONLY when the loop produced at
# least one SUCCESSFUL tool call. On the budget/timeout fallback path
# ``loop_messages`` is None (the round-1 verdict just stands), and a loop
# whose every call errored gathered nothing; treating either as tool
# evidence would launder a non-evidence-backed round-1 TP/FP straight
# past the gate. ``targeted_messages`` is still threaded for tool/pivot
# citation resolution regardless.
targeted_tool = _loop_evidence_marker(ran_investigation_loop, loop_messages)
elif targeted_tool_with_data is not None:
# A Phase-D dispatch returned DISCRIMINATING DATA (tracked per round in
# the bounded loop above — errored or empty dispatches never set it, so
# they can't exempt the hard evidence gate).
targeted_tool = targeted_tool_with_data
# Did this run RETRIEVE anything? The same question the hard evidence gate
# asks, kept as a separate value because auto-acknowledge needs it after the
# Oracle has had its turn, and the Oracle can add retrieval of its own.
run_retrieved_evidence = (
count_successful_tool_calls(targeted_messages) >= 1 or targeted_tool is not None
)
triage_final, validation_audit = _synth_first_post_validate(
triage_final,
enriched,
candidate,
investigator_evidence=loop_evidence_for_report,
targeted_messages=targeted_messages,
targeted_tool_called=targeted_tool,
targeted_tool_results=targeted_results_all,
synthesis_confidence_floor=ctx.settings.synthesis_confidence_floor,
blocklist=ctx.blocklist,
internal_cidrs=classification_cidrs,
)
# Emit validator events in order.
if "investigator_evidence_carried" in validation_audit:
ev = _ev("investigator_evidence_carried", validation_audit["investigator_evidence_carried"])
await _audit(ev)
yield ev
if "template_grounds_adopted" in validation_audit:
ev = _ev("template_grounds_adopted", validation_audit["template_grounds_adopted"])
await _audit(ev)
yield ev
if "citation_validation" in validation_audit:
ev = _ev("citation_validation", {"round": 1, **validation_audit["citation_validation"]})
await _audit(ev)
yield ev
if "citation_cap" in validation_audit:
ev = _ev("citation_cap", {"round": 1, **validation_audit["citation_cap"]})
await _audit(ev)
yield ev
# Emitted in gate order: the floor raise runs after the citation cap and
# before the verdict-floor rewrite in _synth_first_post_validate. Audit-only
# until 2026-08-27 — the eval bundle records YIELDED events, so a raise that
# never emitted was unprovable from the very measurement grading it.
if "confidence_floor_raise" in validation_audit:
ev = _ev("confidence_floor_raise", validation_audit["confidence_floor_raise"])
await _audit(ev)
yield ev
if "verdict_floor_rewrite" in validation_audit:
ev = _ev("verdict_floor_rewrite", validation_audit["verdict_floor_rewrite"])
await _audit(ev)
yield ev
if "icmp_solicited_downgrade" in validation_audit:
ev = _ev("icmp_solicited_downgrade", validation_audit["icmp_solicited_downgrade"])
await _audit(ev)
yield ev
if "ungrounded_host_anchored_tp_downgrade" in validation_audit:
ev = _ev(
"ungrounded_host_anchored_tp_downgrade",
validation_audit["ungrounded_host_anchored_tp_downgrade"],
)
await _audit(ev)
yield ev
if "malware_rule_name_ungrounded_downgrade" in validation_audit:
ev = _ev(
"malware_rule_name_ungrounded_downgrade",
validation_audit["malware_rule_name_ungrounded_downgrade"],
)
await _audit(ev)
yield ev
if "decisive_value_support_cap" in validation_audit:
ev = _ev(
"decisive_value_support_cap",
validation_audit["decisive_value_support_cap"],
)
await _audit(ev)
yield ev
if "unsupported_decisive_value_downgrade" in validation_audit:
ev = _ev(
"unsupported_decisive_value_downgrade",
validation_audit["unsupported_decisive_value_downgrade"],
)
await _audit(ev)
yield ev
if "evidence_gate_downgrade" in validation_audit:
ev = _ev("evidence_gate_downgrade", validation_audit["evidence_gate_downgrade"])
await _audit(ev)
yield ev
# Metric: a zero-tool TP/FP was coerced to needs_more_info by the hard
# evidence gate. Counts how often the gate fires (reliability signal).
await metrics.get_metrics().record_event("zero_tool_verdict_blocked", {})
# The gate's other outcome — a zero-tool verdict exempted because it cites
# correlated prefetched-pivot evidence — was audit-only (same blindness as
# confidence_floor_raise above): "the gate passed this" and "the gate never
# ran" looked identical in a bundle.
if "evidence_gate_pivot_exemption" in validation_audit:
ev = _ev(
"evidence_gate_pivot_exemption",
validation_audit["evidence_gate_pivot_exemption"],
)
await _audit(ev)
yield ev
# ----- Host coverage grounding -----
# A sentence that says a host has no host telemetry is false while the
# prefetch read any plane for that host. Production stored twelve of them
# about one host that shipped host logs and osquery. Each such sentence is
# replaced by the coverage facts. Runs before the Oracle snapshot, so the
# local report the Oracle compares with is the grounded one.
triage_final, host_claims = _ground_host_claims(triage_final, enriched)
if host_claims:
ev = _ev("host_coverage_grounding", {"sentences_rewritten": host_claims})
await _audit(ev)
yield ev
# ----- Oracle escalation (optional, explicit opt-in) -----
# After all post-validators, escalate to the frontier Oracle when the local
# triage needs it (uncertain, malware non-TP, or below-floor confidence).
# The local verdict is preserved in the audit via `local_verdict` in the
# oracle_escalation event so evaluators can compare both.
local_triage_final = triage_final # snapshot before any Oracle override
# True when the Oracle answered with a different class and no evidence: its
# answer is an opinion on the run, and the auto-acknowledge never fires.
oracle_opinion_withheld = False
# The class that ran. A cheap run never reaches the Oracle.
final_run_class = budget.ran_class(class_plan, ran_loop=ran_investigation_loop)
# The Oracle rule mode decides (classic, shadow or uncertainty). The reason
# is computed once, by the predicate itself, so the audit row cannot name a
# rule the predicate did not apply.
rule_decision = oracle_rule_decision(
triage_final,
enriched,
ctx.settings,
candidate=candidate,
ran_loop=ran_investigation_loop,
run_class=final_run_class,
)
escalation_reason = rule_decision.reason
if rule_decision.mode == "shadow" and (
rule_decision.classic_reason is not None or rule_decision.uncertainty_reason is not None
):
# The shadow ledger: what the uncertainty rule would have done, beside
# what the classic rule did. No Oracle call and no verdict change
# comes from it. Detection tuning tallies these rows.
shadow_ev = _ev(
"oracle_shadow",
{
"classic_reason": rule_decision.classic_reason,
"uncertainty_reason": rule_decision.uncertainty_reason,
"would_escalate": rule_decision.uncertainty_reason is not None,
"classic_escalates": rule_decision.classic_reason is not None,
"local_verdict": triage_final.verdict,
"local_confidence": triage_final.confidence,
"template_verdict": getattr(candidate, "verdict", None),
"run_class": final_run_class,
},
)
await _audit(shadow_ev)
yield shadow_ev
# The route pause (soc_ai.oracle.breaker): a route that answered with a
# usage limit, or with a run of server errors, gets no call until the
# reset time. The escalation is recorded as oracle_skipped, never as an
# oracle_escalation: nothing left the box.
oracle_route = _oracle_breaker.route_key(ctx.settings)
oracle_paused_until = (
_oracle_breaker.BREAKER.open_until(oracle_route) if escalation_reason is not None else None
)
if escalation_reason is not None and oracle_paused_until is not None:
pause = _oracle_breaker.BREAKER.state(oracle_route)
skip_ev = _ev(
"oracle_skipped",
{
"reason": "oracle_paused",
"pause_reason": pause.reason,
"paused_until": _oracle_breaker.iso(oracle_paused_until),
"reset_named": pause.reset_named,
"message": pause.message,
"escalation_reason": escalation_reason,
"rule_mode": rule_decision.mode,
"local_verdict": triage_final.verdict,
"local_confidence": triage_final.confidence,
},
)
await _audit(skip_ev)
yield skip_ev
elif escalation_reason is not None:
from soc_ai.agent.gates import gate_band # noqa: PLC0415 - re-exported module
esc_ev = _ev(
"oracle_escalation",
{
# The Oracle ledger: the egress audit row of every escalation
# carries the reason that sent it and the rule mode.
"reason": escalation_reason,
"rule_mode": rule_decision.mode,
"local_verdict": triage_final.verdict,
"local_confidence": triage_final.confidence,
"template_verdict": getattr(candidate, "verdict", None),
"gate_band": list(gate_band()),
"run_class": final_run_class,
},
)
await _audit(esc_ev)
yield esc_ev
# Build a compact text transcript for the Oracle payload — the PROSE
# parts only. The dict-shaped tool results (the backing evidence) travel
# separately via loop_messages → _extract_tool_results in the client,
# STRUCTURED, so sanitize_case's field-aware harvest sees their key
# paths. (b3 finding: this str-only filter used to be the whole payload,
# so every tool result was silently dropped and the Oracle adjudicated
# over unbacked prose.)
transcript_text = ""
if loop_messages is not None:
parts: list[str] = []
for msg in loop_messages:
for part in getattr(msg, "parts", []) or []:
content = getattr(part, "content", None)
if isinstance(content, str) and content.strip():
parts.append(content.strip())
transcript_text = "\n".join(parts)
# Reuse the effective internal-identifier set resolved ONCE at the top of
# the pipeline (env-config union active detected/manual identifiers, minus
# muted) for the Oracle egress sanitizer's suffixes/hosts. DB access stays
# in the caller; the sanitizer stays pure. None ⇒ no DB on ctx (CLI / eval
# / tests) or a resolution failure → the client falls back to the raw
# settings tuples (behavior unchanged).
oracle_suffixes = effective_idents.suffixes if effective_idents is not None else None
oracle_hosts = effective_idents.hosts if effective_idents is not None else None
# `oracle_failure` is the client's failure_out channel (the
# no_propagate_out precedent): adjudicate() keeps its None-on-failure
# contract but names WHY, so the failed-adjudication event below can
# carry a reason instead of a shrug.
oracle_failure: dict[str, Any] = {}
oracle_result = await _oracle_client.adjudicate(
ctx,
enriched=enriched,
local_report=triage_final,
transcript_text=transcript_text,
loop_messages=loop_messages,
evidence_bullets=loop_evidence_bullets,
extra_hosts=oracle_hosts,
extra_suffixes=oracle_suffixes,
failure_out=oracle_failure,
# Change E. A finding run names the claim it judged, so the Oracle
# judges the same claim. None on an alert run and a lead run.
subject_claim=subject.claim() if subject is not None else None,
)
if (
oracle_result is not None
and not oracle_result.override_withheld
and oracle_result.report.verdict != local_triage_final.verdict
and (oracle_result.oracle_tool_calls or 0) < 1
and not oracle_result.oracle_citations
):
# No override without evidence (2026-10-04). The client applies
# this gate on both of its paths; this is the same rule at the
# point where the verdict changes hands, so no client path and no
# stub can land a flip that names no evidence. Production landed
# 6 such flips, NMI to FP, and auto-acknowledged 2 of them.
oracle_result = dataclasses.replace(
oracle_result,
report=local_triage_final,
override_withheld=True,
raw_oracle_verdict=(
oracle_result.raw_oracle_verdict or oracle_result.report.verdict
),
raw_oracle_confidence=(
oracle_result.raw_oracle_confidence
if oracle_result.raw_oracle_confidence is not None
else oracle_result.report.confidence
),
raw_oracle_summary=oracle_result.raw_oracle_summary or oracle_result.report.summary,
)
if oracle_result is not None:
oracle_opinion_withheld = oracle_result.override_withheld
# Fix M2: post-validate the Oracle's output with the same
# deterministic targeted downgrades that ran on the local verdict.
# Closes the path where the Oracle re-introduces a
# solicited-internal-ICMP-echo true_positive that the local
# BPFDoor guard already corrected. Zero egress — deterministic.
oracle_audit: dict[str, Any] = {}
oracle_report = _apply_targeted_downgrades(
oracle_result.report,
enriched,
oracle_audit,
blocklist=ctx.blocklist,
internal_cidrs=classification_cidrs,
)
# No-benign-baseline guard (decoy and every flagged catalog spec) —
# Oracle path parity. A decoy alert
# reaches the Oracle by the local gate's own hand: refusing the
# local false_positive drops confidence to 0.4, under
# oracle_escalate_below_confidence, so the Oracle is asked exactly
# the question the local path just declined to answer benign. An
# Oracle false_positive would land unrefused. Same deterministic,
# zero-egress shape as the two guards around it.
oracle_report = _refuse_benign_verdict_without_baseline(
oracle_report, enriched, oracle_audit
)
# The host history gate (_downgrade_ungrounded_host_anchored_tp)
# does not run on the Oracle report. Owner rule of 2026-10-30: the
# gate must not override an Oracle true positive. On 2026-10-29 the
# call reset two backed Oracle true positives on hunt findings to
# needs_more_info. A class-changing Oracle true positive still needs
# a resolved citation beyond the alert, or a tool call of its own
# (the override gate above and oracle/client.py). That rule accepts
# any prefetched pivot id, so on an alert run the removal is a
# policy choice and not a proof. When the local gate fired,
# local_summary opens with _HOST_ANCHORED_TP_LEAD, so the Oracle
# reads the change. The ICMP (Fix M2) and no-baseline guards stay.
if guard is not None:
# The Oracle desanitizes its OWN labels (its independent
# pipeline/mapping), but the loop-transcript text it was shown
# can carry THIS guard's labels — if it echoed any (e.g.
# HOST_01 quoted from loop evidence), restore them too.
oracle_report = _desanitize_report(oracle_report, guard)
adj_ev = _ev(
"oracle_adjudication",
{
# Normally the adjudicated verdict (post targeted-downgrade,
# unchanged from before). When the override gate WITHHELD a
# flip, oracle_report carries the local class, so surface the
# raw class the Oracle ruled instead — the dissent stays on
# the record.
"oracle_verdict": (
oracle_result.raw_oracle_verdict
if oracle_result.override_withheld
else oracle_report.verdict
),
"oracle_confidence": (
oracle_result.raw_oracle_confidence
if oracle_result.override_withheld
and oracle_result.raw_oracle_confidence is not None
else oracle_report.confidence
),
# The evidence the Oracle named, and how much of it
# resolved. A class change needs one resolved citation or
# one successful tool call of its own.
"oracle_citations": list(oracle_result.oracle_citations),
"oracle_citations_cited": oracle_result.oracle_citations_cited,
"redaction": oracle_result.redaction_summary,
"oracle_model": oracle_result.oracle_model,
# Observability (design §8): how many successful tool calls
# the Oracle's own loop made — 0 on the single-shot path.
"oracle_tool_calls": oracle_result.oracle_tool_calls,
# The allow-known-safe egress backstop's utility cost: how many
# residual unclassified scalars it masked before the wire (only
# surfaced when non-zero, so a clean adjudication's event is
# unchanged).
**(
{"oracle_masked_values": oracle_result.oracle_masked_values}
if oracle_result.oracle_masked_values
else {}
),
# The override gate (design §6): a class-changing verdict with
# no supporting tool call did NOT override; the local verdict
# below stands, and that decision is recorded here.
**(
{
"override_withheld": True,
"withheld_reason": "no_resolving_evidence",
"local_verdict": local_triage_final.verdict,
# The opinion, kept on the run for the analyst.
"oracle_summary": oracle_result.raw_oracle_summary or "",
}
if oracle_result.override_withheld
else {}
),
**({"oracle_targeted_downgrades": oracle_audit} if oracle_audit else {}),
},
)
await _audit(adj_ev)
yield adj_ev
if oracle_result.override_withheld:
# The gate withheld the flip: keep the local verdict unchanged.
# The report already carries the local verdict class (the gate
# applied at the _verdict_to_report seam); do NOT relabel it as an
# Oracle override.
triage_final = local_triage_final
else:
# Mark the Oracle report so UI/audit shows it was adjudicated.
adjudicated_summary = f"[Oracle adjudicated] {oracle_report.summary}"
triage_final = oracle_report.model_copy(update={"summary": adjudicated_summary})
# The Oracle's own loop is retrieval too, and the verdict below
# is now the Oracle's. A single-shot adjudication reports 0 and
# adds nothing.
run_retrieved_evidence = run_retrieved_evidence or (
(oracle_result.oracle_tool_calls or 0) >= 1
)
else:
# Refusal or failure: triage_final stays unchanged and the local
# verdict stands — but say so ON THE RECORD. An escalation whose
# adjudication silently vanished (scenario b8, 2026-08-27
# measurement) was indistinguishable from one still in flight.
# "unknown" = a caller (or test stub) that recorded no reason.
# The failure's truth (soc_ai.oracle.failures): the HTTP status,
# the error class and the gateway's scrubbed message. Absent keys
# stay absent, so "the client recorded no status" never reads as
# a status.
fail_ev = _ev(
"oracle_adjudication_failed",
{
"reason": oracle_failure.get("reason", "unknown"),
**{
key: oracle_failure[key]
for key in ("error_class", "http_status", "message", "paused_until")
if key in oracle_failure
},
"local_verdict": triage_final.verdict,
"local_confidence": triage_final.confidence,
},
)
await _audit(fail_ev)
yield fail_ev
if oracle_failure.get("breaker_opened"):
# The pause opened on this call: one webhook message per pause.
# The bell reads the breaker itself (routes_meta).
from soc_ai import notify # noqa: PLC0415 - lazy, like the runner
pause_event = notify.event_for_oracle_paused(
pause_reason=_oracle_breaker.BREAKER.state(oracle_route).reason,
until=oracle_failure.get("paused_until"),
settings=ctx.settings,
)
if pause_event is not None:
await notify.fire_safe(pause_event, ctx.settings, ctx.audit)
# The Oracle writes its own prose, and the same rule holds for it. Only
# an Oracle report is grounded here: the local one already was, and a
# copy of it would break the identity checks below.
if triage_final is not local_triage_final:
triage_final, oracle_host_claims = _ground_host_claims(triage_final, enriched)
if oracle_host_claims:
ev = _ev(
"host_coverage_grounding",
{"sentences_rewritten": oracle_host_claims, "report": "oracle"},
)
await _audit(ev)
yield ev
# ----- Same-session consistency (last word on the verdict) -----
# After the Oracle, not before: an adjudicated false positive on a session a
# completed run already called malicious is the same contradiction as a
# local one, and this has to run downstream of everything that can still set
# the verdict. Downstream of the auto-acknowledge too, which reads
# ``triage_final`` below and must never see the false positive this holds.
triage_final, session_gate_audit = enforce_session_verdict_consistency(
triage_final, session_digests
)
if session_gate_audit is not None:
session_gate_ev = _ev("session_verdict_conflict", session_gate_audit)
await _audit(session_gate_ev)
yield session_gate_ev
# ----- Final triage emit -----
# W3 A/B provenance on the report itself, so a stored run answers "which
# call wrote this verdict" without a join back across the event stream. The
# report dict only ever carries a ``resolution`` here on the pipeline-
# fallback path (every model-authored report had its own stripped at the
# assignment source), where ``report_path`` is None, so the stamp below is
# a no-op today; it is here so the marker travels with the dict if a future
# path sets both.
if report_path is not None and isinstance(triage_final.resolution, dict):
triage_final = triage_final.model_copy(
update={"resolution": {**triage_final.resolution, "report_path": report_path}}
)
triage_ev = _ev(
"triage_report",
{
"verdict": triage_final.verdict,
"confidence": triage_final.confidence,
"summary": triage_final.summary,
"citations": triage_final.citations,
"recommended_actions": [
a.model_dump(mode="json") for a in triage_final.recommended_actions
],
"field_reconciliation": triage_final.field_reconciliation,
"validator_note": triage_final.validator_note,
# Pipeline-fallback provenance marker (E1.2). Present ONLY on the
# synth-failure path (`_synth_failure_fallback_report`) — a real
# verdict leaves it None and it's dropped below. The persisted report
# dict (recorder captures THIS payload) then carries
# `resolution.provenance == "pipeline_fallback"`, which every
# downstream consumer reads via `is_pipeline_fallback`. Distinct key
# from the manual/chat override's `resolved_via`, so a later manual
# resolution overwrites it cleanly (a resolved run is no longer a
# "pipeline error to retry").
**(
{"resolution": triage_final.resolution}
if triage_final.resolution is not None
else {}
),
# W3 A/B marker: which call wrote this report — ``investigator``
# (the loop's own output, investigator_emits_report ON) or
# ``synth_round2`` (the round-2 synthesis over the transcript). Key
# ABSENT, not null, on every other path (a fallback, a budget-cut
# partial, a round-1-settled or Phase-D run), so "no loop report"
# and "a path that did not record itself" never look alike.
**({"report_path": report_path} if report_path is not None else {}),
# Preserve local verdict in the audit when Oracle overrode it.
"local_verdict": local_triage_final.verdict
if triage_final is not local_triage_final
else None,
# The budget class this run ran in, and why (soc_ai.agent.budget).
# The recorder stamps it on the row; the console shows it as a chip.
"run_class": final_run_class,
"run_class_reason": run_class_reason,
# The swarm triage experiment: which arm produced this report. Key
# ABSENT for ``single``, so the default report keeps its exact
# shape. The swarm counters ride along when a swarm ran.
**({"strategy": loop_strategy} if loop_strategy != SINGLE else {}),
**({"swarm": swarm_counters} if swarm_counters else {}),
},
)
await _audit(triage_ev)
yield triage_ev
# ----- Auto-acknowledge high-confidence false positives (opt-in) -----
# Coverage is only passed on when it describes the report being acked. The
# resolver ran over the LOCAL report's citations; if the Oracle then rewrote
# the verdict, the number is about a different set of citations and would be
# a lie either way it fell. None keeps the empty-citation refusal, which
# reads the final report directly.
ack_coverage: float | None = None
if triage_final is local_triage_final:
cv = validation_audit.get("citation_validation") or {}
raw_coverage = cv.get("coverage_ratio")
if isinstance(raw_coverage, int | float):
ack_coverage = float(raw_coverage)
auto_ack_ev = await _maybe_auto_ack_fp_gated(
triage_final,
alert_id,
alert=enriched.alert,
ctx=ctx,
emit_ev=_ev,
audit_ev=_audit,
investigated=run_retrieved_evidence,
citation_coverage=ack_coverage,
allow_so_writes=allow_so_writes,
oracle_dissent=oracle_opinion_withheld,
)
if auto_ack_ev is not None:
yield auto_ack_ev
# Through _audit like every other event: _audit is what feeds the /metrics
# counters, and socai_investigations_total counts "done". Yielded bare, the
# counter stayed at 0 on every completed run (dogfood 2026-10-01 RA2).
done_ev = _ev("done", {"recommended_count": len(triage_final.recommended_actions)})
await _audit(done_ev)
yield done_ev
__all__ = [
"InvestigationContext",
"InvestigationTranscript",
"RecommendedAction",
"StepEvent",
"TriageReport",
"_should_escalate_to_oracle",
"build_agent",
"build_investigator",
"build_investigator_model",
"build_local_enrichment_context",
"build_model",
"build_synth_first_agent",
"build_synthesizer",
"build_synthesizer_model",
"classic_oracle_escalation_reason",
"investigate",
"maybe_auto_ack_fp",
"oracle_escalation_reason",
"oracle_rule_decision",
]