API reference

Generated from docstrings by mkdocstrings — edit the code, not this page. For the narrative of how these modules fit together, read How it works.

revalid — AI-driven revalidation of pentest findings.

Parses pentest reports, extracts findings and reproduction steps, then drives a human-gated agentic retest against a contained target to verify applied fixes.

Domain schemas shared across ingestion, goal-setting, and agentic retest (ADR-0002).

These Pydantic models are the internal representation every layer speaks. AgenticEvidence and VerdictStatus join this module with the FR-17 retest.

AgenticEvidence

Bases: BaseModel

Flexible proof backing an agentic verdict (FR-17 Slice 6b) — tool-agnostic.

An agentic retest runs arbitrary tooling (not just HTTP probes), so its evidence is the agent's explanation plus the decisive command's real output, not a structured request/response. The orchestrator captures it on conclude from the transcript's last command_output (real data, not the model restating it); command/output are empty when the agent concluded without running a command.

Attributes:

Name	Type	Description
explanation	str	The agent's account of what proves the verdict (its rationale).
command	str	The decisive command the agent ran.
output	str	That command's captured stdout/stderr excerpt (truncated).
exit_code	int | None	The command's exit status, or None when no command ran.
elapsed_ms	float	The command's wall-clock time in milliseconds.

Source code in src/revalid/domain.py

class AgenticEvidence(BaseModel):
    """Flexible proof backing an agentic verdict (FR-17 Slice 6b) — tool-agnostic.

    An agentic retest runs arbitrary tooling (not just HTTP probes), so its
    evidence is the agent's explanation plus the decisive command's real output,
    not a structured request/response. The orchestrator captures it on conclude
    from the transcript's last ``command_output`` (real data, not the model
    restating it); ``command``/``output`` are empty when the agent concluded
    without running a command.

    Attributes:
        explanation: The agent's account of what proves the verdict (its rationale).
        command: The decisive command the agent ran.
        output: That command's captured stdout/stderr excerpt (truncated).
        exit_code: The command's exit status, or ``None`` when no command ran.
        elapsed_ms: The command's wall-clock time in milliseconds.
    """

    model_config = ConfigDict(frozen=True)

    explanation: str
    command: str = ""
    output: str = ""
    exit_code: int | None = None
    elapsed_ms: float = 0.0

CvssCode

Bases: BaseModel

CVSS severity code attached to a finding at ingestion (FR-19).

vector is the CVSS base vector string (e.g. CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H); base_score is the derived 0.0--10.0 base score. inferred records provenance: False when the code was read from the report verbatim, True when the model derived it because the report stated none. An empty vector with inferred=False means the report had no CVSS code and none was derived.

Source code in src/revalid/domain.py

class CvssCode(BaseModel):
    """CVSS severity code attached to a finding at ingestion (FR-19).

    ``vector`` is the CVSS base vector string (e.g.
    ``CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H``); ``base_score`` is the
    derived 0.0--10.0 base score. ``inferred`` records provenance: ``False``
    when the code was read from the report verbatim, ``True`` when the model
    derived it because the report stated none. An empty ``vector`` with
    ``inferred=False`` means the report had no CVSS code and none was derived.
    """

    model_config = ConfigDict(frozen=True)

    vector: str = ""
    base_score: float | None = None
    inferred: bool = False

Finding

Bases: BaseModel

A single pentest finding in the internal model (FR-02/FR-03).

Attributes:

Name	Type	Description
title	str	Short human-readable name of the finding.
severity	Severity	Normalized severity level.
description	str	Free-text description of the vulnerability.
impact	str	What an attacker gains / the business consequence (FR-03).
attack_vector	str	How the vulnerability is reached and exploited (FR-03).
affected_endpoints	tuple[str, ...]	URLs or endpoint identifiers the finding applies to.
reproduction_steps	tuple[str, ...]	Ordered steps to reproduce the issue.
cvss	CvssCode	CVSS severity code, read from the report or derived at ingestion (FR-19). Provenance is on the inferred flag.
mitre	MitreMapping	MITRE ATT&CK technique mapping, read or derived at ingestion (FR-19). Provenance is on the inferred flag.
raw	dict[str, Any]	Complete source payload as ingested, preserving fields the internal model does not map (FR-02 audit criterion). For LLM-extracted findings it also carries extraction lineage — model name and source text — for the audit trail (FR-10 / NFR-02).

Source code in src/revalid/domain.py

class Finding(BaseModel):
    """A single pentest finding in the internal model (FR-02/FR-03).

    Attributes:
        title: Short human-readable name of the finding.
        severity: Normalized severity level.
        description: Free-text description of the vulnerability.
        impact: What an attacker gains / the business consequence (FR-03).
        attack_vector: How the vulnerability is reached and exploited (FR-03).
        affected_endpoints: URLs or endpoint identifiers the finding applies to.
        reproduction_steps: Ordered steps to reproduce the issue.
        cvss: CVSS severity code, read from the report or derived at ingestion
            (FR-19). Provenance is on the ``inferred`` flag.
        mitre: MITRE ATT&CK technique mapping, read or derived at ingestion
            (FR-19). Provenance is on the ``inferred`` flag.
        raw: Complete source payload as ingested, preserving fields the
            internal model does not map (FR-02 audit criterion). For
            LLM-extracted findings it also carries extraction lineage — model
            name and source text — for the audit trail (FR-10 / NFR-02).
    """

    model_config = ConfigDict(frozen=True)

    title: str = Field(min_length=1)
    severity: Severity
    description: str = ""
    impact: str = ""
    attack_vector: str = ""
    affected_endpoints: tuple[str, ...] = ()
    reproduction_steps: tuple[str, ...] = ()
    cvss: CvssCode = Field(default_factory=CvssCode)
    mitre: MitreMapping = Field(default_factory=MitreMapping)
    raw: dict[str, Any] = Field(default_factory=dict)

FindingOrigin

Bases: StrEnum

How a finding version came to be (FR-16, ADR-0024).

Extraction is version 1 — the finding as the LLM/import first produced it; every later version is an operator EDIT. The distinction is audit lineage: which content the machine proposed vs. what a human corrected.

Source code in src/revalid/domain.py

class FindingOrigin(enum.StrEnum):
    """How a finding version came to be (FR-16, ADR-0024).

    Extraction is version 1 — the finding as the LLM/import first produced it;
    every later version is an operator ``EDIT``. The distinction is audit
    lineage: which content the machine proposed vs. what a human corrected.
    """

    EXTRACTION = "extraction"
    EDIT = "edit"

FindingStage

Bases: StrEnum

The pipeline stage a note was written on (FR-16, ADR-0024).

The stages mirror the finding's lifecycle track — extract → goal → retest → verdict — while GENERAL tags a note left from the finding overview rather than any one stage.

PLAN and APPROVE are the retired batch flow's stages (ADR-0033). They are kept readable so a database written before the reshape still loads, but nothing produces them any more: the goal stage tagged its notes plan until issue #113, and existing rows are renamed to goal by the lightweight backfill in :func:revalid.db.create_db_engine. Do not use them for new notes.

Source code in src/revalid/domain.py

class FindingStage(enum.StrEnum):
    """The pipeline stage a note was written on (FR-16, ADR-0024).

    The stages mirror the finding's lifecycle track — extract → goal → retest →
    verdict — while ``GENERAL`` tags a note left from the finding overview
    rather than any one stage.

    ``PLAN`` and ``APPROVE`` are the retired batch flow's stages (ADR-0033).
    They are kept **readable** so a database written before the reshape still
    loads, but nothing produces them any more: the goal stage tagged its notes
    ``plan`` until issue #113, and existing rows are renamed to ``goal`` by the
    lightweight backfill in :func:`revalid.db.create_db_engine`. Do not use them
    for new notes.
    """

    EXTRACT = "extract"
    GOAL = "goal"
    RETEST = "retest"
    VERDICT = "verdict"
    GENERAL = "general"
    #: Legacy, read-only: the retired batch flow's stages (ADR-0033, #113).
    PLAN = "plan"
    APPROVE = "approve"

MitreMapping

Bases: BaseModel

MITRE ATT&CK technique mapping for a finding (FR-19).

techniques are ATT&CK technique IDs (e.g. T1190, T1110) the finding maps onto; inferred is True when the model derived the mapping rather than reading it from the report. Empty techniques with inferred=False means none stated and none derived.

Source code in src/revalid/domain.py

class MitreMapping(BaseModel):
    """MITRE ATT&CK technique mapping for a finding (FR-19).

    ``techniques`` are ATT&CK technique IDs (e.g. ``T1190``,
    ``T1110``) the finding maps onto; ``inferred`` is ``True`` when the model
    derived the mapping rather than reading it from the report. Empty
    ``techniques`` with ``inferred=False`` means none stated and none derived.
    """

    model_config = ConfigDict(frozen=True)

    techniques: tuple[str, ...] = ()
    inferred: bool = False

ReportStatus

Bases: StrEnum

Lifecycle state of an uploaded report's ingest job (FR-01/FR-11).

A report starts EXTRACTING when uploaded and always settles on exactly one terminal state — READY once its findings are persisted, FAILED (with the error recorded) if extraction never completes, or CANCELLED when the operator stops it mid-run (issue #205) — so the UI's status poll is guaranteed to terminate. A CANCELLED report keeps whatever findings had already been extracted before the stop, so it is re-runnable or deletable.

Source code in src/revalid/domain.py

class ReportStatus(enum.StrEnum):
    """Lifecycle state of an uploaded report's ingest job (FR-01/FR-11).

    A report starts ``EXTRACTING`` when uploaded and always settles on exactly
    one terminal state — ``READY`` once its findings are persisted, ``FAILED``
    (with the error recorded) if extraction never completes, or ``CANCELLED``
    when the operator stops it mid-run (issue #205) — so the UI's status poll is
    guaranteed to terminate. A ``CANCELLED`` report keeps whatever findings had
    already been extracted before the stop, so it is re-runnable or deletable.
    """

    EXTRACTING = "extracting"
    READY = "ready"
    FAILED = "failed"
    #: The operator stopped extraction mid-run (issue #205). Terminal like
    #: ``READY``/``FAILED``: any findings extracted before the stop are kept.
    CANCELLED = "cancelled"

RetestSessionStatus

Bases: StrEnum

Lifecycle of an FR-17 agentic retest session — one agent, five live states.

A live session is always in exactly one of: WORKING (a turn in flight), AWAITING_COMMAND (a proposed command awaits the operator's approval), AWAITING_OPERATOR (the agent handed back — a reply, an acknowledgement, a guided one-action report, "I've exhausted my options", or a verdict recommendation), IDLE (created but not yet provisioned — a Restart lands here so it never auto-runs), or STOPPED (the operator paused it, sandbox kept alive). A session reaches a terminal state (CONCLUDED/ENDED/ERROR) only on a determination or an operator action. GIVEN_UP is retired — kept only so any legacy row stays terminal. ADR-0042 collapsed the earlier needs_guidance/running_command/starting/thinking states into these five (needs_guidance folded into awaiting_operator).

Source code in src/revalid/domain.py

class RetestSessionStatus(enum.StrEnum):
    """Lifecycle of an FR-17 agentic retest session — one agent, five live states.

    A live session is always in exactly one of: ``WORKING`` (a turn in flight),
    ``AWAITING_COMMAND`` (a proposed command awaits the operator's approval),
    ``AWAITING_OPERATOR`` (the agent handed back — a reply, an acknowledgement, a
    guided one-action report, "I've exhausted my options", or a verdict recommendation),
    ``IDLE`` (created but not yet provisioned — a ``Restart`` lands here so it never
    auto-runs), or ``STOPPED`` (the operator paused it, sandbox kept alive). A
    session reaches a terminal state (``CONCLUDED``/``ENDED``/``ERROR``) only on a
    determination or an operator action. ``GIVEN_UP`` is retired — kept only so any
    legacy row stays terminal. ADR-0042 collapsed the earlier
    ``needs_guidance``/``running_command``/``starting``/``thinking`` states into
    these five (``needs_guidance`` folded into ``awaiting_operator``).
    """

    #: Created but not started: no sandbox yet, awaiting an operator ``Start`` or a
    #: message. A ``Restart`` opens a session here so the fresh attempt never auto-runs.
    IDLE = "idle"
    #: A turn is in flight — the LLM call and any command it runs happen inside one
    #: turn, so this is the single "busy" state (was ``thinking``/``starting``/the
    #: never-set ``running_command``). The console shows a live indicator; an operator
    #: message is queued and delivered to the same agent at the next turn boundary.
    WORKING = "working"
    #: A command is proposed and awaits the operator's approve/reject — the Claude-Code
    #: permission prompt. A message here withdraws the proposal and steers the agent.
    AWAITING_COMMAND = "awaiting_command"
    #: The agent handed control back without a command or verdict: a conversational
    #: reply, an acknowledgement, a guided "ran X — I'd try Y next" report with a suggested
    #: next step, a verdict *recommendation* to confirm, or "I've exhausted my options"
    #: (the former ``needs_guidance``, now just the agent's own words). Non-terminal:
    #: the sandbox stays alive and the operator's next message resumes it.
    AWAITING_OPERATOR = "awaiting_operator"
    #: The operator paused a running session (issue #150). Non-terminal — the
    #: sandbox stays alive; a message/Resume continues, ``Restart``/``Conclude`` end it.
    STOPPED = "stopped"
    CONCLUDED = "concluded"
    GIVEN_UP = "given_up"
    ENDED = "ended"
    ERROR = "error"

SessionEventKind

Bases: StrEnum

Kinds of append-only transcript event (FR-17 audit trail).

Source code in src/revalid/domain.py

class SessionEventKind(enum.StrEnum):
    """Kinds of append-only transcript event (FR-17 audit trail)."""

    AGENT_MESSAGE = "agent_message"
    COMMAND_PROPOSED = "command_proposed"
    COMMAND_APPROVED = "command_approved"
    COMMAND_REJECTED = "command_rejected"
    COMMAND_OUTPUT = "command_output"
    HUMAN_COMMAND = "human_command"
    HUMAN_MESSAGE = "human_message"
    #: Queued operator message(s) were handed to the agent for this turn (issue
    #: #204). Marks the delivery boundary so the console can stop showing a
    #: "queued" hint on a message the agent has now received.
    MESSAGES_DELIVERED = "messages_delivered"
    #: The operator aborted the in-flight turn and had it re-run to unstick a wedged
    #: model (issue #204). A transcript marker only — the retried turn's real events
    #: follow it.
    TURN_RESTARTED = "turn_restarted"
    # The current guiding goal (FR-17 6b-ii: user-owned; formerly the agent's set_plan).
    PLAN_UPDATED = "plan_updated"
    #: The retest scope set at launch (FR-17): payload ``endpoints`` is the exact
    #: list of target URLs the agent must confine itself to. Emitted once at session
    #: start and never again — reachability is fixed when the sandbox is provisioned,
    #: so changing scope needs a fresh session (Restart), not a live edit.
    TARGET_SET = "target_set"
    STATE_CHANGE = "state_change"
    FREE_LAUNCH_CHANGED = "free_launch_changed"
    VERDICT = "verdict"
    VERDICT_ADJUDICATED = "verdict_adjudicated"
    #: The operator reopened a concluded session (issue #214): the recorded verdict
    #: is cancelled (kept in the transcript, never deleted — FR-10) and the session
    #: returns to ``idle`` so testing can continue. Payload carries the cancelled
    #: verdict's ``status`` for the audit trail.
    VERDICT_CANCELLED = "verdict_cancelled"
    ERROR = "error"

Settings

Bases: BaseModel

User-configurable LLM backend selection (FR-13 / ADR-0021).

Attributes:

Name	Type	Description
model	str	A Pydantic AI provider:model string (e.g. ollama:qwen3.6:27b).
base_url	str | None	Provider base URL for OpenAI-compatible backends (Ollama and friends); None for native providers configured from the environment.
api_key	str | None	Provider API key, or None when supplied via the environment.

Source code in src/revalid/domain.py

class Settings(BaseModel):
    """User-configurable LLM backend selection (FR-13 / ADR-0021).

    Attributes:
        model: A Pydantic AI ``provider:model`` string (e.g. ``ollama:qwen3.6:27b``).
        base_url: Provider base URL for OpenAI-compatible backends (Ollama and
            friends); ``None`` for native providers configured from the environment.
        api_key: Provider API key, or ``None`` when supplied via the environment.
    """

    model_config = ConfigDict(frozen=True, protected_namespaces=())

    model: str = Field(min_length=1)
    base_url: str | None = None
    api_key: str | None = None

Severity

Bases: StrEnum

Normalized severity scale for findings.

Source code in src/revalid/domain.py

class Severity(enum.StrEnum):
    """Normalized severity scale for findings."""

    INFO = "info"
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

VerdictStatus

Bases: StrEnum

Outcome of retesting a finding (FR-09).

Source code in src/revalid/domain.py

class VerdictStatus(enum.StrEnum):
    """Outcome of retesting a finding (FR-09)."""

    STILL_OPEN = "still_open"
    FIXED = "fixed"
    INCONCLUSIVE = "inconclusive"

Report ingestion and understanding

pdf extracts whole-document Markdown; extract turns that into schema-validated findings in one LLM call; ingest maps DefectDojo-style JSON and manual entry with no LLM at all; findings owns versioning, notes and the CVSS/MITRE enrichment every door shares.

Structured-report ingestion by schema mapping — no LLM involved (FR-02).

Initial supported format: DefectDojo-style JSON findings export, i.e. a top-level {"findings": [...]} array where each entry carries at least title and severity. Every source entry is preserved verbatim in Finding.raw so unmapped fields stay auditable.

No LLM is the invariant of this module, and it is why this door is the deterministic, instant, free seeding path for demos and tests. A stated taxonomy is still copied across — cvssv3/cvssv3_score from the DefectDojo format, plus the revalid-specific mitre_techniques the manual form supplies — because copying is not inferring, and both land inferred=False. Deriving a taxonomy the source never stated is the opt-in enrichment pass in :mod:revalid.extract, which the caller runs after mapping (issues #233, #237).

IngestError

Bases: ValueError

Raised when an input document cannot be mapped to the internal model.

Source code in src/revalid/ingest.py

class IngestError(ValueError):
    """Raised when an input document cannot be mapped to the internal model."""

load_defectdojo_export(text)

Parse a DefectDojo-style JSON export string into domain findings.

Parameters:

Name	Type	Description	Default
text	str	Raw JSON document.	required

Returns:

Type	Description
list[Finding]	Findings in document order.

Raises:

Type	Description
IngestError	If the document is not valid JSON or does not map.

Source code in src/revalid/ingest.py

def load_defectdojo_export(text: str) -> list[Finding]:
    """Parse a DefectDojo-style JSON export string into domain findings.

    Args:
        text: Raw JSON document.

    Returns:
        Findings in document order.

    Raises:
        IngestError: If the document is not valid JSON or does not map.
    """
    try:
        data = json.loads(text)
    except json.JSONDecodeError as exc:
        raise IngestError(f"not valid JSON: {exc}") from exc
    return map_defectdojo_export(data)

map_defectdojo_export(data)

Map an already-parsed DefectDojo-style export to domain findings.

Parameters:

Name	Type	Description	Default
data	object	Parsed JSON document; must be an object with a findings array.	required

Returns:

Type	Description
list[Finding]	Findings in document order.

Raises:

Type	Description
IngestError	If the document shape or any entry is invalid.

Source code in src/revalid/ingest.py

def map_defectdojo_export(data: object) -> list[Finding]:
    """Map an already-parsed DefectDojo-style export to domain findings.

    Args:
        data: Parsed JSON document; must be an object with a ``findings`` array.

    Returns:
        Findings in document order.

    Raises:
        IngestError: If the document shape or any entry is invalid.
    """
    if not isinstance(data, dict) or not isinstance(data.get("findings"), list):
        raise IngestError('expected a JSON object with a "findings" array')
    return [_map_finding(item, index) for index, item in enumerate(data["findings"])]

PDF pentest-report ingestion — whole-document text extraction (FR-01).

This module turns report bytes into LLM-ready Markdown text. It is intentionally LLM-free: the semantic step (text -> validated Finding objects) is FR-03's job (see :mod:revalid.extract). Extraction uses PyMuPDF4LLM in its deterministic legacy mode (ADR-0047): the whole document is rendered to GitHub-flavoured Markdown, so headings, tables and lists survive as structure the model can read, and the same bytes always produce the same text (NFR-02). The extractor then hands the entire report to the model in one call — there is no heading segmentation ahead of it (that regex step was removed in ADR-0047 because it was format-bound and silently failed on unfamiliar layouts).

Malformed input fails closed with a clear :class:PdfError rather than crashing: a non-PDF (missing %PDF- header), a structurally corrupt PDF, and a PDF with no extractable text (scanned/image-only — out of scope, no OCR) are all rejected.

PdfError

Bases: ValueError

Raised when a PDF cannot be read or carries no extractable report text.

Source code in src/revalid/pdf.py

class PdfError(ValueError):
    """Raised when a PDF cannot be read or carries no extractable report text."""

PdfPage

Bases: BaseModel

Text extracted from one page of a report.

Attributes:

Name	Type	Description
number	int	1-based page number in document order.
text	str	Reading-order Markdown of the page (empty if the page has none).

Source code in src/revalid/pdf.py

class PdfPage(BaseModel):
    """Text extracted from one page of a report.

    Attributes:
        number: 1-based page number in document order.
        text: Reading-order Markdown of the page (empty if the page has none).
    """

    model_config = ConfigDict(frozen=True)

    number: int = Field(ge=1)
    text: str = ""

PdfReport

Bases: BaseModel

The full deterministic extraction of a PDF report (FR-01 output).

Attributes:

Name	Type	Description
page_count	int	Number of pages in the source document.
pages	tuple[PdfPage, ...]	Per-page extracted Markdown, in document order.
text	str	Whole-report Markdown, pages joined in order — the single input FR-03's LLM structures into findings in one call.

Source code in src/revalid/pdf.py

class PdfReport(BaseModel):
    """The full deterministic extraction of a PDF report (FR-01 output).

    Attributes:
        page_count: Number of pages in the source document.
        pages: Per-page extracted Markdown, in document order.
        text: Whole-report Markdown, pages joined in order — the single input
            FR-03's LLM structures into findings in one call.
    """

    model_config = ConfigDict(frozen=True)

    page_count: int = Field(ge=1)
    pages: tuple[PdfPage, ...]
    text: str = Field(min_length=1)

read_pdf(data)

Extract a PDF report to Markdown text, tolerating common layouts (FR-01).

Parameters:

Name	Type	Description	Default
data	bytes	Raw bytes of the PDF document.	required

Returns:

Type	Description
PdfReport	The extracted report: per-page and whole-document Markdown text.

Raises:

Type	Description
PdfError	If data is not a PDF, is structurally corrupt, or yields no extractable text (e.g. a scanned/image-only document).

Source code in src/revalid/pdf.py

def read_pdf(data: bytes) -> PdfReport:
    """Extract a PDF report to Markdown text, tolerating common layouts (FR-01).

    Args:
        data: Raw bytes of the PDF document.

    Returns:
        The extracted report: per-page and whole-document Markdown text.

    Raises:
        PdfError: If ``data`` is not a PDF, is structurally corrupt, or yields
            no extractable text (e.g. a scanned/image-only document).
    """
    if data[: len(_PDF_MAGIC)] != _PDF_MAGIC:
        raise PdfError("not a PDF (missing %PDF- header)")
    try:
        pages = _extract_pages(data)
    except pymupdf.FileDataError as exc:
        raise PdfError(f"could not parse PDF: {exc}") from exc

    text = "\n\n".join(page.text for page in pages if page.text).strip()
    if not text:
        raise PdfError("no extractable text (is this a scanned or image-only PDF?)")
    return PdfReport(page_count=len(pages), pages=pages, text=text)

Model-agnostic LLM backend selection (FR-13, ADR-0010, ADR-0021).

One switch, REVALID_LLM_MODEL, selects the backend for every LLM-using component: it holds a Pydantic AI model string (provider:model, e.g. ollama:qwen3.5:9b or anthropic:claude-sonnet-5) and defaults to a local-first Ollama backend (ADR-0021) — no API key or network egress required out of the box. Switching backends is configuration-only — no code change. The Ollama backend additionally needs a base URL (OLLAMA_BASE_URL, falling back to :data:DEFAULT_BASE_URL) for its OpenAI-compatible endpoint.

The string is resolved to a concrete model lazily, at the first model call (agents are built with defer_model_check=True), so construction never needs the network and a misconfigured backend surfaces as a clear error on first use.

DEFAULT_BASE_URL = 'http://localhost:11434/v1' module-attribute

Default OpenAI-compatible endpoint for the local-first Ollama backend (ADR-0021).

DEFAULT_MODEL = 'ollama:qwen3.5:9b' module-attribute

Local-first default backend (ADR-0021); used when :data:MODEL_ENV is unset.

A small, responsive model on purpose: the FR-17 agentic console runs a multi-turn reason->command->observe loop, and a heavier model (e.g. ollama:qwen3.6:27b at ~50s+/turn locally) makes the interactive session read as hung. qwen3.5:9b proposes its first step in ~10s here; users can still select a larger model in /settings when they can tolerate the latency.

Not a member of Pydantic AI's KnownModelName literal (Ollama models are open-ended, not a fixed catalog), so this is a plain str.

MODEL_ENV = 'REVALID_LLM_MODEL' module-attribute

Environment variable that selects the Pydantic AI backend (ADR-0010).

agent_model_name(agent)

Return a best-effort stable model identifier for an agent's audit lineage.

Recording which backend produced a finding or plan is required for the audit trail (NFR-02). Works whether the agent was built from a model string or an injected model instance (TestModel/FunctionModel in tests).

Parameters:

Name	Type	Description	Default
agent	Agent[Any, Any]	Any Pydantic AI agent.	required

Returns:

Type	Description
str	The model string, or the instance's model_name (falling back to its
str	repr).

Source code in src/revalid/llm.py

def agent_model_name(agent: Agent[Any, Any]) -> str:
    """Return a best-effort stable model identifier for an agent's audit lineage.

    Recording which backend produced a finding or plan is required for the audit
    trail (NFR-02). Works whether the agent was built from a model string or an
    injected model instance (``TestModel``/``FunctionModel`` in tests).

    Args:
        agent: Any Pydantic AI agent.

    Returns:
        The model string, or the instance's ``model_name`` (falling back to its
        ``repr``).
    """
    model = agent.model
    if isinstance(model, str):
        return model
    return getattr(model, "model_name", str(model))

build_model(cfg)

Construct a concrete Pydantic AI model from a persisted setting (ADR-0021).

• A base_url selects an OpenAI-compatible model (Ollama or any OpenAI-compatible host); the ollama:/openai: provider prefix is stripped from the model name and a placeholder key is used when none is stored (Ollama ignores it).
• Otherwise a native provider with a stored key is built explicitly; with no stored key the bare provider:model string is returned so Pydantic AI resolves credentials from the environment (backward-compatible with FR-13).

Parameters:

Name	Type	Description	Default
cfg	Settings	The persisted settings.	required

Returns:

Type	Description
Model | str	A Pydantic AI :class:~pydantic_ai.models.Model instance, or the model
Model | str	string when the environment should supply credentials.

Source code in src/revalid/llm.py

def build_model(cfg: Settings) -> Model | str:
    """Construct a concrete Pydantic AI model from a persisted setting (ADR-0021).

    - A ``base_url`` selects an OpenAI-compatible model (Ollama or any
      OpenAI-compatible host); the ``ollama:``/``openai:`` provider prefix is
      stripped from the model name and a placeholder key is used when none is
      stored (Ollama ignores it).
    - Otherwise a native provider with a *stored* key is built explicitly; with
      no stored key the bare ``provider:model`` string is returned so Pydantic AI
      resolves credentials from the environment (backward-compatible with FR-13).

    Args:
        cfg: The persisted settings.

    Returns:
        A Pydantic AI :class:`~pydantic_ai.models.Model` instance, or the model
        string when the environment should supply credentials.
    """
    if cfg.base_url:
        name = (
            cfg.model.split(":", 1)[1]
            if cfg.model.startswith(("ollama:", "openai:"))
            else cfg.model
        )
        return OpenAIChatModel(
            name,
            provider=OpenAIProvider(base_url=cfg.base_url, api_key=cfg.api_key or "ollama"),
        )
    provider, _, name = cfg.model.partition(":")
    if provider == "anthropic" and cfg.api_key:
        return AnthropicModel(name, provider=AnthropicProvider(api_key=cfg.api_key))
    return cfg.model

resolve_model()

Return the configured Pydantic AI model string.

Reads :data:MODEL_ENV (REVALID_LLM_MODEL); an unset or blank value falls back to :data:DEFAULT_MODEL.

Returns:

Type	Description
str	A provider:model string for Pydantic AI (validated at first call).

Source code in src/revalid/llm.py

def resolve_model() -> str:
    """Return the configured Pydantic AI model string.

    Reads :data:`MODEL_ENV` (``REVALID_LLM_MODEL``); an unset or blank value
    falls back to :data:`DEFAULT_MODEL`.

    Returns:
        A ``provider:model`` string for Pydantic AI (validated at first call).
    """
    return os.environ.get(MODEL_ENV, "").strip() or DEFAULT_MODEL

LLM extraction of structured findings from report text (FR-03).

Turns the whole-report Markdown produced by FR-01 (pdf.py) into schema-validated domain :class:~revalid.domain.Finding objects using Pydantic AI (ADR-0002, ADR-0047). The entire report is sent to the model in one call, which must return a list of :class:ExtractedFinding objects; output that fails schema validation is retried by Pydantic AI and, if still invalid, flagged as an :class:ExtractionFailure — never silently mapped to a Finding or persisted (FR-03's schema-validation gate). The model is injectable so unit tests drive it with Pydantic AI's TestModel/FunctionModel and never touch the network; when none is passed, the configured backend is used (REVALID_LLM_MODEL, FR-13/ADR-0010).

EnrichmentReport

Bases: BaseModel

Outcome of an opt-in taxonomy enrichment pass (FR-19, issue #233).

Attributes:

Name	Type	Description
findings	tuple[Finding, ...]	The findings in input order, enriched where possible. A finding whose model call failed is present, unchanged.
enriched	int	How many findings actually gained a CVSS vector or ATT&CK techniques. Lower than len(findings) when some already had them.
failed	int	How many model calls failed schema validation. Non-zero means the import succeeded but part of the taxonomy is missing — surfaced to the operator rather than swallowed.

Source code in src/revalid/extract.py

class EnrichmentReport(BaseModel):
    """Outcome of an opt-in taxonomy enrichment pass (FR-19, issue #233).

    Attributes:
        findings: The findings in input order, enriched where possible. A finding
            whose model call failed is present, unchanged.
        enriched: How many findings actually gained a CVSS vector or ATT&CK
            techniques. Lower than ``len(findings)`` when some already had them.
        failed: How many model calls failed schema validation. Non-zero means the
            import succeeded but part of the taxonomy is missing — surfaced to the
            operator rather than swallowed.
    """

    model_config = ConfigDict(frozen=True)

    findings: tuple[Finding, ...]
    enriched: int = 0
    failed: int = 0

ExtractedFinding

Bases: BaseModel

One finding exactly as the model must return it — the FR-03 gate.

Every field is required, so the model must account for all of them; Pydantic AI validates the tool output against this schema and retries on mismatch. Strings may be empty when the report omits a field; title may not.

Source code in src/revalid/extract.py

class ExtractedFinding(BaseModel):
    """One finding exactly as the model must return it — the FR-03 gate.

    Every field is required, so the model must account for all of them; Pydantic
    AI validates the tool output against this schema and retries on mismatch.
    Strings may be empty when the report omits a field; ``title`` may not.
    """

    model_config = ConfigDict(frozen=True)

    title: str = Field(min_length=1)
    severity: Severity
    description: str
    impact: str
    attack_vector: str
    affected_endpoints: tuple[str, ...]
    reproduction_steps: tuple[str, ...]
    cvss: CvssCode = Field(default_factory=CvssCode)
    mitre: MitreMapping = Field(default_factory=MitreMapping)

ExtractionFailure

Bases: BaseModel

A report whose extraction never validated — flagged, not persisted.

Extraction is a single whole-document call, so a failure is all-or-nothing: one record stands for the report, not for an individual finding.

Attributes:

Name	Type	Description
error	str	The reason extraction failed (retries exhausted, etc.).
source_text	str	The report text, kept so the failure is auditable and re-runnable.

Source code in src/revalid/extract.py

class ExtractionFailure(BaseModel):
    """A report whose extraction never validated — flagged, not persisted.

    Extraction is a single whole-document call, so a failure is all-or-nothing:
    one record stands for the report, not for an individual finding.

    Attributes:
        error: The reason extraction failed (retries exhausted, etc.).
        source_text: The report text, kept so the failure is auditable and
            re-runnable.
    """

    model_config = ConfigDict(frozen=True)

    error: str
    source_text: str

ExtractionRegistry

Process-local cancel flags for in-flight extractions (issue #205).

Extraction is a single whole-document model call run as a background task (:func:~revalid.app.run_extraction). This lets the request thread flag a report so the worker settles cooperatively. Thread-safe: the flag is written by the request thread and read by the extraction worker.

A flag carries a reason: "operator" (a Stop) or "deleted" (the report is being removed). A pending delete always wins over a Stop, since the row is going away.

Beyond the flag, the registry holds the event loop + task of an in-flight extraction (attached by the worker) so a cancel can interrupt the running model call cross-thread. Because extraction is one call, interrupting it is the only way to stop it — there is no between-candidates checkpoint — which is exactly what makes Stop work when a local model wedges on a long document.

Source code in src/revalid/extract.py

class ExtractionRegistry:
    """Process-local cancel flags for in-flight extractions (issue #205).

    Extraction is a single whole-document model call run as a background task
    (:func:`~revalid.app.run_extraction`). This lets the request thread flag a
    report so the worker settles cooperatively. Thread-safe: the flag is written
    by the request thread and read by the extraction worker.

    A flag carries a *reason*: ``"operator"`` (a Stop) or ``"deleted"`` (the report
    is being removed). A pending delete always wins over a Stop, since the row is
    going away.

    Beyond the flag, the registry holds the event loop + task of an in-flight
    extraction (attached by the worker) so a cancel can *interrupt* the running
    model call cross-thread. Because extraction is one call, interrupting it is
    the only way to stop it — there is no between-candidates checkpoint — which is
    exactly what makes Stop work when a local model wedges on a long document.
    """

    def __init__(self) -> None:
        """Start with no extraction flagged."""
        self._lock = threading.Lock()
        self._reasons: dict[int, str] = {}
        self._runs: dict[int, tuple[asyncio.AbstractEventLoop, asyncio.Task[ExtractionReport]]] = {}

    def request_cancel(self, report_id: int, reason: str = "operator") -> None:
        """Flag ``report_id`` for cancellation and interrupt its in-flight call.

        Records the reason (a ``"deleted"`` flag always wins over an operator Stop)
        and, if an extraction task is attached, cancels it cross-thread so the
        current model call aborts immediately rather than at the next candidate.
        """
        with self._lock:
            if self._reasons.get(report_id) != "deleted":
                self._reasons[report_id] = reason
            run = self._runs.get(report_id)
        if run is not None:
            loop, task = run
            try:
                loop.call_soon_threadsafe(task.cancel)
            except RuntimeError:  # the loop is already closing — the run is ending anyway
                pass

    def cancel_reason(self, report_id: int) -> str | None:
        """Return the cancel reason flagged for ``report_id``, or ``None`` if not flagged."""
        with self._lock:
            return self._reasons.get(report_id)

    def attach(
        self,
        report_id: int,
        loop: asyncio.AbstractEventLoop,
        task: asyncio.Task[ExtractionReport],
    ) -> None:
        """Register the loop + task of the extraction now running for ``report_id``."""
        with self._lock:
            self._runs[report_id] = (loop, task)

    def clear(self, report_id: int) -> None:
        """Drop any flag + run handle for ``report_id`` (the worker settled)."""
        with self._lock:
            self._reasons.pop(report_id, None)
            self._runs.pop(report_id, None)

__init__()

Start with no extraction flagged.

Source code in src/revalid/extract.py

def __init__(self) -> None:
    """Start with no extraction flagged."""
    self._lock = threading.Lock()
    self._reasons: dict[int, str] = {}
    self._runs: dict[int, tuple[asyncio.AbstractEventLoop, asyncio.Task[ExtractionReport]]] = {}

attach(report_id, loop, task)

Register the loop + task of the extraction now running for report_id.

Source code in src/revalid/extract.py

def attach(
    self,
    report_id: int,
    loop: asyncio.AbstractEventLoop,
    task: asyncio.Task[ExtractionReport],
) -> None:
    """Register the loop + task of the extraction now running for ``report_id``."""
    with self._lock:
        self._runs[report_id] = (loop, task)

cancel_reason(report_id)

Return the cancel reason flagged for report_id, or None if not flagged.

Source code in src/revalid/extract.py

def cancel_reason(self, report_id: int) -> str | None:
    """Return the cancel reason flagged for ``report_id``, or ``None`` if not flagged."""
    with self._lock:
        return self._reasons.get(report_id)

clear(report_id)

Drop any flag + run handle for report_id (the worker settled).

Source code in src/revalid/extract.py

def clear(self, report_id: int) -> None:
    """Drop any flag + run handle for ``report_id`` (the worker settled)."""
    with self._lock:
        self._reasons.pop(report_id, None)
        self._runs.pop(report_id, None)

request_cancel(report_id, reason='operator')

Flag report_id for cancellation and interrupt its in-flight call.

Records the reason (a "deleted" flag always wins over an operator Stop) and, if an extraction task is attached, cancels it cross-thread so the current model call aborts immediately rather than at the next candidate.

Source code in src/revalid/extract.py

def request_cancel(self, report_id: int, reason: str = "operator") -> None:
    """Flag ``report_id`` for cancellation and interrupt its in-flight call.

    Records the reason (a ``"deleted"`` flag always wins over an operator Stop)
    and, if an extraction task is attached, cancels it cross-thread so the
    current model call aborts immediately rather than at the next candidate.
    """
    with self._lock:
        if self._reasons.get(report_id) != "deleted":
            self._reasons[report_id] = reason
        run = self._runs.get(report_id)
    if run is not None:
        loop, task = run
        try:
            loop.call_soon_threadsafe(task.cancel)
        except RuntimeError:  # the loop is already closing — the run is ending anyway
            pass

ExtractionReport

Bases: BaseModel

Outcome of extracting a whole report.

Attributes:

Name	Type	Description
findings	tuple[Finding, ...]	Schema-valid findings, safe to persist.
failures	tuple[ExtractionFailure, ...]	A single flagged failure when the whole-document call never passed the validation gate; empty otherwise.
cancelled	bool	Whether extraction stopped early because the operator asked it to (issue #205). Extraction is one call, so a cancel yields no findings — findings and failures are both empty.

Source code in src/revalid/extract.py

class ExtractionReport(BaseModel):
    """Outcome of extracting a whole report.

    Attributes:
        findings: Schema-valid findings, safe to persist.
        failures: A single flagged failure when the whole-document call never
            passed the validation gate; empty otherwise.
        cancelled: Whether extraction stopped early because the operator asked it
            to (issue #205). Extraction is one call, so a cancel yields no
            findings — ``findings`` and ``failures`` are both empty.
    """

    model_config = ConfigDict(frozen=True)

    findings: tuple[Finding, ...]
    failures: tuple[ExtractionFailure, ...]
    cancelled: bool = False

FindingTaxonomy

Bases: BaseModel

A finding's derived CVSS + ATT&CK classification (FR-19, opt-in enrichment).

Deliberately without an inferred flag: everything this model returns is by definition the model's own derivation, so provenance is stamped server-side (:func:apply_taxonomy sets inferred=True) and the model has no way to express "the report stated this". That is the same rule the finding editor follows — a client, human or machine, never asserts provenance (ADR-0037).

Every field is required, for the same reason :class:ExtractedFinding's are: the model must account for all of them, and Pydantic AI retries when it does not. With defaults, {} was trivially valid — a small model could return nothing, no retry would fire, and the result was indistinguishable from "assessed and found nothing to say" (issue #241, seen live on a 9b local model). An unassessable finding is still expressible, as an explicit empty vector and empty technique list; it just has to be chosen rather than reached by omission.

Source code in src/revalid/extract.py

class FindingTaxonomy(BaseModel):
    """A finding's derived CVSS + ATT&CK classification (FR-19, opt-in enrichment).

    Deliberately **without** an ``inferred`` flag: everything this model returns is
    by definition the model's own derivation, so provenance is stamped server-side
    (:func:`apply_taxonomy` sets ``inferred=True``) and the model has no way to
    express "the report stated this". That is the same rule the finding editor
    follows — a client, human or machine, never asserts provenance (ADR-0037).

    Every field is **required**, for the same reason :class:`ExtractedFinding`'s
    are: the model must account for all of them, and Pydantic AI retries when it
    does not. With defaults, ``{}`` was trivially valid — a small model could
    return nothing, no retry would fire, and the result was indistinguishable from
    "assessed and found nothing to say" (issue #241, seen live on a 9b local
    model). An unassessable finding is still expressible, as an explicit empty
    vector and empty technique list; it just has to be *chosen* rather than
    reached by omission.
    """

    model_config = ConfigDict(frozen=True)

    cvss_vector: str
    cvss_base_score: float | None
    mitre_techniques: tuple[str, ...]

Person

Bases: BaseModel

A person named in the report, with their stated role (#133).

Source code in src/revalid/extract.py

class Person(BaseModel):
    """A person named in the report, with their stated role (#133)."""

    model_config = ConfigDict(frozen=True)

    name: str
    role: str

ReportMetadata

Bases: BaseModel

Document-level metadata extracted from a report (FR-03, #133).

Every field defaults to empty, so an absent value — or a failed extraction on a small local model — yields a blank the operator can fill in, never a guess.

Source code in src/revalid/extract.py

class ReportMetadata(BaseModel):
    """Document-level metadata extracted from a report (FR-03, #133).

    Every field defaults to empty, so an absent value — or a failed extraction on
    a small local model — yields a blank the operator can fill in, never a guess.
    """

    model_config = ConfigDict(frozen=True)

    product: str = ""
    report_date: str = ""
    author: str = ""
    people: tuple[Person, ...] = ()

apply_taxonomy(finding, taxonomy)

Fill a finding's empty CVSS/ATT&CK from taxonomy, flagged as inferred.

Never overwrites a stated value. A finding that already carries a CVSS vector or ATT&CK techniques — mapped verbatim from a DefectDojo export, or typed by the operator — keeps them exactly, with their existing provenance. Only the empty fields are filled, and what this fills is always inferred=True: it is the model's derivation, never a source's claim.

Parameters:

Name	Type	Description	Default
finding	Finding	The finding to enrich.	required
taxonomy	FindingTaxonomy	The model's derived classification.	required

Returns:

Type	Description
Finding	The finding with previously-empty taxonomy fields filled, or the original
Finding	object unchanged when there was nothing to fill.

Source code in src/revalid/extract.py

def apply_taxonomy(finding: Finding, taxonomy: FindingTaxonomy) -> Finding:
    """Fill a finding's empty CVSS/ATT&CK from ``taxonomy``, flagged as inferred.

    **Never overwrites a stated value.** A finding that already carries a CVSS
    vector or ATT&CK techniques — mapped verbatim from a DefectDojo export, or
    typed by the operator — keeps them exactly, with their existing provenance.
    Only the empty fields are filled, and what this fills is always
    ``inferred=True``: it is the model's derivation, never a source's claim.

    Args:
        finding: The finding to enrich.
        taxonomy: The model's derived classification.

    Returns:
        The finding with previously-empty taxonomy fields filled, or the original
        object unchanged when there was nothing to fill.
    """
    update: dict[str, CvssCode | MitreMapping] = {}
    if not finding.cvss.vector and taxonomy.cvss_vector:
        update["cvss"] = CvssCode(
            vector=taxonomy.cvss_vector, base_score=taxonomy.cvss_base_score, inferred=True
        )
    if not finding.mitre.techniques and taxonomy.mitre_techniques:
        update["mitre"] = MitreMapping(techniques=taxonomy.mitre_techniques, inferred=True)
    return finding.model_copy(update=update) if update else finding

build_extraction_agent(model=None)

Build the finding-extraction agent.

Parameters:

Name	Type	Description	Default
model	Model | KnownModelName | str | None	A Pydantic AI model instance or name. When omitted, the configured backend is used (REVALID_LLM_MODEL, Claude by default — FR-13/ADR-0010); tests pass TestModel/ FunctionModel.	None

Returns:

Type	Description
Agent[None, list[ExtractedFinding]]	An agent whose validated output is a list of :class:ExtractedFinding.

Source code in src/revalid/extract.py

def build_extraction_agent(
    model: Model | KnownModelName | str | None = None,
) -> Agent[None, list[ExtractedFinding]]:
    """Build the finding-extraction agent.

    Args:
        model: A Pydantic AI model instance or name. When omitted, the
            configured backend is used (``REVALID_LLM_MODEL``, Claude by
            default — FR-13/ADR-0010); tests pass ``TestModel``/
            ``FunctionModel``.

    Returns:
        An agent whose validated output is a list of :class:`ExtractedFinding`.
    """
    return Agent(
        model if model is not None else resolve_model(),
        output_type=list[ExtractedFinding],
        instructions=_INSTRUCTIONS,
        model_settings=ModelSettings(max_tokens=_MAX_OUTPUT_TOKENS),
        retries=_MAX_OUTPUT_RETRIES,
        defer_model_check=True,
    )

build_metadata_agent(model=None)

Build the document-metadata extraction agent (FR-03, #133).

Source code in src/revalid/extract.py

def build_metadata_agent(
    model: Model | KnownModelName | str | None = None,
) -> Agent[None, ReportMetadata]:
    """Build the document-metadata extraction agent (FR-03, #133)."""
    return Agent(
        model if model is not None else resolve_model(),
        output_type=ReportMetadata,
        instructions=_METADATA_INSTRUCTIONS,
        retries=_MAX_OUTPUT_RETRIES,
        defer_model_check=True,
    )

build_taxonomy_agent(model=None)

Build the opt-in CVSS/ATT&CK enrichment agent (FR-19, issue #233).

The PDF door gets its taxonomy inside the extraction call itself. The FR-02 JSON and manual doors are deliberately LLM-free, so for them enrichment is a separate, opt-in pass driven by this agent — one call per finding, only when the operator asks for it.

Parameters:

Name	Type	Description	Default
model	Model | KnownModelName | str | None	A Pydantic AI model instance or name. When omitted, the configured backend is used (FR-13/ADR-0010); tests pass TestModel/ FunctionModel.	None

Returns:

Type	Description
Agent[None, FindingTaxonomy]	An agent whose validated output is one :class:FindingTaxonomy.

Source code in src/revalid/extract.py

def build_taxonomy_agent(
    model: Model | KnownModelName | str | None = None,
) -> Agent[None, FindingTaxonomy]:
    """Build the opt-in CVSS/ATT&CK enrichment agent (FR-19, issue #233).

    The PDF door gets its taxonomy inside the extraction call itself. The FR-02
    JSON and manual doors are deliberately LLM-free, so for them enrichment is a
    *separate, opt-in* pass driven by this agent — one call per finding, only when
    the operator asks for it.

    Args:
        model: A Pydantic AI model instance or name. When omitted, the configured
            backend is used (FR-13/ADR-0010); tests pass ``TestModel``/
            ``FunctionModel``.

    Returns:
        An agent whose validated output is one :class:`FindingTaxonomy`.
    """
    return Agent(
        model if model is not None else resolve_model(),
        output_type=FindingTaxonomy,
        instructions=_TAXONOMY_INSTRUCTIONS,
        retries=_MAX_OUTPUT_RETRIES,
        defer_model_check=True,
    )

enrich_findings(agent, findings)

Synchronous wrapper over :func:enrich_findings_async (the request path).

Source code in src/revalid/extract.py

def enrich_findings(
    agent: Agent[None, FindingTaxonomy], findings: Sequence[Finding]
) -> EnrichmentReport:
    """Synchronous wrapper over :func:`enrich_findings_async` (the request path)."""
    return asyncio.run(enrich_findings_async(agent, findings))

enrich_findings_async(agent, findings) async

Derive the missing CVSS/ATT&CK for each finding, one model call apiece.

A finding whose call fails schema validation (after Pydantic AI's retries) is left exactly as it was and counted in failed rather than raising: an import must not be lost because a small local model could not produce a CVSS vector. The count is returned so the caller can surface it — a silently unenriched import would look identical to one the operator never asked to enrich.

Parameters:

Name	Type	Description	Default
agent	Agent[None, FindingTaxonomy]	The agent from :func:build_taxonomy_agent.	required
findings	Sequence[Finding]	The findings to enrich, in order.	required

Returns:

Type	Description
EnrichmentReport	The findings in the same order, plus how many were enriched and how many
EnrichmentReport	model calls failed.

Source code in src/revalid/extract.py

async def enrich_findings_async(
    agent: Agent[None, FindingTaxonomy], findings: Sequence[Finding]
) -> EnrichmentReport:
    """Derive the missing CVSS/ATT&CK for each finding, one model call apiece.

    A finding whose call fails schema validation (after Pydantic AI's retries) is
    **left exactly as it was** and counted in ``failed`` rather than raising: an
    import must not be lost because a small local model could not produce a CVSS
    vector. The count is returned so the caller can surface it — a silently
    unenriched import would look identical to one the operator never asked to
    enrich.

    Args:
        agent: The agent from :func:`build_taxonomy_agent`.
        findings: The findings to enrich, in order.

    Returns:
        The findings in the same order, plus how many were enriched and how many
        model calls failed.
    """
    out: list[Finding] = []
    enriched = 0
    failed = 0
    for finding in findings:
        try:
            result = await agent.run(taxonomy_prompt(finding))
        except UnexpectedModelBehavior:
            failed += 1
            out.append(finding)
            continue
        updated = apply_taxonomy(finding, result.output)
        enriched += updated is not finding
        out.append(updated)
    return EnrichmentReport(findings=tuple(out), enriched=enriched, failed=failed)

extract_metadata(agent, report)

Extract document metadata from a report's opening text (FR-03, #133).

Best-effort and non-fatal: metadata lives near the top of a report, so only the first few thousand characters are sent, and any model or validation failure yields empty metadata the operator can edit — it never blocks the report from becoming ready.

Parameters:

Name	Type	Description	Default
agent	Agent[None, ReportMetadata]	The metadata agent (from :func:build_metadata_agent).	required
report	PdfReport	An extracted report from :func:revalid.pdf.read_pdf.	required

Returns:

Type	Description
ReportMetadata	The extracted (or empty-on-failure) :class:ReportMetadata.

Source code in src/revalid/extract.py

def extract_metadata(agent: Agent[None, ReportMetadata], report: PdfReport) -> ReportMetadata:
    """Extract document metadata from a report's opening text (FR-03, #133).

    Best-effort and non-fatal: metadata lives near the top of a report, so only
    the first few thousand characters are sent, and any model or validation
    failure yields empty metadata the operator can edit — it never blocks the
    report from becoming ready.

    Args:
        agent: The metadata agent (from :func:`build_metadata_agent`).
        report: An extracted report from :func:`revalid.pdf.read_pdf`.

    Returns:
        The extracted (or empty-on-failure) :class:`ReportMetadata`.
    """
    try:
        return agent.run_sync(report.text[:6000]).output
    except Exception:
        return ReportMetadata()

extract_report(agent, report, should_cancel=_never_cancel)

Synchronous wrapper over :func:extract_report_async (tests, offline demos).

The production path (:func:~revalid.app.run_extraction) drives the async form directly on a cancellable loop so a Stop can interrupt it; this wrapper runs it to completion on a throwaway loop for callers that do not need cancellation.

Source code in src/revalid/extract.py

def extract_report(
    agent: Agent[None, list[ExtractedFinding]],
    report: PdfReport,
    should_cancel: Callable[[], bool] = _never_cancel,
) -> ExtractionReport:
    """Synchronous wrapper over :func:`extract_report_async` (tests, offline demos).

    The production path (:func:`~revalid.app.run_extraction`) drives the async form
    directly on a cancellable loop so a Stop can interrupt it; this wrapper runs it
    to completion on a throwaway loop for callers that do not need cancellation.
    """
    return asyncio.run(extract_report_async(agent, report, should_cancel))

extract_report_async(agent, report, should_cancel=_never_cancel) async

Extract structured findings from a whole report in one call (FR-03), async.

Sends the entire report to the model in a single await agent.run — the async path — so the run can be interrupted mid-call (issue #205): cancelling the task cancels the in-flight HTTP request, which is what makes Stop work even when a local model wedges on a long document. should_cancel is checked once up front for the case where a stop was already requested before the call began. A cancel yields an empty, cancelled=True result. Valid output is mapped to domain findings; output that never passes the schema gate is flagged as a single failure instead of persisted.

Parameters:

Name	Type	Description	Default
agent	Agent[None, list[ExtractedFinding]]	The extraction agent (from :func:build_extraction_agent).	required
report	PdfReport	An extracted report from :func:revalid.pdf.read_pdf.	required
should_cancel	Callable[[], bool]	Returns True when the operator has asked to stop; checked before the call begins. The task may also be cancelled mid-call.	_never_cancel

Returns:

Type	Description
ExtractionReport	The valid findings and any flagged failure, with cancelled set when the
ExtractionReport	run stopped early.

Source code in src/revalid/extract.py

async def extract_report_async(
    agent: Agent[None, list[ExtractedFinding]],
    report: PdfReport,
    should_cancel: Callable[[], bool] = _never_cancel,
) -> ExtractionReport:
    """Extract structured findings from a whole report in one call (FR-03), async.

    Sends the entire report to the model in a single ``await agent.run`` — the
    async path — so the run can be *interrupted* mid-call (issue #205): cancelling
    the task cancels the in-flight HTTP request, which is what makes Stop work even
    when a local model wedges on a long document. ``should_cancel`` is checked once
    up front for the case where a stop was already requested before the call began.
    A cancel yields an empty, ``cancelled=True`` result. Valid output is mapped to
    domain findings; output that never passes the schema gate is flagged as a
    single failure instead of persisted.

    Args:
        agent: The extraction agent (from :func:`build_extraction_agent`).
        report: An extracted report from :func:`revalid.pdf.read_pdf`.
        should_cancel: Returns ``True`` when the operator has asked to stop; checked
            before the call begins. The task may also be cancelled mid-call.

    Returns:
        The valid findings and any flagged failure, with ``cancelled`` set when the
        run stopped early.
    """
    model_name = agent_model_name(agent)
    if should_cancel():
        return ExtractionReport(findings=(), failures=(), cancelled=True)
    try:
        result = await agent.run(report.text)
    except asyncio.CancelledError:
        # The operator interrupted the model call (Stop / delete): report the stop.
        return ExtractionReport(findings=(), failures=(), cancelled=True)
    except UnexpectedModelBehavior as exc:
        failure = ExtractionFailure(error=str(exc), source_text=report.text)
        return ExtractionReport(findings=(), failures=(failure,))
    findings = tuple(_to_finding(item, model_name) for item in result.output)
    return ExtractionReport(findings=findings, failures=())

taxonomy_prompt(finding)

Render the finding as the prompt the taxonomy agent classifies.

Source code in src/revalid/extract.py

def taxonomy_prompt(finding: Finding) -> str:
    """Render the finding as the prompt the taxonomy agent classifies."""
    return (
        f"Title: {finding.title}\n"
        f"Severity: {finding.severity.value}\n"
        f"Description: {finding.description}\n"
        f"Impact: {finding.impact}\n"
        f"Attack vector: {finding.attack_vector}\n"
        f"Affected endpoints: {', '.join(finding.affected_endpoints)}"
    )

Finding revision & annotation: versioned findings + stage-tagged notes (FR-16, ADR-0024).

A finding is a stable identity (:class:~revalid.db.FindingRecord) plus append-only immutable version rows (:class:~revalid.db.FindingVersionRecord): extraction is version 1, each operator edit a new version — symmetric with the FR-05 plan model (ADR-0012). Plans and verdicts reference the stable identity, so amending a finding never orphans them. Notes (:class:~revalid.db.FindingNoteRecord) are a per-finding, stage-tagged, append-only log. This module owns those lifecycles; nothing mutates a version or a note in place.

add_note(session, finding_id, stage, body, *, author=_ACTOR)

Append a stage-tagged note to the finding's log (append-only).

Source code in src/revalid/findings.py

def add_note(
    session: Session,
    finding_id: int,
    stage: FindingStage,
    body: str,
    *,
    author: str = _ACTOR,
) -> FindingNoteRecord:
    """Append a stage-tagged note to the finding's log (append-only)."""
    record = FindingNoteRecord(finding_id=finding_id, stage=stage.value, body=body, author=author)
    session.add(record)
    session.commit()
    session.refresh(record)
    return record

add_version(session, finding_id, finding, *, edited_by=_ACTOR, reason='')

Append an operator edit as a new immutable version (origin=edit).

Never mutates a prior version — the earlier content stays in history as the correction record (FR-10). The appended version becomes the current one.

Source code in src/revalid/findings.py

def add_version(
    session: Session,
    finding_id: int,
    finding: Finding,
    *,
    edited_by: str = _ACTOR,
    reason: str = "",
) -> FindingVersionRecord:
    """Append an operator edit as a new immutable version (``origin=edit``).

    Never mutates a prior version — the earlier content stays in history as the
    correction record (FR-10). The appended version becomes the current one.
    """
    record = FindingVersionRecord.from_domain(
        finding_id,
        finding,
        version=_next_version(session, finding_id),
        origin=FindingOrigin.EDIT,
        edited_by=edited_by,
        reason=reason,
    )
    session.add(record)
    session.commit()
    session.refresh(record)
    return record

create_finding(session, finding, report_id=None)

Create a finding identity and its version-1 content (origin=extraction).

The single entry point for a newly ingested finding — extraction, FR-02 import, and manual entry all land here. Flushes so the version row can reference the new identity id, but leaves the commit to the caller (findings are created in a batch alongside their report).

Source code in src/revalid/findings.py

def create_finding(
    session: Session, finding: Finding, report_id: int | None = None
) -> FindingRecord:
    """Create a finding identity and its version-1 content (``origin=extraction``).

    The single entry point for a newly ingested finding — extraction, FR-02
    import, and manual entry all land here. Flushes so the version row can
    reference the new identity id, but leaves the ``commit`` to the caller
    (findings are created in a batch alongside their report).
    """
    record = FindingRecord(report_id=report_id)
    session.add(record)
    session.flush()
    session.add(
        FindingVersionRecord.from_domain(
            record.id, finding, version=1, origin=FindingOrigin.EXTRACTION
        )
    )
    return record

current_version(session, finding_id)

Return the finding's current (highest-version) content, or None.

Source code in src/revalid/findings.py

def current_version(session: Session, finding_id: int) -> FindingVersionRecord | None:
    """Return the finding's current (highest-version) content, or ``None``."""
    return session.scalars(
        select(FindingVersionRecord)
        .where(FindingVersionRecord.finding_id == finding_id)
        .order_by(FindingVersionRecord.version.desc())
    ).first()

list_notes(session, finding_id)

Return the finding's notes, newest first.

Source code in src/revalid/findings.py

def list_notes(session: Session, finding_id: int) -> list[FindingNoteRecord]:
    """Return the finding's notes, newest first."""
    return list(
        session.scalars(
            select(FindingNoteRecord)
            .where(FindingNoteRecord.finding_id == finding_id)
            .order_by(FindingNoteRecord.created_at.desc(), FindingNoteRecord.id.desc())
        )
    )

list_versions(session, finding_id)

Return every version of a finding, oldest first (extraction = v1).

Source code in src/revalid/findings.py

def list_versions(session: Session, finding_id: int) -> list[FindingVersionRecord]:
    """Return every version of a finding, oldest first (extraction = v1)."""
    return list(
        session.scalars(
            select(FindingVersionRecord)
            .where(FindingVersionRecord.finding_id == finding_id)
            .order_by(FindingVersionRecord.version)
        )
    )

Retest goal and agentic session

plan generates the retest goal (FR-04, repurposed by ADR-0032); sandbox provides the egress-locked execution environment and scope parses the host it is provisioned against; retest_agent is the Pydantic AI agent and its two tools; retest_session is the orchestrator that owns the lifecycle, the transcript and the approval gate; deltas is the transient reasoning-token channel that deliberately never reaches that transcript.

Retest-goal generation from findings (FR-04, repurposed — ADR-0032).

FR-04's plan generator was repurposed (FR-17 6b-ii) from producing HTTP probes into producing a retest goal: a few concise, tool-agnostic verification steps the agentic console's agent works to. A Pydantic AI agent proposes a :class:GeneratedGoal from the finding (ADR-0009 schema-gate pattern; model chosen by REVALID_LLM_MODEL — FR-13). The old batch probe-plan path was removed with the batch execution in FR-17 6b-iii.

GeneratedGoal

Bases: BaseModel

A short, tool-agnostic retest goal — the steps the agent works to (FR-17 6b-ii).

A few concise natural-language verification steps for any finding, not HTTP probes (the batch probe plan retired with the batch path in FR-17 6b-iii).

Source code in src/revalid/plan.py

class GeneratedGoal(BaseModel):
    """A short, tool-agnostic retest goal — the steps the agent works to (FR-17 6b-ii).

    A few concise natural-language verification steps for any finding, not HTTP
    probes (the batch probe plan retired with the batch path in FR-17 6b-iii).
    """

    model_config = ConfigDict(frozen=True)

    steps: tuple[str, ...] = Field(default=(), max_length=6)

build_goal_agent(model=None)

Build the retest-goal agent (FR-17 6b-ii): a generic, tool-agnostic goal generator.

Parameters:

Name	Type	Description	Default
model	Model | KnownModelName | str | None	A Pydantic AI model instance or name. When omitted, the configured backend is used (REVALID_LLM_MODEL, FR-13); tests pass a stand-in.	None

Returns:

Type	Description
Agent[None, GeneratedGoal]	An agent whose validated output is a :class:GeneratedGoal.

Source code in src/revalid/plan.py

def build_goal_agent(
    model: Model | KnownModelName | str | None = None,
) -> Agent[None, GeneratedGoal]:
    """Build the retest-goal agent (FR-17 6b-ii): a generic, tool-agnostic goal generator.

    Args:
        model: A Pydantic AI model instance or name. When omitted, the configured
            backend is used (``REVALID_LLM_MODEL``, FR-13); tests pass a stand-in.

    Returns:
        An agent whose validated output is a :class:`GeneratedGoal`.
    """
    return Agent(
        model if model is not None else resolve_model(),
        output_type=GeneratedGoal,
        instructions=_GOAL_INSTRUCTIONS,
        retries=_MAX_OUTPUT_RETRIES,
        defer_model_check=True,
    )

finding_prompt(finding)

Render a finding as the model-facing context both retest prompts are built from.

One renderer, deliberately (issue #249): this used to exist twice — here and in app.py — and the copies had drifted, so the goal generator saw the finding's severity and attack vector while the retest agent, which acts on it, did not.

Every field is omitted when the finding does not state it, rather than rendered as a placeholder: a report ingested through the JSON or manual door often carries only a title and steps, and telling the model "Attack vector:" followed by nothing invites it to invent one.

Parameters:

Name	Type	Description	Default
finding	Finding	The finding to describe.	required

Returns:

Type	Description
str	The finding's identity, classification and original reproduction steps as plain
str	text — the shared basis for goal generation (FR-04) and for the retest agent's
str	opening prompt (FR-17).

Source code in src/revalid/plan.py

def finding_prompt(finding: Finding) -> str:
    """Render a finding as the model-facing context both retest prompts are built from.

    One renderer, deliberately (issue #249): this used to exist twice — here and in
    ``app.py`` — and the copies had drifted, so the goal generator saw the finding's
    severity and attack vector while the retest agent, which acts on it, did not.

    Every field is omitted when the finding does not state it, rather than rendered as
    a placeholder: a report ingested through the JSON or manual door often carries only
    a title and steps, and telling the model "Attack vector:" followed by nothing invites
    it to invent one.

    Args:
        finding: The finding to describe.

    Returns:
        The finding's identity, classification and original reproduction steps as plain
        text — the shared basis for goal generation (FR-04) and for the retest agent's
        opening prompt (FR-17).
    """
    lines = [f"Title: {finding.title}", f"Severity: {finding.severity.value}"]
    if finding.description:
        lines.append(f"Description: {finding.description}")
    if finding.attack_vector:
        lines.append(f"Attack vector: {finding.attack_vector}")
    if finding.affected_endpoints:
        lines.append("Affected endpoints: " + ", ".join(finding.affected_endpoints))
    if finding.reproduction_steps:
        steps = "\n".join(f"{i}. {s}" for i, s in enumerate(finding.reproduction_steps, 1))
        lines.append(f"Reproduction steps:\n{steps}")
    return "\n".join(lines)

generate_goal(agent, finding)

Generate a generic retest goal for finding (FR-17 6b-ii).

Best-effort: on a model failure it returns an empty tuple so session start never blocks — the agent then falls back to the finding context alone.

Parameters:

Name	Type	Description	Default
agent	Agent[None, GeneratedGoal]	The goal agent (from :func:build_goal_agent).	required
finding	Finding	The finding to derive a retest goal for.	required

Returns:

Type	Description
tuple[str, ...]	The generated goal steps, or () if the model produced nothing usable.

Source code in src/revalid/plan.py

def generate_goal(agent: Agent[None, GeneratedGoal], finding: Finding) -> tuple[str, ...]:
    """Generate a generic retest goal for ``finding`` (FR-17 6b-ii).

    Best-effort: on a model failure it returns an empty tuple so session start
    never blocks — the agent then falls back to the finding context alone.

    Args:
        agent: The goal agent (from :func:`build_goal_agent`).
        finding: The finding to derive a retest goal for.

    Returns:
        The generated goal steps, or ``()`` if the model produced nothing usable.
    """
    try:
        return agent.run_sync(finding_prompt(finding)).output.steps
    except UnexpectedModelBehavior:
        return ()

Retest scope parsing (FR-17 / FR-06, issue #208).

The retest scope is written by the operator at launch as one or more target endpoints (or defaulted from the finding's affected endpoints). The sandbox is provisioned against the host of that scope, not the specific page: a finding about https://domain.com/#/login retests domain.com at any path, so the agent can follow the vulnerability wherever it lives under that host — not only the one URL in the report.

This module holds the pure parsing: an endpoint string in, its host (host or host:port) out. Provisioning (lab vs. online egress) lives in the sandbox.

scope_host(endpoint)

Parse one scope endpoint down to its host (host or host:port).

Keeps the port (it is part of the reachable target — the lab is localhost:3000) but drops scheme, userinfo, path, query and fragment, including SPA hash routes. Sub-domains are preserved (they are distinct hosts); only the path is stripped.

Parameters:

Name	Type	Description	Default
endpoint	str	A target string — a full URL, a scheme-less host/path or a bare host:port. SPA hash routes (domain.com/#/login) are handled.	required

Returns:

Type	Description
str | None	The lower-cased host (or host:port), or None when the string
str | None	carries no parseable host.

Examples:

>>> scope_host("https://domain.com/#/login")
'domain.com'
>>> scope_host("http://domain.com:8080/a/b?x=1")
'domain.com:8080'
>>> scope_host("domain.com/login")
'domain.com'
>>> scope_host("http://localhost:3000/rest/user/login")
'localhost:3000'
>>> scope_host("   ") is None
True

Source code in src/revalid/scope.py

def scope_host(endpoint: str) -> str | None:
    """Parse one scope endpoint down to its host (``host`` or ``host:port``).

    Keeps the port (it is part of the reachable target — the lab is
    ``localhost:3000``) but drops scheme, userinfo, path, query and fragment,
    including SPA hash routes. Sub-domains are preserved (they are distinct
    hosts); only the path is stripped.

    Args:
        endpoint: A target string — a full URL, a scheme-less ``host/path`` or a
            bare ``host:port``. SPA hash routes (``domain.com/#/login``) are
            handled.

    Returns:
        The lower-cased ``host`` (or ``host:port``), or ``None`` when the string
        carries no parseable host.

    Examples:
        >>> scope_host("https://domain.com/#/login")
        'domain.com'
        >>> scope_host("http://domain.com:8080/a/b?x=1")
        'domain.com:8080'
        >>> scope_host("domain.com/login")
        'domain.com'
        >>> scope_host("http://localhost:3000/rest/user/login")
        'localhost:3000'
        >>> scope_host("   ") is None
        True
    """
    raw = endpoint.strip()
    if not raw:
        return None
    # A scheme-less input ("domain.com/login") parses with an empty netloc, so
    # force the authority form; a real scheme already yields the netloc.
    if "://" not in raw:
        raw = "//" + raw.lstrip("/")
    netloc = urlsplit(raw).netloc
    # Strip any userinfo ("user:pass@host") — the authority is host[:port].
    host = netloc.rsplit("@", 1)[-1].strip()
    return host.lower() or None

scope_hosts(endpoints)

Parse a scope's endpoints to a de-duplicated, order-preserving host tuple.

Parameters:

Name	Type	Description	Default
endpoints	tuple[str, ...]	The launch-time scope endpoints (target_set).	required

Returns:

Type	Description
str	Each distinct parseable host, first-seen order preserved. Unparseable
...	entries are dropped.

Source code in src/revalid/scope.py

def scope_hosts(endpoints: tuple[str, ...]) -> tuple[str, ...]:
    """Parse a scope's endpoints to a de-duplicated, order-preserving host tuple.

    Args:
        endpoints: The launch-time scope endpoints (``target_set``).

    Returns:
        Each distinct parseable host, first-seen order preserved. Unparseable
        entries are dropped.
    """
    seen: dict[str, None] = {}
    for endpoint in endpoints:
        host = scope_host(endpoint)
        if host is not None and host not in seen:
            seen[host] = None
    return tuple(seen)

FR-17 / M6 egress-locked retest sandbox (ADR-0025, Slice 0).

An ephemeral Docker container in which the retest agent runs one approved command at a time, provisioned into one of two topologies according to the session's scope (ADR-0041): a lab target gets an --internal network with the target attached as its only other member (no host/internet route at all), while an online target gets a per-session L3 egress gateway whose network namespace the sandbox joins and whose iptables allowlist it holds no capability to change (ADR-0045). Either way containment is topology, never command inspection.

The pure surface (CommandResult, FakeSandbox, the mode/resolution/firewall helpers) is unit-tested; the live DockerSandbox needs the optional sandbox extra and is covered only by the nightly system test.

CommandResult

Bases: BaseModel

The captured result of one command run in the sandbox.

Source code in src/revalid/sandbox.py

class CommandResult(BaseModel):
    """The captured result of one command run in the sandbox."""

    model_config = ConfigDict(frozen=True)

    stdout: str
    stderr: str
    exit_code: int
    elapsed_ms: int

DockerSandbox

A real ephemeral Docker sandbox on an egress-locked --internal network.

Source code in src/revalid/sandbox.py

class DockerSandbox:  # pragma: no cover - drives a live Docker daemon; covered by the system test
    """A real ephemeral Docker sandbox on an egress-locked ``--internal`` network."""

    def __init__(
        self,
        session_id: int,
        *,
        image: str | None = None,
        lab_container: str = DEFAULT_LAB_CONTAINER,
    ) -> None:
        """Bind this sandbox to ``session_id`` (scopes its per-session resource names)."""
        self._session_id = session_id
        self._image = image if image is not None else sandbox_image()
        self._lab_container = lab_container
        self._container: Container | None = None
        self._gateway: Container | None = None
        self._network_name = internal_network_name(session_id)
        self._sandbox_name = sandbox_container_name(session_id)
        self._gateway_name = gateway_container_name(session_id)

    def start(self, scope_hosts: tuple[str, ...] = ()) -> None:
        """Provision the sandbox for this session's scope (ADR-0025/0041/0045).

        Lab scope (empty, or every host is the lab host) keeps the unchanged
        ``--internal`` network with the lab container attached — the target is a
        named neighbour, reachable by construction and nothing else. Any other host
        is an online target: provision a per-session **egress gateway** that holds an
        L3 iptables allowlist for the scoped IP(s), and run the sandbox inside the
        gateway's network namespace so every tool (not just HTTP) can reach the
        scoped host and nothing else — and cannot alter that, because it holds no
        ``NET_ADMIN`` (ADR-0045).
        """
        try:
            import docker
        except ImportError as exc:
            raise SandboxUnavailableError(
                "the sandbox extra is required: `uv sync --extra sandbox`"
            ) from exc
        client = docker.from_env()
        self._require_image(client)
        self._clear_stale(client)
        if is_lab_scope(scope_hosts):
            self._start_lab(client)
        else:
            self._start_online(client, online_scope_hosts(scope_hosts))

    def _start_lab(self, client: docker.DockerClient) -> None:
        """Lab provisioning (unchanged): internal network + attached lab container."""
        network = client.networks.create(self._network_name, driver="bridge", internal=True)
        network.connect(self._lab_container)  # allowlist == network membership (FR-06)
        self._container = client.containers.run(
            self._image,
            name=self._sandbox_name,
            command="sleep infinity",
            network=self._network_name,
            detach=True,
            auto_remove=False,
            network_disabled=False,
        )

    def _start_online(self, client: docker.DockerClient, hosts: tuple[str, ...]) -> None:
        """Online provisioning (ADR-0045): a per-session L3 egress gateway, fail-closed.

        Resolve the scope host(s) to their IP(s), then provision two containers on
        a per-session bridge network:

        - a **gateway** that holds ``NET_ADMIN`` and installs an iptables OUTPUT
          allowlist permitting only the scoped IPs and one DNS resolver — default
          drop otherwise — then blocks to keep its network namespace alive;
        - the **sandbox** itself, run *inside the gateway's network namespace*
          (``network_mode=container:<gateway>``) with ``NET_RAW`` but **not**
          ``NET_ADMIN``. So the sandbox's every packet is filtered by the gateway's
          rules, all tools work (raw sockets, ICMP, any port on the scoped host),
          and no command it runs can change egress — the capability lives in a
          different container it cannot reach.

        The scoped host is pinned in the shared ``/etc/hosts`` (via the gateway's
        ``extra_hosts``) so name resolution needs no network; the one allowed
        resolver covers tools that resolve independently (nmap). Any provisioning
        failure tears everything down and raises — never a half-open route.
        """
        import docker.errors

        try:
            resolved = resolve_scope_ips(hosts)
            scope_ips = tuple(ip for ips in resolved.values() for ip in ips)
            if not scope_ips:
                raise SandboxUnavailableError(
                    f"could not resolve any scope host to an IP: {', '.join(hosts)}"
                )
            dns_ip = dns_resolver()
            client.networks.create(self._network_name, driver="bridge")
            self._gateway = client.containers.run(
                self._image,
                name=self._gateway_name,
                # `entrypoint`, not `command`: the run-command path must own PID 1
                # so its `exec sleep infinity` keeps the netns alive; the script is
                # a plain multi-line `sh -c` argument (docker-py passes argv, so no
                # shell requoting) and interpolates only numeric IPs — no injection.
                entrypoint=["sh", "-c", egress_firewall_script(scope_ips, dns_ip)],
                network=self._network_name,
                cap_add=["NET_ADMIN"],
                dns=[dns_ip],
                # Pin the scope host(s) so glibc-based tools resolve with zero DNS;
                # this /etc/hosts is shared into the sandbox via the netns join.
                extra_hosts={host: ips[0] for host, ips in resolved.items() if ips},
                detach=True,
                auto_remove=False,
            )
            self._container = client.containers.run(
                self._image,
                name=self._sandbox_name,
                command="sleep infinity",
                # Share the gateway's network namespace: the sandbox has no network
                # of its own, so the gateway's OUTPUT allowlist is the sandbox's
                # only route out. `network_mode=container` forbids network/dns/
                # extra_hosts kwargs — they belong to the namespace owner above.
                network_mode=f"container:{self._gateway_name}",
                cap_add=["NET_RAW"],  # SYN scans etc.; NOT NET_ADMIN (can't edit rules)
                detach=True,
                auto_remove=False,
            )
        except docker.errors.APIError as exc:
            self.stop()  # fail closed: never leave a half-provisioned open route
            raise SandboxUnavailableError(
                f"online egress gateway provisioning failed: {exc}"
            ) from exc

    def _require_image(self, client: docker.DockerClient) -> None:
        """Fail early and actionably when the toolbox image has not been built.

        The sandbox image is built locally (``make sandbox-image``), not pulled,
        so a fresh clone has no copy of it. Checking here turns a Docker
        ``ImageNotFound`` raised mid-launch into a message naming the command
        that fixes it.

        Deliberately not falling back to a smaller image: the agent would then
        silently lose nmap, sqlmap and the rest, and a retest that concludes
        ``fixed`` because its tool was missing is precisely the confidently-wrong
        verdict this project is built to avoid.
        """
        import docker.errors

        try:
            client.images.get(self._image)
        except docker.errors.ImageNotFound as exc:
            raise SandboxUnavailableError(
                f"sandbox image {self._image!r} is not built — run `make sandbox-image` "
                f"(or set ${SANDBOX_IMAGE_ENV} to an image you already have)"
            ) from exc

    def _clear_stale(self, client: docker.DockerClient) -> None:
        """Reap this session's leftover containers + network from a crashed prior run.

        ``start()`` is not re-entrant across a crash: a session killed before it
        reaches ``stop()`` (process kill, daemon restart) leaves its sandbox
        container, gateway container and/or network behind. All three are
        session-scoped by *name*, so retrying ``start()`` for the same
        ``session_id`` would hit a 409 name conflict forever. Self-heal by tearing
        down the same-named resources first — exactly the by-name teardown
        ``stop()`` performs.

        Safety assumption (ADR-0008, single trusted user): this targets leftovers
        from a *prior, crashed* run of the *same* ``session_id``. That is safe only
        because ``session_id`` is a unique DB row id and sessions run sequentially
        under the single-user model; two sandboxes sharing a ``session_id`` must
        never run concurrently (e.g. the system test's fixed sentinel id).
        """
        _teardown_by_name(
            client,
            self._network_name,
            (self._sandbox_name, self._gateway_name),
            self._lab_container,
        )

    def exec(self, command: str, *, timeout: float) -> CommandResult:
        """Run ``command`` inside the live container, capped at ``timeout`` seconds.

        Docker's ``exec_run`` is blocking (the model always sees the *complete*
        output before its next turn) but has **no native timeout**, so the cap is
        enforced in-container by wrapping the command with ``timeout`` — present on
        both the alpine/busybox base of the default image and a coreutils-based
        Kali image (#105). A command that overruns is killed and exits non-zero
        (124 on coreutils, 143/SIGTERM on busybox); ``run_command`` surfaces that to
        the agent, which chose the limit and can retry with a narrower scope. Without
        this, a hanging or unbounded command (e.g. an nmap sweep, or one blocked on
        stdin) would wedge the session at ``working`` forever.
        """
        import time

        if self._container is None:
            raise SandboxUnavailableError("sandbox not started")
        start = time.monotonic()
        wrapped = ["timeout", str(int(timeout)), "sh", "-c", command]
        code, output = self._container.exec_run(wrapped, demux=True)
        elapsed_ms = int((time.monotonic() - start) * 1000)
        stdout, stderr = output
        return CommandResult(
            stdout=(stdout or b"").decode(errors="replace"),
            stderr=(stderr or b"").decode(errors="replace"),
            exit_code=code,
            elapsed_ms=elapsed_ms,
        )

    def stop(self) -> None:
        """Tear down this session's containers and network by name (best-effort).

        Removal is keyed on the session-scoped resource *names*, not on the
        object's held references, so a freshly constructed ``DockerSandbox`` can
        reap resources it never created — the case that matters when a report is
        deleted after a backend restart has forgotten the live session. Every step
        tolerates "already gone".
        """
        import docker

        self._container = None
        self._gateway = None
        _teardown_by_name(
            docker.from_env(),
            self._network_name,
            (self._sandbox_name, self._gateway_name),
            self._lab_container,
        )

__init__(session_id, *, image=None, lab_container=DEFAULT_LAB_CONTAINER)

Bind this sandbox to session_id (scopes its per-session resource names).

Source code in src/revalid/sandbox.py

def __init__(
    self,
    session_id: int,
    *,
    image: str | None = None,
    lab_container: str = DEFAULT_LAB_CONTAINER,
) -> None:
    """Bind this sandbox to ``session_id`` (scopes its per-session resource names)."""
    self._session_id = session_id
    self._image = image if image is not None else sandbox_image()
    self._lab_container = lab_container
    self._container: Container | None = None
    self._gateway: Container | None = None
    self._network_name = internal_network_name(session_id)
    self._sandbox_name = sandbox_container_name(session_id)
    self._gateway_name = gateway_container_name(session_id)

exec(command, *, timeout)

Run command inside the live container, capped at timeout seconds.

Docker's exec_run is blocking (the model always sees the complete output before its next turn) but has no native timeout, so the cap is enforced in-container by wrapping the command with timeout — present on both the alpine/busybox base of the default image and a coreutils-based Kali image (#105). A command that overruns is killed and exits non-zero (124 on coreutils, 143/SIGTERM on busybox); run_command surfaces that to the agent, which chose the limit and can retry with a narrower scope. Without this, a hanging or unbounded command (e.g. an nmap sweep, or one blocked on stdin) would wedge the session at working forever.

Source code in src/revalid/sandbox.py

def exec(self, command: str, *, timeout: float) -> CommandResult:
    """Run ``command`` inside the live container, capped at ``timeout`` seconds.

    Docker's ``exec_run`` is blocking (the model always sees the *complete*
    output before its next turn) but has **no native timeout**, so the cap is
    enforced in-container by wrapping the command with ``timeout`` — present on
    both the alpine/busybox base of the default image and a coreutils-based
    Kali image (#105). A command that overruns is killed and exits non-zero
    (124 on coreutils, 143/SIGTERM on busybox); ``run_command`` surfaces that to
    the agent, which chose the limit and can retry with a narrower scope. Without
    this, a hanging or unbounded command (e.g. an nmap sweep, or one blocked on
    stdin) would wedge the session at ``working`` forever.
    """
    import time

    if self._container is None:
        raise SandboxUnavailableError("sandbox not started")
    start = time.monotonic()
    wrapped = ["timeout", str(int(timeout)), "sh", "-c", command]
    code, output = self._container.exec_run(wrapped, demux=True)
    elapsed_ms = int((time.monotonic() - start) * 1000)
    stdout, stderr = output
    return CommandResult(
        stdout=(stdout or b"").decode(errors="replace"),
        stderr=(stderr or b"").decode(errors="replace"),
        exit_code=code,
        elapsed_ms=elapsed_ms,
    )

start(scope_hosts=())

Provision the sandbox for this session's scope (ADR-0025/0041/0045).

Lab scope (empty, or every host is the lab host) keeps the unchanged --internal network with the lab container attached — the target is a named neighbour, reachable by construction and nothing else. Any other host is an online target: provision a per-session egress gateway that holds an L3 iptables allowlist for the scoped IP(s), and run the sandbox inside the gateway's network namespace so every tool (not just HTTP) can reach the scoped host and nothing else — and cannot alter that, because it holds no NET_ADMIN (ADR-0045).

Source code in src/revalid/sandbox.py

def start(self, scope_hosts: tuple[str, ...] = ()) -> None:
    """Provision the sandbox for this session's scope (ADR-0025/0041/0045).

    Lab scope (empty, or every host is the lab host) keeps the unchanged
    ``--internal`` network with the lab container attached — the target is a
    named neighbour, reachable by construction and nothing else. Any other host
    is an online target: provision a per-session **egress gateway** that holds an
    L3 iptables allowlist for the scoped IP(s), and run the sandbox inside the
    gateway's network namespace so every tool (not just HTTP) can reach the
    scoped host and nothing else — and cannot alter that, because it holds no
    ``NET_ADMIN`` (ADR-0045).
    """
    try:
        import docker
    except ImportError as exc:
        raise SandboxUnavailableError(
            "the sandbox extra is required: `uv sync --extra sandbox`"
        ) from exc
    client = docker.from_env()
    self._require_image(client)
    self._clear_stale(client)
    if is_lab_scope(scope_hosts):
        self._start_lab(client)
    else:
        self._start_online(client, online_scope_hosts(scope_hosts))

stop()

Tear down this session's containers and network by name (best-effort).

Removal is keyed on the session-scoped resource names, not on the object's held references, so a freshly constructed DockerSandbox can reap resources it never created — the case that matters when a report is deleted after a backend restart has forgotten the live session. Every step tolerates "already gone".

Source code in src/revalid/sandbox.py

def stop(self) -> None:
    """Tear down this session's containers and network by name (best-effort).

    Removal is keyed on the session-scoped resource *names*, not on the
    object's held references, so a freshly constructed ``DockerSandbox`` can
    reap resources it never created — the case that matters when a report is
    deleted after a backend restart has forgotten the live session. Every step
    tolerates "already gone".
    """
    import docker

    self._container = None
    self._gateway = None
    _teardown_by_name(
        docker.from_env(),
        self._network_name,
        (self._sandbox_name, self._gateway_name),
        self._lab_container,
    )

FakeSandbox

A scripted in-memory sandbox for unit/integration tests (no Docker).

Source code in src/revalid/sandbox.py

class FakeSandbox:
    """A scripted in-memory sandbox for unit/integration tests (no Docker)."""

    def __init__(self, script: list[CommandResult] | Callable[[str], CommandResult]) -> None:
        """Store the scripted results (or callable) to replay in :meth:`exec`."""
        self._script = script
        self.commands: list[str] = []
        #: The per-command timeouts passed to :meth:`exec`, in order — lets tests
        #: assert the agent-chosen ``timeout_seconds`` reaches the sandbox (#150).
        self.timeouts: list[float] = []
        self.started = False
        self.stopped = False
        #: The scope hosts the orchestrator provisioned with — lets tests assert the
        #: parsed finding scope reaches the sandbox (ADR-0041).
        self.scope_hosts: tuple[str, ...] = ()

    def start(self, scope_hosts: tuple[str, ...] = ()) -> None:
        """Mark the fake as started, recording the scope it was provisioned for."""
        self.started = True
        self.scope_hosts = scope_hosts

    def exec(self, command: str, *, timeout: float) -> CommandResult:
        """Return the next scripted result (or apply the callable)."""
        self.commands.append(command)
        self.timeouts.append(timeout)
        if callable(self._script):
            return self._script(command)
        if not self._script:
            raise SandboxUnavailableError("FakeSandbox script exhausted")
        return self._script.pop(0)

    def stop(self) -> None:
        """Mark the fake as stopped."""
        self.stopped = True

__init__(script)

Store the scripted results (or callable) to replay in :meth:exec.

Source code in src/revalid/sandbox.py

def __init__(self, script: list[CommandResult] | Callable[[str], CommandResult]) -> None:
    """Store the scripted results (or callable) to replay in :meth:`exec`."""
    self._script = script
    self.commands: list[str] = []
    #: The per-command timeouts passed to :meth:`exec`, in order — lets tests
    #: assert the agent-chosen ``timeout_seconds`` reaches the sandbox (#150).
    self.timeouts: list[float] = []
    self.started = False
    self.stopped = False
    #: The scope hosts the orchestrator provisioned with — lets tests assert the
    #: parsed finding scope reaches the sandbox (ADR-0041).
    self.scope_hosts: tuple[str, ...] = ()

exec(command, *, timeout)

Return the next scripted result (or apply the callable).

Source code in src/revalid/sandbox.py

def exec(self, command: str, *, timeout: float) -> CommandResult:
    """Return the next scripted result (or apply the callable)."""
    self.commands.append(command)
    self.timeouts.append(timeout)
    if callable(self._script):
        return self._script(command)
    if not self._script:
        raise SandboxUnavailableError("FakeSandbox script exhausted")
    return self._script.pop(0)

start(scope_hosts=())

Mark the fake as started, recording the scope it was provisioned for.

Source code in src/revalid/sandbox.py

def start(self, scope_hosts: tuple[str, ...] = ()) -> None:
    """Mark the fake as started, recording the scope it was provisioned for."""
    self.started = True
    self.scope_hosts = scope_hosts

stop()

Mark the fake as stopped.

Source code in src/revalid/sandbox.py

def stop(self) -> None:
    """Mark the fake as stopped."""
    self.stopped = True

Sandbox

Bases: Protocol

One ephemeral, egress-locked execution environment for a retest session.

Source code in src/revalid/sandbox.py

class Sandbox(Protocol):
    """One ephemeral, egress-locked execution environment for a retest session."""

    def start(self, scope_hosts: tuple[str, ...] = ()) -> None:
        """Provision the environment for ``scope_hosts`` (idempotent).

        ``scope_hosts`` is the session's parsed scope (ADR-0041): empty or the lab
        host keeps lab provisioning; any other host provisions egress to it.
        """

    def exec(self, command: str, *, timeout: float) -> CommandResult:
        """Run ``command`` and capture its result."""

    def stop(self) -> None:
        """Tear the environment down; nothing persists."""

exec(command, *, timeout)

Run command and capture its result.

Source code in src/revalid/sandbox.py

def exec(self, command: str, *, timeout: float) -> CommandResult:
    """Run ``command`` and capture its result."""

start(scope_hosts=())

Provision the environment for scope_hosts (idempotent).

scope_hosts is the session's parsed scope (ADR-0041): empty or the lab host keeps lab provisioning; any other host provisions egress to it.

Source code in src/revalid/sandbox.py

def start(self, scope_hosts: tuple[str, ...] = ()) -> None:
    """Provision the environment for ``scope_hosts`` (idempotent).

    ``scope_hosts`` is the session's parsed scope (ADR-0041): empty or the lab
    host keeps lab provisioning; any other host provisions egress to it.
    """

stop()

Tear the environment down; nothing persists.

Source code in src/revalid/sandbox.py

def stop(self) -> None:
    """Tear the environment down; nothing persists."""

SandboxUnavailableError

Bases: Exception

Raised when a sandbox is required but the runtime cannot provide one.

Source code in src/revalid/sandbox.py

class SandboxUnavailableError(Exception):
    """Raised when a sandbox is required but the runtime cannot provide one."""

dns_resolver()

Return the allowed DNS resolver ($REVALID_DNS_RESOLVER or the default).

Source code in src/revalid/sandbox.py

def dns_resolver() -> str:
    """Return the allowed DNS resolver (``$REVALID_DNS_RESOLVER`` or the default)."""
    return os.environ.get(DNS_RESOLVER_ENV, DEFAULT_DNS_RESOLVER)

egress_firewall_script(scope_ips, dns_ip)

Build the gateway's iptables egress allowlist as a shell script (ADR-0045).

Default-drop OUTPUT (IPv4), then allow only: loopback, established/related return traffic, DNS to the one permitted resolver, and each scoped IP (any protocol, so ICMP/UDP/TCP scans all work). IPv6 is blanket-dropped — the scope IPs are v4 (see :func:_getaddrinfo_ips) and Docker's bridge is v4-only, so v6 is both unneeded and a leak to close; the drop is best-effort (|| true) so a kernel without a v6 stack does not abort the script. Everything else is dropped, so the sandbox sharing this namespace reaches the scoped host and nothing else. The gateway then blocks on sleep infinity to keep the namespace (and thus its rules) alive for the sandbox's lifetime.

Only resolved IPs and a resolver IP are interpolated — never a hostname — so there is no shell-injection surface (all inputs are numeric).

Parameters:

Name	Type	Description	Default
scope_ips	tuple[str, ...]	The resolved (IPv4) scope IPs to permit.	required
dns_ip	str	The single DNS resolver to permit on port 53.	required

Returns:

Type	Description
str	A sh-executable script string.

Source code in src/revalid/sandbox.py

def egress_firewall_script(scope_ips: tuple[str, ...], dns_ip: str) -> str:
    """Build the gateway's iptables egress allowlist as a shell script (ADR-0045).

    Default-drop OUTPUT (IPv4), then allow only: loopback, established/related
    return traffic, DNS to the one permitted resolver, and each scoped IP (any
    protocol, so ICMP/UDP/TCP scans all work). IPv6 is blanket-dropped — the
    scope IPs are v4 (see :func:`_getaddrinfo_ips`) and Docker's bridge is v4-only,
    so v6 is both unneeded and a leak to close; the drop is best-effort (``|| true``)
    so a kernel without a v6 stack does not abort the script. Everything else is
    dropped, so the sandbox sharing this namespace reaches the scoped host and
    nothing else. The gateway then blocks on ``sleep infinity`` to keep the
    namespace (and thus its rules) alive for the sandbox's lifetime.

    Only resolved IPs and a resolver IP are interpolated — never a hostname — so
    there is no shell-injection surface (all inputs are numeric).

    Args:
        scope_ips: The resolved (IPv4) scope IPs to permit.
        dns_ip: The single DNS resolver to permit on port 53.

    Returns:
        A ``sh``-executable script string.
    """
    lines = [
        "set -e",
        "iptables -P OUTPUT DROP",
        "iptables -A OUTPUT -o lo -j ACCEPT",
        "iptables -A OUTPUT -m conntrack --ctstate ESTABLISHED,RELATED -j ACCEPT",
        f"iptables -A OUTPUT -d {dns_ip} -p udp --dport 53 -j ACCEPT",
        f"iptables -A OUTPUT -d {dns_ip} -p tcp --dport 53 -j ACCEPT",
    ]
    lines += [f"iptables -A OUTPUT -d {ip} -j ACCEPT" for ip in scope_ips]
    # Close IPv6 entirely (best-effort: a v6-less kernel has no ip6tables table).
    lines.append("ip6tables -P OUTPUT DROP 2>/dev/null || true")
    lines.append("exec sleep infinity")
    return "\n".join(lines)

gateway_container_name(session_id)

Return the per-session egress-gateway container name (ADR-0045).

The gateway owns the network namespace and the iptables egress allowlist; the sandbox joins that namespace but cannot alter it (it holds no NET_ADMIN).

Source code in src/revalid/sandbox.py

def gateway_container_name(session_id: int) -> str:
    """Return the per-session egress-gateway container name (ADR-0045).

    The gateway owns the network namespace and the iptables egress allowlist; the
    sandbox joins that namespace but cannot alter it (it holds no ``NET_ADMIN``).
    """
    return f"revalid-retest-gw-{session_id}"

internal_network_name(session_id)

Return the per-session Docker network name.

One name for both topologies, because it is the same session resource either way — created with internal=True for a lab scope (hence the function's name), and as an ordinary routable bridge for an online one, where egress is bounded by the gateway's firewall rather than by the network having no route out (ADR-0045).

Source code in src/revalid/sandbox.py

def internal_network_name(session_id: int) -> str:
    """Return the per-session Docker network name.

    One name for both topologies, because it is the same session resource either
    way — created with ``internal=True`` for a lab scope (hence the function's
    name), and as an ordinary routable bridge for an online one, where egress is
    bounded by the gateway's firewall rather than by the network having no route
    out (ADR-0045).
    """
    return f"revalid-retest-{session_id}"

is_lab_scope(scope_hosts)

Whether a scope stays on the lab (empty, or every host is the lab host).

Lab scope keeps the unchanged --internal + attached-lab-container provisioning; any other host is an online target that needs the L3 egress gateway (ADR-0041, re-mechanised by ADR-0045). An empty scope defaults to the lab.

Source code in src/revalid/sandbox.py

def is_lab_scope(scope_hosts: tuple[str, ...]) -> bool:
    """Whether a scope stays on the lab (empty, or every host is the lab host).

    Lab scope keeps the unchanged ``--internal`` + attached-lab-container
    provisioning; any other host is an online target that needs the L3 egress
    gateway (ADR-0041, re-mechanised by ADR-0045). An empty scope defaults to
    the lab.
    """
    lab = lab_host()
    return all(host == lab for host in scope_hosts)

lab_base_url()

Return the lab target base URL ($REVALID_LAB_BASE_URL or the default).

Source code in src/revalid/sandbox.py

def lab_base_url() -> str:
    """Return the lab target base URL (``$REVALID_LAB_BASE_URL`` or the default)."""
    return os.environ.get(LAB_BASE_URL_ENV, DEFAULT_LAB_BASE_URL)

lab_host()

Return the lab target's host (host or host:port) from the lab base URL.

Source code in src/revalid/sandbox.py

def lab_host() -> str:
    """Return the lab target's host (``host`` or ``host:port``) from the lab base URL."""
    from revalid.scope import scope_host

    return scope_host(lab_base_url()) or ""

online_scope_hosts(scope_hosts)

The non-lab hosts in a scope — the online targets to allowlist (ADR-0041).

Source code in src/revalid/sandbox.py

def online_scope_hosts(scope_hosts: tuple[str, ...]) -> tuple[str, ...]:
    """The non-lab hosts in a scope — the online targets to allowlist (ADR-0041)."""
    lab = lab_host()
    return tuple(host for host in scope_hosts if host != lab)

resolve_scope_ips(hosts, resolver=None)

Resolve each online scope host to its IP addresses (ADR-0045).

The IPs are pinned into the sandbox at launch — allowlisted in the egress firewall and written to /etc/hosts — so the L3 gateway can permit exactly the scoped host and nothing else. All A/AAAA records are taken, but they are a snapshot: a target behind rotating CDN IPs may present addresses not seen here (the inherent limit of an IP allowlist vs. the retired L7 proxy — stated in ADR-0045).

Parameters:

Name	Type	Description	Default
hosts	tuple[str, ...]	The online scope hosts (host or host:port); the port is dropped for resolution.	required
resolver	Callable[[str], list[str]] | None	Injection seam for tests — maps a bare host to its IP strings. Defaults to a real DNS lookup via :func:socket.getaddrinfo.	None

Returns:

Type	Description
dict[str, tuple[str, ...]]	A mapping of bare host → its resolved IPs (in stable, de-duplicated order).
dict[str, tuple[str, ...]]	A host that does not resolve maps to an empty tuple (the caller fails
dict[str, tuple[str, ...]]	closed on it rather than opening egress to a guessed address).

Source code in src/revalid/sandbox.py

def resolve_scope_ips(
    hosts: tuple[str, ...],
    resolver: Callable[[str], list[str]] | None = None,
) -> dict[str, tuple[str, ...]]:
    """Resolve each online scope host to its IP addresses (ADR-0045).

    The IPs are pinned into the sandbox at launch — allowlisted in the egress
    firewall and written to ``/etc/hosts`` — so the L3 gateway can permit exactly
    the scoped host and nothing else. All A/AAAA records are taken, but they are a
    *snapshot*: a target behind rotating CDN IPs may present addresses not seen
    here (the inherent limit of an IP allowlist vs. the retired L7 proxy — stated
    in ADR-0045).

    Args:
        hosts: The online scope hosts (``host`` or ``host:port``); the port is
            dropped for resolution.
        resolver: Injection seam for tests — maps a bare host to its IP strings.
            Defaults to a real DNS lookup via :func:`socket.getaddrinfo`.

    Returns:
        A mapping of bare host → its resolved IPs (in stable, de-duplicated order).
        A host that does not resolve maps to an empty tuple (the caller fails
        closed on it rather than opening egress to a guessed address).
    """
    resolve = resolver if resolver is not None else _getaddrinfo_ips
    resolved: dict[str, tuple[str, ...]] = {}
    for host in hosts:
        bare = _bare_host(host)
        if not bare:
            continue
        ips = tuple(dict.fromkeys(resolve(bare)))  # de-dupe, preserve order
        resolved[bare] = ips
    return resolved

sandbox_container_name(session_id)

Return the per-session sandbox container name (ADR-0045).

Named (rather than anonymous) so teardown can find and remove it by name even when no live DockerSandbox object holds a reference — e.g. reaping a session orphaned by a backend restart when its report is deleted.

Source code in src/revalid/sandbox.py

def sandbox_container_name(session_id: int) -> str:
    """Return the per-session sandbox container name (ADR-0045).

    Named (rather than anonymous) so teardown can find and remove it by name even
    when no live ``DockerSandbox`` object holds a reference — e.g. reaping a
    session orphaned by a backend restart when its report is deleted.
    """
    return f"revalid-retest-sbx-{session_id}"

sandbox_image()

Return the sandbox image to run ($REVALID_SANDBOX_IMAGE or the default).

The default is the locally-built Kali toolbox (issue #105); the override exists so an operator can point the agent at their own image without a code change — the egress lock is enforced by the network, not by the image, so swapping it changes the tools available and nothing about containment.

Source code in src/revalid/sandbox.py

def sandbox_image() -> str:
    """Return the sandbox image to run (``$REVALID_SANDBOX_IMAGE`` or the default).

    The default is the locally-built Kali toolbox (issue #105); the override
    exists so an operator can point the agent at their own image without a code
    change — the egress lock is enforced by the network, not by the image, so
    swapping it changes the tools available and nothing about containment.
    """
    return os.environ.get(SANDBOX_IMAGE_ENV, DEFAULT_SANDBOX_IMAGE)

FR-17 / M6 agentic retest agent (ADR-0025, Slice 0).

Two tools — a gated run_command (Pydantic AI deferred approval) and an ungated respond for prose — over a three-way output union: a ConcludeOutput verdict, an AwaitOperator hand-back (ADR-0039), or a DeferredToolRequests pause while a proposed command awaits approval. The orchestrator (retest_session.py) runs the agent step-by-step, pausing on each proposed command for human approval and resuming with ToolApproved/ToolDenied. Its persona branches per turn on deps.free_launch: guided (one action, then hand back) or autonomous (drive to a verdict) — ADR-0040.

AwaitOperator

Bases: BaseModel

The agent replied and is handing control back to the operator (issue #204).

The turn ends without a command or a verdict: the agent answered the operator conversationally (a greeting, a small-talk reply, an acknowledgement) and is now waiting for them, sandbox kept alive. Lighter than an inconclusive conclusion — it is not "I'm stuck, please guide me", just a pause: the agent has said its piece and waits. The orchestrator surfaces message as an agent chat bubble and parks the session in awaiting_operator; the operator's next message resumes it.

Source code in src/revalid/retest_agent.py

class AwaitOperator(BaseModel):
    """The agent replied and is handing control back to the operator (issue #204).

    The turn ends without a command or a verdict: the agent answered the operator
    conversationally (a greeting, a small-talk reply, an acknowledgement) and is
    now waiting for them, sandbox kept alive. Lighter than an ``inconclusive``
    conclusion — it is not "I'm stuck, please guide me", just a pause: the agent has
    said its piece and waits. The
    orchestrator surfaces ``message`` as an agent chat bubble and parks the session
    in ``awaiting_operator``; the operator's next message resumes it.
    """

    model_config = ConfigDict(frozen=True)

    message: str = Field(min_length=1)

ConcludeOutput

Bases: BaseModel

The agent's terminal verdict for a retest session.

Source code in src/revalid/retest_agent.py

class ConcludeOutput(BaseModel):
    """The agent's terminal verdict for a retest session."""

    model_config = ConfigDict(frozen=True)

    status: VerdictStatus
    rationale: str = Field(min_length=1)

RetestSessionDeps dataclass

Runtime dependencies injected into the retest agent's tools.

Source code in src/revalid/retest_agent.py

@dataclass
class RetestSessionDeps:
    """Runtime dependencies injected into the retest agent's tools."""

    sandbox: Sandbox
    emit_output: Callable[[str, CommandResult], None]
    #: Returns (and clears) any manual operator commands run since the agent's
    #: last turn, so the agent observes what the human did (FR-17 Slice 2). The
    #: default surfaces nothing — the human-command path (`!`) injects the real
    #: drain via the orchestrator's :func:`~revalid.retest_session._make_deps`.
    drain_observations: Callable[[], list[str]] = _no_observations
    #: Records the agent's prose replies to the operator (FR-17 Slice 4). Invoked
    #: by the non-gated ``respond`` tool; the orchestrator wires this to append an
    #: ``agent_message`` transcript event. The default drops it (agent-unit tests).
    emit_message: Callable[[str], None] = _no_emit_message
    #: Whether the operator has handed over the wheel (Auto-run / free-launch). It
    #: selects the agent's persona via dynamic instructions (ADR-0040): guided
    #: (one action then hand back) when ``False``, autonomous (drive to a verdict)
    #: when ``True``. Rebuilt fresh each turn from the live session, so a live
    #: Auto-run toggle takes effect on the agent's next turn. Default ``False`` —
    #: the guided persona — so agent-unit constructions need not set it.
    free_launch: bool = False
    #: This session's parsed scope hosts (issue #247). Drives the reachability line in
    #: the instructions, so an online-scope session is told the truth about what it can
    #: reach (ADR-0045) instead of the old hardcoded "lab only". Rebuilt each turn like
    #: the rest of the deps; the default — empty, i.e. the lab — keeps agent-unit
    #: constructions and lab sessions on the original wording.
    scope_hosts: tuple[str, ...] = ()

build_retest_agent(model=None)

Build the FR-17 retest agent: a gated run_command tool + a verdict.

Parameters:

Name	Type	Description	Default
model	Model | KnownModelName | str | None	A Pydantic AI model instance or name. When omitted, the configured backend is used (REVALID_LLM_MODEL, Claude by default — FR-13); tests pass TestModel/FunctionModel.	None

Returns:

Type	Description
RetestAgent	An agent whose output is a :class:ConcludeOutput verdict, an
RetestAgent	class:AwaitOperator conversational hand-back, or — while a gated
RetestAgent	run_command call awaits human approval — a
RetestAgent	class:~pydantic_ai.DeferredToolRequests.

Source code in src/revalid/retest_agent.py

def build_retest_agent(
    model: Model | KnownModelName | str | None = None,
) -> RetestAgent:
    """Build the FR-17 retest agent: a gated ``run_command`` tool + a verdict.

    Args:
        model: A Pydantic AI model instance or name. When omitted, the
            configured backend is used (``REVALID_LLM_MODEL``, Claude by
            default — FR-13); tests pass ``TestModel``/``FunctionModel``.

    Returns:
        An agent whose output is a :class:`ConcludeOutput` verdict, an
        :class:`AwaitOperator` conversational hand-back, or — while a gated
        ``run_command`` call awaits human approval — a
        :class:`~pydantic_ai.DeferredToolRequests`.
    """
    agent: RetestAgent = Agent(
        model if model is not None else resolve_model(),
        deps_type=RetestSessionDeps,
        output_type=[ConcludeOutput, AwaitOperator, DeferredToolRequests],
        instructions=_BASE_INSTRUCTIONS,
        retries=_MAX_TOOL_RETRIES,
        defer_model_check=True,
    )

    @agent.instructions
    def _mode_guidance(ctx: RunContext[RetestSessionDeps]) -> str:
        """Append the persona for this turn's mode (ADR-0040).

        Evaluated per run against freshly-built deps, so a live Auto-run toggle
        switches the agent between driving itself to a verdict (autonomous) and
        doing one action then handing back (guided) on its very next turn.
        """
        return _AUTONOMOUS_GUIDANCE if ctx.deps.free_launch else _GUIDED_GUIDANCE

    @agent.instructions
    def _scope_guidance(ctx: RunContext[RetestSessionDeps]) -> str:
        """State what this session can reach (issue #247).

        Dynamic for the same reason as the persona above: the scope is a property of
        the session, not of the build, so a lab and an online retest must be told
        different — and true — things about their reachability.
        """
        return _scope_reachability(ctx.deps.scope_hosts)

    @agent.tool(requires_approval=True)
    def run_command(
        ctx: RunContext[RetestSessionDeps],
        command: str,
        rationale: str,
        timeout_seconds: int = DEFAULT_COMMAND_TIMEOUT,
    ) -> str:
        """Run one shell command in the egress-locked sandbox and return its output.

        Args:
            ctx: The run context carrying the sandbox + output-emit callback.
            command: The exact shell command to execute (against the scoped target).
            rationale: A one-line reason this command advances the retest.
            timeout_seconds: How long the command may run before it is killed —
                pick a value that fits it (a few seconds for a curl, more for a
                scan). Clamped to at most ``MAX_COMMAND_TIMEOUT`` seconds.

        Returns:
            The command's exit code, timing, stdout and stderr as text; a note is
            appended when the command was killed for exceeding its timeout.
        """
        timeout = clamp_timeout(timeout_seconds)
        result = ctx.deps.sandbox.exec(command, timeout=timeout)
        ctx.deps.emit_output(command, result)
        text = _format_result(result)
        if result.exit_code in TIMEOUT_EXIT_CODES:
            text += f"\n[terminated: the command exceeded its {timeout}s timeout]"
        return text + format_observations(ctx.deps.drain_observations())

    @agent.tool
    def respond(ctx: RunContext[RetestSessionDeps], message: str) -> str:
        """Send a short prose message to the operator (e.g. answer a question).

        Use this to reply to the operator or give a brief status note — not to
        narrate every step. It runs nothing; after it you continue with a command
        or a verdict.

        Args:
            ctx: The run context carrying the message-emit callback.
            message: The prose to show the operator in the chat.

        Returns:
            A short confirmation the message was delivered.
        """
        ctx.deps.emit_message(message)
        return "Delivered to the operator."

    return agent

clamp_timeout(seconds)

Clamp an agent-requested per-command timeout to [1, MAX_COMMAND_TIMEOUT].

Source code in src/revalid/retest_agent.py

def clamp_timeout(seconds: int) -> int:
    """Clamp an agent-requested per-command timeout to ``[1, MAX_COMMAND_TIMEOUT]``."""
    return max(1, min(seconds, MAX_COMMAND_TIMEOUT))

format_observations(observations)

Render buffered operator activity as a block the agent reads on its next turn.

Returns an empty string when there is nothing to surface, so callers can append it unconditionally.

Parameters:

Name	Type	Description	Default
observations	list[str]	Human-run command summaries buffered since the last turn.	required

Returns:

Type	Description
str	A labelled block to append to the next tool result, or "" if empty.

Source code in src/revalid/retest_agent.py

def format_observations(observations: list[str]) -> str:
    """Render buffered operator activity as a block the agent reads on its next turn.

    Returns an empty string when there is nothing to surface, so callers can
    append it unconditionally.

    Args:
        observations: Human-run command summaries buffered since the last turn.

    Returns:
        A labelled block to append to the next tool result, or ``""`` if empty.
    """
    if not observations:
        return ""
    return "\n\n--- operator activity while you waited ---\n" + "\n".join(observations)

FR-17 / M6 retest-session persistence + orchestration (ADR-0025, Slice 0).

An agentic retest session is a :class:~revalid.db.RetestSessionRecord row plus its append-only transcript of :class:~revalid.db.SessionEventRecord rows, symmetric with how :mod:revalid.findings splits identity from immutable history (Task 3: create_session, append_event, load_events_after, set_status, record_verdict).

Task 5 adds the orchestration layer that drives the Task 4 agent (:mod:revalid.retest_agent) step-by-step: a process-local :class:SessionRegistry of :class:LiveSession state, start_and_step/ apply_decision to pause on each proposed command for human approval and resume it. When the agent exhausts the options it can think of, it hands back to the operator (awaiting_operator, ADR-0034/0042) rather than running forever.

LiveSession dataclass

In-memory live state for one active session (not restart-safe, Slice 0).

Deliberately does NOT cache a :class:~revalid.retest_agent.RetestSessionDeps: start_and_step and apply_decision may run in separate background tasks against separate DB sessions (Task 6's async driver), and deps' emit_output closure captures whichever Session built it. Caching deps here would let a later call write command_output events through an already-closed session. Callers build deps fresh via _make_deps immediately before each agent.run_sync.

Attributes:

Name	Type	Description
agent	RetestAgent	The built retest agent driving this session.
sandbox	Sandbox	The persistent sandbox handle for this session's lifetime.
messages	list[ModelMessage]	The full Pydantic AI message history so far (resumed on each step via message_history).
pending_call_id	str | None	The tool_call_id awaiting a human decision, or None when no command is currently proposed.
lock	Lock	Guards the compare-and-swap on pending_call_id in apply_decision so two concurrent decisions (e.g. a double-click on Approve before the REST 202 re-enables the button) can't both observe the same pending call and both resume the agent run.

Source code in src/revalid/retest_session.py

@dataclass
class LiveSession:
    """In-memory live state for one active session (not restart-safe, Slice 0).

    Deliberately does NOT cache a :class:`~revalid.retest_agent.RetestSessionDeps`:
    ``start_and_step`` and ``apply_decision`` may run in separate background
    tasks against separate DB sessions (Task 6's async driver), and deps'
    ``emit_output`` closure captures whichever ``Session`` built it. Caching
    deps here would let a later call write ``command_output`` events through
    an already-closed session. Callers build deps fresh via ``_make_deps``
    immediately before each ``agent.run_sync``.

    Attributes:
        agent: The built retest agent driving this session.
        sandbox: The persistent sandbox handle for this session's lifetime.
        messages: The full Pydantic AI message history so far (resumed on
            each step via ``message_history``).
        pending_call_id: The ``tool_call_id`` awaiting a human decision, or
            ``None`` when no command is currently proposed.
        lock: Guards the compare-and-swap on ``pending_call_id`` in
            ``apply_decision`` so two concurrent decisions (e.g. a double-click
            on Approve before the REST 202 re-enables the button) can't both
            observe the same pending call and both resume the agent run.
    """

    agent: RetestAgent
    sandbox: Sandbox
    messages: list[ModelMessage] = field(default_factory=list)
    pending_call_id: str | None = None
    #: Whether the agent's commands auto-run without a per-command human approval
    #: (FR-17 Slice 5) — the one deliberate relaxation of the gate; the egress lock
    #: is unaffected. Toggled live by ``set_free_launch``; the loop lives in
    #: ``_advance``, which also delivers queued operator messages (ADR-0042).
    free_launch: bool = False
    #: The operator pressed Stop (issue #150): a cooperative pause. Set by
    #: ``stop_session``, cleared by ``resume_session``. An in-flight step that
    #: finishes while this is ``True`` parks the session in ``stopped`` instead of
    #: advancing, and the free-launch loop halts — the sandbox is kept alive.
    stopped: bool = False
    lock: threading.Lock = field(default_factory=threading.Lock)
    #: Manual operator commands run since the agent's last turn (from the console
    #: terminal's ``operator$`` prompt), buffered here and surfaced to the agent on
    #: its next turn so it observes what the human did (FR-17 Slice 2, ADR-0026).
    #: Guarded by ``lock`` — appended by the human-
    #: command worker, drained by the next agent resume, on separate threads.
    observations: list[str] = field(default_factory=list)

    def observe(self, summary: str) -> None:
        """Buffer one operator-command summary for the agent's next turn (thread-safe)."""
        with self.lock:
            self.observations.append(summary)

    def drain(self) -> list[str]:
        """Atomically return and clear the buffered operator observations (thread-safe)."""
        with self.lock:
            drained = list(self.observations)
            self.observations.clear()
            return drained

    #: Free-text operator chat messages (FR-17 Slice 4) queued since the agent's
    #: last turn. Delivered as a first-class ``user_prompt`` on the next agent
    #: resume (approve/reject) — the operator's *voice*, distinct from
    #: ``observations`` (`!` command *results* folded into a tool return).
    #: Guarded by ``lock`` — appended by the message worker, drained by the next
    #: agent resume, on separate threads.
    human_messages: list[str] = field(default_factory=list)

    def receive_message(self, text: str) -> None:
        """Queue one operator chat message for the agent's next turn (thread-safe)."""
        with self.lock:
            self.human_messages.append(text)

    def drain_messages(self) -> list[str]:
        """Atomically return and clear the queued operator chat messages (thread-safe)."""
        with self.lock:
            drained = list(self.human_messages)
            self.human_messages.clear()
            return drained

    def has_queued_messages(self) -> bool:
        """Whether operator chat messages are waiting for the next turn (thread-safe).

        Peeked (not drained) by :func:`_advance` to decide whether to deliver them at
        a turn boundary — the actual drain happens inside the resume it then runs.
        """
        with self.lock:
            return bool(self.human_messages)

    #: The current goal (FR-17 6b-ii) queued by an operator edit since the agent's
    #: last turn, delivered as a user turn on the next resume (like human_messages).
    #: ``None`` means no pending change. Guarded by ``lock``.
    pending_goal: list[str] | None = None

    def set_pending_goal(self, steps: list[str]) -> None:
        """Queue a goal change for the agent's next turn (thread-safe)."""
        with self.lock:
            self.pending_goal = list(steps)

    def drain_goal(self) -> list[str] | None:
        """Atomically return and clear the queued goal change (thread-safe)."""
        with self.lock:
            drained = self.pending_goal
            self.pending_goal = None
            return drained

    #: The event loop + task driving the in-flight agent turn, when one is running
    #: (issue #204). Held so another thread — the ``restart-model`` endpoint or a
    #: teardown — can cancel a wedged turn cross-thread via
    #: ``loop.call_soon_threadsafe(task.cancel)``. Both ``None`` between turns.
    #: Guarded by ``lock``; the run thread attaches on start, detaches on completion.
    _run_loop: asyncio.AbstractEventLoop | None = None
    _run_task: asyncio.Task[Any] | None = None
    #: Set when the operator asked to abort-and-*retry* the in-flight turn (unstick,
    #: issue #204). The run thread, on catching the cancellation, consumes this and
    #: re-runs the same turn instead of failing. A plain cancel (teardown) leaves it
    #: ``False`` so the cancellation settles the session. Guarded by ``lock``.
    abort_retry: bool = False

    def attach_run(self, loop: asyncio.AbstractEventLoop, task: asyncio.Task[Any]) -> None:
        """Register the loop + task of the turn now starting (thread-safe).

        Resets ``abort_retry`` so a stale flag from a prior turn never bleeds into
        this one — each turn starts with clean cancellation state.
        """
        with self.lock:
            self._run_loop = loop
            self._run_task = task
            self.abort_retry = False

    def detach_run(self) -> None:
        """Clear the in-flight-turn handle once the turn completes (thread-safe)."""
        with self.lock:
            self._run_loop = None
            self._run_task = None

    def request_restart(self) -> bool:
        """Abort the in-flight turn and mark it to be re-run — unstick (issue #204).

        Cancels the run's asyncio task from this (foreign) thread and flags the run
        thread to retry the same turn rather than fail. Returns whether a turn was
        actually in flight to abort (``False`` = nothing to unstick).
        """
        with self.lock:
            loop, task = self._run_loop, self._run_task
            if loop is None or task is None:
                return False
            self.abort_retry = True
        try:
            loop.call_soon_threadsafe(task.cancel)
        except RuntimeError:  # the loop is already closing — the turn is ending anyway
            with self.lock:
                self.abort_retry = False
            return False
        return True

    def request_cancel(self) -> bool:
        """Abort the in-flight turn WITHOUT a retry — teardown (issues #204/#205).

        Like :meth:`request_restart` but leaves ``abort_retry`` ``False`` so the
        cancellation propagates out of the run thread (which then settles the
        already-terminal session) instead of re-running. Used when ending or
        deleting a session so a wedged turn's thread does not linger on a hung call.
        """
        with self.lock:
            loop, task = self._run_loop, self._run_task
        if loop is None or task is None:
            return False
        try:
            loop.call_soon_threadsafe(task.cancel)
        except RuntimeError:  # the loop is already closing — nothing to cancel
            return False
        return True

    def consume_restart(self) -> bool:
        """Return + clear whether the aborted turn should be re-run (thread-safe)."""
        with self.lock:
            requested = self.abort_retry
            self.abort_retry = False
            return requested

attach_run(loop, task)

Register the loop + task of the turn now starting (thread-safe).

Resets abort_retry so a stale flag from a prior turn never bleeds into this one — each turn starts with clean cancellation state.

Source code in src/revalid/retest_session.py

def attach_run(self, loop: asyncio.AbstractEventLoop, task: asyncio.Task[Any]) -> None:
    """Register the loop + task of the turn now starting (thread-safe).

    Resets ``abort_retry`` so a stale flag from a prior turn never bleeds into
    this one — each turn starts with clean cancellation state.
    """
    with self.lock:
        self._run_loop = loop
        self._run_task = task
        self.abort_retry = False

consume_restart()

Return + clear whether the aborted turn should be re-run (thread-safe).

Source code in src/revalid/retest_session.py

def consume_restart(self) -> bool:
    """Return + clear whether the aborted turn should be re-run (thread-safe)."""
    with self.lock:
        requested = self.abort_retry
        self.abort_retry = False
        return requested

detach_run()

Clear the in-flight-turn handle once the turn completes (thread-safe).

Source code in src/revalid/retest_session.py

def detach_run(self) -> None:
    """Clear the in-flight-turn handle once the turn completes (thread-safe)."""
    with self.lock:
        self._run_loop = None
        self._run_task = None

drain()

Atomically return and clear the buffered operator observations (thread-safe).

Source code in src/revalid/retest_session.py

def drain(self) -> list[str]:
    """Atomically return and clear the buffered operator observations (thread-safe)."""
    with self.lock:
        drained = list(self.observations)
        self.observations.clear()
        return drained

drain_goal()

Atomically return and clear the queued goal change (thread-safe).

Source code in src/revalid/retest_session.py

def drain_goal(self) -> list[str] | None:
    """Atomically return and clear the queued goal change (thread-safe)."""
    with self.lock:
        drained = self.pending_goal
        self.pending_goal = None
        return drained

drain_messages()

Atomically return and clear the queued operator chat messages (thread-safe).

Source code in src/revalid/retest_session.py

def drain_messages(self) -> list[str]:
    """Atomically return and clear the queued operator chat messages (thread-safe)."""
    with self.lock:
        drained = list(self.human_messages)
        self.human_messages.clear()
        return drained

has_queued_messages()

Whether operator chat messages are waiting for the next turn (thread-safe).

Peeked (not drained) by :func:_advance to decide whether to deliver them at a turn boundary — the actual drain happens inside the resume it then runs.

Source code in src/revalid/retest_session.py

def has_queued_messages(self) -> bool:
    """Whether operator chat messages are waiting for the next turn (thread-safe).

    Peeked (not drained) by :func:`_advance` to decide whether to deliver them at
    a turn boundary — the actual drain happens inside the resume it then runs.
    """
    with self.lock:
        return bool(self.human_messages)

observe(summary)

Buffer one operator-command summary for the agent's next turn (thread-safe).

Source code in src/revalid/retest_session.py

def observe(self, summary: str) -> None:
    """Buffer one operator-command summary for the agent's next turn (thread-safe)."""
    with self.lock:
        self.observations.append(summary)

receive_message(text)

Queue one operator chat message for the agent's next turn (thread-safe).

Source code in src/revalid/retest_session.py

def receive_message(self, text: str) -> None:
    """Queue one operator chat message for the agent's next turn (thread-safe)."""
    with self.lock:
        self.human_messages.append(text)

request_cancel()

Abort the in-flight turn WITHOUT a retry — teardown (issues #204/#205).

Like :meth:request_restart but leaves abort_retry False so the cancellation propagates out of the run thread (which then settles the already-terminal session) instead of re-running. Used when ending or deleting a session so a wedged turn's thread does not linger on a hung call.

Source code in src/revalid/retest_session.py

def request_cancel(self) -> bool:
    """Abort the in-flight turn WITHOUT a retry — teardown (issues #204/#205).

    Like :meth:`request_restart` but leaves ``abort_retry`` ``False`` so the
    cancellation propagates out of the run thread (which then settles the
    already-terminal session) instead of re-running. Used when ending or
    deleting a session so a wedged turn's thread does not linger on a hung call.
    """
    with self.lock:
        loop, task = self._run_loop, self._run_task
    if loop is None or task is None:
        return False
    try:
        loop.call_soon_threadsafe(task.cancel)
    except RuntimeError:  # the loop is already closing — nothing to cancel
        return False
    return True

request_restart()

Abort the in-flight turn and mark it to be re-run — unstick (issue #204).

Cancels the run's asyncio task from this (foreign) thread and flags the run thread to retry the same turn rather than fail. Returns whether a turn was actually in flight to abort (False = nothing to unstick).

Source code in src/revalid/retest_session.py

def request_restart(self) -> bool:
    """Abort the in-flight turn and mark it to be re-run — unstick (issue #204).

    Cancels the run's asyncio task from this (foreign) thread and flags the run
    thread to retry the same turn rather than fail. Returns whether a turn was
    actually in flight to abort (``False`` = nothing to unstick).
    """
    with self.lock:
        loop, task = self._run_loop, self._run_task
        if loop is None or task is None:
            return False
        self.abort_retry = True
    try:
        loop.call_soon_threadsafe(task.cancel)
    except RuntimeError:  # the loop is already closing — the turn is ending anyway
        with self.lock:
            self.abort_retry = False
        return False
    return True

set_pending_goal(steps)

Queue a goal change for the agent's next turn (thread-safe).

Source code in src/revalid/retest_session.py

def set_pending_goal(self, steps: list[str]) -> None:
    """Queue a goal change for the agent's next turn (thread-safe)."""
    with self.lock:
        self.pending_goal = list(steps)

SessionRegistry

Process-local registry of live sessions (one per app instance).

Source code in src/revalid/retest_session.py

class SessionRegistry:
    """Process-local registry of live sessions (one per app instance)."""

    def __init__(self) -> None:
        """Start with an empty registry."""
        self._live: dict[int, LiveSession] = {}

    def put(self, session_id: int, live: LiveSession) -> None:
        """Register ``live`` as the active state for ``session_id``."""
        self._live[session_id] = live

    def get(self, session_id: int) -> LiveSession | None:
        """Return the live state for ``session_id``, or ``None`` if not live."""
        return self._live.get(session_id)

    def drop(self, session_id: int) -> None:
        """Remove ``session_id`` from the registry (no-op if already absent)."""
        self._live.pop(session_id, None)

__init__()

Start with an empty registry.

Source code in src/revalid/retest_session.py

def __init__(self) -> None:
    """Start with an empty registry."""
    self._live: dict[int, LiveSession] = {}

drop(session_id)

Remove session_id from the registry (no-op if already absent).

Source code in src/revalid/retest_session.py

def drop(self, session_id: int) -> None:
    """Remove ``session_id`` from the registry (no-op if already absent)."""
    self._live.pop(session_id, None)

get(session_id)

Return the live state for session_id, or None if not live.

Source code in src/revalid/retest_session.py

def get(self, session_id: int) -> LiveSession | None:
    """Return the live state for ``session_id``, or ``None`` if not live."""
    return self._live.get(session_id)

put(session_id, live)

Register live as the active state for session_id.

Source code in src/revalid/retest_session.py

def put(self, session_id: int, live: LiveSession) -> None:
    """Register ``live`` as the active state for ``session_id``."""
    self._live[session_id] = live

adjudicate_verdict(session, session_id, status, rationale)

Record a human adjudication of a concluded session's verdict (FR-17 Slice 6a).

The human accepts or overrides the agent's conclusion. Either way this appends a superseding agentic verdict (actor="operator") — the agent's record is never mutated (append-only; FR-10 intact) — plus a verdict_adjudicated transcript event, and updates the session row so its view shows the final call. Latest-per-finding (highest verdict id) is authoritative, so the operator record supersedes the agent's.

A pure DB operation: the session is already terminal (torn down), so the live registry is never touched. A no-op if the session doesn't exist or has no agent verdict yet (nothing to adjudicate) — the guard also makes a premature or duplicate call safe.

Parameters:

Name	Type	Description	Default
session	Session	Active DB session for this call.	required
session_id	int	The concluded retest session being adjudicated.	required
status	VerdictStatus	The human's verdict (may equal or differ from the agent's).	required
rationale	str	The human's justification.	required

Source code in src/revalid/retest_session.py

def adjudicate_verdict(
    session: Session, session_id: int, status: VerdictStatus, rationale: str
) -> None:
    """Record a human adjudication of a concluded session's verdict (FR-17 Slice 6a).

    The human accepts or overrides the agent's conclusion. Either way this
    **appends** a superseding agentic verdict (``actor="operator"``) — the agent's
    record is never mutated (append-only; FR-10 intact) — plus a
    ``verdict_adjudicated`` transcript event, and updates the session row so its
    view shows the final call. Latest-per-finding (highest verdict id) is
    authoritative, so the operator record supersedes the agent's.

    A pure DB operation: the session is already terminal (torn down), so the live
    registry is never touched. A no-op if the session doesn't exist or has no
    agent verdict yet (nothing to adjudicate) — the guard also makes a premature
    or duplicate call safe.

    Args:
        session: Active DB session for this call.
        session_id: The concluded retest session being adjudicated.
        status: The human's verdict (may equal or differ from the agent's).
        rationale: The human's justification.
    """
    record = session.get(RetestSessionRecord, session_id)
    if record is None or record.verdict_status is None:
        return
    append_event(
        session,
        session_id,
        SessionEventKind.VERDICT_ADJUDICATED,
        {"status": status.value, "rationale": rationale},
    )
    session.add(
        VerdictRecord.agentic(
            finding_id=record.finding_id,
            session_id=session_id,
            status=status,
            rationale=rationale,
            actor="operator",
            reason_code="operator_adjudication",
        )
    )
    record.verdict_status = status.value
    record.verdict_rationale = rationale
    session.commit()

append_event(session, session_id, kind, payload)

Append one transcript event with the next seq and commit.

Source code in src/revalid/retest_session.py

def append_event(
    session: Session, session_id: int, kind: SessionEventKind, payload: dict[str, Any]
) -> SessionEventRecord:
    """Append one transcript event with the next ``seq`` and commit."""
    event = SessionEventRecord(
        session_id=session_id, seq=_next_seq(session, session_id), kind=kind.value, payload=payload
    )
    session.add(event)
    session.commit()
    session.refresh(event)
    return event

apply_decision(session, registry, session_id, *, approved, reason='', command_id)

Resume a paused run with a human decision on the pending command.

An approval runs the command and resumes the agent; when the agent later exhausts the options it can think of it hands back for operator guidance (:func:_dispatch_output, ADR-0034) rather than running forever.

command_id must match the session's currently pending tool_call_id (validated + consumed atomically under live.lock, see _consume_pending_call); a stale, duplicate, or mismatched decision is a silent no-op. This closes a double-approve race: without it, two concurrent decisions on the same command could both pass the guard, both resume the agent (running the approved command twice in the sandbox) and both append transcript events under colliding seq numbers.

Parameters:

Name	Type	Description	Default
session	Session	Active DB session for this call — always freshly obtained by the caller, never held across separate orchestration calls.	required
registry	SessionRegistry	The live-session registry.	required
session_id	int	The retest session to resume.	required
approved	bool	Whether the pending command was approved.	required
reason	str	Optional human-supplied reason, recorded and (on rejection) surfaced back to the model as the tool's denial message.	''
command_id	str	The cid from the approve/reject URL; must match the session's pending tool_call_id or the call no-ops.	required

Source code in src/revalid/retest_session.py

def apply_decision(
    session: Session,
    registry: SessionRegistry,
    session_id: int,
    *,
    approved: bool,
    reason: str = "",
    command_id: str,
) -> None:
    """Resume a paused run with a human decision on the pending command.

    An approval runs the command and resumes the agent; when the agent later
    exhausts the options it can think of it hands back for operator guidance
    (:func:`_dispatch_output`, ADR-0034) rather than running forever.

    ``command_id`` must match the session's currently pending ``tool_call_id``
    (validated + consumed atomically under ``live.lock``, see
    ``_consume_pending_call``); a stale, duplicate, or mismatched decision is a
    silent no-op. This closes a double-approve race: without it, two
    concurrent decisions on the same command could both pass the guard, both
    resume the agent (running the approved command twice in the sandbox) and
    both append transcript events under colliding ``seq`` numbers.

    Args:
        session: Active DB session for this call — always freshly obtained by
            the caller, never held across separate orchestration calls.
        registry: The live-session registry.
        session_id: The retest session to resume.
        approved: Whether the pending command was approved.
        reason: Optional human-supplied reason, recorded and (on rejection)
            surfaced back to the model as the tool's denial message.
        command_id: The ``cid`` from the approve/reject URL; must match the
            session's pending ``tool_call_id`` or the call no-ops.
    """
    live = registry.get(session_id)
    if live is None:
        return
    call_id = _consume_pending_call(live, command_id)
    if call_id is None:
        return  # stale, duplicate, or mismatched decision: no-op

    kind = _decision_event_kind(approved=approved)
    append_event(session, session_id, kind, {"reason": reason} if reason else {})
    _resume_with_decision(
        session, registry, session_id, live, call_id, approved=approved, reason=reason
    )
    # Drive the boundary: deliver a queued message, or (free-launch) auto-approve a
    # new proposal; in gated mode with nothing queued this returns immediately.
    _advance(session, registry, session_id)

conclude_session(session, registry, session_id, status, rationale)

Operator manually concludes a session with a determination — ADR-0034.

The operator's own verdict, recordable at ANY live point in the retest (issue

150) — not only at an awaiting_operator hand-back: while a command awaits approval,

or even while the agent is mid-step. Writes the verdict (actor="operator", the only path that can record inconclusive) and tears down the sandbox. A no-op if the session is already terminal. Works on an orphaned session too (the verdict is recorded; the teardown no-ops). When invoked mid-step, the in-flight agent turn's resulting failure is swallowed rather than clobbering this verdict (see :func:_fail).

Parameters:

Name	Type	Description	Default
session	Session	Active DB session for this call.	required
registry	SessionRegistry	The live-session registry.	required
session_id	int	The session to conclude.	required
status	VerdictStatus	The operator's determination.	required
rationale	str	The operator's justification.	required

Source code in src/revalid/retest_session.py

def conclude_session(
    session: Session,
    registry: SessionRegistry,
    session_id: int,
    status: VerdictStatus,
    rationale: str,
) -> None:
    """Operator manually concludes a session with a determination — ADR-0034.

    The operator's own verdict, recordable at ANY live point in the retest (issue
    #150) — not only at an ``awaiting_operator`` hand-back: while a command awaits approval,
    or even while the agent is mid-step. Writes the verdict (``actor="operator"``,
    the only path that can record ``inconclusive``) and tears down the sandbox. A
    no-op if the session is already terminal. Works on an orphaned session too (the
    verdict is recorded; the teardown no-ops). When invoked mid-step, the in-flight
    agent turn's resulting failure is swallowed rather than clobbering this verdict
    (see :func:`_fail`).

    Args:
        session: Active DB session for this call.
        registry: The live-session registry.
        session_id: The session to conclude.
        status: The operator's determination.
        rationale: The operator's justification.
    """
    record = session.get(RetestSessionRecord, session_id)
    if record is None or RetestSessionStatus(record.status) in _TERMINAL:
        return
    record_verdict(
        session,
        session_id,
        status,
        rationale,
        actor="operator",
        reason_code="operator_conclusion",
    )
    _teardown(registry, session_id)

continue_session(session, registry, session_id)

Resume a session the agent handed back — ADR-0034 "Keep going" / reply (#204).

A no-op unless the session is handed back in :data:_RESUMABLE_ON_MESSAGE (awaiting_operator) with a live agent — a handed-back session that outlived a backend restart has no sandbox to resume, so the operator restarts it instead. The agent only ever hands back between turns (never with a command still pending), so continuing re-runs it, folding in any queued goal/chat steering, then drives the boundary that follows (:func:_advance).

Parameters:

Name	Type	Description	Default
session	Session	Active DB session for this call.	required
registry	SessionRegistry	The live-session registry.	required
session_id	int	The paused retest session to resume.	required

Source code in src/revalid/retest_session.py

def continue_session(session: Session, registry: SessionRegistry, session_id: int) -> None:
    """Resume a session the agent handed back — ADR-0034 "Keep going" / reply (#204).

    A no-op unless the session is handed back in :data:`_RESUMABLE_ON_MESSAGE`
    (``awaiting_operator``) with a live agent — a handed-back session that outlived a
    backend restart has no sandbox to resume, so the operator restarts it instead. The
    agent only ever hands back between turns (never with a command still pending), so
    continuing re-runs it, folding in any queued goal/chat steering, then drives the
    boundary that follows (:func:`_advance`).

    Args:
        session: Active DB session for this call.
        registry: The live-session registry.
        session_id: The paused retest session to resume.
    """
    record = session.get(RetestSessionRecord, session_id)
    if record is None or RetestSessionStatus(record.status) not in _RESUMABLE_ON_MESSAGE:
        return
    live = registry.get(session_id)
    if live is None:
        return
    _resume_run(session, registry, session_id, live)
    _advance(session, registry, session_id)

create_session(session, *, finding_id, model, free_launch=False, deferred=False)

Insert a session row and return it.

Parameters:

Name	Type	Description	Default
session	Session	Active DB session.	required
finding_id	int	The finding identity (FR-16) this session retests.	required
model	str	The resolved LLM model string driving the agent.	required
free_launch	bool	Whether the agent's commands auto-run without a per-command human approval. The gate only ever carries a run_command. FR-17 Slice 5.	False
deferred	bool	When True, open the session idle (created but not started) so it waits for an operator Start instead of auto-running — the Restart path (issue #150). Default False opens it working.	False

Source code in src/revalid/retest_session.py

def create_session(
    session: Session,
    *,
    finding_id: int,
    model: str,
    free_launch: bool = False,
    deferred: bool = False,
) -> RetestSessionRecord:
    """Insert a session row and return it.

    Args:
        session: Active DB session.
        finding_id: The finding identity (FR-16) this session retests.
        model: The resolved LLM model string driving the agent.
        free_launch: Whether the agent's commands auto-run without a per-command
            human approval. The gate only ever carries a ``run_command``. FR-17 Slice 5.
        deferred: When ``True``, open the session ``idle`` (created but not started)
            so it waits for an operator ``Start`` instead of auto-running — the
            Restart path (issue #150). Default ``False`` opens it ``working``.
    """
    status = RetestSessionStatus.IDLE if deferred else RetestSessionStatus.WORKING
    record = RetestSessionRecord(
        finding_id=finding_id,
        status=status.value,
        model=model,
        free_launch=free_launch,
    )
    session.add(record)
    session.commit()
    session.refresh(record)
    return record

end_session(session, registry, session_id)

Operator-initiated end: tear down and mark ended (no-op if already terminal).

Acquires live.lock around the teardown for consistency with apply_decision's registry-mutating critical section, even though end_session doesn't touch pending_call_id itself. Cancels any in-flight turn first (issue #204) so a wedged model call does not leave the run thread lingering after the session is gone; the row is already ended (terminal), so the cancelled turn's :class:_TurnAbortedError is swallowed by :func:_fail.

Source code in src/revalid/retest_session.py

def end_session(session: Session, registry: SessionRegistry, session_id: int) -> None:
    """Operator-initiated end: tear down and mark ``ended`` (no-op if already terminal).

    Acquires ``live.lock`` around the teardown for consistency with
    ``apply_decision``'s registry-mutating critical section, even though
    ``end_session`` doesn't touch ``pending_call_id`` itself. Cancels any in-flight
    turn first (issue #204) so a wedged model call does not leave the run thread
    lingering after the session is gone; the row is already ``ended`` (terminal), so
    the cancelled turn's :class:`_TurnAbortedError` is swallowed by :func:`_fail`.
    """
    record = session.get(RetestSessionRecord, session_id)
    if record is None or RetestSessionStatus(record.status) in _TERMINAL:
        return
    set_status(session, session_id, RetestSessionStatus.ENDED)
    live = registry.get(session_id)
    if live is None:
        return
    live.request_cancel()
    with live.lock:
        _teardown(registry, session_id)

is_terminal(status)

Return whether status is one of the terminal retest-session states.

Parameters:

Name	Type	Description	Default
status	RetestSessionStatus	The status to test.	required

Returns:

Type	Description
bool	True for concluded/given_up/ended/error.

Source code in src/revalid/retest_session.py

def is_terminal(status: RetestSessionStatus) -> bool:
    """Return whether ``status`` is one of the terminal retest-session states.

    Args:
        status: The status to test.

    Returns:
        ``True`` for ``concluded``/``given_up``/``ended``/``error``.
    """
    return status in _TERMINAL

load_events_after(session, session_id, after_seq)

Return transcript events with seq > after_seq in order, as plain dicts.

Source code in src/revalid/retest_session.py

def load_events_after(session: Session, session_id: int, after_seq: int) -> list[dict[str, Any]]:
    """Return transcript events with ``seq > after_seq`` in order, as plain dicts."""
    rows = session.scalars(
        select(SessionEventRecord)
        .where(SessionEventRecord.session_id == session_id, SessionEventRecord.seq > after_seq)
        .order_by(SessionEventRecord.seq)
    ).all()
    return [{"seq": r.seq, "kind": r.kind, "payload": r.payload} for r in rows]

record_verdict(session, session_id, status, rationale, *, actor='agent', reason_code='agentic_conclusion')

Persist a determination on the session row + its transcript events.

Appends the VERDICT event, then a STATE_CHANGE event to concluded, BOTH before committing the terminal row (event-before- terminal-status invariant, cf. _fail which already appends its ERROR event before set_status). The WS stream handler closes on terminal AND no new events, polling concurrently with this function on a separate DB session; without this ordering a poll landing between the event appends and the row commit could observe the terminal status with one of the events not yet visible, and close the stream without ever sending it.

The STATE_CHANGE event is required because the frontend derives the session's displayed status only from the latest such event, not from polling the row directly — without it, a conclude would leave the UI showing the pre-verdict status forever. Both terminal producers route here: the agent concluding fixed/still_open (actor="agent") and the operator manually concluding a paused session (actor="operator", the only path that writes inconclusive under ADR-0034).

Source code in src/revalid/retest_session.py

def record_verdict(
    session: Session,
    session_id: int,
    status: VerdictStatus,
    rationale: str,
    *,
    actor: str = "agent",
    reason_code: str = "agentic_conclusion",
) -> None:
    """Persist a determination on the session row + its transcript events.

    Appends the ``VERDICT`` event, then a ``STATE_CHANGE`` event to
    ``concluded``, BOTH before committing the terminal row (event-before-
    terminal-status invariant, cf. ``_fail`` which already appends its
    ``ERROR`` event before ``set_status``). The WS stream handler closes on
    ``terminal AND no new events``, polling concurrently with this function on
    a separate DB session; without this ordering a poll landing between the
    event appends and the row commit could observe the terminal status with
    one of the events not yet visible, and close the stream without ever
    sending it.

    The ``STATE_CHANGE`` event is required because the frontend derives the
    session's displayed status only from the latest such event, not from
    polling the row directly — without it, a conclude would leave the UI showing
    the pre-verdict status forever. Both terminal producers route here: the agent
    concluding ``fixed``/``still_open`` (``actor="agent"``) and the operator
    manually concluding a paused session (``actor="operator"``, the only path that
    writes ``inconclusive`` under ADR-0034).
    """
    record = session.get(RetestSessionRecord, session_id)
    if record is None:
        return
    append_event(
        session,
        session_id,
        SessionEventKind.VERDICT,
        {"status": status.value, "rationale": rationale},
    )
    append_event(
        session,
        session_id,
        SessionEventKind.STATE_CHANGE,
        {"to": RetestSessionStatus.CONCLUDED.value},
    )
    record.status = RetestSessionStatus.CONCLUDED.value
    record.verdict_status = status.value
    record.verdict_rationale = rationale
    record.ended_at = func.now()
    # FR-09/Slice 6a: the conclusion becomes a queryable agentic verdict so a
    # session's outcome reaches the `verdicts` table, the FR-10 audit, and the
    # FR-12 export — the agent's own (actor="agent") with no human action, or the
    # operator's manual conclude (actor="operator").
    session.add(
        VerdictRecord.agentic(
            finding_id=record.finding_id,
            session_id=session_id,
            status=status,
            rationale=rationale,
            actor=actor,
            reason_code=reason_code,
            evidence=_build_agentic_evidence(session, session_id, rationale).model_dump(),
        )
    )
    session.commit()

reopen_session(session, session_id)

Reopen a concluded session so the operator can keep testing (issue #214).

Withdraws the recorded verdict and returns the session to idle, whose wake path re-provisions the sandbox and continues from the transcript (goal + scope reconstructed there). The verdict is kept in the transcript — the VERDICT event plus a new VERDICT_CANCELLED event — which is the append-only audit for an agentic session (ADR-0025); its row in the queryable verdicts projection is removed because a withdrawn verdict is no longer a current determination, so the finding stops showing an outcome the operator retracted.

A pure DB operation (the session is already torn down, so the live registry is never touched). A no-op unless the session is concluded — so a premature or duplicate call is safe.

Parameters:

Name	Type	Description	Default
session	Session	Active DB session for this call.	required
session_id	int	The concluded retest session to reopen.	required

Source code in src/revalid/retest_session.py

def reopen_session(session: Session, session_id: int) -> None:
    """Reopen a concluded session so the operator can keep testing (issue #214).

    Withdraws the recorded verdict and returns the session to ``idle``, whose wake
    path re-provisions the sandbox and continues from the transcript (goal + scope
    reconstructed there). The verdict is **kept in the transcript** — the ``VERDICT``
    event plus a new ``VERDICT_CANCELLED`` event — which is the append-only audit
    for an agentic session (ADR-0025); its row in the queryable ``verdicts``
    projection is removed because a withdrawn verdict is no longer a current
    determination, so the finding stops showing an outcome the operator retracted.

    A pure DB operation (the session is already torn down, so the live registry is
    never touched). A no-op unless the session is ``concluded`` — so a premature or
    duplicate call is safe.

    Args:
        session: Active DB session for this call.
        session_id: The concluded retest session to reopen.
    """
    record = session.get(RetestSessionRecord, session_id)
    if record is None or RetestSessionStatus(record.status) is not RetestSessionStatus.CONCLUDED:
        return
    append_event(
        session,
        session_id,
        SessionEventKind.VERDICT_CANCELLED,
        {"status": record.verdict_status},
    )
    append_event(
        session,
        session_id,
        SessionEventKind.STATE_CHANGE,
        {"to": RetestSessionStatus.IDLE.value},
    )
    for verdict in session.scalars(
        select(VerdictRecord).where(VerdictRecord.session_id == session_id)
    ):
        session.delete(verdict)
    record.status = RetestSessionStatus.IDLE.value
    record.verdict_status = None
    record.verdict_rationale = None
    record.ended_at = None
    session.commit()

restart_model(session, registry, session_id)

Abort the in-flight turn and re-run it to unstick a wedged model (issue #204).

The "restart model" console action. Asks the live session to cancel its current turn and re-run it (:meth:LiveSession.request_restart); a turn_restarted marker is appended so the transcript shows the operator intervened. A no-op when the session is not live or no turn is in flight (nothing to unstick) — the console only offers the action while the agent is working.

Parameters:

Name	Type	Description	Default
session	Session	Active DB session for this call.	required
registry	SessionRegistry	The live-session registry.	required
session_id	int	The retest session whose turn to restart.	required

Source code in src/revalid/retest_session.py

def restart_model(session: Session, registry: SessionRegistry, session_id: int) -> None:
    """Abort the in-flight turn and re-run it to unstick a wedged model (issue #204).

    The "restart model" console action. Asks the live session to cancel its current
    turn and re-run it (:meth:`LiveSession.request_restart`); a ``turn_restarted``
    marker is appended so the transcript shows the operator intervened. A no-op when
    the session is not live or no turn is in flight (nothing to unstick) — the
    console only offers the action while the agent is working.

    Args:
        session: Active DB session for this call.
        registry: The live-session registry.
        session_id: The retest session whose turn to restart.
    """
    live = registry.get(session_id)
    if live is None:
        return
    if live.request_restart():
        append_event(session, session_id, SessionEventKind.TURN_RESTARTED, {})

resume_session(session, registry, session_id)

Operator resumes a stopped session — Resume (issue #150).

Clears the stopped flag and continues where the pause left off: if a command was held pending when the operator stopped, the gate re-opens (awaiting_command, then the free-launch loop drives it if enabled); otherwise the agent is re-run for its next step. A no-op unless the session is in STOPPED with a live agent (a stopped session that outlived a backend restart has no sandbox — the operator restarts it instead).

Source code in src/revalid/retest_session.py

def resume_session(session: Session, registry: SessionRegistry, session_id: int) -> None:
    """Operator resumes a stopped session — Resume (issue #150).

    Clears the ``stopped`` flag and continues where the pause left off: if a
    command was held pending when the operator stopped, the gate re-opens
    (``awaiting_command``, then the free-launch loop drives it if enabled);
    otherwise the agent is re-run for its next step. A no-op unless the session
    is in ``STOPPED`` with a live agent (a stopped session that outlived a backend
    restart has no sandbox — the operator restarts it instead).
    """
    record = session.get(RetestSessionRecord, session_id)
    if record is None or RetestSessionStatus(record.status) is not RetestSessionStatus.STOPPED:
        return
    live = registry.get(session_id)
    if live is None:
        return
    live.stopped = False
    if live.pending_call_id is not None:
        set_status(session, session_id, RetestSessionStatus.AWAITING_COMMAND)
        _advance(session, registry, session_id)
    else:
        _resume_run(session, registry, session_id, live)
        _advance(session, registry, session_id)

resume_with_message_at_gate(session, registry, session_id)

Steer a command awaiting approval with an operator message (Claude-Code gate).

The message-routing rule's gate case (ADR-0042): a message sent instead of approving withdraws the pending command and re-runs the agent with the message as a first-class user turn, then drives the boundary that follows. A no-op if the session is not live or a concurrent approve/reject already took the command — the message stays buffered (recorded by :func:submit_message) for the next boundary.

Parameters:

Name	Type	Description	Default
session	Session	Active DB session for this call.	required
registry	SessionRegistry	The live-session registry.	required
session_id	int	The retest session whose gate the message steers.	required

Source code in src/revalid/retest_session.py

def resume_with_message_at_gate(
    session: Session, registry: SessionRegistry, session_id: int
) -> None:
    """Steer a command awaiting approval with an operator message (Claude-Code gate).

    The message-routing rule's gate case (ADR-0042): a message sent instead of
    approving withdraws the pending command and re-runs the agent with the message as
    a first-class user turn, then drives the boundary that follows. A no-op if the
    session is not live or a concurrent approve/reject already took the command — the
    message stays buffered (recorded by :func:`submit_message`) for the next boundary.

    Args:
        session: Active DB session for this call.
        registry: The live-session registry.
        session_id: The retest session whose gate the message steers.
    """
    live = registry.get(session_id)
    if live is None:
        return
    if _steer_pending_command(session, registry, session_id, live):
        _advance(session, registry, session_id)

run_agent_step(agent, user_prompt, *, session_id, deps, message_history=None, deferred_tool_results=None, channel=DELTAS, live=None)

Run one agent turn, streaming its live tokens to the console (issue #140).

Replaces agent.run_sync at every step site. The turn's result is unchanged — same output union, same message history, same deferred-tool handling — so the orchestrator's state machine is untouched; the only addition is that tokens are published to channel as they arrive.

What actually streams. This agent's output is structured (ConcludeOutput) or a gated tool request, and its commands and prose reach the operator as tool arguments, which the model emits whole rather than incrementally. What it does stream token-by-token is its reasoning: measured against a live ollama:qwen3:14b, one turn produced 746 thinking deltas and zero text or tool-argument deltas. So the reasoning is what the console shows while a turn is in flight — which is exactly the stretch the operator previously spent watching a motionless spinner.

Text deltas are forwarded too, for models that narrate in plain parts rather than a thinking part. Tool-argument deltas are deliberately not: they arrive as partial JSON ({"rationale": "I will che), and rendering that would show the operator half-escaped syntax rather than a sentence.

Runs the async stream on its own event loop via :func:_run_cancellable_turn, which also carries the issue #204 cancel/retry: an operator unstick (:meth:LiveSession.request_restart) re-runs the same turn from the top, while a cancel for teardown surfaces as :class:_TurnAbortedError. Step sites are already background/worker threads with no loop running, and the orchestrator around them is synchronous — converting the whole state machine to async would be a far larger change for no behavioural gain here.

Parameters:

Name	Type	Description	Default
agent	RetestAgent	The retest agent to run.	required
user_prompt	str | None	The turn's prompt (None when resuming from a tool result).	required
session_id	int	The session whose console receives the tokens.	required
deps	RetestSessionDeps	The tool dependencies for this turn.	required
message_history	list[ModelMessage] | None	Prior turns, when continuing a run.	None
deferred_tool_results	DeferredToolResults | None	Approvals/denials resuming a gated call.	None
channel	DeltaChannel	The live-token channel (injectable for tests).	DELTAS
live	LiveSession | None	The live session, when its turn should be cancellable (issue #204). None (agent-unit tests) runs a plain, non-cancellable turn.	None

Returns:

Type	Description
AgentRunResult[RetestOutput]	The completed run result, exactly as run_sync would have returned it.

Raises:

Type	Description
RuntimeError	If the stream ends without producing a run result, which would otherwise surface as a confusing None far from here.
_TurnAbortedError	If the turn was cancelled for teardown (not to retry).

Source code in src/revalid/retest_session.py

def run_agent_step(
    agent: RetestAgent,
    user_prompt: str | None,
    *,
    session_id: int,
    deps: RetestSessionDeps,
    message_history: list[ModelMessage] | None = None,
    deferred_tool_results: DeferredToolResults | None = None,
    channel: DeltaChannel = DELTAS,
    live: LiveSession | None = None,
) -> AgentRunResult[RetestOutput]:
    """Run one agent turn, streaming its live tokens to the console (issue #140).

    Replaces ``agent.run_sync`` at every step site. The turn's *result* is
    unchanged — same output union, same message history, same deferred-tool
    handling — so the orchestrator's state machine is untouched; the only
    addition is that tokens are published to ``channel`` as they arrive.

    **What actually streams.** This agent's output is structured
    (``ConcludeOutput``) or a gated tool request, and its commands and prose
    reach the operator as *tool arguments*, which the model emits whole rather
    than incrementally. What it does stream token-by-token is its **reasoning**:
    measured against a live ``ollama:qwen3:14b``, one turn produced 746 thinking
    deltas and zero text or tool-argument deltas. So the reasoning is what the
    console shows while a turn is in flight — which is exactly the stretch the
    operator previously spent watching a motionless spinner.

    Text deltas are forwarded too, for models that narrate in plain parts rather
    than a thinking part. Tool-argument deltas are deliberately **not**: they
    arrive as partial JSON (``{"rationale": "I will che``), and rendering that
    would show the operator half-escaped syntax rather than a sentence.

    Runs the async stream on its own event loop via :func:`_run_cancellable_turn`,
    which also carries the issue #204 cancel/retry: an operator *unstick*
    (:meth:`LiveSession.request_restart`) re-runs the same turn from the top, while
    a cancel for teardown surfaces as :class:`_TurnAbortedError`. Step sites are
    already background/worker threads with no loop running, and the orchestrator
    around them is synchronous — converting the whole state machine to async would
    be a far larger change for no behavioural gain here.

    Args:
        agent: The retest agent to run.
        user_prompt: The turn's prompt (``None`` when resuming from a tool result).
        session_id: The session whose console receives the tokens.
        deps: The tool dependencies for this turn.
        message_history: Prior turns, when continuing a run.
        deferred_tool_results: Approvals/denials resuming a gated call.
        channel: The live-token channel (injectable for tests).
        live: The live session, when its turn should be cancellable (issue #204).
            ``None`` (agent-unit tests) runs a plain, non-cancellable turn.

    Returns:
        The completed run result, exactly as ``run_sync`` would have returned it.

    Raises:
        RuntimeError: If the stream ends without producing a run result, which
            would otherwise surface as a confusing ``None`` far from here.
        _TurnAbortedError: If the turn was cancelled for teardown (not to retry).
    """

    async def drive() -> AgentRunResult[RetestOutput]:
        result: AgentRunResult[RetestOutput] | None = None
        async with agent.run_stream_events(
            user_prompt,
            deps=deps,
            message_history=message_history,
            deferred_tool_results=deferred_tool_results,
        ) as events:
            async for event in events:
                if isinstance(event, PartDeltaEvent) and isinstance(
                    event.delta, ThinkingPartDelta | TextPartDelta
                ):
                    channel.publish(session_id, event.delta.content_delta or "")
                elif isinstance(event, AgentRunResultEvent):
                    result = event.result
        if result is None:  # pragma: no cover - the library always ends with a result
            raise RuntimeError("agent stream ended without a result")
        return result

    return _run_cancellable_turn(drive, session_id=session_id, channel=channel, live=live)

session_goal(session, session_id)

Return the session's current goal steps (latest plan_updated), or empty.

Reads the goal back from the transcript so a deferred (idle) session's Start (issue #150) reconstructs the goal recorded at create time, without any in-memory carry that a backend restart would lose.

Source code in src/revalid/retest_session.py

def session_goal(session: Session, session_id: int) -> tuple[str, ...]:
    """Return the session's current goal steps (latest ``plan_updated``), or empty.

    Reads the goal back from the transcript so a deferred (``idle``) session's
    Start (issue #150) reconstructs the goal recorded at create time, without any
    in-memory carry that a backend restart would lose.
    """
    payload = _latest_payload(
        load_events_after(session, session_id, 0), SessionEventKind.PLAN_UPDATED
    )
    steps = payload.get("steps") if payload else None
    return tuple(str(s) for s in steps) if isinstance(steps, list) else ()

session_scope(session, session_id)

Return the session's retest scope (the launch target_set endpoints), or empty.

Source code in src/revalid/retest_session.py

def session_scope(session: Session, session_id: int) -> tuple[str, ...]:
    """Return the session's retest scope (the launch ``target_set`` endpoints), or empty."""
    payload = _latest_payload(
        load_events_after(session, session_id, 0), SessionEventKind.TARGET_SET
    )
    endpoints = payload.get("endpoints") if payload else None
    return tuple(str(e) for e in endpoints) if isinstance(endpoints, list) else ()

set_free_launch(session, registry, session_id, enabled)

Toggle free-launch on a live session (FR-17 Slice 5).

Updates the persisted mode + the live flag, records a free_launch_changed transcript event, and — when enabling with a command already pending — auto-approves it (and any that follow) via :func:_advance. A no-op if the session is not live (already ended/concluded, or never started): there is nothing to steer once torn down, and the persisted mode is fixed at that point.

Parameters:

Name	Type	Description	Default
session	Session	Active DB session for this call.	required
registry	SessionRegistry	The live-session registry.	required
session_id	int	The retest session to toggle.	required
enabled	bool	The new free-launch state.	required

Source code in src/revalid/retest_session.py

def set_free_launch(
    session: Session, registry: SessionRegistry, session_id: int, enabled: bool
) -> None:
    """Toggle free-launch on a live session (FR-17 Slice 5).

    Updates the persisted mode + the live flag, records a ``free_launch_changed``
    transcript event, and — when enabling with a command already pending —
    auto-approves it (and any that follow) via :func:`_advance`. A no-op if
    the session is not live (already ended/concluded, or never started): there is
    nothing to steer once torn down, and the persisted mode is fixed at that point.

    Args:
        session: Active DB session for this call.
        registry: The live-session registry.
        session_id: The retest session to toggle.
        enabled: The new free-launch state.
    """
    live = registry.get(session_id)
    if live is None:
        return
    record = session.get(RetestSessionRecord, session_id)
    if record is None:
        return
    record.free_launch = enabled
    session.commit()
    live.free_launch = enabled
    append_event(session, session_id, SessionEventKind.FREE_LAUNCH_CHANGED, {"enabled": enabled})
    if enabled:
        _advance(session, registry, session_id)

set_goal(session, registry, session_id, steps)

Set the user-owned goal on a non-terminal session (FR-17 6b-ii).

Appends a plan_updated transcript event so the "Current goal" panel reflects the edit (and it replays); when a live agent is attached, the goal is also queued and delivered to it as a first-class user turn on the next approve/reject (:func:_resume_with_decision) — pure-queue, never interrupting a run.

The event is emitted for any non-terminal session, live or not: the live orchestration state is process-local, so a session that outlives a backend restart is non-terminal yet has no live agent. Gating the panel update on liveness (the prior behaviour) made such an edit silently vanish — the endpoint still returned 202 but the panel kept the old steps. Emitting regardless keeps the edit visible; queuing stays conditional on a live agent existing to receive it. A no-op only when the session is unknown or already terminal.

Parameters:

Name	Type	Description	Default
session	Session	Active DB session for this call.	required
registry	SessionRegistry	The live-session registry (holds the goal buffer).	required
session_id	int	The retest session whose goal to set.	required
steps	list[str]	The operator's goal steps (replaces the whole goal).	required

Source code in src/revalid/retest_session.py

def set_goal(
    session: Session, registry: SessionRegistry, session_id: int, steps: list[str]
) -> None:
    """Set the user-owned goal on a non-terminal session (FR-17 6b-ii).

    Appends a ``plan_updated`` transcript event so the "Current goal" panel reflects
    the edit (and it replays); when a live agent is attached, the goal is also queued
    and delivered to it as a first-class user turn on the next approve/reject
    (:func:`_resume_with_decision`) — pure-queue, never interrupting a run.

    The event is emitted for **any** non-terminal session, live or not: the live
    orchestration state is process-local, so a session that outlives a backend
    restart is non-terminal yet has no live agent. Gating the panel update on
    liveness (the prior behaviour) made such an edit silently vanish — the endpoint
    still returned 202 but the panel kept the old steps. Emitting regardless keeps
    the edit visible; queuing stays conditional on a live agent existing to receive
    it. A no-op only when the session is unknown or already terminal.

    Args:
        session: Active DB session for this call.
        registry: The live-session registry (holds the goal buffer).
        session_id: The retest session whose goal to set.
        steps: The operator's goal steps (replaces the whole goal).
    """
    record = session.get(RetestSessionRecord, session_id)
    if record is None or is_terminal(RetestSessionStatus(record.status)):
        return
    append_event(session, session_id, SessionEventKind.PLAN_UPDATED, {"steps": list(steps)})
    live = registry.get(session_id)
    if live is not None:
        live.set_pending_goal(steps)

set_status(session, session_id, status)

Move a session to status and record a state_change transcript event.

Source code in src/revalid/retest_session.py

def set_status(session: Session, session_id: int, status: RetestSessionStatus) -> None:
    """Move a session to ``status`` and record a ``state_change`` transcript event."""
    record = session.get(RetestSessionRecord, session_id)
    if record is None:
        return
    record.status = status.value
    session.commit()
    append_event(session, session_id, SessionEventKind.STATE_CHANGE, {"to": status.value})

start_and_step(session, registry, session_id, agent, sandbox, finding_prompt, *, free_launch=False)

Start the sandbox and run the retest agent's first step.

Parameters:

Name	Type	Description	Default
session	Session	Active DB session for this call.	required
registry	SessionRegistry	The live-session registry; a new :class:LiveSession is registered here for session_id.	required
session_id	int	The already-created (working) retest session to drive.	required
agent	RetestAgent	The built retest agent (Task 4).	required
sandbox	Sandbox	The not-yet-started sandbox for this session.	required
finding_prompt	str	The user prompt describing the finding to retest.	required
free_launch	bool	Whether the agent's commands auto-run without a per-command human approval (FR-17 Slice 5). The gate only ever carries a command.	False

Source code in src/revalid/retest_session.py

def start_and_step(
    session: Session,
    registry: SessionRegistry,
    session_id: int,
    agent: RetestAgent,
    sandbox: Sandbox,
    finding_prompt: str,
    *,
    free_launch: bool = False,
) -> None:
    """Start the sandbox and run the retest agent's first step.

    Args:
        session: Active DB session for this call.
        registry: The live-session registry; a new :class:`LiveSession` is
            registered here for ``session_id``.
        session_id: The already-created (``working``) retest session to drive.
        agent: The built retest agent (Task 4).
        sandbox: The not-yet-started sandbox for this session.
        finding_prompt: The user prompt describing the finding to retest.
        free_launch: Whether the agent's commands auto-run without a per-command
            human approval (FR-17 Slice 5). The gate only ever carries a command.
    """
    # Provision against the session's scope (ADR-0041): the launch `target_set`
    # endpoints parsed to their hosts. Lab scope keeps the unchanged internal
    # network; an online host provisions the L3 egress gateway (ADR-0045).
    sandbox.start(scope_hosts(session_scope(session, session_id)))
    live = LiveSession(agent=agent, sandbox=sandbox, free_launch=free_launch)
    registry.put(session_id, live)
    set_status(session, session_id, RetestSessionStatus.WORKING)
    deps = _make_deps(session, session_id, live)
    try:
        result = run_agent_step(agent, finding_prompt, session_id=session_id, deps=deps, live=live)
    except Exception as exc:  # broad on purpose: orchestration boundary, records + tears down
        _fail(session, registry, session_id, str(exc))
        return
    _dispatch_output(session, registry, session_id, result)
    # Deliver any queued message and, in free-launch, auto-approve; else parks.
    _advance(session, registry, session_id)

stop_session(session, registry, session_id)

Operator pauses a running session — Stop (issue #150).

A cooperative pause: sets the live stopped flag and moves the row to the non-terminal STOPPED state, keeping the sandbox alive. A command already running finishes (its output is recorded) and an in-flight agent step, on completion, parks in stopped rather than advancing (see :func:_dispatch_output); the free-launch loop halts (:func:_advance). A no-op if the session is not live or is already terminal or stopped.

Source code in src/revalid/retest_session.py

def stop_session(session: Session, registry: SessionRegistry, session_id: int) -> None:
    """Operator pauses a running session — Stop (issue #150).

    A cooperative pause: sets the live ``stopped`` flag and moves the row to the
    non-terminal ``STOPPED`` state, keeping the sandbox alive. A command already
    running finishes (its output is recorded) and an in-flight agent step, on
    completion, parks in ``stopped`` rather than advancing (see
    :func:`_dispatch_output`); the free-launch loop halts (:func:`_advance`).
    A no-op if the session is not live or is already terminal or stopped.
    """
    record = session.get(RetestSessionRecord, session_id)
    if record is None or RetestSessionStatus(record.status) in _TERMINAL:
        return
    live = registry.get(session_id)
    if live is None or RetestSessionStatus(record.status) is RetestSessionStatus.STOPPED:
        return
    live.stopped = True
    set_status(session, session_id, RetestSessionStatus.STOPPED)

submit_human_command(session, registry, session_id, command, *, timeout=MAX_COMMAND_TIMEOUT)

Run a manual operator command (!) in the live session's sandbox (FR-17 Slice 2).

The human's own commands are ungated (single trusted user, ADR-0008) and run through the same discrete sandbox.exec the agent uses — no shared PTY (ADR-0026). The command + its result are recorded as a HUMAN_COMMAND transcript event (so the terminal shows it) and buffered on the live session so the agent observes it on its next turn.

A no-op if the session is not live (already ended/concluded, or never started) — there is no sandbox to run in once torn down.

Parameters:

Name	Type	Description	Default
session	Session	Active DB session for this call.	required
registry	SessionRegistry	The live-session registry (holds the sandbox + observation buffer).	required
session_id	int	The retest session to run the command in.	required
command	str	The exact shell command the operator submitted (without the !).	required
timeout	int	Per-command cap in seconds. Defaults to the hard ceiling — the operator ran it deliberately (it may be a slow scan) — and is clamped to that ceiling so even a manual command can never wedge the sandbox.	MAX_COMMAND_TIMEOUT

Source code in src/revalid/retest_session.py

def submit_human_command(
    session: Session,
    registry: SessionRegistry,
    session_id: int,
    command: str,
    *,
    timeout: int = MAX_COMMAND_TIMEOUT,
) -> None:
    """Run a manual operator command (`!`) in the live session's sandbox (FR-17 Slice 2).

    The human's own commands are **ungated** (single trusted user, ADR-0008) and
    run through the *same* discrete ``sandbox.exec`` the agent uses — no shared
    PTY (ADR-0026). The command + its result are recorded as a ``HUMAN_COMMAND``
    transcript event (so the terminal shows it) and buffered on the live session
    so the agent observes it on its next turn.

    A no-op if the session is not live (already ended/concluded, or never
    started) — there is no sandbox to run in once torn down.

    Args:
        session: Active DB session for this call.
        registry: The live-session registry (holds the sandbox + observation buffer).
        session_id: The retest session to run the command in.
        command: The exact shell command the operator submitted (without the `!`).
        timeout: Per-command cap in seconds. Defaults to the hard ceiling — the
            operator ran it deliberately (it may be a slow scan) — and is clamped
            to that ceiling so even a manual command can never wedge the sandbox.
    """
    live = registry.get(session_id)
    if live is None:
        return
    result = live.sandbox.exec(command, timeout=clamp_timeout(timeout))
    append_event(
        session,
        session_id,
        SessionEventKind.HUMAN_COMMAND,
        {
            "command": command,
            "stdout": result.stdout,
            "stderr": result.stderr,
            "exit_code": result.exit_code,
            "elapsed_ms": result.elapsed_ms,
        },
    )
    live.observe(_summarize_human_command(command, result))

submit_message(session, registry, session_id, text)

Record an operator chat message and buffer it for the agent's next turn (FR-17).

Always recorded as a HUMAN_MESSAGE transcript event (so the chat shows it and it replays) — even for a session that outlived a backend restart and has no live agent, so a message is never silently lost (ADR-0042). When the session is live it is also buffered for delivery to the agent as a first-class user turn at the next turn boundary (:func:_advance / :func:_resume_with_decision) — the operator's voice, distinct from the ! command path (:func:submit_human_command).

Parameters:

Name	Type	Description	Default
session	Session	Active DB session for this call.	required
registry	SessionRegistry	The live-session registry (holds the message buffer).	required
session_id	int	The retest session to message.	required
text	str	The exact operator message.	required

Source code in src/revalid/retest_session.py

def submit_message(session: Session, registry: SessionRegistry, session_id: int, text: str) -> None:
    """Record an operator chat message and buffer it for the agent's next turn (FR-17).

    Always recorded as a ``HUMAN_MESSAGE`` transcript event (so the chat shows it and
    it replays) — even for a session that outlived a backend restart and has no live
    agent, so a message is never silently lost (ADR-0042). When the session is live it
    is also buffered for delivery to the agent as a first-class user turn at the next
    turn boundary (:func:`_advance` / :func:`_resume_with_decision`) — the operator's
    *voice*, distinct from the `!` command path (:func:`submit_human_command`).

    Args:
        session: Active DB session for this call.
        registry: The live-session registry (holds the message buffer).
        session_id: The retest session to message.
        text: The exact operator message.
    """
    append_event(session, session_id, SessionEventKind.HUMAN_MESSAGE, {"text": text})
    live = registry.get(session_id)
    if live is not None:
        live.receive_message(text)

Transient live-token channel for the FR-17 retest console (issue #140).

The retest transcript (session_events) is the durable, append-only record a verdict is re-derived from (FR-10). This channel is the opposite: an in-memory, per-session buffer of the tokens the model emits while it is thinking, so the console can show the reasoning as it is written instead of a static spinner for the length of an LLM call.

Deliberately not persisted. A half-finished thought is not evidence, and writing it to the transcript would put text into the audit trail that no verdict was ever derived from. Deltas are dropped as soon as the turn they belong to lands as a real transcript event, and the whole buffer is dropped when the session ends. A reader that misses them (a browser opened mid-turn) loses nothing that matters — the persisted events still tell the whole story.

DeltaChannel

Thread-safe fan-out of live model tokens, keyed by retest session id.

One writer (the agent step running on a worker thread) and one reader (the WebSocket loop on the event loop), so a plain lock is enough; readers pull with a cursor rather than being pushed to, which keeps the WebSocket handler a simple poll and avoids needing an async queue per connection.

Source code in src/revalid/deltas.py

class DeltaChannel:
    """Thread-safe fan-out of live model tokens, keyed by retest session id.

    One writer (the agent step running on a worker thread) and one reader (the
    WebSocket loop on the event loop), so a plain lock is enough; readers pull
    with a cursor rather than being pushed to, which keeps the WebSocket handler
    a simple poll and avoids needing an async queue per connection.
    """

    def __init__(self) -> None:
        """Create an empty channel."""
        self._buffers: dict[int, _Buffer] = {}
        self._lock = threading.Lock()

    def publish(self, session_id: int, chunk: str) -> None:
        """Append one token chunk for ``session_id`` (ignored when empty)."""
        if not chunk:
            return
        with self._lock:
            buffer = self._buffers.setdefault(session_id, _Buffer())
            buffer.chunks.append(chunk)
            overflow = len(buffer.chunks) - MAX_BUFFERED_DELTAS
            if overflow > 0:
                del buffer.chunks[:overflow]
                buffer.dropped += overflow

    def read_after(self, session_id: int, cursor: int) -> tuple[str, int]:
        """Return the text buffered after ``cursor``, and the new cursor.

        Args:
            session_id: The session to read.
            cursor: The index returned by the previous call (0 to start).

        Returns:
            The concatenated new chunks (empty when there are none) and the
            cursor to pass next time. Chunks are joined here rather than sent
            individually because the console appends them to one growing string
            anyway, and one frame per token would flood the socket.
        """
        with self._lock:
            buffer = self._buffers.get(session_id)
            if buffer is None:
                return "", cursor
            total = buffer.dropped + len(buffer.chunks)
            if cursor >= total:
                return "", total
            start = max(cursor - buffer.dropped, 0)
            return "".join(buffer.chunks[start:]), total

    def clear(self, session_id: int) -> None:
        """Drop a session's buffer — its turn landed, or the session ended."""
        with self._lock:
            self._buffers.pop(session_id, None)

__init__()

Create an empty channel.

Source code in src/revalid/deltas.py

def __init__(self) -> None:
    """Create an empty channel."""
    self._buffers: dict[int, _Buffer] = {}
    self._lock = threading.Lock()

clear(session_id)

Drop a session's buffer — its turn landed, or the session ended.

Source code in src/revalid/deltas.py

def clear(self, session_id: int) -> None:
    """Drop a session's buffer — its turn landed, or the session ended."""
    with self._lock:
        self._buffers.pop(session_id, None)

publish(session_id, chunk)

Append one token chunk for session_id (ignored when empty).

Source code in src/revalid/deltas.py

def publish(self, session_id: int, chunk: str) -> None:
    """Append one token chunk for ``session_id`` (ignored when empty)."""
    if not chunk:
        return
    with self._lock:
        buffer = self._buffers.setdefault(session_id, _Buffer())
        buffer.chunks.append(chunk)
        overflow = len(buffer.chunks) - MAX_BUFFERED_DELTAS
        if overflow > 0:
            del buffer.chunks[:overflow]
            buffer.dropped += overflow

read_after(session_id, cursor)

Return the text buffered after cursor, and the new cursor.

Parameters:

Name	Type	Description	Default
session_id	int	The session to read.	required
cursor	int	The index returned by the previous call (0 to start).	required

Returns:

Type	Description
str	The concatenated new chunks (empty when there are none) and the
int	cursor to pass next time. Chunks are joined here rather than sent
tuple[str, int]	individually because the console appends them to one growing string
tuple[str, int]	anyway, and one frame per token would flood the socket.

Source code in src/revalid/deltas.py

def read_after(self, session_id: int, cursor: int) -> tuple[str, int]:
    """Return the text buffered after ``cursor``, and the new cursor.

    Args:
        session_id: The session to read.
        cursor: The index returned by the previous call (0 to start).

    Returns:
        The concatenated new chunks (empty when there are none) and the
        cursor to pass next time. Chunks are joined here rather than sent
        individually because the console appends them to one growing string
        anyway, and one frame per token would flood the socket.
    """
    with self._lock:
        buffer = self._buffers.get(session_id)
        if buffer is None:
            return "", cursor
        total = buffer.dropped + len(buffer.chunks)
        if cursor >= total:
            return "", total
        start = max(cursor - buffer.dropped, 0)
        return "".join(buffer.chunks[start:]), total

Verdicts, audit and export

Audit-trail verdict re-derivation (FR-10, NFR-02, ADR-0025, ADR-0030).

A persisted (agentic) verdict must be reproducible from the audit trail alone, with no re-execution. An agentic verdict (FR-17) is a human-adjudicated judgment over a whole session, so its audit trail is that session's append-only transcript (ADR-0025's NFR-02 shift). It re-derives by re-projecting the authoritative transcript event — the verdict event for the agent's record, the latest verdict_adjudicated event for an operator adjudication — and confirming the stored row still equals it (a denormalization-integrity check).

:func:rederive_run recomputes every stored verdict and diffs it against storage. A clean run proves reproducibility (FR-10 acceptance / NFR-02); any discrepancy flags a verdict that has drifted from the transcript it was derived from. (The old FR-04/05/07-09 batch verdict re-derivation was removed with the batch path in FR-17 6b-iii.)

AuditReport dataclass

Outcome of re-deriving every stored verdict (FR-10 acceptance).

Attributes:

Name	Type	Description
total	int	Number of stored verdicts examined.
reproduced	int	How many re-derived identically to storage.
discrepancies	tuple[Discrepancy, ...]	The verdicts that did not (empty on a clean run).

Source code in src/revalid/audit.py

@dataclass(frozen=True)
class AuditReport:
    """Outcome of re-deriving every stored verdict (FR-10 acceptance).

    Attributes:
        total: Number of stored verdicts examined.
        reproduced: How many re-derived identically to storage.
        discrepancies: The verdicts that did not (empty on a clean run).
    """

    total: int
    reproduced: int
    discrepancies: tuple[Discrepancy, ...] = ()

    @property
    def ok(self) -> bool:
        """True iff every stored verdict re-derived exactly from its transcript."""
        return not self.discrepancies

ok property

True iff every stored verdict re-derived exactly from its transcript.

Discrepancy dataclass

A stored verdict whose re-derivation no longer matches the transcript.

Attributes:

Name	Type	Description
verdict_id	int	The stored verdict's row id.
finding_id	int	The finding the verdict belongs to.
stored	str	The stored status/rationale.
rederived	str	The status/rationale re-projected from the transcript.

Source code in src/revalid/audit.py

@dataclass(frozen=True)
class Discrepancy:
    """A stored verdict whose re-derivation no longer matches the transcript.

    Attributes:
        verdict_id: The stored verdict's row id.
        finding_id: The finding the verdict belongs to.
        stored: The stored ``status/rationale``.
        rederived: The ``status/rationale`` re-projected from the transcript.
    """

    verdict_id: int
    finding_id: int
    stored: str
    rederived: str

rederive_run(session)

Re-derive every stored verdict from the audit trail and diff against storage.

Reproduces each agentic verdict from its session transcript (FR-10 AC) — no re-execution. Returns counts and any discrepancies (empty when the transcript fully reproduces the verdicts).

Source code in src/revalid/audit.py

def rederive_run(session: Session) -> AuditReport:
    """Re-derive every stored verdict from the audit trail and diff against storage.

    Reproduces each agentic verdict from its session transcript (FR-10 AC) — no
    re-execution. Returns counts and any discrepancies (empty when the transcript
    fully reproduces the verdicts).
    """
    records = list(session.scalars(select(VerdictRecord).order_by(VerdictRecord.id)))
    discrepancies: list[Discrepancy] = []
    for record in records:
        discrepancy = _rederive_agentic(session, record)
        if discrepancy is not None:
            discrepancies.append(discrepancy)
    return AuditReport(
        total=len(records),
        reproduced=len(records) - len(discrepancies),
        discrepancies=tuple(discrepancies),
    )

Versioned, schema-validated run export (FR-12).

A run is the full state a revalidation produces — the uploaded reports, the findings extracted from them, every retest-plan version, and every evidence-backed verdict. :func:build_export assembles that state into a single :class:RunExport: a self-contained, versioned JSON document the evaluation harness (FR-15) consumes.

The document is versioned by :data:SCHEMA_VERSION — bumped whenever the export shape changes — so a consumer can tell which contract a file follows. :func:export_schema emits the JSON Schema the document validates against; it is generated from these models (never hand-written) and published to docs/reference/schemas/ via make export-schema, guarded against drift by tests/unit/test_export.py.

FindingExport

Bases: BaseModel

A persisted finding: current content plus full version history and notes (FR-02/03/16).

finding is the current version's content (the field pre-FR-16 consumers read); version names which version that is; versions is the complete append-only history (oldest first, extraction = v1) and notes the stage-tagged annotation log (chronological) — the audit-grade record (FR-10).

Source code in src/revalid/export.py

class FindingExport(BaseModel):
    """A persisted finding: current content plus full version history and notes (FR-02/03/16).

    ``finding`` is the *current* version's content (the field pre-FR-16 consumers
    read); ``version`` names which version that is; ``versions`` is the complete
    append-only history (oldest first, extraction = v1) and ``notes`` the
    stage-tagged annotation log (chronological) — the audit-grade record (FR-10).
    """

    model_config = ConfigDict(frozen=True)

    id: int
    report_id: int | None
    version: int
    finding: Finding
    versions: tuple[FindingVersionExport, ...]
    notes: tuple[NoteExport, ...]

FindingVersionExport

Bases: BaseModel

One immutable version of a finding's content (FR-16).

origin is extraction for version 1 and edit for operator revisions; edited_by/reason carry the edit lineage.

Source code in src/revalid/export.py

class FindingVersionExport(BaseModel):
    """One immutable version of a finding's content (FR-16).

    ``origin`` is ``extraction`` for version 1 and ``edit`` for operator
    revisions; ``edited_by``/``reason`` carry the edit lineage.
    """

    model_config = ConfigDict(frozen=True)

    version: int
    origin: str
    edited_by: str | None
    reason: str
    created_at: datetime
    finding: Finding

Generator

Bases: BaseModel

Provenance of the tool that produced the export (NFR-02 lineage).

Source code in src/revalid/export.py

class Generator(BaseModel):
    """Provenance of the tool that produced the export (NFR-02 lineage)."""

    model_config = ConfigDict(frozen=True)

    tool: str
    version: str

NoteExport

Bases: BaseModel

One stage-tagged operator note on a finding (FR-16).

Source code in src/revalid/export.py

class NoteExport(BaseModel):
    """One stage-tagged operator note on a finding (FR-16)."""

    model_config = ConfigDict(frozen=True)

    id: int
    stage: str
    body: str
    author: str
    created_at: datetime

ReportExport

Bases: BaseModel

An uploaded report and its ingest outcome (FR-01).

Source code in src/revalid/export.py

class ReportExport(BaseModel):
    """An uploaded report and its ingest outcome (FR-01)."""

    model_config = ConfigDict(frozen=True)

    id: int
    filename: str
    status: str
    model: str
    finding_count: int
    created_at: datetime

RunExport

Bases: BaseModel

A complete revalidation run as one versioned JSON document (FR-12).

Source code in src/revalid/export.py

class RunExport(BaseModel):
    """A complete revalidation run as one versioned JSON document (FR-12)."""

    model_config = ConfigDict(frozen=True)

    schema_version: str
    generated_at: datetime
    generator: Generator
    reports: tuple[ReportExport, ...]
    findings: tuple[FindingExport, ...]
    verdicts: tuple[VerdictExport, ...]
    metrics: RunMetrics

RunMetrics

Bases: BaseModel

Descriptive counts and timing over the exported run (FR-15 denominators).

Neutral facts about the run, not correctness scores: the tool has no ground truth, so grading (correct/wrong) is the evaluation harness's job (FR-15). verdicts_by_status always carries every :class:VerdictStatus key so the shape is stable across runs.

Source code in src/revalid/export.py

class RunMetrics(BaseModel):
    """Descriptive counts and timing over the exported run (FR-15 denominators).

    Neutral facts about the run, not correctness scores: the tool has no ground
    truth, so grading (correct/wrong) is the evaluation harness's job (FR-15).
    ``verdicts_by_status`` always carries every :class:`VerdictStatus` key so the
    shape is stable across runs.
    """

    model_config = ConfigDict(frozen=True)

    reports: int
    findings: int
    verdicts: int
    verdicts_by_status: dict[str, int]
    total_elapsed_ms: float
    mean_elapsed_ms: float

VerdictExport

Bases: BaseModel

An agentic verdict with its audit stamps (FR-09/FR-10/FR-17).

Every verdict is a retest-session conclusion (the batch verdict path retired in FR-17 6b-iii): session_id links the session, evidence is the flexible :class:~revalid.domain.AgenticEvidence proof (or None for a human adjudication), and actor is "agent"/"operator".

Source code in src/revalid/export.py

class VerdictExport(BaseModel):
    """An agentic verdict with its audit stamps (FR-09/FR-10/FR-17).

    Every verdict is a retest-session conclusion (the batch verdict path retired
    in FR-17 6b-iii): ``session_id`` links the session, ``evidence`` is the
    flexible :class:`~revalid.domain.AgenticEvidence` proof (or ``None`` for a
    human adjudication), and ``actor`` is ``"agent"``/``"operator"``.
    """

    model_config = ConfigDict(frozen=True)

    id: int
    finding_id: int
    actor: str
    created_at: datetime
    session_id: int | None
    status: VerdictStatus
    reason_code: str
    rationale: str
    matched_indicators: tuple[str, ...]
    evidence: AgenticEvidence | None

build_export(session, *, generated_at=None)

Assemble the full run — reports, findings, verdicts — for export (FR-12).

Reads every entity in id order (deterministic output) and derives the run metrics. Purely a read: it opens no network and mutates nothing.

Parameters:

Name	Type	Description	Default
session	Session	Session over the database to export.	required
generated_at	datetime | None	Stamp for the document; defaults to the current UTC time (injectable so tests and re-runs are deterministic).	None

Returns:

Type	Description
RunExport	The complete run as a :class:RunExport, valid against
RunExport	func:export_schema.

Source code in src/revalid/export.py

def build_export(session: Session, *, generated_at: datetime | None = None) -> RunExport:
    """Assemble the full run — reports, findings, verdicts — for export (FR-12).

    Reads every entity in id order (deterministic output) and derives the run
    metrics. Purely a read: it opens no network and mutates nothing.

    Args:
        session: Session over the database to export.
        generated_at: Stamp for the document; defaults to the current UTC time
            (injectable so tests and re-runs are deterministic).

    Returns:
        The complete run as a :class:`RunExport`, valid against
        :func:`export_schema`.
    """
    reports = tuple(
        _report_export(r) for r in session.scalars(select(ReportRecord).order_by(ReportRecord.id))
    )
    findings = tuple(
        export
        for r in session.scalars(select(FindingRecord).order_by(FindingRecord.id))
        if (export := _finding_export(session, r)) is not None
    )
    verdicts = tuple(
        _verdict_export(r)
        for r in session.scalars(select(VerdictRecord).order_by(VerdictRecord.id))
    )
    return RunExport(
        schema_version=SCHEMA_VERSION,
        generated_at=generated_at if generated_at is not None else datetime.now(UTC),
        generator=Generator(tool="revalid", version=__version__),
        reports=reports,
        findings=findings,
        verdicts=verdicts,
        metrics=_metrics(reports, findings, verdicts),
    )

export_schema()

Return the published JSON Schema the export validates against (FR-12).

Generated from :class:RunExport so the contract can never drift from the document it describes.

Source code in src/revalid/export.py

def export_schema() -> dict[str, Any]:
    """Return the published JSON Schema the export validates against (FR-12).

    Generated from :class:`RunExport` so the contract can never drift from the
    document it describes.
    """
    return RunExport.model_json_schema()

Evaluation harness: score a run's verdicts against ground truth (FR-15 / NFR-01).

The harness consumes an FR-12 run export (:class:revalid.export.RunExport) and a ground-truth file — one expected verdict per evaluation-set finding, plus an ambiguous flag marking findings whose only defensible outcome is inconclusive. For each ground-truth finding it takes the system's latest verdict from the export and classifies it:

• correct — the verdict matches the expected verdict.
• inconclusive — the system hedged (returned inconclusive) where the truth was a definite verdict: a safe miss, not a confident error.
• wrong — the system returned a confident verdict that contradicts the truth. This is the dangerous case; per NFR-01 it counts double in the analysis, and on an ambiguous finding it violates the hard constraint.

:func:evaluate returns an :class:EvalReport with the per-finding rows, the totals, timing, and the NFR-01 pass decision; :func:format_table renders the metrics table for the thesis Results chapter. Nothing here touches the network — scoring is a pure function of the export and the ground truth.

Classification

Bases: StrEnum

How a finding's actual verdict scored against its expected verdict.

Source code in src/revalid/eval.py

class Classification(enum.StrEnum):
    """How a finding's actual verdict scored against its expected verdict."""

    CORRECT = "correct"
    INCONCLUSIVE = "inconclusive"
    WRONG = "wrong"
    NO_VERDICT = "no_verdict"

EvalReport dataclass

The scored evaluation run (FR-15 metrics table / NFR-01 decision).

Attributes:

Name	Type	Description
rows	tuple[EvalRow, ...]	One scored row per matched ground-truth finding.
unmatched_findings	tuple[str, ...]	Export finding titles with no ground-truth entry.
unmatched_ground_truth	tuple[str, ...]	Ground-truth titles absent from the export.

Source code in src/revalid/eval.py

@dataclass(frozen=True)
class EvalReport:
    """The scored evaluation run (FR-15 metrics table / NFR-01 decision).

    Attributes:
        rows: One scored row per matched ground-truth finding.
        unmatched_findings: Export finding titles with no ground-truth entry.
        unmatched_ground_truth: Ground-truth titles absent from the export.
    """

    rows: tuple[EvalRow, ...]
    unmatched_findings: tuple[str, ...] = ()
    unmatched_ground_truth: tuple[str, ...] = ()

    @property
    def total(self) -> int:
        """Number of scored findings."""
        return len(self.rows)

    def _count(self, classification: Classification) -> int:
        return sum(1 for row in self.rows if row.classification is classification)

    @property
    def correct(self) -> int:
        """Findings whose verdict matched the expected verdict."""
        return self._count(Classification.CORRECT)

    @property
    def inconclusive(self) -> int:
        """Findings where the system safely hedged (returned inconclusive)."""
        return self._count(Classification.INCONCLUSIVE)

    @property
    def wrong(self) -> int:
        """Findings given a confident verdict that contradicts the truth."""
        return self._count(Classification.WRONG)

    @property
    def no_verdict(self) -> int:
        """Ground-truth findings that were never retested in this run."""
        return self._count(Classification.NO_VERDICT)

    @property
    def correct_pct(self) -> float:
        """Fraction of scored findings that were correct (0.0 when none)."""
        return self.correct / self.total if self.total else 0.0

    @property
    def confidently_wrong(self) -> int:
        """Count of confident, contradicting verdicts (NFR-01 counts these double)."""
        return self.wrong

    @property
    def weighted_error(self) -> int:
        """Confident errors weighted double, per the NFR-01 analysis rule."""
        return 2 * self.wrong

    @property
    def wrong_on_ambiguous(self) -> int:
        """Confident verdicts on ambiguous findings — the NFR-01 hard-constraint breaches."""
        return sum(1 for row in self.rows if row.ambiguous and row.confidently_wrong)

    @property
    def total_elapsed_ms(self) -> float:
        """Total round-trip time across the scored verdicts."""
        return sum(row.elapsed_ms for row in self.rows)

    @property
    def mean_elapsed_ms(self) -> float:
        """Mean round-trip time across the scored verdicts (0.0 when none)."""
        return self.total_elapsed_ms / self.total if self.total else 0.0

    @property
    def nfr01_pass(self) -> bool:
        """NFR-01: ≥70% correct AND no ambiguous finding given a confident verdict."""
        return self.correct_pct >= NFR01_MIN_CORRECT and self.wrong_on_ambiguous == 0

confidently_wrong property

Count of confident, contradicting verdicts (NFR-01 counts these double).

correct property

Findings whose verdict matched the expected verdict.

correct_pct property

Fraction of scored findings that were correct (0.0 when none).

inconclusive property

Findings where the system safely hedged (returned inconclusive).

mean_elapsed_ms property

Mean round-trip time across the scored verdicts (0.0 when none).

nfr01_pass property

NFR-01: ≥70% correct AND no ambiguous finding given a confident verdict.

no_verdict property

Ground-truth findings that were never retested in this run.

total property

Number of scored findings.

total_elapsed_ms property

Total round-trip time across the scored verdicts.

weighted_error property

Confident errors weighted double, per the NFR-01 analysis rule.

wrong property

Findings given a confident verdict that contradicts the truth.

wrong_on_ambiguous property

Confident verdicts on ambiguous findings — the NFR-01 hard-constraint breaches.

EvalRow dataclass

One scored evaluation-set finding.

Attributes:

Name	Type	Description
finding	str	The ground-truth finding title.
expected	VerdictStatus	The expected verdict.
actual	VerdictStatus | None	The system's latest verdict, or None if it was never retested in this run.
ambiguous	bool	Whether the finding is an NFR-01 hard-constraint case.
classification	Classification	The scoring bucket this finding fell into.
elapsed_ms	float	Round-trip time of the scored verdict's evidence.

Source code in src/revalid/eval.py

@dataclass(frozen=True)
class EvalRow:
    """One scored evaluation-set finding.

    Attributes:
        finding: The ground-truth finding title.
        expected: The expected verdict.
        actual: The system's latest verdict, or ``None`` if it was never
            retested in this run.
        ambiguous: Whether the finding is an NFR-01 hard-constraint case.
        classification: The scoring bucket this finding fell into.
        elapsed_ms: Round-trip time of the scored verdict's evidence.
    """

    finding: str
    expected: VerdictStatus
    actual: VerdictStatus | None
    ambiguous: bool
    classification: Classification
    elapsed_ms: float

    @property
    def confidently_wrong(self) -> bool:
        """True when the system gave a confident verdict that contradicts truth."""
        return self.classification is Classification.WRONG

confidently_wrong property

True when the system gave a confident verdict that contradicts truth.

GroundTruth

Bases: BaseModel

The evaluation set's expected verdicts (FR-15).

Attributes:

Name	Type	Description
target	str	The system under retest the verdicts are expected against (e.g. the pinned vulnerable Juice Shop version).
source_report	str	Provenance of the pentest report the findings came from.
findings	tuple[GroundTruthEntry, ...]	One expected verdict per evaluation-set finding.

Source code in src/revalid/eval.py

class GroundTruth(BaseModel):
    """The evaluation set's expected verdicts (FR-15).

    Attributes:
        target: The system under retest the verdicts are expected against
            (e.g. the pinned vulnerable Juice Shop version).
        source_report: Provenance of the pentest report the findings came from.
        findings: One expected verdict per evaluation-set finding.
    """

    model_config = ConfigDict(frozen=True)

    target: str
    source_report: str
    findings: tuple[GroundTruthEntry, ...]

GroundTruthEntry

Bases: BaseModel

The expected verdict for one evaluation-set finding (FR-15 ground truth).

Attributes:

Name	Type	Description
finding	str	The finding title, matched to the export case-insensitively and whitespace-normalized (see :func:normalize_title).
expected	VerdictStatus	The verdict a correct system should return. For an ambiguous finding this must be inconclusive.
ambiguous	bool	Whether this finding's only defensible outcome is inconclusive (the NFR-01 hard-constraint cases).
note	str	Free-text rationale for the expected verdict (kept in the report).

Source code in src/revalid/eval.py

class GroundTruthEntry(BaseModel):
    """The expected verdict for one evaluation-set finding (FR-15 ground truth).

    Attributes:
        finding: The finding title, matched to the export case-insensitively and
            whitespace-normalized (see :func:`normalize_title`).
        expected: The verdict a correct system should return. For an
            ``ambiguous`` finding this must be ``inconclusive``.
        ambiguous: Whether this finding's only defensible outcome is
            ``inconclusive`` (the NFR-01 hard-constraint cases).
        note: Free-text rationale for the expected verdict (kept in the report).
    """

    model_config = ConfigDict(frozen=True)

    finding: str
    expected: VerdictStatus
    ambiguous: bool = False
    note: str = ""

classify(expected, actual, *, ambiguous)

Score one finding's actual verdict against its expected verdict (NFR-01).

Parameters:

Name	Type	Description	Default
expected	VerdictStatus	The verdict a correct system should return.	required
actual	VerdictStatus | None	The system's verdict, or None if the finding was not retested.	required
ambiguous	bool	Whether the finding's only defensible outcome is inconclusive (unused in the logic — an ambiguous entry simply has expected = inconclusive — but named for call-site clarity).	required

Returns:

Type	Description
Classification	The scoring bucket: NO_VERDICT when nothing ran, CORRECT on a
Classification	match, INCONCLUSIVE when the system safely hedged, else WRONG.

Source code in src/revalid/eval.py

def classify(
    expected: VerdictStatus, actual: VerdictStatus | None, *, ambiguous: bool
) -> Classification:
    """Score one finding's actual verdict against its expected verdict (NFR-01).

    Args:
        expected: The verdict a correct system should return.
        actual: The system's verdict, or ``None`` if the finding was not retested.
        ambiguous: Whether the finding's only defensible outcome is inconclusive
            (unused in the logic — an ambiguous entry simply has ``expected`` =
            ``inconclusive`` — but named for call-site clarity).

    Returns:
        The scoring bucket: ``NO_VERDICT`` when nothing ran, ``CORRECT`` on a
        match, ``INCONCLUSIVE`` when the system safely hedged, else ``WRONG``.
    """
    del ambiguous  # expressed through `expected`; kept for call-site readability
    if actual is None:
        return Classification.NO_VERDICT
    if actual == expected:
        return Classification.CORRECT
    if actual is VerdictStatus.INCONCLUSIVE:
        return Classification.INCONCLUSIVE
    return Classification.WRONG

evaluate(export, ground_truth)

Score every ground-truth finding against the run export (FR-15).

Matches ground-truth entries to export findings by normalized title, takes each finding's latest verdict, and classifies it. Findings present on only one side are surfaced (unmatched_*) rather than silently dropped — an unmatched entry means the ground truth and the run disagree on the finding set and the score would otherwise be quietly wrong.

Parameters:

Name	Type	Description	Default
export	RunExport	The FR-12 run export to score.	required
ground_truth	GroundTruth	The evaluation set's expected verdicts.	required

Returns:

Type	Description
EvalReport	The scored :class:EvalReport.

Source code in src/revalid/eval.py

def evaluate(export: RunExport, ground_truth: GroundTruth) -> EvalReport:
    """Score every ground-truth finding against the run export (FR-15).

    Matches ground-truth entries to export findings by normalized title, takes
    each finding's latest verdict, and classifies it. Findings present on only
    one side are surfaced (``unmatched_*``) rather than silently dropped — an
    unmatched entry means the ground truth and the run disagree on the finding
    set and the score would otherwise be quietly wrong.

    Args:
        export: The FR-12 run export to score.
        ground_truth: The evaluation set's expected verdicts.

    Returns:
        The scored :class:`EvalReport`.
    """
    findings_by_key = {normalize_title(f.finding.title): f for f in export.findings}
    latest = latest_verdict_by_finding(export)
    matched_keys: set[str] = set()
    rows: list[EvalRow] = []
    unmatched_gt: list[str] = []

    for entry in ground_truth.findings:
        key = normalize_title(entry.finding)
        finding = findings_by_key.get(key)
        if finding is None:
            unmatched_gt.append(entry.finding)
            continue
        matched_keys.add(key)
        verdict = latest.get(finding.id)
        actual = verdict.status if verdict is not None else None
        # Agentic verdicts carry no single-request evidence (timing is in the transcript).
        elapsed_ms = (
            verdict.evidence.elapsed_ms
            if verdict is not None and verdict.evidence is not None
            else 0.0
        )
        rows.append(
            EvalRow(
                finding=entry.finding,
                expected=entry.expected,
                actual=actual,
                ambiguous=entry.ambiguous,
                classification=classify(entry.expected, actual, ambiguous=entry.ambiguous),
                elapsed_ms=elapsed_ms,
            )
        )

    unmatched_findings = tuple(
        f.finding.title for key, f in findings_by_key.items() if key not in matched_keys
    )
    return EvalReport(
        rows=tuple(rows),
        unmatched_findings=unmatched_findings,
        unmatched_ground_truth=tuple(unmatched_gt),
    )

format_table(report)

Render the FR-15 metrics table (correct / wrong / inconclusive, totals, timing).

Source code in src/revalid/eval.py

def format_table(report: EvalReport) -> str:
    """Render the FR-15 metrics table (correct / wrong / inconclusive, totals, timing)."""
    lines = ["Evaluation — verdict reliability (FR-15 / NFR-01)", ""]
    lines.extend(_row_line(row) for row in report.rows)
    lines.append("")
    lines.append(
        f"  total={report.total}  correct={report.correct}  "
        f"inconclusive={report.inconclusive}  wrong={report.wrong}"
        + (f"  no_verdict={report.no_verdict}" if report.no_verdict else "")
    )
    lines.append(
        f"  correct={report.correct_pct:.0%}  confidently_wrong={report.confidently_wrong} "
        f"(weighted x2 = {report.weighted_error})  wrong_on_ambiguous={report.wrong_on_ambiguous}"
    )
    lines.append(
        f"  timing: total={report.total_elapsed_ms:.0f}ms  mean={report.mean_elapsed_ms:.0f}ms"
    )
    if report.unmatched_ground_truth:
        lines.append(
            f"  ! ground-truth findings not in the run: {list(report.unmatched_ground_truth)}"
        )
    if report.unmatched_findings:
        lines.append(f"  ! run findings not in the ground truth: {list(report.unmatched_findings)}")
    verdict = "PASS" if report.nfr01_pass else "FAIL"
    lines.append("")
    lines.append(
        f"  NFR-01: {verdict}  (need ≥{NFR01_MIN_CORRECT:.0%} correct and zero "
        f"confident verdicts on ambiguous findings)"
    )
    return "\n".join(lines)

ground_truth_skeleton(export)

Build a fill-in-the-blanks ground-truth skeleton from a run export (FR-15 aid).

Emits one entry per export finding, its finding title already keyed so it matches the run exactly, with expected set to :data:GROUND_TRUTH_TODO for the author to replace. Because that sentinel is not a valid verdict, an unfilled skeleton won't load — the author cannot accidentally score a run against placeholders.

Parameters:

Name	Type	Description	Default
export	RunExport	The FR-12 run export to seed the ground truth from.	required

Returns:

Type	Description
dict[str, Any]	A (skeleton, duplicate_titles) pair. skeleton is JSON-serializable
tuple[str, ...]	and shaped like :class:GroundTruth; duplicate_titles lists finding
tuple[dict[str, Any], tuple[str, ...]]	titles that collapse to the same match key (they would collide in scoring,
tuple[dict[str, Any], tuple[str, ...]]	so the author must disambiguate them).

Source code in src/revalid/eval.py

def ground_truth_skeleton(export: RunExport) -> tuple[dict[str, Any], tuple[str, ...]]:
    """Build a fill-in-the-blanks ground-truth skeleton from a run export (FR-15 aid).

    Emits one entry per export finding, its ``finding`` title already keyed so it
    matches the run exactly, with ``expected`` set to :data:`GROUND_TRUTH_TODO` for
    the author to replace. Because that sentinel is not a valid verdict, an
    unfilled skeleton won't load — the author cannot accidentally score a run
    against placeholders.

    Args:
        export: The FR-12 run export to seed the ground truth from.

    Returns:
        A ``(skeleton, duplicate_titles)`` pair. ``skeleton`` is JSON-serializable
        and shaped like :class:`GroundTruth`; ``duplicate_titles`` lists finding
        titles that collapse to the same match key (they would collide in scoring,
        so the author must disambiguate them).
    """
    seen: set[str] = set()
    duplicates: list[str] = []
    entries: list[dict[str, Any]] = []
    for finding in export.findings:
        title = finding.finding.title
        key = normalize_title(title)
        if key in seen:
            duplicates.append(title)
        seen.add(key)
        entries.append(
            {"finding": title, "expected": GROUND_TRUTH_TODO, "ambiguous": False, "note": ""}
        )
    skeleton = {
        "target": "<pin the evaluated target, e.g. OWASP Juice Shop v17.1.1>",
        "source_report": export.reports[0].filename if export.reports else "<pentest report>",
        "findings": entries,
    }
    return skeleton, tuple(duplicates)

latest_verdict_by_finding(export)

Map each finding id to its latest verdict in the export (highest verdict id).

Source code in src/revalid/eval.py

def latest_verdict_by_finding(export: RunExport) -> dict[int, VerdictExport]:
    """Map each finding id to its latest verdict in the export (highest verdict id)."""
    latest: dict[int, VerdictExport] = {}
    for verdict in export.verdicts:
        current = latest.get(verdict.finding_id)
        if current is None or verdict.id > current.id:
            latest[verdict.finding_id] = verdict
    return latest

load_export(path)

Load and validate an FR-12 run export from disk.

Source code in src/revalid/eval.py

def load_export(path: Path) -> RunExport:
    """Load and validate an FR-12 run export from disk."""
    return RunExport.model_validate(json.loads(path.read_text()))

load_ground_truth(path)

Load and validate a ground-truth file (FR-15).

Source code in src/revalid/eval.py

def load_ground_truth(path: Path) -> GroundTruth:
    """Load and validate a ground-truth file (FR-15)."""
    return GroundTruth.model_validate_json(path.read_text())

normalize_title(title)

Normalize a finding title to its match key (lowercased, whitespace-collapsed).

The single key both scoring (:func:evaluate) and the ground-truth authoring aid (:func:ground_truth_skeleton) use to line findings up, so they can never disagree on what counts as the same finding.

Source code in src/revalid/eval.py

def normalize_title(title: str) -> str:
    """Normalize a finding title to its match key (lowercased, whitespace-collapsed).

    The single key both scoring (:func:`evaluate`) and the ground-truth authoring
    aid (:func:`ground_truth_skeleton`) use to line findings up, so they can never
    disagree on what counts as the same finding.
    """
    return " ".join(title.lower().split())

Corpus chat

FR-18 reports chat: a read-only analytics agent over the report corpus.

A Pydantic AI agent with typed, read-only DB query tools (corpus counts, report list, finding search, finding detail) so it answers natural-language questions about ingested reports, findings, and verdicts with exact data pulled from SQLite — never mutating anything and never launching a retest. It reuses the FR-13 configured backend like every other agent.

The tools are the source of truth, so the agent re-queries the DB on every turn and only the prose turns of a conversation are persisted (:class:~revalid.db.ChatSessionRecord / :class:~revalid.db.ChatMessageRecord). The query functions here are plain, session-taking helpers — unit-testable without an LLM — that the agent's tools wrap thinly.

CorpusOverview

Bases: BaseModel

Whole-corpus counts (reports, findings, verdicts) for the assistant.

Source code in src/revalid/reports_chat.py

class CorpusOverview(BaseModel):
    """Whole-corpus counts (reports, findings, verdicts) for the assistant."""

    reports_total: int
    reports_by_status: dict[str, int]
    findings_total: int
    findings_by_severity: dict[str, int]
    verdicts_total: int
    verdicts_by_status: dict[str, int]

FindingBrief

Bases: BaseModel

A compact finding row (current version) for search results.

Source code in src/revalid/reports_chat.py

class FindingBrief(BaseModel):
    """A compact finding row (current version) for search results."""

    id: int
    report_id: int | None
    title: str
    severity: str
    affected_endpoints: list[str]

FindingDetail

Bases: BaseModel

Full current content of one finding plus its latest verdict, if any.

Source code in src/revalid/reports_chat.py

class FindingDetail(BaseModel):
    """Full current content of one finding plus its latest verdict, if any."""

    id: int
    report_id: int | None
    title: str
    severity: str
    description: str
    impact: str
    attack_vector: str
    affected_endpoints: list[str]
    reproduction_steps: list[str]
    latest_verdict: str | None
    latest_verdict_rationale: str | None

FindingSearch

Bases: BaseModel

A finding search result: the exact total plus a (possibly capped) list.

Source code in src/revalid/reports_chat.py

class FindingSearch(BaseModel):
    """A finding search result: the exact ``total`` plus a (possibly capped) list."""

    total: int
    shown: int
    findings: list[FindingBrief]

ReportBrief

Bases: BaseModel

A compact report row for the assistant's report list.

Source code in src/revalid/reports_chat.py

class ReportBrief(BaseModel):
    """A compact report row for the assistant's report list."""

    id: int
    filename: str
    status: str
    finding_count: int
    archived: bool

ReportsChatDeps dataclass

Runtime dependency injected into the reports agent's tools: a DB session.

Attributes:

Name	Type	Description
session	Session	The read-only DB session every tool queries through.
lock	AbstractContextManager[bool]	Serialises the tool bodies that share session (issue #156). Pydantic AI runs sync tools in worker threads and runs them concurrently when one model turn emits several tool calls — the normal case for a corpus question. A SQLAlchemy Session is not thread-safe, and neither engine this app builds stops the overlap: the in-memory engine shares one connection through StaticPool with check_same_thread=False, and the pysqlite dialect disables that same guard for file databases. Overlapping use therefore corrupts the connection's result/parameter state rather than raising, surfacing as InterfaceError or an IndexError from the result proxy. Concurrency buys nothing here — these are short local SQLite reads — so the tools take turns.

Source code in src/revalid/reports_chat.py

@dataclass
class ReportsChatDeps:
    """Runtime dependency injected into the reports agent's tools: a DB session.

    Attributes:
        session: The read-only DB session every tool queries through.
        lock: Serialises the tool bodies that share ``session`` (issue #156).
            Pydantic AI runs **sync** tools in worker threads and runs them
            *concurrently* when one model turn emits several tool calls — the
            normal case for a corpus question. A SQLAlchemy ``Session`` is not
            thread-safe, and neither engine this app builds stops the overlap:
            the in-memory engine shares one connection through ``StaticPool``
            with ``check_same_thread=False``, and the pysqlite dialect disables
            that same guard for file databases. Overlapping use therefore
            corrupts the connection's result/parameter state rather than
            raising, surfacing as ``InterfaceError`` or an ``IndexError`` from
            the result proxy. Concurrency buys nothing here — these are short
            local SQLite reads — so the tools take turns.
    """

    session: Session
    lock: AbstractContextManager[bool] = field(default_factory=threading.Lock)

answer_question(agent, session, chat, question)

Record the user turn, answer it with the read-only agent, persist the reply (FR-18).

The prior turns become the agent's message history; the agent answers by calling its read-only DB tools over session and the reply is appended to the thread. The first question also sets the thread title. All in one session, committed once the reply is stored.

Parameters:

Name	Type	Description	Default
agent	Agent[ReportsChatDeps, str]	The FR-18 reports agent (a stand-in model in tests).	required
session	Session	The active DB session (shared by persistence and the tools).	required
chat	ChatSessionRecord	The thread to append to.	required
question	str	The operator's new message.	required

Returns:

Type	Description
ChatMessageRecord	The persisted assistant reply row.

Source code in src/revalid/reports_chat.py

def answer_question(
    agent: Agent[ReportsChatDeps, str],
    session: Session,
    chat: ChatSessionRecord,
    question: str,
) -> ChatMessageRecord:
    """Record the user turn, answer it with the read-only agent, persist the reply (FR-18).

    The prior turns become the agent's message history; the agent answers by
    calling its read-only DB tools over ``session`` and the reply is appended to
    the thread. The first question also sets the thread title. All in one session,
    committed once the reply is stored.

    Args:
        agent: The FR-18 reports agent (a stand-in model in tests).
        session: The active DB session (shared by persistence and the tools).
        chat: The thread to append to.
        question: The operator's new message.

    Returns:
        The persisted assistant reply row.
    """
    prior = list_messages(session, chat.id)
    session.add(ChatMessageRecord(chat_id=chat.id, role=USER, content=question))
    if not prior:
        chat.title = _title_from(question)
    session.commit()

    result = agent.run_sync(
        question, deps=ReportsChatDeps(session=session), message_history=_history(prior)
    )
    answer = result.output.strip() or "(no answer)"

    chat.model = agent_model_name(agent)
    reply = ChatMessageRecord(chat_id=chat.id, role=ASSISTANT, content=answer)
    session.add(reply)
    session.commit()
    session.refresh(reply)
    return reply

build_reports_agent(model=None)

Build the FR-18 reports assistant: read-only corpus-query tools + a prose answer.

Parameters:

Name	Type	Description	Default
model	Model | KnownModelName | str | None	A Pydantic AI model instance or name. When omitted, the configured backend is used (REVALID_LLM_MODEL/settings — FR-13); tests pass TestModel/FunctionModel.	None

Returns:

Type	Description
Agent[ReportsChatDeps, str]	An agent whose output is the plain-text answer to show the operator. Its
Agent[ReportsChatDeps, str]	tools query the DB carried in :class:ReportsChatDeps and mutate nothing.

Source code in src/revalid/reports_chat.py

def build_reports_agent(
    model: Model | KnownModelName | str | None = None,
) -> Agent[ReportsChatDeps, str]:
    """Build the FR-18 reports assistant: read-only corpus-query tools + a prose answer.

    Args:
        model: A Pydantic AI model instance or name. When omitted, the configured
            backend is used (``REVALID_LLM_MODEL``/settings — FR-13); tests pass
            ``TestModel``/``FunctionModel``.

    Returns:
        An agent whose output is the plain-text answer to show the operator. Its
        tools query the DB carried in :class:`ReportsChatDeps` and mutate nothing.
    """
    agent: Agent[ReportsChatDeps, str] = Agent(
        model if model is not None else resolve_model(),
        deps_type=ReportsChatDeps,
        output_type=str,
        instructions=_INSTRUCTIONS,
        defer_model_check=True,
    )

    # Every tool body goes through `_read`: the tools may run concurrently in
    # worker threads and share one non-thread-safe Session (issue #156).

    @agent.tool
    def get_corpus_overview(ctx: RunContext[ReportsChatDeps]) -> CorpusOverview:
        """Return whole-corpus counts: reports by status, findings by severity, verdicts."""
        return _read(ctx, corpus_overview)

    @agent.tool
    def list_all_reports(
        ctx: RunContext[ReportsChatDeps], include_archived: bool = False
    ) -> list[ReportBrief]:
        """List reports (id, filename, status, finding count); archived excluded by default."""
        return _read(ctx, lambda s: list_reports(s, include_archived=include_archived))

    @agent.tool
    def search_findings(
        ctx: RunContext[ReportsChatDeps],
        query: str = "",
        severity: str | None = None,
        report_id: int | None = None,
    ) -> FindingSearch:
        """Search current findings by keyword/severity/report; returns an exact total."""
        return _read(
            ctx, lambda s: find_findings(s, query=query, severity=severity, report_id=report_id)
        )

    @agent.tool
    def finding_detail(ctx: RunContext[ReportsChatDeps], finding_id: int) -> FindingDetail | None:
        """Return one finding's full content and its latest verdict, or null if unknown."""
        return _read(ctx, lambda s: get_finding(s, finding_id))

    return agent

corpus_overview(session)

Summarise the whole corpus: report/finding/verdict counts (FR-18).

Parameters:

Name	Type	Description	Default
session	Session	An active read-only DB session.	required

Returns:

Type	Description
CorpusOverview	Totals plus per-status / per-severity breakdowns. Verdicts are counted
CorpusOverview	latest-per-finding (see :func:_latest_verdicts).

Source code in src/revalid/reports_chat.py

def corpus_overview(session: Session) -> CorpusOverview:
    """Summarise the whole corpus: report/finding/verdict counts (FR-18).

    Args:
        session: An active read-only DB session.

    Returns:
        Totals plus per-status / per-severity breakdowns. Verdicts are counted
        latest-per-finding (see :func:`_latest_verdicts`).
    """
    reports = list(session.scalars(select(ReportRecord)))
    versions = _current_versions(session)
    verdicts = _latest_verdicts(session)
    return CorpusOverview(
        reports_total=len(reports),
        reports_by_status=dict(Counter(r.status for r in reports)),
        findings_total=len(versions),
        findings_by_severity=dict(Counter(v.severity for v in versions)),
        verdicts_total=len(verdicts),
        verdicts_by_status=dict(Counter(v.status for v in verdicts)),
    )

create_chat(session)

Create an empty chat thread and return it (committed).

Source code in src/revalid/reports_chat.py

def create_chat(session: Session) -> ChatSessionRecord:
    """Create an empty chat thread and return it (committed)."""
    record = ChatSessionRecord()
    session.add(record)
    session.commit()
    session.refresh(record)
    return record

delete_chat(session, chat_id)

Delete a thread and all its messages (committed); a no-op if it's gone.

Source code in src/revalid/reports_chat.py

def delete_chat(session: Session, chat_id: int) -> None:
    """Delete a thread and all its messages (committed); a no-op if it's gone."""
    record = session.get(ChatSessionRecord, chat_id)
    if record is None:
        return
    session.execute(delete(ChatMessageRecord).where(ChatMessageRecord.chat_id == chat_id))
    session.delete(record)
    session.commit()

find_findings(session, *, query='', severity=None, report_id=None)

Search current findings by keyword / severity / report (FR-18, read-only).

A finding matches when the (case-insensitive) query occurs anywhere in its title, description, impact, attack vector, endpoints, or reproduction steps — so "how many findings relate to SQL injection?" is answerable. severity and report_id further constrain the set.

Parameters:

Name	Type	Description	Default
session	Session	An active read-only DB session.	required
query	str	Case-insensitive substring to match; empty matches everything.	''
severity	str | None	Exact severity to filter by (critical…info), or None.	None
report_id	int | None	Restrict to one report, or None for all reports.	None

Returns:

Type	Description
FindingSearch	The exact total of matches plus up to :data:_SEARCH_LIMIT rows.

Source code in src/revalid/reports_chat.py

def find_findings(
    session: Session,
    *,
    query: str = "",
    severity: str | None = None,
    report_id: int | None = None,
) -> FindingSearch:
    """Search current findings by keyword / severity / report (FR-18, read-only).

    A finding matches when the (case-insensitive) ``query`` occurs anywhere in its
    title, description, impact, attack vector, endpoints, or reproduction steps —
    so "how many findings relate to SQL injection?" is answerable. ``severity`` and
    ``report_id`` further constrain the set.

    Args:
        session: An active read-only DB session.
        query: Case-insensitive substring to match; empty matches everything.
        severity: Exact severity to filter by (``critical``…``info``), or ``None``.
        report_id: Restrict to one report, or ``None`` for all reports.

    Returns:
        The exact ``total`` of matches plus up to :data:`_SEARCH_LIMIT` rows.
    """
    needle = query.strip().lower()
    want_sev = severity.strip().lower() if severity else None
    matches: list[FindingBrief] = []
    for finding_id in session.scalars(select(FindingRecord.id).order_by(FindingRecord.id)):
        version = current_version(session, finding_id)
        if version is None:
            continue
        identity = session.get(FindingRecord, finding_id)
        if identity is None:  # pragma: no cover - id came from the same table
            continue
        if report_id is not None and identity.report_id != report_id:
            continue
        if want_sev is not None and version.severity != want_sev:
            continue
        if needle and needle not in _haystack(version):
            continue
        matches.append(
            FindingBrief(
                id=finding_id,
                report_id=identity.report_id,
                title=version.title,
                severity=version.severity,
                affected_endpoints=list(version.affected_endpoints),
            )
        )
    return FindingSearch(
        total=len(matches),
        shown=min(len(matches), _SEARCH_LIMIT),
        findings=matches[:_SEARCH_LIMIT],
    )

get_finding(session, finding_id)

Return one finding's full current content + latest verdict, or None (FR-18).

Source code in src/revalid/reports_chat.py

def get_finding(session: Session, finding_id: int) -> FindingDetail | None:
    """Return one finding's full current content + latest verdict, or ``None`` (FR-18)."""
    version = current_version(session, finding_id)
    if version is None:
        return None
    identity = session.get(FindingRecord, finding_id)
    verdict = session.scalars(
        select(VerdictRecord)
        .where(VerdictRecord.finding_id == finding_id)
        .order_by(VerdictRecord.id.desc())
    ).first()
    return FindingDetail(
        id=finding_id,
        report_id=identity.report_id if identity else None,
        title=version.title,
        severity=version.severity,
        description=version.description,
        impact=version.impact,
        attack_vector=version.attack_vector,
        affected_endpoints=list(version.affected_endpoints),
        reproduction_steps=list(version.reproduction_steps),
        latest_verdict=verdict.status if verdict else None,
        latest_verdict_rationale=verdict.rationale if verdict else None,
    )

list_chats(session)

Return chat threads, most-recently-updated first.

Source code in src/revalid/reports_chat.py

def list_chats(session: Session) -> list[ChatSessionRecord]:
    """Return chat threads, most-recently-updated first."""
    return list(
        session.scalars(
            select(ChatSessionRecord).order_by(
                ChatSessionRecord.updated_at.desc(), ChatSessionRecord.id.desc()
            )
        )
    )

list_messages(session, chat_id)

Return a thread's messages in insert order (oldest first).

Source code in src/revalid/reports_chat.py

def list_messages(session: Session, chat_id: int) -> list[ChatMessageRecord]:
    """Return a thread's messages in insert order (oldest first)."""
    return list(
        session.scalars(
            select(ChatMessageRecord)
            .where(ChatMessageRecord.chat_id == chat_id)
            .order_by(ChatMessageRecord.id)
        )
    )

list_reports(session, *, include_archived=False)

List reports newest first; active only unless include_archived (FR-18).

Source code in src/revalid/reports_chat.py

def list_reports(session: Session, *, include_archived: bool = False) -> list[ReportBrief]:
    """List reports newest first; active only unless ``include_archived`` (FR-18)."""
    stmt = select(ReportRecord).order_by(ReportRecord.id.desc())
    if not include_archived:
        stmt = stmt.where(ReportRecord.archived == False)  # noqa: E712 — SQL boolean, not `is`
    return [
        ReportBrief(
            id=r.id,
            filename=r.filename,
            status=r.status,
            finding_count=r.finding_count,
            archived=r.archived,
        )
        for r in session.scalars(stmt)
    ]

stream_answer(agent, session, chat, question) async

Answer like :func:answer_question but yield the reply's text as it's generated.

Same contract — record the user turn (title on the first), run the read-only agent over session, append the completed reply — but the reply's tokens are yielded as they stream from the model so a streaming transport can show the answer live. Persistence and title-setting happen after the stream drains, so the stored thread is identical to the blocking path.

This is an async generator so it runs in the request's own event loop: the sync run_stream_sync binds its worker to the calling thread, which breaks when a streaming response iterates the generator across threadpool threads, so the async :meth:~pydantic_ai.Agent.run_stream is used instead.

Parameters:

Name	Type	Description	Default
agent	Agent[ReportsChatDeps, str]	The FR-18 reports agent (a stand-in model in tests).	required
session	Session	The active DB session (shared by persistence and the tools).	required
chat	ChatSessionRecord	The thread to append to.	required
question	str	The operator's new message.	required

Yields:

Type	Description
AsyncIterator[str]	Successive text deltas of the assistant's reply, in order.

Source code in src/revalid/reports_chat.py

async def stream_answer(
    agent: Agent[ReportsChatDeps, str],
    session: Session,
    chat: ChatSessionRecord,
    question: str,
) -> AsyncIterator[str]:
    """Answer like :func:`answer_question` but yield the reply's text as it's generated.

    Same contract — record the user turn (title on the first), run the read-only
    agent over ``session``, append the completed reply — but the reply's tokens are
    yielded as they stream from the model so a streaming transport can show the
    answer live. Persistence and title-setting happen after the stream drains, so
    the stored thread is identical to the blocking path.

    This is an **async** generator so it runs in the request's own event loop: the
    sync ``run_stream_sync`` binds its worker to the calling thread, which breaks
    when a streaming response iterates the generator across threadpool threads, so
    the async :meth:`~pydantic_ai.Agent.run_stream` is used instead.

    Args:
        agent: The FR-18 reports agent (a stand-in model in tests).
        session: The active DB session (shared by persistence and the tools).
        chat: The thread to append to.
        question: The operator's new message.

    Yields:
        Successive text deltas of the assistant's reply, in order.
    """
    prior = list_messages(session, chat.id)
    session.add(ChatMessageRecord(chat_id=chat.id, role=USER, content=question))
    if not prior:
        chat.title = _title_from(question)
    session.commit()

    async with agent.run_stream(
        question, deps=ReportsChatDeps(session=session), message_history=_history(prior)
    ) as result:
        async for delta in result.stream_text(delta=True):
            if delta:
                yield delta
        answer = (await result.get_output()).strip() or "(no answer)"

    chat.model = agent_model_name(agent)
    session.add(ChatMessageRecord(chat_id=chat.id, role=ASSISTANT, content=answer))
    session.commit()

Application and configuration

Persisted, runtime-editable model/provider setting (FR-13, ADR-0021).

A single settings row is the source of truth for LLM backend selection. On a fresh database it is seeded once from the environment (REVALID_LLM_MODEL / OLLAMA_BASE_URL) or the local-first default; thereafter the stored row is authoritative and the environment no longer overrides it.

ANTHROPIC_MODELS_URL = 'https://api.anthropic.com/v1/models' module-attribute

Anthropic model-list endpoint (authenticated differently from OpenAI's).

ANTHROPIC_VERSION = '2023-06-01' module-attribute

Required anthropic-version header for the Anthropic REST API.

SETTINGS_ID = 1 module-attribute

Primary key of the singleton settings row.

ProbeResult

Bases: BaseModel

Outcome of a provider connection probe / model discovery (ADR-0021).

Source code in src/revalid/settings.py

class ProbeResult(BaseModel):
    """Outcome of a provider connection probe / model discovery (ADR-0021)."""

    reachable: bool
    models: tuple[str, ...] = ()
    error: str | None = None

discover_models(provider, base_url, api_key, *, client=None)

Discover a provider's models, dispatching on its authentication scheme.

anthropic uses the Anthropic model list (:func:probe_anthropic); every other provider (ollama, openai, or any OpenAI-compatible host) uses the OpenAI-compatible {base_url}/models endpoint (:func:probe_provider).

Parameters:

Name	Type	Description	Default
provider	str | None	The selected provider id (ollama / anthropic / openai); None falls back to the OpenAI-compatible probe.	required
base_url	str | None	Provider base URL (used by the OpenAI-compatible path).	required
api_key	str | None	Provider API key (required for Anthropic/OpenAI).	required
client	Client | None	Injectable HTTP client (tests pass a MockTransport client).	None

Returns:

Name	Type	Description
A	ProbeResult	class:ProbeResult — never raises.

Source code in src/revalid/settings.py

def discover_models(
    provider: str | None,
    base_url: str | None,
    api_key: str | None,
    *,
    client: httpx.Client | None = None,
) -> ProbeResult:
    """Discover a provider's models, dispatching on its authentication scheme.

    ``anthropic`` uses the Anthropic model list (:func:`probe_anthropic`); every
    other provider (``ollama``, ``openai``, or any OpenAI-compatible host) uses
    the OpenAI-compatible ``{base_url}/models`` endpoint (:func:`probe_provider`).

    Args:
        provider: The selected provider id (``ollama`` / ``anthropic`` / ``openai``);
            ``None`` falls back to the OpenAI-compatible probe.
        base_url: Provider base URL (used by the OpenAI-compatible path).
        api_key: Provider API key (required for Anthropic/OpenAI).
        client: Injectable HTTP client (tests pass a ``MockTransport`` client).

    Returns:
        A :class:`ProbeResult` — never raises.
    """
    if provider == "anthropic":
        return probe_anthropic(api_key, client=client)
    return probe_provider(base_url, api_key, client=client)

load_or_seed(session)

Return the current setting, seeding the singleton row on first use.

Parameters:

Name	Type	Description	Default
session	Session	An open SQLAlchemy session.	required

Returns:

Type	Description
Settings	The persisted :class:~revalid.domain.Settings.

Source code in src/revalid/settings.py

def load_or_seed(session: Session) -> Settings:
    """Return the current setting, seeding the singleton row on first use.

    Args:
        session: An open SQLAlchemy session.

    Returns:
        The persisted :class:`~revalid.domain.Settings`.
    """
    record = session.get(SettingsRecord, SETTINGS_ID)
    if record is None:
        record = SettingsRecord.from_domain(_seed_from_env())
        record.id = SETTINGS_ID
        session.add(record)
        session.commit()
        session.refresh(record)
    return record.to_domain()

probe_anthropic(api_key, *, client=None)

Discover Claude models from the Anthropic API (ADR-0021).

Anthropic's model list authenticates with x-api-key + anthropic-version headers rather than OpenAI's bearer scheme, so it needs its own probe. A key is required — without one this reports unreachable rather than calling out.

Parameters:

Name	Type	Description	Default
api_key	str | None	The Anthropic API key.	required
client	Client | None	Injectable HTTP client (tests pass a MockTransport client).	None

Returns:

Name	Type	Description
A	ProbeResult	class:ProbeResult; reachable is false with an error message
	ProbeResult	on any failure (no exception escapes).

Source code in src/revalid/settings.py

def probe_anthropic(
    api_key: str | None,
    *,
    client: httpx.Client | None = None,
) -> ProbeResult:
    """Discover Claude models from the Anthropic API (ADR-0021).

    Anthropic's model list authenticates with ``x-api-key`` + ``anthropic-version``
    headers rather than OpenAI's bearer scheme, so it needs its own probe. A key
    is required — without one this reports unreachable rather than calling out.

    Args:
        api_key: The Anthropic API key.
        client: Injectable HTTP client (tests pass a ``MockTransport`` client).

    Returns:
        A :class:`ProbeResult`; ``reachable`` is false with an ``error`` message
        on any failure (no exception escapes).
    """
    if not api_key:
        return ProbeResult(reachable=False, error="an Anthropic API key is required")
    owns = client is None
    client = client or httpx.Client(timeout=10.0)
    try:
        response = client.get(
            ANTHROPIC_MODELS_URL,
            headers={"x-api-key": api_key, "anthropic-version": ANTHROPIC_VERSION},
        )
        response.raise_for_status()
        return ProbeResult(reachable=True, models=_extract_model_ids(response.json()))
    except (httpx.HTTPError, ValueError, KeyError) as exc:
        return ProbeResult(reachable=False, error=str(exc))
    finally:
        if owns:
            client.close()

probe_provider(base_url, api_key=None, *, client=None)

Probe an OpenAI-compatible provider and list its models (ADR-0021).

Hits {base_url}/models (e.g. Ollama's or OpenAI's OpenAI-compatible endpoint). This deliberately bypasses the FR-06 allowlist: the LLM host is infrastructure the operator configures, not a pentest target (ADR-0008).

Parameters:

Name	Type	Description	Default
base_url	str | None	The provider base URL (must already include any /v1 suffix).	required
api_key	str | None	Optional bearer token for hosts that require one.	None
client	Client | None	Injectable HTTP client (tests pass a MockTransport client).	None

Returns:

Name	Type	Description
A	ProbeResult	class:ProbeResult; reachable is false with an error message
	ProbeResult	on any failure (no exception escapes).

Source code in src/revalid/settings.py

def probe_provider(
    base_url: str | None,
    api_key: str | None = None,
    *,
    client: httpx.Client | None = None,
) -> ProbeResult:
    """Probe an OpenAI-compatible provider and list its models (ADR-0021).

    Hits ``{base_url}/models`` (e.g. Ollama's or OpenAI's OpenAI-compatible
    endpoint). This deliberately bypasses the FR-06 allowlist: the LLM host is
    infrastructure the operator configures, not a pentest target (ADR-0008).

    Args:
        base_url: The provider base URL (must already include any ``/v1`` suffix).
        api_key: Optional bearer token for hosts that require one.
        client: Injectable HTTP client (tests pass a ``MockTransport`` client).

    Returns:
        A :class:`ProbeResult`; ``reachable`` is false with an ``error`` message
        on any failure (no exception escapes).
    """
    if not base_url:
        return ProbeResult(reachable=False, error="set a base URL to discover models")
    owns = client is None
    client = client or httpx.Client(timeout=5.0)
    try:
        headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
        response = client.get(f"{base_url.rstrip('/')}/models", headers=headers)
        response.raise_for_status()
        return ProbeResult(reachable=True, models=_extract_model_ids(response.json()))
    except (httpx.HTTPError, ValueError, KeyError) as exc:
        return ProbeResult(reachable=False, error=str(exc))
    finally:
        if owns:
            client.close()

save(session, *, model, base_url, api_key, clear_key=False)

Persist an updated setting and return it.

The API key is sticky: a blank/None api_key leaves the stored key unchanged (so the UI never has to re-enter it); clear_key explicitly removes it.

Parameters:

Name	Type	Description	Default
session	Session	An open SQLAlchemy session.	required
model	str	The Pydantic AI provider:model string.	required
base_url	str | None	Provider base URL, or None for env-configured providers.	required
api_key	str | None	A new key to store, or blank/None to keep the existing one.	required
clear_key	bool	When true, delete the stored key.	False

Returns:

Type	Description
Settings	The persisted :class:~revalid.domain.Settings.

Source code in src/revalid/settings.py

def save(
    session: Session,
    *,
    model: str,
    base_url: str | None,
    api_key: str | None,
    clear_key: bool = False,
) -> Settings:
    """Persist an updated setting and return it.

    The API key is *sticky*: a blank/``None`` ``api_key`` leaves the stored key
    unchanged (so the UI never has to re-enter it); ``clear_key`` explicitly
    removes it.

    Args:
        session: An open SQLAlchemy session.
        model: The Pydantic AI ``provider:model`` string.
        base_url: Provider base URL, or ``None`` for env-configured providers.
        api_key: A new key to store, or blank/``None`` to keep the existing one.
        clear_key: When true, delete the stored key.

    Returns:
        The persisted :class:`~revalid.domain.Settings`.
    """
    record = session.get(SettingsRecord, SETTINGS_ID)
    if record is None:
        record = SettingsRecord.from_domain(_seed_from_env())
        record.id = SETTINGS_ID
        session.add(record)
    record.model = model
    record.base_url = base_url or None
    if clear_key:
        record.api_key = None
    elif api_key:
        record.api_key = api_key
    session.commit()
    session.refresh(record)
    return record.to_domain()

SQLite persistence layer via SQLAlchemy 2.0 (ADR-0002).

Single-file zero-ops storage. Findings, retest sessions, verdicts, and the audit trail (FR-10) all live here; only findings exist in the walking skeleton.

Base

Bases: DeclarativeBase

Declarative base for all ORM models.

Source code in src/revalid/db.py

class Base(DeclarativeBase):
    """Declarative base for all ORM models."""

ChatMessageRecord

Bases: Base

One append-only turn in a reports-chat thread (FR-18).

role is "user" or "assistant"; content is the plain text. Rows are ordered by id (monotonic insert order). The assistant re-queries the DB via its read-only tools on every turn, so only the prose is stored — no tool-call parts — and nothing here is ever mutated or deleted in place.

Source code in src/revalid/db.py

class ChatMessageRecord(Base):
    """One append-only turn in a reports-chat thread (FR-18).

    ``role`` is ``"user"`` or ``"assistant"``; ``content`` is the plain text.
    Rows are ordered by ``id`` (monotonic insert order). The assistant re-queries
    the DB via its read-only tools on every turn, so only the prose is stored — no
    tool-call parts — and nothing here is ever mutated or deleted in place.
    """

    __tablename__ = "chat_messages"

    id: Mapped[int] = mapped_column(primary_key=True)
    chat_id: Mapped[int] = mapped_column(ForeignKey("chat_sessions.id"))
    role: Mapped[str] = mapped_column(String(16))
    content: Mapped[str]
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

ChatSessionRecord

Bases: Base

A persisted reports-chat conversation thread (FR-18).

The read-only reports assistant answers natural-language questions about the whole corpus (reports, findings, verdicts). A thread is a lightweight container for an append-only sequence of :class:ChatMessageRecord turns, persisted so a conversation survives a page reload. title is a short human label (the first question, truncated) shown in the thread list; model records the LLM backend that answered (NFR-02 lineage).

Source code in src/revalid/db.py

class ChatSessionRecord(Base):
    """A persisted reports-chat conversation thread (FR-18).

    The read-only reports assistant answers natural-language questions about the
    whole corpus (reports, findings, verdicts). A thread is a lightweight
    container for an append-only sequence of :class:`ChatMessageRecord` turns,
    persisted so a conversation survives a page reload. ``title`` is a short
    human label (the first question, truncated) shown in the thread list;
    ``model`` records the LLM backend that answered (NFR-02 lineage).
    """

    __tablename__ = "chat_sessions"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200), default="New chat")
    model: Mapped[str] = mapped_column(String(128), default="")
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now(), onupdate=func.now()
    )

FindingNoteRecord

Bases: Base

One append-only, stage-tagged note on a finding (FR-16, ADR-0024).

Notes are the operator's reasoning trail: free text, tagged with the pipeline stage it was written on (:class:~revalid.domain.FindingStage) and never edited or deleted — history is kept, like every other record here.

Source code in src/revalid/db.py

class FindingNoteRecord(Base):
    """One append-only, stage-tagged note on a finding (FR-16, ADR-0024).

    Notes are the operator's reasoning trail: free text, tagged with the pipeline
    stage it was written on (:class:`~revalid.domain.FindingStage`) and never
    edited or deleted — history is kept, like every other record here.
    """

    __tablename__ = "finding_notes"

    id: Mapped[int] = mapped_column(primary_key=True)
    finding_id: Mapped[int] = mapped_column(ForeignKey("findings.id"))
    stage: Mapped[str] = mapped_column(String(16))
    body: Mapped[str]
    author: Mapped[str] = mapped_column(String(32), default="user")
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

FindingRecord

Bases: Base

Stable identity of a finding (FR-16, ADR-0024).

The finding's content lives in append-only :class:FindingVersionRecord rows; this row is the stable handle that :attr:VerdictRecord.finding_id references, so amending a finding (appending a new version) never orphans its verdicts. Notes link back via :attr:FindingNoteRecord.finding_id.

Source code in src/revalid/db.py

class FindingRecord(Base):
    """Stable identity of a finding (FR-16, ADR-0024).

    The finding's *content* lives in append-only :class:`FindingVersionRecord`
    rows; this row is the stable handle that :attr:`VerdictRecord.finding_id`
    references, so amending a finding (appending a new version) never orphans its
    verdicts. Notes link back via :attr:`FindingNoteRecord.finding_id`.
    """

    __tablename__ = "findings"

    id: Mapped[int] = mapped_column(primary_key=True)
    report_id: Mapped[int | None] = mapped_column(ForeignKey("reports.id"), default=None)
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

FindingVersionRecord

Bases: Base

One immutable version of a finding's content (FR-16, ADR-0024).

Extraction/import lands version 1 (origin=extraction); each operator edit appends a new version (origin=edit). The current version is the highest version — older ones are kept, never mutated (append-only version history, FR-16). edited_by/reason capture the edit lineage (FR-10); they stay None/empty on the extraction version.

Source code in src/revalid/db.py

class FindingVersionRecord(Base):
    """One immutable version of a finding's content (FR-16, ADR-0024).

    Extraction/import lands version 1 (``origin=extraction``); each operator edit
    appends a new version (``origin=edit``). The *current* version is the highest
    ``version`` — older ones are kept, never mutated (append-only version history,
    FR-16). ``edited_by``/``reason`` capture the edit lineage (FR-10); they stay
    ``None``/empty on the extraction version.
    """

    __tablename__ = "finding_versions"

    id: Mapped[int] = mapped_column(primary_key=True)
    finding_id: Mapped[int] = mapped_column(ForeignKey("findings.id"))
    version: Mapped[int]
    origin: Mapped[str] = mapped_column(String(16))
    title: Mapped[str] = mapped_column(String(500))
    severity: Mapped[str] = mapped_column(String(16))
    description: Mapped[str]
    impact: Mapped[str] = mapped_column(default="")
    attack_vector: Mapped[str] = mapped_column(default="")
    affected_endpoints: Mapped[list[str]] = mapped_column(JSON)
    reproduction_steps: Mapped[list[str]] = mapped_column(JSON)
    cvss: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict)
    mitre: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict)
    raw: Mapped[dict[str, Any]] = mapped_column(JSON)
    edited_by: Mapped[str | None] = mapped_column(String(32), default=None)
    reason: Mapped[str] = mapped_column(default="")
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

    @classmethod
    def from_domain(
        cls,
        finding_id: int,
        finding: Finding,
        *,
        version: int,
        origin: FindingOrigin,
        edited_by: str | None = None,
        reason: str = "",
    ) -> FindingVersionRecord:
        """Build a version row from a domain finding (extraction or an edit)."""
        return cls(
            finding_id=finding_id,
            version=version,
            origin=origin.value,
            title=finding.title,
            severity=finding.severity.value,
            description=finding.description,
            impact=finding.impact,
            attack_vector=finding.attack_vector,
            affected_endpoints=list(finding.affected_endpoints),
            reproduction_steps=list(finding.reproduction_steps),
            cvss=finding.cvss.model_dump(mode="json"),
            mitre=finding.mitre.model_dump(mode="json"),
            raw=finding.raw,
            edited_by=edited_by,
            reason=reason,
        )

    def to_domain(self) -> Finding:
        """Convert this version's content back to a domain finding."""
        return Finding(
            title=self.title,
            severity=Severity(self.severity),
            description=self.description,
            impact=self.impact,
            attack_vector=self.attack_vector,
            affected_endpoints=tuple(self.affected_endpoints),
            reproduction_steps=tuple(self.reproduction_steps),
            cvss=CvssCode.model_validate(self.cvss or {}),
            mitre=MitreMapping.model_validate(self.mitre or {}),
            raw=self.raw,
        )

from_domain(finding_id, finding, *, version, origin, edited_by=None, reason='') classmethod

Build a version row from a domain finding (extraction or an edit).

Source code in src/revalid/db.py

@classmethod
def from_domain(
    cls,
    finding_id: int,
    finding: Finding,
    *,
    version: int,
    origin: FindingOrigin,
    edited_by: str | None = None,
    reason: str = "",
) -> FindingVersionRecord:
    """Build a version row from a domain finding (extraction or an edit)."""
    return cls(
        finding_id=finding_id,
        version=version,
        origin=origin.value,
        title=finding.title,
        severity=finding.severity.value,
        description=finding.description,
        impact=finding.impact,
        attack_vector=finding.attack_vector,
        affected_endpoints=list(finding.affected_endpoints),
        reproduction_steps=list(finding.reproduction_steps),
        cvss=finding.cvss.model_dump(mode="json"),
        mitre=finding.mitre.model_dump(mode="json"),
        raw=finding.raw,
        edited_by=edited_by,
        reason=reason,
    )

to_domain()

Convert this version's content back to a domain finding.

Source code in src/revalid/db.py

def to_domain(self) -> Finding:
    """Convert this version's content back to a domain finding."""
    return Finding(
        title=self.title,
        severity=Severity(self.severity),
        description=self.description,
        impact=self.impact,
        attack_vector=self.attack_vector,
        affected_endpoints=tuple(self.affected_endpoints),
        reproduction_steps=tuple(self.reproduction_steps),
        cvss=CvssCode.model_validate(self.cvss or {}),
        mitre=MitreMapping.model_validate(self.mitre or {}),
        raw=self.raw,
    )

ReportRecord

Bases: Base

An uploaded pentest report and its ingest-job status (FR-01/FR-11).

One row per uploaded PDF; it doubles as the ingest job the UI polls (:class:~revalid.domain.ReportStatus). model records the LLM backend used (NFR-02 lineage); error holds the failure message when status is failed. Its findings link back via :attr:FindingRecord.report_id.

Source code in src/revalid/db.py

class ReportRecord(Base):
    """An uploaded pentest report and its ingest-job status (FR-01/FR-11).

    One row per uploaded PDF; it doubles as the ingest job the UI polls
    (:class:`~revalid.domain.ReportStatus`). ``model`` records the LLM backend
    used (NFR-02 lineage); ``error`` holds the failure message when
    ``status`` is ``failed``. Its findings link back via
    :attr:`FindingRecord.report_id`.
    """

    __tablename__ = "reports"

    id: Mapped[int] = mapped_column(primary_key=True)
    filename: Mapped[str] = mapped_column(String(500))
    status: Mapped[str] = mapped_column(String(16))
    model: Mapped[str] = mapped_column(String(128))
    error: Mapped[str | None] = mapped_column(default=None)
    finding_count: Mapped[int] = mapped_column(default=0)
    #: Soft-hidden from the overview but kept (reversible); deletable (FR-11, #128).
    archived: Mapped[bool] = mapped_column(default=False)
    #: SHA-256 of the uploaded bytes, for duplicate-upload detection (FR-01, #134).
    content_hash: Mapped[str | None] = mapped_column(String(64), default=None)
    #: Document-level metadata extracted from the report, operator-editable (#133).
    doc_metadata: Mapped[dict[str, Any] | None] = mapped_column(JSON, default=None)
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

RetestSessionRecord

Bases: Base

An FR-17 agentic retest session (parent of its append-only transcript).

Mirrors :class:VerdictRecord's finding link (finding_id FK) but tracks a live, in-progress agent run rather than a concluded outcome: status moves through :class:~revalid.domain.RetestSessionStatus until a terminal state, at which point ended_at is set. verdict_status/verdict_rationale are populated only once the session concludes with a verdict.

Source code in src/revalid/db.py

class RetestSessionRecord(Base):
    """An FR-17 agentic retest session (parent of its append-only transcript).

    Mirrors :class:`VerdictRecord`'s finding link (``finding_id`` FK) but tracks a
    live, in-progress agent run rather than a concluded outcome: ``status`` moves
    through :class:`~revalid.domain.RetestSessionStatus` until a terminal state,
    at which point ``ended_at`` is set. ``verdict_status``/``verdict_rationale``
    are populated only once the session concludes with a verdict.
    """

    __tablename__ = "retest_sessions"

    id: Mapped[int] = mapped_column(primary_key=True)
    finding_id: Mapped[int] = mapped_column(ForeignKey("findings.id"))
    status: Mapped[str] = mapped_column(String(16))
    model: Mapped[str] = mapped_column(String(128))
    verdict_status: Mapped[str | None] = mapped_column(String(16), default=None)
    verdict_rationale: Mapped[str | None] = mapped_column(default=None)
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    ended_at: Mapped[datetime | None] = mapped_column(DateTime, default=None)
    free_launch: Mapped[bool] = mapped_column(default=False)

SessionEventRecord

Bases: Base

One append-only transcript event for a retest session (FR-17 audit).

The full record of what an agentic session did: each proposed/approved/ rejected command, its output, state transitions, and the final verdict are all rows here, ordered by seq (monotonic per session, assigned by :func:revalid.retest_session.append_event) so the transcript replays deterministically regardless of wall-clock timestamp resolution.

Source code in src/revalid/db.py

class SessionEventRecord(Base):
    """One append-only transcript event for a retest session (FR-17 audit).

    The full record of what an agentic session did: each proposed/approved/
    rejected command, its output, state transitions, and the final verdict are
    all rows here, ordered by ``seq`` (monotonic per session, assigned by
    :func:`revalid.retest_session.append_event`) so the transcript replays
    deterministically regardless of wall-clock timestamp resolution.
    """

    __tablename__ = "session_events"

    id: Mapped[int] = mapped_column(primary_key=True)
    session_id: Mapped[int] = mapped_column(ForeignKey("retest_sessions.id"))
    seq: Mapped[int]
    kind: Mapped[str] = mapped_column(String(32))
    payload: Mapped[dict[str, Any]] = mapped_column(JSON)
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

SettingsRecord

Bases: Base

The single-row persisted model/provider setting (FR-13 / ADR-0021).

One row (id == 1) holds the runtime backend selection. The API key is stored here in the gitignored SQLite file (ADR-0008) but is never returned by the API (write-only, masked on read).

Source code in src/revalid/db.py

class SettingsRecord(Base):
    """The single-row persisted model/provider setting (FR-13 / ADR-0021).

    One row (``id == 1``) holds the runtime backend selection. The API key is
    stored here in the gitignored SQLite file (ADR-0008) but is never returned
    by the API (write-only, masked on read).
    """

    __tablename__ = "settings"

    id: Mapped[int] = mapped_column(primary_key=True)
    model: Mapped[str] = mapped_column(String(128))
    base_url: Mapped[str | None] = mapped_column(String(256), default=None)
    api_key: Mapped[str | None] = mapped_column(String(256), default=None)
    updated_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now(), onupdate=func.now()
    )

    @classmethod
    def from_domain(cls, cfg: Settings) -> SettingsRecord:
        """Build the singleton row from a domain settings object."""
        return cls(
            model=cfg.model,
            base_url=cfg.base_url,
            api_key=cfg.api_key,
        )

    def to_domain(self) -> Settings:
        """Convert this row back to a domain settings object."""
        return Settings(
            model=self.model,
            base_url=self.base_url,
            api_key=self.api_key,
        )

from_domain(cfg) classmethod

Build the singleton row from a domain settings object.

Source code in src/revalid/db.py

@classmethod
def from_domain(cls, cfg: Settings) -> SettingsRecord:
    """Build the singleton row from a domain settings object."""
    return cls(
        model=cfg.model,
        base_url=cfg.base_url,
        api_key=cfg.api_key,
    )

to_domain()

Convert this row back to a domain settings object.

Source code in src/revalid/db.py

def to_domain(self) -> Settings:
    """Convert this row back to a domain settings object."""
    return Settings(
        model=self.model,
        base_url=self.base_url,
        api_key=self.api_key,
    )

VerdictRecord

Bases: Base

A persisted agentic retest verdict, linked to its finding + session (FR-09/FR-17).

Every verdict is the conclusion of an agentic retest session (the batch verdict path retired in FR-17 6b-iii): session_id links the session whose append-only transcript justifies it, and evidence is the flexible :class:~revalid.domain.AgenticEvidence proof the agent pinned on conclude (6b-i), or NULL for a human adjudication that ran no command. actor is "agent" for the auto-persisted conclusion or "operator" for an adjudication that supersedes it (latest id wins, FR-10 append-only).

Source code in src/revalid/db.py

class VerdictRecord(Base):
    """A persisted agentic retest verdict, linked to its finding + session (FR-09/FR-17).

    Every verdict is the conclusion of an agentic retest session (the batch
    verdict path retired in FR-17 6b-iii): ``session_id`` links the session whose
    append-only transcript justifies it, and ``evidence`` is the flexible
    :class:`~revalid.domain.AgenticEvidence` proof the agent pinned on conclude
    (6b-i), or ``NULL`` for a human adjudication that ran no command. ``actor`` is
    ``"agent"`` for the auto-persisted conclusion or ``"operator"`` for an
    adjudication that supersedes it (latest id wins, FR-10 append-only).
    """

    __tablename__ = "verdicts"

    id: Mapped[int] = mapped_column(primary_key=True)
    finding_id: Mapped[int] = mapped_column(ForeignKey("findings.id"))
    session_id: Mapped[int | None] = mapped_column(ForeignKey("retest_sessions.id"), default=None)
    status: Mapped[str] = mapped_column(String(16))
    reason_code: Mapped[str] = mapped_column(String(64))
    rationale: Mapped[str]
    matched_indicators: Mapped[list[str]] = mapped_column(JSON)
    #: The flexible :class:`~revalid.domain.AgenticEvidence` proof, or ``NULL``.
    evidence: Mapped[dict[str, Any] | None] = mapped_column(JSON, default=None)
    #: ``"agent"`` (auto-persisted conclusion) or ``"operator"`` (adjudication).
    actor: Mapped[str] = mapped_column(String(32), default="agent")
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

    @classmethod
    def agentic(
        cls,
        *,
        finding_id: int,
        session_id: int,
        status: VerdictStatus,
        rationale: str,
        actor: str,
        reason_code: str,
        evidence: dict[str, Any] | None = None,
    ) -> VerdictRecord:
        """Build a verdict row for a retest session's conclusion (FR-17).

        ``evidence`` is the flexible :class:`~revalid.domain.AgenticEvidence` proof
        the agent pinned on conclude (6b-i), or ``None`` when unavailable (e.g. a
        human adjudication). ``actor`` is ``"agent"`` for the auto-persisted
        conclusion or ``"operator"`` for a human adjudication that supersedes it.
        """
        return cls(
            finding_id=finding_id,
            status=status.value,
            reason_code=reason_code,
            rationale=rationale,
            matched_indicators=[],
            evidence=evidence,
            session_id=session_id,
            actor=actor,
        )

agentic(*, finding_id, session_id, status, rationale, actor, reason_code, evidence=None) classmethod

Build a verdict row for a retest session's conclusion (FR-17).

evidence is the flexible :class:~revalid.domain.AgenticEvidence proof the agent pinned on conclude (6b-i), or None when unavailable (e.g. a human adjudication). actor is "agent" for the auto-persisted conclusion or "operator" for a human adjudication that supersedes it.

Source code in src/revalid/db.py

@classmethod
def agentic(
    cls,
    *,
    finding_id: int,
    session_id: int,
    status: VerdictStatus,
    rationale: str,
    actor: str,
    reason_code: str,
    evidence: dict[str, Any] | None = None,
) -> VerdictRecord:
    """Build a verdict row for a retest session's conclusion (FR-17).

    ``evidence`` is the flexible :class:`~revalid.domain.AgenticEvidence` proof
    the agent pinned on conclude (6b-i), or ``None`` when unavailable (e.g. a
    human adjudication). ``actor`` is ``"agent"`` for the auto-persisted
    conclusion or ``"operator"`` for a human adjudication that supersedes it.
    """
    return cls(
        finding_id=finding_id,
        status=status.value,
        reason_code=reason_code,
        rationale=rationale,
        matched_indicators=[],
        evidence=evidence,
        session_id=session_id,
        actor=actor,
    )

create_db_engine(path='revalid.db')

Create the SQLite engine and ensure the schema exists.

Parameters:

Name	Type	Description	Default
path	str	Database file path, or :data:IN_MEMORY for an in-memory database (shared across threads, for tests).	'revalid.db'

Returns:

Type	Description
Engine	A ready-to-use engine with all tables created.

Source code in src/revalid/db.py

def create_db_engine(path: str = "revalid.db") -> Engine:
    """Create the SQLite engine and ensure the schema exists.

    Args:
        path: Database file path, or :data:`IN_MEMORY` for an in-memory
            database (shared across threads, for tests).

    Returns:
        A ready-to-use engine with all tables created.
    """
    if path == IN_MEMORY:
        # One shared connection so the in-memory db survives FastAPI's
        # per-request worker threads.
        engine = create_engine(
            f"sqlite:///{IN_MEMORY}",
            poolclass=StaticPool,
            connect_args={"check_same_thread": False},
        )
    else:
        engine = create_engine(f"sqlite:///{path}")
    Base.metadata.create_all(engine)
    _ensure_columns(engine)
    _backfill_note_stages(engine)
    return engine

session_factory(engine)

Return a sessionmaker bound to the given engine.

Source code in src/revalid/db.py

def session_factory(engine: Engine) -> sessionmaker[Session]:
    """Return a sessionmaker bound to the given engine."""
    return sessionmaker(bind=engine)

FastAPI application factory (ADR-0002: local single-user web app).

Run locally with::

uv run uvicorn --factory revalid.app:create_app --host 127.0.0.1

The app must only ever bind to 127.0.0.1 (NFR-03); there is no authentication in TFG scope.

AdjudicateRequest

Bases: BaseModel

Body for a human verdict adjudication of a concluded session (FR-17 Slice 6a).

status is the human's call — equal to the agent's when accepting, or a different value when overriding; rationale is their justification.

Source code in src/revalid/app.py

class AdjudicateRequest(BaseModel):
    """Body for a human verdict adjudication of a concluded session (FR-17 Slice 6a).

    ``status`` is the human's call — equal to the agent's when accepting, or a
    different value when overriding; ``rationale`` is their justification.
    """

    status: VerdictStatus
    rationale: str = ""

AuditOut

Bases: BaseModel

Result of re-deriving every verdict from the stored audit trail (FR-10).

Source code in src/revalid/app.py

class AuditOut(BaseModel):
    """Result of re-deriving every verdict from the stored audit trail (FR-10)."""

    total: int
    reproduced: int
    ok: bool
    discrepancies: list[DiscrepancyOut]

BackendStatusOut

Bases: BaseModel

Live LLM-backend reachability + active model, for the sidebar status pill.

Source code in src/revalid/app.py

class BackendStatusOut(BaseModel):
    """Live LLM-backend reachability + active model, for the sidebar status pill."""

    model_config = ConfigDict(protected_namespaces=())

    connected: bool
    model: str

ChatDetailOut

Bases: ChatOut

A chat thread plus its full ordered transcript (FR-18).

Source code in src/revalid/app.py

class ChatDetailOut(ChatOut):
    """A chat thread plus its full ordered transcript (FR-18)."""

    messages: list[ChatMessageOut] = []

    @classmethod
    def of(cls, record: ChatSessionRecord, messages: list[ChatMessageRecord]) -> "ChatDetailOut":
        """Build the thread + messages view from a chat row and its turns."""
        return cls(
            id=record.id,
            title=record.title,
            model=record.model,
            created_at=record.created_at,
            updated_at=record.updated_at,
            messages=[ChatMessageOut.from_record(m) for m in messages],
        )

of(record, messages) classmethod

Build the thread + messages view from a chat row and its turns.

Source code in src/revalid/app.py

@classmethod
def of(cls, record: ChatSessionRecord, messages: list[ChatMessageRecord]) -> "ChatDetailOut":
    """Build the thread + messages view from a chat row and its turns."""
    return cls(
        id=record.id,
        title=record.title,
        model=record.model,
        created_at=record.created_at,
        updated_at=record.updated_at,
        messages=[ChatMessageOut.from_record(m) for m in messages],
    )

ChatMessageIn

Bases: BaseModel

Body for posting an operator question to a chat thread (FR-18).

Source code in src/revalid/app.py

class ChatMessageIn(BaseModel):
    """Body for posting an operator question to a chat thread (FR-18)."""

    content: str = Field(min_length=1)

ChatMessageOut

Bases: BaseModel

One persisted chat turn as returned by the API (FR-18).

Source code in src/revalid/app.py

class ChatMessageOut(BaseModel):
    """One persisted chat turn as returned by the API (FR-18)."""

    id: int
    role: str
    content: str
    created_at: datetime

    @classmethod
    def from_record(cls, record: ChatMessageRecord) -> "ChatMessageOut":
        """Build the message view from a persisted chat-message row."""
        return cls(
            id=record.id,
            role=record.role,
            content=record.content,
            created_at=record.created_at,
        )

from_record(record) classmethod

Build the message view from a persisted chat-message row.

Source code in src/revalid/app.py

@classmethod
def from_record(cls, record: ChatMessageRecord) -> "ChatMessageOut":
    """Build the message view from a persisted chat-message row."""
    return cls(
        id=record.id,
        role=record.role,
        content=record.content,
        created_at=record.created_at,
    )

ChatOut

Bases: BaseModel

A reports-chat thread summary as returned by the API (FR-18).

Source code in src/revalid/app.py

class ChatOut(BaseModel):
    """A reports-chat thread summary as returned by the API (FR-18)."""

    id: int
    title: str
    model: str
    created_at: datetime
    updated_at: datetime

    @classmethod
    def from_record(cls, record: ChatSessionRecord) -> "ChatOut":
        """Build the thread-summary view from a persisted chat row."""
        return cls(
            id=record.id,
            title=record.title,
            model=record.model,
            created_at=record.created_at,
            updated_at=record.updated_at,
        )

from_record(record) classmethod

Build the thread-summary view from a persisted chat row.

Source code in src/revalid/app.py

@classmethod
def from_record(cls, record: ChatSessionRecord) -> "ChatOut":
    """Build the thread-summary view from a persisted chat row."""
    return cls(
        id=record.id,
        title=record.title,
        model=record.model,
        created_at=record.created_at,
        updated_at=record.updated_at,
    )

ConcludeRequest

Bases: BaseModel

Body for an operator's manual conclusion of a paused session (ADR-0034).

Source code in src/revalid/app.py

class ConcludeRequest(BaseModel):
    """Body for an operator's manual conclusion of a paused session (ADR-0034)."""

    status: VerdictStatus
    rationale: str = ""

CvssIn

Bases: BaseModel

An operator-supplied CVSS code on a finding edit (FR-19).

Carries no inferred flag: provenance is not the client's to assert. The server derives it by comparing against the current version — a value the operator changed is author-stated, not model-derived.

Source code in src/revalid/app.py

class CvssIn(BaseModel):
    """An operator-supplied CVSS code on a finding edit (FR-19).

    Carries no ``inferred`` flag: provenance is not the client's to assert. The
    server derives it by comparing against the current version — a value the
    operator changed is author-stated, not model-derived.
    """

    vector: str = ""
    base_score: float | None = Field(default=None, ge=0.0, le=10.0)

DiscrepancyOut

Bases: BaseModel

A stored verdict that no longer re-derives from its evidence (FR-10).

Source code in src/revalid/app.py

class DiscrepancyOut(BaseModel):
    """A stored verdict that no longer re-derives from its evidence (FR-10)."""

    verdict_id: int
    finding_id: int
    stored: str
    rederived: str

FindingEditIn

Bases: BaseModel

An operator edit of a finding's content → a new immutable version (FR-16).

Source code in src/revalid/app.py

class FindingEditIn(BaseModel):
    """An operator edit of a finding's content → a new immutable version (FR-16)."""

    title: str = Field(min_length=1)
    severity: Severity
    description: str = ""
    impact: str = ""
    attack_vector: str = ""
    affected_endpoints: tuple[str, ...] = ()
    reproduction_steps: tuple[str, ...] = ()
    #: Omit to leave the taxonomy untouched; send a value to set it (FR-19).
    cvss: CvssIn | None = None
    mitre: MitreIn | None = None
    reason: str = ""

    def to_finding(self, current: Finding) -> Finding:
        """Build the domain finding for this edit, carrying forward prior lineage.

        ``current`` is the version being edited. Its ``raw`` (extraction lineage)
        is kept so editing never discards how the finding was originally produced
        (FR-10), and so are its CVSS/ATT&CK values when the payload omits them —
        without that, every edit silently wiped the taxonomy an extraction had
        derived, because the fields simply were not carried across.
        """
        return Finding(
            title=self.title,
            severity=self.severity,
            description=self.description,
            impact=self.impact,
            attack_vector=self.attack_vector,
            affected_endpoints=self.affected_endpoints,
            reproduction_steps=self.reproduction_steps,
            cvss=self._resolved_cvss(current.cvss),
            mitre=self._resolved_mitre(current.mitre),
            raw=current.raw,
        )

    def _resolved_cvss(self, current: CvssCode) -> CvssCode:
        """Resolve the edit's CVSS against the current one, deriving provenance.

        Omitted → keep the current value untouched, provenance included. Supplied
        and identical → likewise (a round-trip of an inferred value must not
        launder it into an author-stated one). Supplied and different → the
        operator authored it, so ``inferred`` becomes ``False``.
        """
        if self.cvss is None:
            return current
        proposed = CvssCode(
            vector=self.cvss.vector, base_score=self.cvss.base_score, inferred=current.inferred
        )
        if proposed == current:
            return current
        return proposed.model_copy(update={"inferred": False})

    def _resolved_mitre(self, current: MitreMapping) -> MitreMapping:
        """Resolve the edit's ATT&CK mapping against the current one (see above)."""
        if self.mitre is None:
            return current
        proposed = MitreMapping(techniques=self.mitre.techniques, inferred=current.inferred)
        if proposed == current:
            return current
        return proposed.model_copy(update={"inferred": False})

to_finding(current)

Build the domain finding for this edit, carrying forward prior lineage.

current is the version being edited. Its raw (extraction lineage) is kept so editing never discards how the finding was originally produced (FR-10), and so are its CVSS/ATT&CK values when the payload omits them — without that, every edit silently wiped the taxonomy an extraction had derived, because the fields simply were not carried across.

Source code in src/revalid/app.py

def to_finding(self, current: Finding) -> Finding:
    """Build the domain finding for this edit, carrying forward prior lineage.

    ``current`` is the version being edited. Its ``raw`` (extraction lineage)
    is kept so editing never discards how the finding was originally produced
    (FR-10), and so are its CVSS/ATT&CK values when the payload omits them —
    without that, every edit silently wiped the taxonomy an extraction had
    derived, because the fields simply were not carried across.
    """
    return Finding(
        title=self.title,
        severity=self.severity,
        description=self.description,
        impact=self.impact,
        attack_vector=self.attack_vector,
        affected_endpoints=self.affected_endpoints,
        reproduction_steps=self.reproduction_steps,
        cvss=self._resolved_cvss(current.cvss),
        mitre=self._resolved_mitre(current.mitre),
        raw=current.raw,
    )

FindingOut

Bases: Finding

A persisted finding as returned by the API — the current version's content.

version is the current version number (extraction = 1); it bumps on every operator edit (FR-16). id is the stable finding identity verdicts and retest sessions reference.

Source code in src/revalid/app.py

class FindingOut(Finding):
    """A persisted finding as returned by the API — the *current* version's content.

    ``version`` is the current version number (extraction = 1); it bumps on every
    operator edit (FR-16). ``id`` is the stable finding identity verdicts and
    retest sessions reference.
    """

    id: int
    report_id: int | None = None
    version: int = 1

FindingVersionOut

Bases: Finding

One immutable finding version as returned by the API (FR-16).

Source code in src/revalid/app.py

class FindingVersionOut(Finding):
    """One immutable finding version as returned by the API (FR-16)."""

    version: int
    origin: str
    edited_by: str | None = None
    reason: str = ""
    created_at: datetime

    @classmethod
    def from_record(cls, record: FindingVersionRecord) -> "FindingVersionOut":
        """Build the API view from a persisted finding-version row."""
        return cls(
            version=record.version,
            origin=record.origin,
            edited_by=record.edited_by,
            reason=record.reason,
            created_at=record.created_at,
            **record.to_domain().model_dump(),
        )

from_record(record) classmethod

Build the API view from a persisted finding-version row.

Source code in src/revalid/app.py

@classmethod
def from_record(cls, record: FindingVersionRecord) -> "FindingVersionOut":
    """Build the API view from a persisted finding-version row."""
    return cls(
        version=record.version,
        origin=record.origin,
        edited_by=record.edited_by,
        reason=record.reason,
        created_at=record.created_at,
        **record.to_domain().model_dump(),
    )

FreeLaunchRequest

Bases: BaseModel

Body for the live free-launch toggle (FR-17 Slice 5).

Source code in src/revalid/app.py

class FreeLaunchRequest(BaseModel):
    """Body for the live free-launch toggle (FR-17 Slice 5)."""

    enabled: bool

GoalDraftOut

Bases: BaseModel

A generated retest-goal draft for a finding, pre-session (FR-17 6b-iii-b).

Source code in src/revalid/app.py

class GoalDraftOut(BaseModel):
    """A generated retest-goal draft for a finding, pre-session (FR-17 6b-iii-b)."""

    steps: list[str]

GoalRequest

Bases: BaseModel

Body for a user-owned goal edit (FR-17 6b-ii).

Source code in src/revalid/app.py

class GoalRequest(BaseModel):
    """Body for a user-owned goal edit (FR-17 6b-ii)."""

    steps: list[str]

HumanCommandRequest

Bases: BaseModel

Body for a manual operator command (!): the exact command to run (FR-17).

Source code in src/revalid/app.py

class HumanCommandRequest(BaseModel):
    """Body for a manual operator command (`!`): the exact command to run (FR-17)."""

    command: str = Field(min_length=1)

ImportResult

Bases: BaseModel

Outcome of a findings import.

enriched/enrichment_failed are 0 unless the caller asked for the opt-in FR-19 taxonomy pass (issue #233). A non-zero enrichment_failed means the import itself succeeded but some findings came back without a taxonomy — reported rather than swallowed, so a partially-enriched import is never mistaken for a complete one.

Source code in src/revalid/app.py

class ImportResult(BaseModel):
    """Outcome of a findings import.

    ``enriched``/``enrichment_failed`` are 0 unless the caller asked for the
    opt-in FR-19 taxonomy pass (issue #233). A non-zero ``enrichment_failed``
    means the import itself succeeded but some findings came back without a
    taxonomy — reported rather than swallowed, so a partially-enriched import is
    never mistaken for a complete one.
    """

    imported: int
    enriched: int = 0
    enrichment_failed: int = 0

MessageRequest

Bases: BaseModel

Body for an operator chat message to the agent (FR-17 Slice 4).

Source code in src/revalid/app.py

class MessageRequest(BaseModel):
    """Body for an operator chat message to the agent (FR-17 Slice 4)."""

    text: str = Field(min_length=1)

MitreIn

Bases: BaseModel

An operator-supplied MITRE ATT&CK mapping on a finding edit (FR-19).

Source code in src/revalid/app.py

class MitreIn(BaseModel):
    """An operator-supplied MITRE ATT&CK mapping on a finding edit (FR-19)."""

    techniques: tuple[str, ...] = ()

NoteIn

Bases: BaseModel

An operator note on a finding, tagged with the stage it was written on (FR-16).

Source code in src/revalid/app.py

class NoteIn(BaseModel):
    """An operator note on a finding, tagged with the stage it was written on (FR-16)."""

    stage: FindingStage = FindingStage.GENERAL
    body: str = Field(min_length=1)

NoteOut

Bases: BaseModel

A persisted finding note as returned by the API (FR-16).

Source code in src/revalid/app.py

class NoteOut(BaseModel):
    """A persisted finding note as returned by the API (FR-16)."""

    id: int
    finding_id: int
    stage: str
    body: str
    author: str
    created_at: datetime

    @classmethod
    def from_record(cls, record: FindingNoteRecord) -> "NoteOut":
        """Build the API view from a persisted note row."""
        return cls(
            id=record.id,
            finding_id=record.finding_id,
            stage=record.stage,
            body=record.body,
            author=record.author,
            created_at=record.created_at,
        )

from_record(record) classmethod

Build the API view from a persisted note row.

Source code in src/revalid/app.py

@classmethod
def from_record(cls, record: FindingNoteRecord) -> "NoteOut":
    """Build the API view from a persisted note row."""
    return cls(
        id=record.id,
        finding_id=record.finding_id,
        stage=record.stage,
        body=record.body,
        author=record.author,
        created_at=record.created_at,
    )

ProbeIn

Bases: BaseModel

Probe request: which provider/endpoint (and optional key) to discover from.

provider selects the discovery scheme (anthropic uses the Anthropic model list; anything else uses the OpenAI-compatible {base_url}/models).

Source code in src/revalid/app.py

class ProbeIn(BaseModel):
    """Probe request: which provider/endpoint (and optional key) to discover from.

    ``provider`` selects the discovery scheme (``anthropic`` uses the Anthropic
    model list; anything else uses the OpenAI-compatible ``{base_url}/models``).
    """

    provider: str | None = None
    base_url: str | None = None
    api_key: str | None = None

RejectRequest

Bases: BaseModel

Optional body for a command rejection: the operator's reason (FR-17).

Source code in src/revalid/app.py

class RejectRequest(BaseModel):
    """Optional body for a command rejection: the operator's reason (FR-17)."""

    reason: str = ""

ReportOut

Bases: BaseModel

A persisted report / ingest job as returned by the API (FR-01/FR-11).

Source code in src/revalid/app.py

class ReportOut(BaseModel):
    """A persisted report / ingest job as returned by the API (FR-01/FR-11)."""

    id: int
    filename: str
    status: str
    model: str
    error: str | None
    finding_count: int
    archived: bool
    content_hash: str | None
    metadata: ReportMetadata | None
    created_at: datetime

    @classmethod
    def from_record(cls, record: ReportRecord) -> "ReportOut":
        """Build the API view from a persisted report row."""
        return cls(
            id=record.id,
            filename=record.filename,
            status=record.status,
            model=record.model,
            error=record.error,
            finding_count=record.finding_count,
            archived=record.archived,
            content_hash=record.content_hash,
            metadata=(
                ReportMetadata.model_validate(record.doc_metadata) if record.doc_metadata else None
            ),
            created_at=record.created_at,
        )

from_record(record) classmethod

Build the API view from a persisted report row.

Source code in src/revalid/app.py

@classmethod
def from_record(cls, record: ReportRecord) -> "ReportOut":
    """Build the API view from a persisted report row."""
    return cls(
        id=record.id,
        filename=record.filename,
        status=record.status,
        model=record.model,
        error=record.error,
        finding_count=record.finding_count,
        archived=record.archived,
        content_hash=record.content_hash,
        metadata=(
            ReportMetadata.model_validate(record.doc_metadata) if record.doc_metadata else None
        ),
        created_at=record.created_at,
    )

ReportPatchIn

Bases: BaseModel

Body for archiving / unarchiving a report (FR-11, #128).

Source code in src/revalid/app.py

class ReportPatchIn(BaseModel):
    """Body for archiving / unarchiving a report (FR-11, #128)."""

    archived: bool

RetestSessionOut

Bases: BaseModel

A persisted agentic retest session + its transcript as returned by the API (FR-17).

Source code in src/revalid/app.py

class RetestSessionOut(BaseModel):
    """A persisted agentic retest session + its transcript as returned by the API (FR-17)."""

    id: int
    finding_id: int
    status: str
    model: str
    verdict_status: str | None
    verdict_rationale: str | None
    free_launch: bool
    events: list[SessionEventOut] = []

    @classmethod
    def from_record(
        cls, record: RetestSessionRecord, events: list[dict[str, Any]]
    ) -> "RetestSessionOut":
        """Build the API view from a session row and its ordered transcript events."""
        return cls(
            id=record.id,
            finding_id=record.finding_id,
            status=record.status,
            model=record.model,
            verdict_status=record.verdict_status,
            verdict_rationale=record.verdict_rationale,
            free_launch=record.free_launch,
            events=[SessionEventOut(**e) for e in events],
        )

from_record(record, events) classmethod

Build the API view from a session row and its ordered transcript events.

Source code in src/revalid/app.py

@classmethod
def from_record(
    cls, record: RetestSessionRecord, events: list[dict[str, Any]]
) -> "RetestSessionOut":
    """Build the API view from a session row and its ordered transcript events."""
    return cls(
        id=record.id,
        finding_id=record.finding_id,
        status=record.status,
        model=record.model,
        verdict_status=record.verdict_status,
        verdict_rationale=record.verdict_rationale,
        free_launch=record.free_launch,
        events=[SessionEventOut(**e) for e in events],
    )

RetestSessionSummary

Bases: BaseModel

A compact retest-session row for a finding's session list (FR-17 6b-iii-b).

Source code in src/revalid/app.py

class RetestSessionSummary(BaseModel):
    """A compact retest-session row for a finding's session list (FR-17 6b-iii-b)."""

    id: int
    finding_id: int
    status: str
    verdict_status: str | None
    created_at: datetime

    @classmethod
    def from_record(cls, record: RetestSessionRecord) -> "RetestSessionSummary":
        """Build the compact list-row view from a full session record."""
        return cls(
            id=record.id,
            finding_id=record.finding_id,
            status=record.status,
            verdict_status=record.verdict_status,
            created_at=record.created_at,
        )

from_record(record) classmethod

Build the compact list-row view from a full session record.

Source code in src/revalid/app.py

@classmethod
def from_record(cls, record: RetestSessionRecord) -> "RetestSessionSummary":
    """Build the compact list-row view from a full session record."""
    return cls(
        id=record.id,
        finding_id=record.finding_id,
        status=record.status,
        verdict_status=record.verdict_status,
        created_at=record.created_at,
    )

SessionEventOut

Bases: BaseModel

One append-only transcript event as returned by the API (FR-17).

Source code in src/revalid/app.py

class SessionEventOut(BaseModel):
    """One append-only transcript event as returned by the API (FR-17)."""

    seq: int
    kind: str
    payload: dict[str, Any]

SettingsOut

Bases: BaseModel

Public view of the model/provider setting; the key is write-only (ADR-0021).

Source code in src/revalid/app.py

class SettingsOut(BaseModel):
    """Public view of the model/provider setting; the key is write-only (ADR-0021)."""

    model_config = ConfigDict(protected_namespaces=())

    model: str
    base_url: str | None
    api_key_set: bool
    api_key_hint: str | None

    @classmethod
    def from_domain(cls, cfg: Settings) -> "SettingsOut":
        """Build the masked view: the key becomes a boolean + last-4 hint only."""
        key = cfg.api_key or ""
        return cls(
            model=cfg.model,
            base_url=cfg.base_url,
            api_key_set=bool(key),
            api_key_hint=key[-4:] if key else None,
        )

from_domain(cfg) classmethod

Build the masked view: the key becomes a boolean + last-4 hint only.

Source code in src/revalid/app.py

@classmethod
def from_domain(cls, cfg: Settings) -> "SettingsOut":
    """Build the masked view: the key becomes a boolean + last-4 hint only."""
    key = cfg.api_key or ""
    return cls(
        model=cfg.model,
        base_url=cfg.base_url,
        api_key_set=bool(key),
        api_key_hint=key[-4:] if key else None,
    )

SettingsUpdateIn

Bases: BaseModel

Settings update payload; a blank api_key keeps the stored one (ADR-0021).

Source code in src/revalid/app.py

class SettingsUpdateIn(BaseModel):
    """Settings update payload; a blank ``api_key`` keeps the stored one (ADR-0021)."""

    model_config = ConfigDict(protected_namespaces=())

    model: str = Field(min_length=1)
    base_url: str | None = None
    api_key: str | None = None
    clear_key: bool = False

StartSessionRequest

Bases: BaseModel

Optional body for starting a session: free-launch + seed goal (FR-17 Slice 5).

Source code in src/revalid/app.py

class StartSessionRequest(BaseModel):
    """Optional body for starting a session: free-launch + seed goal (FR-17 Slice 5)."""

    free_launch: bool = False
    # A user-owned goal drafted before the session (FR-17 6b-iii-b): when present,
    # the session seeds it verbatim instead of generating one at start.
    initial_goal: list[str] | None = None
    # The retest scope — the exact target URL(s) the agent may hit (FR-17). Set at
    # launch (reachability is fixed when the sandbox is provisioned), so there is no
    # live-edit path; changing scope means a fresh session. Defaults to the finding's
    # affected endpoints when omitted.
    target_endpoints: list[str] | None = None
    # Open the session `idle` (created but not started) instead of auto-running —
    # the Restart path (issue #150). The goal + scope are recorded so the idle
    # console shows them; the operator presses Start to provision and begin.
    deferred: bool = False

VerdictOut

Bases: BaseModel

An agentic verdict as returned by the API (FR-09/FR-17).

Every verdict is a retest-session conclusion (the batch verdict path retired in FR-17 6b-iii); the shape mirrors the FR-12 :class:~revalid.export.VerdictExport.

Source code in src/revalid/app.py

class VerdictOut(BaseModel):
    """An agentic verdict as returned by the API (FR-09/FR-17).

    Every verdict is a retest-session conclusion (the batch verdict path retired
    in FR-17 6b-iii); the shape mirrors the FR-12 :class:`~revalid.export.VerdictExport`.
    """

    id: int
    finding_id: int
    session_id: int | None
    actor: str
    status: VerdictStatus
    reason_code: str
    rationale: str
    matched_indicators: tuple[str, ...]
    evidence: AgenticEvidence | None

    @classmethod
    def from_record(cls, record: VerdictRecord) -> "VerdictOut":
        """Build the API view from a stored verdict row."""
        return cls(
            id=record.id,
            finding_id=record.finding_id,
            session_id=record.session_id,
            actor=record.actor,
            status=VerdictStatus(record.status),
            reason_code=record.reason_code,
            rationale=record.rationale,
            matched_indicators=tuple(record.matched_indicators),
            evidence=AgenticEvidence(**record.evidence) if record.evidence is not None else None,
        )

from_record(record) classmethod

Build the API view from a stored verdict row.

Source code in src/revalid/app.py

@classmethod
def from_record(cls, record: VerdictRecord) -> "VerdictOut":
    """Build the API view from a stored verdict row."""
    return cls(
        id=record.id,
        finding_id=record.finding_id,
        session_id=record.session_id,
        actor=record.actor,
        status=VerdictStatus(record.status),
        reason_code=record.reason_code,
        rationale=record.rationale,
        matched_indicators=tuple(record.matched_indicators),
        evidence=AgenticEvidence(**record.evidence) if record.evidence is not None else None,
    )

create_app(db_path='revalid.db', engine=None)

Build the application with its own database engine.

Parameters:

Name	Type	Description	Default
db_path	str	SQLite file backing this instance; ignored when engine is given (tests inject an in-memory engine).	'revalid.db'
engine	Engine | None	Pre-built engine to use instead of opening db_path.	None

Returns:

Type	Description
FastAPI	The configured FastAPI application.

Source code in src/revalid/app.py

def create_app(db_path: str = "revalid.db", engine: Engine | None = None) -> FastAPI:
    """Build the application with its own database engine.

    Args:
        db_path: SQLite file backing this instance; ignored when ``engine``
            is given (tests inject an in-memory engine).
        engine: Pre-built engine to use instead of opening ``db_path``.

    Returns:
        The configured FastAPI application.
    """
    db_engine = engine if engine is not None else create_db_engine(db_path)
    sessions = session_factory(db_engine)
    _fail_orphaned_extractions(sessions)
    app = FastAPI(title="revalid", version=__version__)
    app.state.sessions = sessions
    registry = SessionRegistry()
    app.state.registry = registry
    extractions = ExtractionRegistry()
    app.state.extractions = extractions

    api = APIRouter(prefix="/api")
    _register_core_routes(api, sessions)
    _register_finding_routes(api, sessions)
    _register_finding_retest_routes(api, sessions)
    _register_report_routes(api, sessions, extractions)
    _register_report_cancel_route(api, sessions, extractions)
    _register_report_admin_routes(api, sessions, registry, extractions)
    _register_verdict_routes(api, sessions)
    _register_session_routes(api, sessions, registry)
    _register_launch_route(api, sessions, registry)
    _register_free_launch_route(api, sessions, registry)
    _register_guidance_routes(api, sessions, registry)
    _register_lifecycle_routes(api, sessions, registry)
    _register_goal_routes(api, sessions, registry)
    _register_adjudicate_route(api, sessions)
    _register_session_stream_route(api, sessions)
    _register_export_routes(api, sessions)
    _register_settings_routes(api, sessions)
    _register_chat_routes(api, sessions)
    _register_chat_message_route(api, sessions)
    app.include_router(api)
    _register_api_fallback(app)
    _mount_spa(app)

    return app

get_extraction_agent(settings)

Yield the FR-03 extraction agent built from the persisted setting (ADR-0021).

Source code in src/revalid/app.py

def get_extraction_agent(settings: SettingsDep) -> Agent[None, list[ExtractedFinding]]:
    """Yield the FR-03 extraction agent built from the persisted setting (ADR-0021)."""
    return build_extraction_agent(build_model(settings))

get_goal_agent(settings)

Yield the FR-17 retest-goal agent built from the persisted setting (ADR-0021).

Source code in src/revalid/app.py

def get_goal_agent(settings: SettingsDep) -> Agent[None, GeneratedGoal]:
    """Yield the FR-17 retest-goal agent built from the persisted setting (ADR-0021)."""
    return build_goal_agent(build_model(settings))

get_metadata_agent(settings)

Yield the FR-03 document-metadata agent built from the persisted setting (#133).

Source code in src/revalid/app.py

def get_metadata_agent(settings: SettingsDep) -> Agent[None, ReportMetadata]:
    """Yield the FR-03 document-metadata agent built from the persisted setting (#133)."""
    return build_metadata_agent(build_model(settings))

get_reports_agent(settings)

Yield the FR-18 read-only reports assistant built from the persisted setting.

Source code in src/revalid/app.py

def get_reports_agent(settings: SettingsDep) -> Agent[ReportsChatDeps, str]:
    """Yield the FR-18 read-only reports assistant built from the persisted setting."""
    return build_reports_agent(build_model(settings))

get_retest_agent(settings)

Yield the FR-17 agentic retest agent built from the persisted setting (ADR-0021).

Source code in src/revalid/app.py

def get_retest_agent(settings: SettingsDep) -> RetestAgent:
    """Yield the FR-17 agentic retest agent built from the persisted setting (ADR-0021)."""
    return build_retest_agent(build_model(settings))

get_sandbox_factory()

Yield the production sandbox factory: a fresh egress-locked Docker sandbox per session.

The returned factory is bound to the retest session's id (which scopes the sandbox's internal Docker network, FR-06). Tests override this with a factory that returns a :class:~revalid.sandbox.FakeSandbox, so the HTTP flow runs without Docker.

Source code in src/revalid/app.py

def get_sandbox_factory() -> SandboxFactory:
    """Yield the production sandbox factory: a fresh egress-locked Docker sandbox per session.

    The returned factory is bound to the retest session's id (which scopes the
    sandbox's internal Docker network, FR-06). Tests override this with a factory
    that returns a :class:`~revalid.sandbox.FakeSandbox`, so the HTTP flow runs
    without Docker.
    """
    return lambda sid: DockerSandbox(sid)

get_settings_dep(request)

Load the persisted model/provider setting, seeding a fresh DB (ADR-0021).

Source code in src/revalid/app.py

def get_settings_dep(request: Request) -> Settings:
    """Load the persisted model/provider setting, seeding a fresh DB (ADR-0021)."""
    sessions = cast("sessionmaker[Session]", request.app.state.sessions)
    with sessions() as session:
        return load_or_seed(session)

get_taxonomy_agent(settings)

Yield the FR-19 opt-in taxonomy-enrichment agent (issue #233).

Constructed per request like every other agent, but only invoked when the caller passed enrich — building it costs no model call, so the LLM-free doors stay LLM-free by default (ADR-0021 for the backend selection).

Source code in src/revalid/app.py

def get_taxonomy_agent(settings: SettingsDep) -> Agent[None, FindingTaxonomy]:
    """Yield the FR-19 opt-in taxonomy-enrichment agent (issue #233).

    Constructed per request like every other agent, but only *invoked* when the
    caller passed ``enrich`` — building it costs no model call, so the LLM-free
    doors stay LLM-free by default (ADR-0021 for the backend selection).
    """
    return build_taxonomy_agent(build_model(settings))

run_conclude(sessions, registry, session_id, status, rationale)

Record the operator's manual conclusion + tear down (ADR-0034 background task).

Parameters:

Name	Type	Description	Default
sessions	sessionmaker[Session]	The app's session factory (each task opens a fresh session).	required
registry	SessionRegistry	The process-local live-session registry.	required
session_id	int	The session to conclude.	required
status	VerdictStatus	The operator's determination.	required
rationale	str	The operator's justification.	required

Source code in src/revalid/app.py

def run_conclude(
    sessions: sessionmaker[Session],
    registry: SessionRegistry,
    session_id: int,
    status: VerdictStatus,
    rationale: str,
) -> None:
    """Record the operator's manual conclusion + tear down (ADR-0034 background task).

    Args:
        sessions: The app's session factory (each task opens a fresh session).
        registry: The process-local live-session registry.
        session_id: The session to conclude.
        status: The operator's determination.
        rationale: The operator's justification.
    """
    with sessions() as session:
        conclude_session(session, registry, session_id, status, rationale)

run_continue(sessions, registry, session_id)

Resume a paused session (ADR-0034 "Keep going", background task).

Runs in the background because resuming drives further agent turns. A no-op unless the session is parked in awaiting_operator with a live agent (needs_guidance folded into that state in ADR-0042).

Parameters:

Name	Type	Description	Default
sessions	sessionmaker[Session]	The app's session factory (each task opens a fresh session).	required
registry	SessionRegistry	The process-local live-session registry.	required
session_id	int	The paused retest session to resume.	required

Source code in src/revalid/app.py

def run_continue(
    sessions: sessionmaker[Session],
    registry: SessionRegistry,
    session_id: int,
) -> None:
    """Resume a paused session (ADR-0034 "Keep going", background task).

    Runs in the background because resuming drives further agent turns. A no-op
    unless the session is parked in ``awaiting_operator`` with a live agent
    (``needs_guidance`` folded into that state in ADR-0042).

    Args:
        sessions: The app's session factory (each task opens a fresh session).
        registry: The process-local live-session registry.
        session_id: The paused retest session to resume.
    """
    with sessions() as session:
        continue_session(session, registry, session_id)

run_decision(sessions, registry, session_id, approved, reason, command_id)

Resume a paused session with the operator's decision (FR-17 background task).

Parameters:

Name	Type	Description	Default
sessions	sessionmaker[Session]	The app's session factory (each task opens a fresh session).	required
registry	SessionRegistry	The process-local live-session registry.	required
session_id	int	The retest session to resume.	required
approved	bool	Whether the pending command was approved.	required
reason	str	Optional operator reason (surfaced to the model on rejection).	required
command_id	str	The cid path param from the approve/reject URL; must match the session's pending tool_call_id or the decision is a no-op (guards against a double-click resuming the run twice).	required

Source code in src/revalid/app.py

def run_decision(
    sessions: sessionmaker[Session],
    registry: SessionRegistry,
    session_id: int,
    approved: bool,
    reason: str,
    command_id: str,
) -> None:
    """Resume a paused session with the operator's decision (FR-17 background task).

    Args:
        sessions: The app's session factory (each task opens a fresh session).
        registry: The process-local live-session registry.
        session_id: The retest session to resume.
        approved: Whether the pending command was approved.
        reason: Optional operator reason (surfaced to the model on rejection).
        command_id: The ``cid`` path param from the approve/reject URL; must
            match the session's pending ``tool_call_id`` or the decision is a
            no-op (guards against a double-click resuming the run twice).
    """
    with sessions() as session:
        apply_decision(
            session, registry, session_id, approved=approved, reason=reason, command_id=command_id
        )

run_extraction(sessions, report_id, data, agent, metadata_agent, extractions)

Extract findings from an uploaded PDF and persist them (FR-01/FR-03/FR-11).

Runs as a FastAPI background task — a sync function Starlette dispatches to its threadpool, so it must open its own session (the request session is already closed once the 202 was sent) and it never blocks the event loop. The report is always moved out of extracting: to ready with its findings persisted, failed with the error recorded, or — when the operator stopped it (issue #205) — cancelled, so the UI's status poll always terminates.

Cancellation (issue #205): a Stop or delete interrupts the single in-flight model call via extractions. Because extraction is one whole-document call, a Stop lands cancelled with no findings; a report delete flags "deleted" so this persists nothing into the row being removed.

Parameters:

Name	Type	Description	Default
sessions	sessionmaker[Session]	The app's session factory (each task opens a fresh session).	required
report_id	int	The extracting report row to fill in.	required
data	bytes	The uploaded PDF bytes.	required
agent	Agent[None, list[ExtractedFinding]]	The extraction agent (a stand-in model in tests).	required
metadata_agent	Agent[None, ReportMetadata]	The document-metadata agent (a stand-in model in tests).	required
extractions	ExtractionRegistry	The process-local extraction cancel registry.	required

Source code in src/revalid/app.py

def run_extraction(
    sessions: sessionmaker[Session],
    report_id: int,
    data: bytes,
    agent: Agent[None, list[ExtractedFinding]],
    metadata_agent: Agent[None, ReportMetadata],
    extractions: ExtractionRegistry,
) -> None:
    """Extract findings from an uploaded PDF and persist them (FR-01/FR-03/FR-11).

    Runs as a FastAPI background task — a sync function Starlette dispatches to
    its threadpool, so it must open its **own** session (the request session is
    already closed once the ``202`` was sent) and it never blocks the event
    loop. The report is always moved out of ``extracting``: to ``ready`` with its
    findings persisted, ``failed`` with the error recorded, or — when the operator
    stopped it (issue #205) — ``cancelled``, so the UI's status poll always
    terminates.

    Cancellation (issue #205): a Stop or delete interrupts the single in-flight
    model call via ``extractions``. Because extraction is one whole-document call, a
    Stop lands ``cancelled`` with no findings; a report delete flags ``"deleted"`` so
    this persists nothing into the row being removed.

    Args:
        sessions: The app's session factory (each task opens a fresh session).
        report_id: The ``extracting`` report row to fill in.
        data: The uploaded PDF bytes.
        agent: The extraction agent (a stand-in model in tests).
        metadata_agent: The document-metadata agent (a stand-in model in tests).
        extractions: The process-local extraction cancel registry.
    """
    with sessions() as session:
        report = session.get(ReportRecord, report_id)
        if report is None:  # pragma: no cover - the row was just committed
            extractions.clear(report_id)
            return
        try:
            pdf = read_pdf(data)
            result = _run_cancellable_extraction(agent, pdf, report_id, extractions)
        except PdfError as exc:
            report.status, report.error = ReportStatus.FAILED.value, str(exc)
            session.commit()
            extractions.clear(report_id)
            return
        except Exception as exc:
            report.status, report.error = ReportStatus.FAILED.value, f"extraction failed: {exc}"
            session.commit()
            extractions.clear(report_id)
            return
        if result.cancelled:
            _settle_cancelled_extraction(session, report_id, result.findings, extractions)
            return
        # The report may have been deleted mid-run (a concurrent delete on another
        # session); re-check before writing so findings are never orphaned (#205).
        if session.get(ReportRecord, report_id) is None:  # pragma: no cover - narrow race
            extractions.clear(report_id)
            return
        _persist_findings(session, result.findings, report_id=report_id)
        # Best-effort document metadata (#133) — never fails the report.
        report.doc_metadata = extract_metadata(metadata_agent, pdf).model_dump()
        report.status = ReportStatus.READY.value
        report.finding_count = len(result.findings)
        session.commit()
        extractions.clear(report_id)

run_first_step(sessions, registry, session_id, agent, make_sandbox, finding, goal_agent, initial_goal=None, target_endpoints=None)

Build the sandbox and run the retest agent's first step (FR-17 background task).

Runs as a FastAPI background task — a sync function on Starlette's threadpool — so it opens its own session (the request session closed with the 202). It must let no exception escape: make_sandbox may raise (SandboxUnavailableError when the sandbox extra is absent, or a real Docker error) and :func:~revalid.retest_session.start_and_step calls sandbox.start() outside its own guard, so both are wrapped here. On any failure the sandbox (if one was created) is best-effort torn down and the session is settled to error — never stranded in working.

It also seeds the goal (FR-17 6b-ii): a caller-supplied initial_goal (a goal drafted before the session started, FR-17 6b-iii-b) is used verbatim; otherwise the goal agent generates a generic retest goal. Either way the result is emitted as the initial plan_updated (the "Current goal" panel) and prepended to the agent's prompt. Goal generation is best-effort — a failure degrades to an empty goal, never blocking start.

Parameters:

Name	Type	Description	Default
sessions	sessionmaker[Session]	The app's session factory (each task opens a fresh session).	required
registry	SessionRegistry	The process-local live-session registry.	required
session_id	int	The already-created (working) retest session to drive.	required
agent	RetestAgent	The built retest agent (a stand-in model in tests).	required
make_sandbox	SandboxFactory	The session-scoped sandbox factory.	required
finding	Finding	The finding to retest — the agent's goal is derived from it.	required
goal_agent	Agent[None, GeneratedGoal]	The FR-17 goal agent (a stand-in model in tests).	required
initial_goal	tuple[str, ...] | None	A pre-start goal drafted by the user (FR-17 6b-iii-b). None means none was supplied, so one is generated; a supplied goal — including an empty one — is seeded verbatim and generation is skipped (#113 F3).	None
target_endpoints	tuple[str, ...] | None	The launch-time retest scope (FR-17) — the exact URL(s) the agent may hit; defaults to the finding's endpoints when omitted.	None

Source code in src/revalid/app.py

def run_first_step(
    sessions: sessionmaker[Session],
    registry: SessionRegistry,
    session_id: int,
    agent: RetestAgent,
    make_sandbox: SandboxFactory,
    finding: Finding,
    goal_agent: Agent[None, GeneratedGoal],
    initial_goal: tuple[str, ...] | None = None,
    target_endpoints: tuple[str, ...] | None = None,
) -> None:
    """Build the sandbox and run the retest agent's first step (FR-17 background task).

    Runs as a FastAPI background task — a sync function on Starlette's threadpool
    — so it opens its **own** session (the request session closed with the
    ``202``). It must let **no** exception escape: ``make_sandbox`` may raise
    (``SandboxUnavailableError`` when the ``sandbox`` extra is absent, or a real
    Docker error) and :func:`~revalid.retest_session.start_and_step` calls
    ``sandbox.start()`` outside its own guard, so both are wrapped here. On any
    failure the sandbox (if one was created) is best-effort torn down and the
    session is settled to ``error`` — never stranded in ``working``.

    It also **seeds the goal** (FR-17 6b-ii): a caller-supplied ``initial_goal``
    (a goal drafted before the session started, FR-17 6b-iii-b) is used verbatim;
    otherwise the goal agent generates a generic retest goal. Either way the
    result is emitted as the initial ``plan_updated`` (the "Current goal" panel)
    and prepended to the agent's prompt. Goal generation is best-effort — a
    failure degrades to an empty goal, never blocking start.

    Args:
        sessions: The app's session factory (each task opens a fresh session).
        registry: The process-local live-session registry.
        session_id: The already-created (``working``) retest session to drive.
        agent: The built retest agent (a stand-in model in tests).
        make_sandbox: The session-scoped sandbox factory.
        finding: The finding to retest — the agent's goal is derived from it.
        goal_agent: The FR-17 goal agent (a stand-in model in tests).
        initial_goal: A pre-start goal drafted by the user (FR-17 6b-iii-b).
            ``None`` means none was supplied, so one is generated; a supplied
            goal — **including an empty one** — is seeded verbatim and
            generation is skipped (#113 F3).
        target_endpoints: The launch-time retest scope (FR-17) — the exact URL(s)
            the agent may hit; defaults to the finding's endpoints when omitted.
    """
    with sessions() as session:
        sandbox: Sandbox | None = None
        try:
            sandbox = make_sandbox(session_id)
            record = session.get(RetestSessionRecord, session_id)
            free_launch = record.free_launch if record else False
            # Scope is set once, at launch: the operator's endpoints when supplied,
            # else the finding's. It's recorded (TARGET_SET) for the read-only cockpit
            # display and injected authoritatively into the prompt.
            endpoints = tuple(target_endpoints) if target_endpoints else finding.affected_endpoints
            if endpoints:
                append_event(
                    session, session_id, SessionEventKind.TARGET_SET, {"endpoints": list(endpoints)}
                )
            # `None` = no goal supplied, so draft one (best-effort; never blocks).
            # An explicitly empty goal is the operator's choice to start goal-less
            # and steer by message, and is left alone (#113 F3).
            goal: tuple[str, ...] = () if initial_goal is None else tuple(initial_goal)
            if initial_goal is None:
                with contextlib.suppress(Exception):
                    goal = generate_goal(goal_agent, finding)
            if goal:
                append_event(
                    session, session_id, SessionEventKind.PLAN_UPDATED, {"steps": list(goal)}
                )
            start_and_step(
                session,
                registry,
                session_id,
                agent,
                sandbox,
                _target_preamble(endpoints) + _goal_prompt(goal, finding),
                free_launch=free_launch,
            )
        except Exception as exc:  # broad on purpose: no failure may strand the session
            if sandbox is not None:
                with contextlib.suppress(Exception):  # best-effort teardown only
                    sandbox.stop()
            _fail(session, registry, session_id, str(exc))

run_free_launch(sessions, registry, session_id, enabled)

Toggle free-launch on a session (FR-17 Slice 5 background task).

Runs in the background because enabling may drive the auto-approve loop (successive agent turns). A no-op if the session is no longer live.

Parameters:

Name	Type	Description	Default
sessions	sessionmaker[Session]	The app's session factory (each task opens a fresh session).	required
registry	SessionRegistry	The process-local live-session registry.	required
session_id	int	The retest session to toggle.	required
enabled	bool	The new free-launch state.	required

Source code in src/revalid/app.py

def run_free_launch(
    sessions: sessionmaker[Session],
    registry: SessionRegistry,
    session_id: int,
    enabled: bool,
) -> None:
    """Toggle free-launch on a session (FR-17 Slice 5 background task).

    Runs in the background because enabling may drive the auto-approve loop
    (successive agent turns). A no-op if the session is no longer live.

    Args:
        sessions: The app's session factory (each task opens a fresh session).
        registry: The process-local live-session registry.
        session_id: The retest session to toggle.
        enabled: The new free-launch state.
    """
    with sessions() as session:
        set_free_launch(session, registry, session_id, enabled)

run_goal(sessions, registry, session_id, steps)

Set the user-owned goal on a session (FR-17 6b-ii background task).

Source code in src/revalid/app.py

def run_goal(
    sessions: sessionmaker[Session], registry: SessionRegistry, session_id: int, steps: list[str]
) -> None:
    """Set the user-owned goal on a session (FR-17 6b-ii background task)."""
    with sessions() as session:
        set_goal(session, registry, session_id, steps)

run_human_command(sessions, registry, session_id, command)

Run a manual operator command (!) in the session's sandbox (FR-17 background task).

Parameters:

Name	Type	Description	Default
sessions	sessionmaker[Session]	The app's session factory (each task opens a fresh session).	required
registry	SessionRegistry	The process-local live-session registry.	required
session_id	int	The retest session to run the command in.	required
command	str	The exact shell command the operator submitted (without the !).	required

Source code in src/revalid/app.py

def run_human_command(
    sessions: sessionmaker[Session],
    registry: SessionRegistry,
    session_id: int,
    command: str,
) -> None:
    """Run a manual operator command (`!`) in the session's sandbox (FR-17 background task).

    Args:
        sessions: The app's session factory (each task opens a fresh session).
        registry: The process-local live-session registry.
        session_id: The retest session to run the command in.
        command: The exact shell command the operator submitted (without the `!`).
    """
    with sessions() as session:
        submit_human_command(session, registry, session_id, command)

run_message(sessions, registry, session_id, text, agent, make_sandbox)

Deliver an operator chat message to the one agent — the Claude-Code model (#163).

The chat is the lifecycle control: talking to a parked agent is how you start or continue it, so there is no separate Resume/Wake/Keep-going button and no second voice (ADR-0042 removed the parallel read-only Q&A). What the message does depends only on where the session is:

• idle — provision the sandbox and run the first turn, message folded into the opening prompt.
• awaiting_operator — the agent handed back (a reply, a guided report, "I've exhausted my options"); resume it with the message.
• stopped — resume the paused session with the message.
• awaiting_command — a command awaits approval; the message withdraws it and steers the agent (Claude Code's "type at the permission prompt").
• working — the agent is mid-turn; the message is queued and the same agent answers it at the next turn boundary (:func:~revalid.retest_session._advance).
• terminal — a finished session cannot be messaged; a no-op.

Whether the message is a question or an instruction is the agent's call: it answers with the ungated respond tool and acts with the approval-gated run_command, so waking it can never execute anything the operator has not approved.

Parameters:

Name	Type	Description	Default
sessions	sessionmaker[Session]	The app's session factory (each task opens a fresh session).	required
registry	SessionRegistry	The process-local live-session registry.	required
session_id	int	The retest session to message.	required
text	str	The exact operator message.	required
agent	RetestAgent	The built retest agent, to wake an idle session.	required
make_sandbox	SandboxFactory	The session-scoped sandbox factory, likewise.	required

Source code in src/revalid/app.py

def run_message(
    sessions: sessionmaker[Session],
    registry: SessionRegistry,
    session_id: int,
    text: str,
    agent: RetestAgent,
    make_sandbox: SandboxFactory,
) -> None:
    """Deliver an operator chat message to the one agent — the Claude-Code model (#163).

    The chat *is* the lifecycle control: talking to a parked agent is how you start or
    continue it, so there is no separate Resume/Wake/Keep-going button and no second
    voice (ADR-0042 removed the parallel read-only Q&A). What the message does depends
    only on where the session is:

    * ``idle`` — provision the sandbox and run the first turn, message folded into the
      opening prompt.
    * ``awaiting_operator`` — the agent handed back (a reply, a guided report, "I've
      exhausted my options"); resume it with the message.
    * ``stopped`` — resume the paused session with the message.
    * ``awaiting_command`` — a command awaits approval; the message withdraws it and
      steers the agent (Claude Code's "type at the permission prompt").
    * ``working`` — the agent is mid-turn; the message is queued and the *same* agent
      answers it at the next turn boundary (:func:`~revalid.retest_session._advance`).
    * terminal — a finished session cannot be messaged; a no-op.

    Whether the message is a question or an instruction is the **agent's** call: it
    answers with the ungated ``respond`` tool and acts with the approval-gated
    ``run_command``, so waking it can never execute anything the operator has not
    approved.

    Args:
        sessions: The app's session factory (each task opens a fresh session).
        registry: The process-local live-session registry.
        session_id: The retest session to message.
        text: The exact operator message.
        agent: The built retest agent, to wake an ``idle`` session.
        make_sandbox: The session-scoped sandbox factory, likewise.
    """
    with sessions() as session:
        record = session.get(RetestSessionRecord, session_id)
        if record is None:
            return
        status = RetestSessionStatus(record.status)
        if is_terminal(status):
            return  # a concluded/ended/errored session cannot be messaged
        if status is RetestSessionStatus.IDLE:
            # No live session yet, so the message cannot be buffered on one: record it
            # here and hand it to the opening prompt instead.
            append_event(session, session_id, SessionEventKind.HUMAN_MESSAGE, {"text": text})
            _start_idle(session, registry, session_id, agent, make_sandbox, steer=text)
            return
        submit_message(session, registry, session_id, text)
        if status is RetestSessionStatus.STOPPED:
            resume_session(session, registry, session_id)
        elif status is RetestSessionStatus.AWAITING_OPERATOR:
            continue_session(session, registry, session_id)
        elif status is RetestSessionStatus.AWAITING_COMMAND:
            resume_with_message_at_gate(session, registry, session_id)

run_regenerate_goal(sessions, registry, session_id, goal_agent, finding)

Regenerate + set the goal for a session (FR-17 6b-ii background task).

Source code in src/revalid/app.py

def run_regenerate_goal(
    sessions: sessionmaker[Session],
    registry: SessionRegistry,
    session_id: int,
    goal_agent: Agent[None, GeneratedGoal],
    finding: Finding,
) -> None:
    """Regenerate + set the goal for a session (FR-17 6b-ii background task)."""
    with sessions() as session:
        set_goal(session, registry, session_id, list(generate_goal(goal_agent, finding)))

run_reopen(sessions, session_id)

Reopen a concluded session so testing can continue (issue #214 background task).

A pure DB op (the session is already torn down); the operator wakes the now-idle session to re-provision the sandbox and continue.

Parameters:

Name	Type	Description	Default
sessions	sessionmaker[Session]	The app's session factory (each task opens a fresh session).	required
session_id	int	The concluded session to reopen.	required

Source code in src/revalid/app.py

def run_reopen(sessions: sessionmaker[Session], session_id: int) -> None:
    """Reopen a concluded session so testing can continue (issue #214 background task).

    A pure DB op (the session is already torn down); the operator wakes the now-idle
    session to re-provision the sandbox and continue.

    Args:
        sessions: The app's session factory (each task opens a fresh session).
        session_id: The concluded session to reopen.
    """
    with sessions() as session:
        reopen_session(session, session_id)

run_restart_model(sessions, registry, session_id)

Abort + re-run the in-flight turn to unstick a wedged model (issue #204).

Runs as a background task: it cancels the current turn from a fresh thread while the wedged turn's own thread re-runs it. A no-op when nothing is in flight.

Source code in src/revalid/app.py

def run_restart_model(
    sessions: sessionmaker[Session], registry: SessionRegistry, session_id: int
) -> None:
    """Abort + re-run the in-flight turn to unstick a wedged model (issue #204).

    Runs as a background task: it cancels the current turn from a fresh thread while
    the wedged turn's own thread re-runs it. A no-op when nothing is in flight.
    """
    with sessions() as session:
        restart_model(session, registry, session_id)

run_resume(sessions, registry, session_id)

Resume a stopped session — Resume (issue #150, background task).

Runs in the background because resuming may drive further agent turns.

Source code in src/revalid/app.py

def run_resume(sessions: sessionmaker[Session], registry: SessionRegistry, session_id: int) -> None:
    """Resume a stopped session — Resume (issue #150, background task).

    Runs in the background because resuming may drive further agent turns.
    """
    with sessions() as session:
        resume_session(session, registry, session_id)

run_start(sessions, registry, session_id, agent, make_sandbox)

Provision the sandbox and run the first step of an idle session (issue #150).

The deferred-start counterpart of :func:run_first_step: a Restart opened the session idle with its goal + scope already recorded, so this reads them back from the transcript (never re-recording — the target_set invariant is "emitted once") and drives the first agent turn. A no-op unless the session is still idle, so a double-click Start cannot double-provision. Lets no exception escape: on any failure the sandbox is torn down and the session settles to error.

Parameters:

Name	Type	Description	Default
sessions	sessionmaker[Session]	The app's session factory (each task opens a fresh session).	required
registry	SessionRegistry	The process-local live-session registry.	required
session_id	int	The idle session to start.	required
agent	RetestAgent	The built retest agent (a stand-in model in tests).	required
make_sandbox	SandboxFactory	The session-scoped sandbox factory.	required

Source code in src/revalid/app.py

def run_start(
    sessions: sessionmaker[Session],
    registry: SessionRegistry,
    session_id: int,
    agent: RetestAgent,
    make_sandbox: SandboxFactory,
) -> None:
    """Provision the sandbox and run the first step of an ``idle`` session (issue #150).

    The deferred-start counterpart of :func:`run_first_step`: a ``Restart`` opened
    the session ``idle`` with its goal + scope already recorded, so this reads them
    back from the transcript (never re-recording — the ``target_set`` invariant is
    "emitted once") and drives the first agent turn. A no-op unless the session is
    still ``idle``, so a double-click Start cannot double-provision. Lets no
    exception escape: on any failure the sandbox is torn down and the session
    settles to ``error``.

    Args:
        sessions: The app's session factory (each task opens a fresh session).
        registry: The process-local live-session registry.
        session_id: The ``idle`` session to start.
        agent: The built retest agent (a stand-in model in tests).
        make_sandbox: The session-scoped sandbox factory.
    """
    with sessions() as session:
        _start_idle(session, registry, session_id, agent, make_sandbox)

run_stop(sessions, registry, session_id)

Pause a running session — Stop (issue #150, background task).

Source code in src/revalid/app.py

def run_stop(sessions: sessionmaker[Session], registry: SessionRegistry, session_id: int) -> None:
    """Pause a running session — Stop (issue #150, background task)."""
    with sessions() as session:
        stop_session(session, registry, session_id)