"""Output schema for the agent's final triage report (neutral, shared types). The agent's ``output_type=TriageReport`` constrains its final answer to this shape - PydanticAI ensures the model emits valid JSON or retries. Recommended actions are **suggestions only** in v1 - they map to write tools (``ack_alert`` / ``escalate_to_case`` / ``add_case_comment``) but never auto-execute. They surface in the report, and the analyst executes them selectively through the actions API (``POST /api/v1/investigations/{id}/actions/{index}/execute``). These models live at the package root (below both ``soc_ai.agent`` and ``soc_ai.oracle``) so consumers like the Oracle client can import the report types without pulling in the whole agent package — importing ``soc_ai.agent.triage`` executes ``soc_ai.agent.__init__`` and therefore the orchestrator, which created an oracle↔agent import cycle. ``soc_ai.agent.triage`` re-exports everything here for compatibility. """ from __future__ import annotations import json import re from typing import Any, Literal from pydantic import BaseModel, Field, field_validator WriteToolName = Literal["ack_alert", "escalate_to_case", "add_case_comment"] # ``inconclusive`` is a TERMINAL non-committed verdict produced ONLY by the # orchestrator's flag-gated self-consistency vote (a split across N samples). # The model must never emit it directly (the field description says so); it is # in the Literal so voted reports validate/persist/read back like any verdict. Verdict = Literal["true_positive", "false_positive", "needs_more_info", "inconclusive"] # Null sentinels a serving model writes into an Optional container field instead # of emitting JSON `null`. Matched case-insensitively after stripping. Kept # DELIBERATELY SHORT: only tokens that unambiguously mean "nothing here". A real # sentence must still fail validation, or a genuine model error would be silently # swallowed as None (see test_meaningful_strings_are_not_swallowed). _NULL_SENTINELS = frozenset({"", "none", "null", "nil", "n/a", "na", "nothing", "undefined"}) def _normalize_verdict(v: Any) -> Any: """Fold formatting variants of the canonical verdicts to canonical form. Serving models pick the right verdict and fumble the FORMAT — "False Positive", "false-positive", " true_positive ". Rejecting those burns a schema retry the model may not recover from (it chose correctly; the error message tells it nothing new). Only case / whitespace / hyphen variants of the four canonical values fold; anything else — including synonyms like "benign" or "malicious" — passes through UNCHANGED to fail the ``Verdict`` Literal, because mapping meaning (rather than formatting) would put a verdict in the model's mouth. """ if isinstance(v, str): folded = re.sub(r"[\s\-]+", "_", v.strip().lower()) if folded in ("true_positive", "false_positive", "needs_more_info", "inconclusive"): return folded return v def _decode_string_list(v: Any) -> Any: """Rescue ``list[str]`` fields from the shapes weak models actually emit. - null sentinel string ("None", "null", "") → ``[]`` - stringified JSON array → the decoded list (pre-existing behavior) - any other bare string → wrapped as a one-element list. For citations this is trust-free lenience: every entry is validated by the downstream citation resolver anyway, so a wrapped bad string fails THERE with a per-citation reason instead of failing schema validation wholesale. """ if isinstance(v, str): decoded = _decode_stringified_json(v) if decoded is None: return [] if isinstance(decoded, list): return decoded if isinstance(decoded, str): return [decoded] return v # a dict for a list[str] field: fall through to the normal error return v def _decode_object_list(v: Any) -> Any: """Rescue ``list[Model]`` fields: bare object → one-element list. A model recommending ONE action often emits the object without the array wrapper (both bare and stringified). Sentinels mean "no actions" → ``[]``. Prose stays a string and fails validation normally — item-level contracts (e.g. the WriteToolName Literal) are untouched, so this widens accepted PACKAGING only, never accepted content. """ if isinstance(v, dict): return [v] if isinstance(v, str): decoded = _decode_stringified_json(v) if decoded is None: return [] if isinstance(decoded, list): return decoded if isinstance(decoded, dict): return [decoded] return v return v def _null_sentinel_to_none(v: Any) -> Any: """Map a null-sentinel STRING to None; pass everything else through. The conservative half of :func:`_decode_stringified_json` — used on fields where accepting a model-supplied container would be unsafe, so only the "the model meant null" case is rescued. """ if isinstance(v, str) and v.strip().lower() in _NULL_SENTINELS: return None return v def _decode_stringified_json(v: Any) -> Any: """Decode a JSON-encoded string into its container (dict/list), else pass through. Serving models recurrently emit nested container fields as JSON-encoded strings instead of objects/arrays: Nemotron-30B did it to ``InvestigationTranscript.rubric_coverage`` (Phase 2 smoke), and deepseek-v4-flash does it to ``TriageReport.gap_for_investigator`` (prod run 01KY0T3ZDPX5MXD1TYMPVDQ5ZH, 2026-07-20 — all 3 schema retries failed identically, so retry feedback does not recover it). The payload inside the string is well-formed; auto-parsing it is strictly more accepting. Strings that don't parse to a container fall through to Pydantic's normal error. Lower-tier models (the Qwen / Laguna fallback tier) add a second shape: the STRING ``"None"`` (or ``"null"``, ``"N/A"``) where the schema wants ``null``. Observed live from qwen3.6-35b-instruct on ``TriageReport. resolution``, 2026-08-03. That never parsed here — ``json.loads("None")`` raises, and even valid ``"null"`` decodes to ``None``, which failed the container check and fell through as the original *string*. Both now map to None so the field's own ``| None`` accepts it instead of burning a schema retry that the model has no idea how to satisfy. """ if isinstance(v, str): if v.strip().lower() in _NULL_SENTINELS: return None try: parsed = json.loads(v) except (json.JSONDecodeError, TypeError, ValueError): return v if isinstance(parsed, (dict, list)): return parsed # `json.loads("null")` -> None: a real JSON null arriving as a string. if parsed is None: return None return v class RecommendedAction(BaseModel): """One write-tool invocation the agent recommends for the analyst to execute.""" tool_name: WriteToolName = Field(description="Which write tool the analyst should invoke.") tool_args: dict[str, Any] = Field( default_factory=dict, description="Arguments to pass to the tool. Must match the tool's signature.", ) rationale: str = Field(description="One-line justification visible to the analyst.") _decode_tool_args = field_validator("tool_args", mode="before")(_decode_stringified_json) class InvestigationTranscript(BaseModel): """Output of the investigator (fast model) phase. The investigator's job is to gather evidence with the read tools and hand a concise, structured summary to the synthesizer (heavy model). It does NOT decide a verdict or recommend write actions — those live in :class:`TriageReport`. """ evidence: list[str] = Field( default_factory=list, description=( "Bullet-point findings backed by tool results. Each item should be " "a single fact + the ES `_id` or SOC API id that supports it. " "Add the @timestamp of that document when the result shows one." ), ) tentative_summary: str = Field( description=( "2-4 sentence neutral narrative of what happened, written for the " "synthesizer (NOT the analyst). No verdict; no recommendations." ) ) open_questions: list[str] = Field( default_factory=list, description=( "Specific gaps the investigator could not close (missing logs, " "unenriched indicators, ambiguous behavior). Used to retask the " "investigator if the synthesizer comes back low-confidence." ), ) # ``rubric_coverage`` used to live here. It was removed on 2026-09-05 # because nothing read it: the coverage cap it described was deleted from # the gates, and the prompt went on promising the model that confidence # would be capped at 0.6 if it answered honestly. That is worse than dead # code — the model was being told a consequence existed, and anyone # auditing the prompt would have believed soc-ai enforced coverage # discipline it had stopped enforcing. # # Safe to drop from the schema: this model does not set ``extra="forbid"``, # so a model still emitting the field from a cached prompt has it ignored # rather than rejected (verified before removal). # # The evidence gates took over the job and do it better, on evidence the # run actually RETRIEVED rather than on the model's self-report of what it # had done. The ``coverage_cap`` audit kind and its timeline label are # deliberately KEPT so historical audit records still render. class TargetedGap(BaseModel): """Synth's request for ONE specific tool call to close ONE specific gap. Phase D of the synth-first redesign. When a synth round-1 emits a non-null gap, the orchestrator dispatches the 30B targeted-investigator with the exact tool call specified, then runs synth round 2 with the result appended. Hard cap: one Phase D call per investigation. The orchestrator enforces this by rejecting a non-null gap on synth round 2. """ question: str = Field( ..., description="Human-readable: 'What was the SSL SNI for community_id X?'" ) tool_name: Literal[ "t_query_zeek_logs", "t_query_events_oql", "t_enrich_ip", "t_enrich_domain", "t_enrich_hash", "t_get_playbooks", "t_lookup_runbook", "t_query_cases", "t_query_detections", "t_get_rule_content", "t_get_event_raw", "t_get_pcap", "t_web_search", "t_crawl_page", ] = Field(..., description="Which tool the targeted-investigator should call.") tool_args: dict[str, Any] = Field( default_factory=dict, description="Exact args; the targeted-investigator runs them verbatim.", ) why_this_matters: str = Field(..., description="One line: how the answer changes the verdict.") _decode_tool_args = field_validator("tool_args", mode="before")(_decode_stringified_json) class TriageReport(BaseModel): """The agent's final structured triage output.""" verdict: Verdict = Field( description=( "true_positive: confirmed malicious / suspicious; " "false_positive: benign / expected; " "needs_more_info: insufficient evidence to decide. " "(inconclusive is reserved for the orchestrator's self-consistency " "vote — NEVER emit it directly; use needs_more_info when unsure.)" ) ) confidence: float = Field( ge=0.0, le=1.0, description=( "Confidence in the verdict, 0.0-1.0. Below 0.6 means 'needs_more_info' " "rather than a guess." ), ) summary: str = Field( description=( "Plain-English narrative summarizing what happened, in 3-6 sentences. " "Written for the on-call analyst." ) ) citations: list[str] = Field( default_factory=list, description=( "Specific event/case/detection IDs that support the conclusions. " "Each item is an ES _id or SOC API id." ), ) recommended_actions: list[RecommendedAction] = Field( default_factory=list, description=( "Write-tool invocations recommended for the analyst to execute. " "DO NOT execute these automatically - each runs only on an " "explicit analyst action." ), ) field_reconciliation: str | None = Field( default=None, description=( "Structured-output reconciliation field. When the alert has typed " "fields whose plain-English interpretation could appear to " "contradict each other — most commonly layered protocols where " "alert.proto != the conn transport (`network.transport`, or " "`zeek.conn.proto` on older SO) for the same community_id (ICMP " "T3/C4 PMTUD referring to a UDP flow), or alert.alert_action=" "allowed but high severity_score — the synthesizer MUST emit a " "one-line reconciliation note here. Example: 'alert.proto=ICMP " "refers to the UDP flow at community_id X (PMTUD unreachable, " "not a TCP connection)'. Leave empty when no apparent " "contradictions exist." ), ) gap_for_investigator: TargetedGap | None = Field( default=None, description=( "If non-null on synth round 1, the orchestrator dispatches the 30B " "targeted-investigator with these exact args and feeds the result " "back as a synth round 2. Hard cap: one Phase D call per " "investigation. Synth round 2 MUST emit gap_for_investigator=None." ), ) validator_note: str | None = Field( default=None, description=( "Machine/analyst-facing note explaining any deterministic " "post-validator override. Set by the orchestrator when a " "post-validator corrects the verdict; the summary is then " "rewritten to lead with the correct conclusion. Contains the " "override reason and the original agent summary so nothing is " "lost, just relocated." ), ) resolution: dict[str, Any] | None = Field( default=None, description=( "Provenance marker for a report the pipeline did NOT reason its way " "to. Set ONLY on the synth-failure fallback path (model truncation, " "gateway 5xx, schema-validation exhaustion) to " "``{'provenance': 'pipeline_fallback', 'phase', 'error_type', " "'hint'}`` — so a failure-driven needs_more_info renders distinctly " "from a genuine one, is filterable, and is excluded from the " "Needs-info KPI. The verdict stays needs_more_info; only the " "rendering differs. Distinct from the manual/chat override " "``report['resolution']`` (which carries ``resolved_via``, added " "post-hoc by the resolve endpoint) — the two never conflate because " "``provenance`` and ``resolved_via`` are different keys." ), ) _decode_gap = field_validator("gap_for_investigator", mode="before")(_decode_stringified_json) # List fields get list-aware rescue: sentinel → [], bare item → [item]. _decode_citations = field_validator("citations", mode="before")(_decode_string_list) _decode_actions = field_validator("recommended_actions", mode="before")(_decode_object_list) # Formatting-only fold of the canonical verdicts (never synonym mapping). _normalize_verdict_fmt = field_validator("verdict", mode="before")(_normalize_verdict) # ``resolution`` gets the null-sentinel rescue ONLY — deliberately NOT the # full stringified-JSON decode the fields above receive. # # The bug this fixes is narrow and observed: a model emitted the STRING # "None" for this nullable dict, which failed ``dict_type`` and burned the # run's whole schema-retry budget (2026-08-03, deepseek-v4-flash serving the # qwen3.6-35b-instruct route). # # Decoding stringified JSON here would be strictly WORSE, because this field # is privileged: ``resolution['provenance'] == 'pipeline_fallback'`` is the # marker that tells the rest of the system "the pipeline failed, this is not # a model opinion". It suppresses Oracle escalation # (``orchestrator._should_escalate_to_oracle``), excludes the run from the # Needs-info KPI, and renders the row as an infrastructure error. Only the # orchestrator's own ``_synth_failure_fallback_report`` may set it. Nothing # downstream strips a model-supplied value, so widening what the MODEL can # get accepted into this field widens a spoofing surface for zero observed # benefit — no model has been seen emitting a stringified resolution object. _decode_resolution = field_validator("resolution", mode="before")(_null_sentinel_to_none) # Sentinel written into ``TriageReport.resolution['provenance']`` (and, downstream, # the persisted ``report['resolution']`` dict) by the synth-failure fallback path. PIPELINE_FALLBACK_PROVENANCE = "pipeline_fallback" def is_pipeline_fallback(report: dict[str, Any] | None) -> bool: """True iff a persisted report dict is a pipeline-failure fallback. A run is a "pipeline fallback" iff its report's ``resolution`` marker carries ``provenance == "pipeline_fallback"`` — set only by ``_synth_failure_fallback_report`` when the synth path raises (model truncation, gateway 5xx, schema-validation exhaustion). This is the single shared predicate every downstream consumer (timeline builder, investigation row, alerts badge, dashboard KPI) derives its boolean from, so the notion of "failure vs. genuine needs_more_info" lives in exactly one place. Defensive: a non-dict ``report`` / ``resolution`` (a manual override's ``resolution`` is always a dict, but old rows or a mangled column may not be) returns False rather than raising — a rendering predicate must never break a page. """ if not isinstance(report, dict): return False resolution = report.get("resolution") if not isinstance(resolution, dict): return False return resolution.get("provenance") == PIPELINE_FALLBACK_PROVENANCE __all__ = [ "PIPELINE_FALLBACK_PROVENANCE", "InvestigationTranscript", "RecommendedAction", "TargetedGap", "TriageReport", "Verdict", "WriteToolName", "is_pipeline_fallback", ]