← Files Codex SecurityARCHIVED FILE
scripts/workbench_saved_results.py
116 KB · Oct 7, 2026 · 06:02 UTC
"""Retain semantic scan drafts without rerunning analysis or changing run state."""
from __future__ import annotations
import argparse
import base64
import copy
import hashlib
import json
import os
import re
import sqlite3
import stat
import sys
from collections.abc import Iterator
from contextlib import contextmanager
from pathlib import Path
from typing import Any
sys.path.insert(0, str(Path(__file__).resolve().parent))
from finalize_scan_contract import (
ContractError,
_finding_strength,
_populate_unsealed_artifact_envelope,
_populate_unsealed_manifest_envelope,
_prepare_scan_finalization,
_read_json,
_read_saved_threat_model,
_read_scan_local_json,
_read_scan_local_json_bytes,
_read_scan_local_json_with_metadata,
_recover_unsealed_findings,
_remove_scan_local_file_if_exists,
_validate_completion_binding,
_validate_resolved_deferred,
_validate_schema_node,
_write_prepared_scan_finalization,
finalize_scan,
finding_candidate_id,
open_scan_local_file_descriptor,
write_scan_local_bytes,
write_threat_model_projection_if_possible,
)
from workbench_constants import PHASES
from workbench_target import committed_diff_snapshot_digest
from workbench_validation import path_within_scope
_PUBLISHED_OUTPUTS = (
"findings.json",
"coverage.json",
"scan-manifest.json",
"report.md",
"threatmodel.md",
"report.html",
"exports/results.sarif",
)
_PUBLICATION_FOLLOW_UP_WARNING = (
"Saved scan evidence remains on disk; result publication needs follow-up:"
)
_RESERVED_ARTIFACT_PATHS = json.loads(
Path(__file__).with_name("reserved_artifact_paths.json").read_text(encoding="utf-8")
)
def threat_model_fields(db: Any, scan: sqlite3.Row) -> dict[str, Any]:
scan_dir = Path(scan["scan_dir"])
fields: dict[str, Any] = {"threatModelAvailable": False}
try:
db.require_recorded_manifest_digest(scan, scan_dir)
db.verify_manifest_binding(
scan, _read_scan_local_json(scan_dir, db.ARTIFACTS["manifest"], "scan manifest")
)
saved_model = _read_saved_threat_model(scan_dir)
if saved_model is not None:
description = saved_model[0]
fields.update(
threatModelAvailable=True,
threatModelProvenance=description["provenance"],
)
if description["path"] is not None:
fields["threatModelPath"] = description["path"]
except (ContractError, OSError, SystemExit):
# Unavailable optional model data must not prevent reading the saved scan.
pass
return fields
def refresh_completed_scan(
db: Any,
connection: sqlite3.Connection,
scan: sqlite3.Row,
cost_json: str | None,
) -> dict[str, Any]:
warnings = json.loads(scan["completion_warnings_json"])
scan_dir = db.require_canonical_scan_directory(Path(scan["scan_dir"]))
db.require_recorded_manifest_digest(scan, scan_dir)
db.verify_manifest_binding(scan, db.read_json_object(scan_dir / db.ARTIFACTS["manifest"]))
try:
manifest, _, _ = finalize_scan(
scan_dir,
expected_coverage_mode=db.expected_coverage_mode(scan),
projection_warnings=warnings,
)
except ContractError as exc:
raise SystemExit(str(exc)) from exc
db.verify_manifest_binding(scan, manifest)
manifest_digest = db.published_manifest_digest(scan_dir, manifest)
db.pin_legacy_manifest_digest(connection, scan["id"], manifest_digest)
if cost_json is not None and scan["recipe_json"] is not None:
db.scan_usage.reconcile_completed_scan_cost(connection, scan, cost_json)
if warnings != json.loads(scan["completion_warnings_json"]):
with connection:
connection.execute(
"UPDATE scans SET completion_warnings_json = ? WHERE id = ?",
(json.dumps(warnings), scan["id"]),
)
return db.scan_context(connection, scan["id"])
def _encoded(value: Any) -> bytes:
return json.dumps(
value, ensure_ascii=True, allow_nan=False, sort_keys=True, separators=(",", ":")
).encode()
def _digest(value: Any) -> str:
return hashlib.sha256(_encoded(value)).hexdigest()
def _children(scan_dir: Path, relative: str) -> list[str]:
cursor = scan_dir
for part in Path(relative).parts:
if part in {"..", "."}:
return []
cursor = cursor / part
try:
if not stat.S_ISDIR(cursor.lstat().st_mode):
return []
except FileNotFoundError:
return []
return sorted(child.name for child in cursor.iterdir())
def _latest_successful_reducer(workers: list[Any]) -> Any | None:
return max(
(
worker
for worker in workers
if worker["kind"] == "dedup"
and worker["status"] == "succeeded"
and worker["result_manifest_path"]
),
key=lambda worker: (worker["completed_at"] or "", worker["id"]),
default=None,
)
def _checkpoint_paths(scan_dir: Path, directory: str) -> list[str]:
return [
f"{directory}/{name}"
for name in _children(scan_dir, directory)
if re.fullmatch(r"[0-9a-f]{64}\.json", name)
]
def _worker_outputs(scan_dir: Path, worker: Any) -> list[tuple[str, int]]:
output = Path(worker["artifact_dir"]).relative_to(scan_dir)
attempts = (output.parent if output.name == "output" else output) / "attempts"
archived = [
((attempts / name).as_posix(), int(name.split("-")[1]))
for name in _children(scan_dir, attempts.as_posix())
if re.fullmatch(r"attempt-\d+", name)
]
attempt = int(worker["attempt"] or 0) if "attempt" in worker.keys() else 0
if worker["kind"] == "discovery" and not attempt:
attempt = max((attempt for _, attempt in archived), default=0) + 1
return [(output.as_posix(), attempt), *archived]
def _saved_result_paths(scan_dir: Path, workers: list[Any]) -> Iterator[tuple[str, str | None]]:
latest_reducer = _latest_successful_reducer(workers)
yield "checkpoint-head.json", None
for directory in ("checkpoint-heads", "checkpoints"):
yield from ((path, None) for path in _checkpoint_paths(scan_dir, directory))
for worker in workers:
if worker["kind"] not in {"dedup", "discovery"}:
continue
try:
outputs = _worker_outputs(scan_dir, worker)
except (TypeError, ValueError):
continue
for directory, _ in outputs:
if worker["kind"] == "discovery":
yield f"{directory}/checkpoint-head.json", worker["kind"]
yield from (
(path, worker["kind"])
for path in _checkpoint_paths(scan_dir, f"{directory}/checkpoint-heads")
)
checkpoint_paths = _checkpoint_paths(scan_dir, f"{directory}/checkpoints")
if worker["kind"] == "discovery" or checkpoint_paths:
yield f"{directory}/result.json", worker["kind"]
yield from ((path, worker["kind"]) for path in checkpoint_paths)
if worker["result_manifest_path"] and (
worker["kind"] == "discovery"
or (latest_reducer is not None and worker["id"] == latest_reducer["id"])
):
try:
yield (
Path(worker["result_manifest_path"]).relative_to(scan_dir).as_posix(),
worker["kind"],
)
except ValueError:
continue
def _checkpoint_head_directory(relative: str) -> Path | None:
path = Path(relative)
if path.name == "checkpoint-head.json":
return path.parent
if path.parent.name == "checkpoint-heads":
return path.parent.parent
return None
def _capture_saved_source(
scan_dir: Path,
relative: str,
scan_id: str,
*,
kind: str | None = None,
snapshot_head: bool = True,
write: bool = True,
) -> dict[str, tuple[str, int]]:
if not snapshot_head or Path(relative).name != "checkpoint-head.json":
_, digest, observed = _read_saved_result(scan_dir, relative, scan_id, kind=kind)
return {relative: (digest, observed)}
head, _, observed = _read_saved_result(scan_dir, relative, scan_id)
observation = {"checkpoint": head["checkpoint"], "observedAtNs": str(observed)}
directory = Path(relative).parent
selected = (directory / "checkpoints" / observation["checkpoint"]).as_posix()
_, selected_digest, selected_time = _read_saved_result(scan_dir, selected, scan_id)
digest = _digest(observation)
snapshot = (directory / "checkpoint-heads" / f"{digest}.json").as_posix()
# Capture the selected file even if the worker created it after directory enumeration.
if write and not (scan_dir / snapshot).exists():
write_scan_local_bytes(scan_dir, snapshot, _encoded(observation))
return {
snapshot: (digest, int(observation["observedAtNs"])),
selected: (selected_digest, selected_time),
}
def _is_source_order_snapshot(relative: str) -> bool:
path = Path(relative)
return path.parent == Path("source-order") and bool(
re.fullmatch(r"[0-9a-f]{64}\.json", path.name)
)
def _read_saved_result(
scan_dir: Path, relative: str, scan_id: str, *, kind: str | None = None
) -> tuple[dict[str, Any], str, int]:
draft, _, metadata = _read_scan_local_json_with_metadata(
scan_dir, relative, "Saved scan checkpoint"
)
directory = _checkpoint_head_directory(relative)
if directory is not None:
checkpoint = draft.get("checkpoint")
if not isinstance(checkpoint, str) or not re.fullmatch(r"[0-9a-f]{64}\.json", checkpoint):
raise ContractError("checkpoint head does not name a saved checkpoint")
_read_saved_result(scan_dir, (directory / "checkpoints" / checkpoint).as_posix(), scan_id)
if Path(relative).name == "checkpoint-head.json":
return draft, _digest([draft, metadata.st_mtime_ns]), metadata.st_mtime_ns
observed = draft.get("observedAtNs")
if not isinstance(observed, str) or not re.fullmatch(r"-?[0-9]+", observed):
raise ContractError("checkpoint head has no observation time")
return draft, _digest(draft), int(observed)
if draft.get("scanId") != scan_id:
raise ContractError("checkpoint belongs to a different scan")
if not _is_source_order_snapshot(relative) and (
not isinstance(draft.get("findings"), list)
or not isinstance(draft.get("coverage", {} if kind == "dedup" else None), dict)
):
raise ContractError("checkpoint has no semantic findings or coverage")
return draft, _digest(draft), metadata.st_mtime_ns
def _frozen_source_times(scan_dir: Path, scan_id: str, sources: dict[str, str]) -> dict[str, int]:
times: dict[str, int] = {}
snapshots = [path for path in sources if _is_source_order_snapshot(path)]
for path in snapshots:
record, digest, _ = _read_saved_result(scan_dir, path, scan_id)
if digest != sources[path] or not isinstance(record.get("sources"), dict):
raise ContractError("saved source ordering changed after the scan stopped")
for relative, observation in record["sources"].items():
if (
not isinstance(observation, dict)
or relative not in sources
or observation.get("digest") != sources[relative]
or not isinstance(observation.get("observedAtNs"), str)
or not re.fullmatch(r"-?[0-9]+", observation["observedAtNs"])
):
raise ContractError("saved source ordering does not match its frozen sources")
observed = int(observation["observedAtNs"])
if relative in times and times[relative] != observed:
raise ContractError("saved source ordering has conflicting observations")
times[relative] = observed
if snapshots and sources.keys() - set(snapshots) - times.keys():
raise ContractError("saved source ordering is incomplete")
return times
def _freeze_source_times(
scan_dir: Path, scan_id: str, sources: dict[str, str], times: dict[str, int]
) -> None:
# Identical result rewrites must not change the order of frozen review evidence.
observations = {
path: {"digest": digest, "observedAtNs": str(times[path])}
for path, digest in sources.items()
if not _is_source_order_snapshot(path)
}
if not observations:
return
record = {"scanId": scan_id, "sources": observations}
digest = _digest(record)
path = f"source-order/{digest}.json"
if not (scan_dir / path).exists():
write_scan_local_bytes(scan_dir, path, _encoded(record))
if _read_saved_result(scan_dir, path, scan_id)[1] != digest:
raise ContractError("saved source ordering does not match its digest")
sources[path] = digest
def _parent_scan_draft(
scan_id: str,
parent_scan: dict[str, Any],
findings: dict[str, Any],
coverage: dict[str, Any],
) -> dict[str, Any]:
parent = {
"scanId": scan_id,
"findings": findings.get("findings"),
"coverage": coverage,
**{
key: parent_scan[key]
for key in ("scope", "threatModel", "complete")
if key in parent_scan
},
}
if not isinstance(parent["findings"], list):
raise ContractError("Saved parent draft has no findings array")
return parent
def _read_saved_parent_result(
scan_dir: Path, scan_id: str
) -> tuple[dict[str, Any], dict[str, Any]]:
manifest = _read_scan_local_json(scan_dir, "scan-manifest.json", "Saved parent manifest")
findings = _read_scan_local_json(scan_dir, "findings.json", "Saved parent findings")
coverage = _read_scan_local_json(scan_dir, "coverage.json", "Saved parent coverage")
parent_scan = manifest.get("scan")
if not isinstance(parent_scan, dict):
raise ContractError("Saved parent manifest has no scan object")
if (parent_scan.get("sealedAt") or parent_scan.get("artifacts")) and (
parent_scan.get("id", scan_id) != scan_id
or findings.get("scanId", scan_id) != scan_id
or coverage.get("scanId", scan_id) != scan_id
):
raise ContractError("Saved parent documents belong to a different scan")
return manifest, _parent_scan_draft(scan_id, parent_scan, findings, coverage)
def _source_digests(value: Any, error: str) -> dict[str, str]:
if not isinstance(value, dict) or not all(
isinstance(relative, str) and isinstance(digest, str) for relative, digest in value.items()
):
raise ContractError(error)
return value
def _retained_source_state(value: Any) -> tuple[dict[str, str], str | None]:
if isinstance(value, dict) and isinstance(value.get("sources"), dict):
sources = _source_digests(
value["sources"], "Saved stopped-scan source digests are malformed."
)
model_source = value.get("threatModelSource")
if not isinstance(model_source, str) or model_source not in sources:
raise ContractError("Saved stopped-scan model source is outside its checkpoint set.")
return sources, model_source
return _source_digests(value, "Saved stopped-scan source digests are malformed."), None
def _encode_retained_sources(sources: dict[str, str], model_source: list[str]) -> str:
state = {"sources": sources, "threatModelSource": model_source[0]} if model_source else sources
return json.dumps(state, sort_keys=True)
def _saved_results_changed(db: Any, connection: Any, scan: Any) -> bool:
try:
scan_dir = db.require_canonical_scan_directory(Path(scan["scan_dir"]))
manifest_path = db.artifact_path(scan_dir, db.ARTIFACTS["manifest"], required=False)
workers = connection.execute(
"SELECT id, kind, status, completed_at, artifact_dir, result_manifest_path "
"FROM deep_scan_workers WHERE scan_id = ?",
(scan["id"],),
).fetchall()
paths = dict(_saved_result_paths(scan_dir, workers))
frozen_sources = scan["retained_source_digests_json"]
def has_saved_source() -> bool:
for path in paths:
try:
_read_saved_result(scan_dir, path, scan["id"], kind=paths[path])
return True
except (ContractError, OSError, ValueError):
continue
return False
if manifest_path is None:
if frozen_sources is not None:
return bool(_retained_source_state(json.loads(frozen_sources))[0])
return has_saved_source()
if scan["seal_manifest_digest"] is None:
try:
_read_saved_parent_result(scan_dir, scan["id"])
return True
except (ContractError, OSError, ValueError):
pass
return has_saved_source()
manifest = _read_scan_local_json(
scan_dir,
manifest_path.relative_to(scan_dir).as_posix(),
"Saved scan manifest",
)
manifest_scan = manifest.get("scan")
if not isinstance(manifest_scan, dict):
return True
published_sources = _source_digests(
manifest_scan.get("preservedSources", {}),
"Published scan source digests are malformed.",
)
current_sources = dict(published_sources)
paths.update({path: None for path in published_sources if _is_source_order_snapshot(path)})
for path in paths:
try:
captured = _capture_saved_source(
scan_dir,
path,
scan["id"],
kind=paths[path],
snapshot_head=path not in published_sources,
write=False,
)
current_sources.update({path: value[0] for path, value in captured.items()})
except (ContractError, OSError, ValueError):
continue
return current_sources != published_sources
except (ContractError, OSError, SystemExit, ValueError):
return False
def _recovery_source_digests(db: Any, connection: Any, scan: Any) -> tuple[dict[str, str], bool]:
scan_dir = db.require_canonical_scan_directory(Path(scan["scan_dir"]))
frozen_sources: dict[str, str] | None = None
include_parent = True
raw_frozen_sources = scan["retained_source_digests_json"]
if raw_frozen_sources is not None:
frozen_sources, _ = _retained_source_state(json.loads(raw_frozen_sources))
include_parent = False
manifest_path = db.artifact_path(scan_dir, db.ARTIFACTS["manifest"], required=False)
if manifest_path is not None:
manifest = _read_scan_local_json(
scan_dir,
manifest_path.relative_to(scan_dir).as_posix(),
"Saved scan manifest",
)
manifest_scan = manifest.get("scan")
if not isinstance(manifest_scan, dict):
raise ContractError("Saved scan manifest has no scan object")
if scan["seal_manifest_digest"] is not None or (
manifest_scan.get("sealedAt") is not None or manifest_scan.get("artifacts") is not None
):
if "preservedSources" in manifest_scan:
published_sources = _source_digests(
manifest_scan["preservedSources"],
"Published scan source digests are malformed.",
)
include_parent = not published_sources
if published_sources:
if frozen_sources is not None and frozen_sources != published_sources:
raise ContractError(
"Stopped scan sources changed after terminal publication."
)
frozen_sources = published_sources
elif frozen_sources is None:
frozen_sources = {}
else:
include_parent = True
workers = connection.execute(
"SELECT id, kind, status, completed_at, artifact_dir, result_manifest_path "
"FROM deep_scan_workers WHERE scan_id = ?",
(scan["id"],),
).fetchall()
paths = dict(_saved_result_paths(scan_dir, workers))
recovery_sources = dict(frozen_sources or {})
source_times = _frozen_source_times(scan_dir, scan["id"], recovery_sources)
for relative, expected_digest in recovery_sources.items():
try:
_, digest, observed = _read_saved_result(
scan_dir, relative, scan["id"], kind=paths.get(relative)
)
except (ContractError, OSError, ValueError) as exc:
raise ContractError("Frozen stopped-scan checkpoint set is incomplete.") from exc
if digest != expected_digest:
raise ContractError("checkpoint changed after the scan stopped")
if not _is_source_order_snapshot(relative):
source_times.setdefault(relative, observed)
for relative in paths.keys() - recovery_sources.keys():
try:
captured = _capture_saved_source(scan_dir, relative, scan["id"], kind=paths[relative])
except (ContractError, OSError, ValueError):
continue
for path, (digest, observed) in captured.items():
if path in recovery_sources and recovery_sources[path] != digest:
raise ContractError("checkpoint changed after the scan stopped")
recovery_sources[path] = digest
source_times.setdefault(path, observed)
_freeze_source_times(scan_dir, scan["id"], recovery_sources, source_times)
return recovery_sources, include_parent
def scan_results_recovery_needed(db: Any, connection: Any, scan: Any) -> bool:
if scan["status"] != "failed" or scan["canceled_at"] is not None:
return False
warnings = json.loads(scan["completion_warnings_json"])
if any(
isinstance(warning, str) and warning.startswith(_PUBLICATION_FOLLOW_UP_WARNING)
for warning in warnings
):
return True
publication_error = connection.execute(
"SELECT publication_error_message FROM deep_scan_runs WHERE scan_id = ?",
(scan["id"],),
).fetchone()
if publication_error is not None and publication_error["publication_error_message"]:
return True
return _saved_results_changed(db, connection, scan)
def _finding_key(finding: dict[str, Any]) -> str:
# Wording and evidence may improve between checkpoints; distinct source locations
# must not collide merely because two workers chose the same semantic identity.
provenance = finding.get("provenance")
identity = (
provenance.get("preservedIdentity", finding.get("identity"))
if isinstance(provenance, dict)
else finding.get("identity")
)
if not isinstance(identity, dict):
normalized = dict(finding)
normalized.pop("identity", None)
_ensure_finding_identity(normalized)
identity = normalized["identity"]
locations = finding.get("locations", [])
if not isinstance(locations, list):
locations = []
return _digest(
[
finding.get("ruleId"),
identity,
sorted(
(
(
location.get("path"),
location.get("startLine"),
location.get("endLine", location.get("startLine")),
)
for location in locations
if isinstance(location, dict)
),
key=_encoded,
),
]
)
def _worker_candidate_key(
worker_id: str, candidate_id: str, finding: dict[str, Any]
) -> tuple[str, str, Any, Any, Any]:
"""Identify one worker-local candidate without merging unrelated locations."""
provenance = finding.get("provenance")
identity = (
provenance.get("preservedIdentity", finding.get("identity"))
if isinstance(provenance, dict)
else finding.get("identity")
)
if not isinstance(identity, dict):
normalized = dict(finding)
_ensure_finding_identity(normalized)
identity = normalized.get("identity")
anchor = identity.get("anchor") if isinstance(identity, dict) else None
instance = identity.get("instance") if isinstance(identity, dict) else None
return worker_id, candidate_id, finding.get("ruleId"), anchor, instance
def _finding_content(finding: dict[str, Any]) -> dict[str, Any]:
"""Return substantive finding content without generated identity or provenance."""
return {
key: value
for key, value in finding.items()
if key not in {"findingId", "occurrenceId", "fingerprints", "identity", "provenance"}
}
def _ensure_finding_identity(finding: Any, *, candidate_only: bool = False) -> None:
if not isinstance(finding, dict) or "identity" in finding:
return
if candidate_only and not finding_candidate_id(finding):
return
extensions = finding.get("extensions")
source = str(
(extensions.get("candidateId") if isinstance(extensions, dict) else None)
or finding.get("title")
or "finding"
)
anchor = re.sub(r"[^a-z0-9._/-]+", "-", source.lower()).strip("._/-") or "finding"
finding["identity"] = {"anchor": anchor}
def _retained_findings(finding: dict[str, Any]) -> Iterator[dict[str, Any]]:
"""Yield canonical and historical findings without trusting candidate IDs."""
pending = [finding]
seen: set[int] = set()
while pending:
current = pending.pop()
marker = id(current)
if marker in seen:
continue
seen.add(marker)
yield current
provenance = current.get("provenance")
if not isinstance(provenance, dict):
continue
previous = provenance.get("previousFindings")
if isinstance(previous, list):
pending.extend(item for item in reversed(previous) if isinstance(item, dict))
sources = provenance.get("sourceFindings")
if isinstance(sources, list):
pending.extend(
source["finding"]
for source in reversed(sources)
if isinstance(source, dict) and isinstance(source.get("finding"), dict)
)
def _deferred_rows(coverage: dict[str, Any]) -> list[Any]:
rows = coverage.get("deferred", [])
return rows if isinstance(rows, list) else []
def _resolved_deferred_rows(draft: dict[str, Any], schema: dict[str, Any]) -> list[dict[str, Any]]:
if draft.get("complete") is False:
return []
coverage = draft["coverage"]
rows = coverage.get("resolvedDeferred", [])
try:
# Invalid closure metadata cannot discard the evidence it names.
_validate_schema_node(rows, schema, "coverage.resolvedDeferred")
_validate_resolved_deferred({**coverage, "deferred": _deferred_rows(coverage)})
except ContractError:
return []
return rows
def _merge_tied_parent_observations(
current: dict[str, Any], previous: dict[str, Any]
) -> dict[str, Any]:
merged = copy.deepcopy(current)
for finding in previous["findings"]:
if finding not in merged["findings"]:
merged["findings"].append(copy.deepcopy(finding))
coverage = merged["coverage"]
for field in ("surfaces", "explicitExclusions", "deferred", "openQuestions"):
rows = previous["coverage"].get(field, [])
output = coverage.setdefault(field, [])
if isinstance(rows, list) and isinstance(output, list):
for row in rows:
if row not in output:
output.append(copy.deepcopy(row))
pending_ids = {
identity
for row in _deferred_rows(coverage)
if isinstance(row, dict)
for identity in (row.get("id"), row.get("candidateId"))
if isinstance(identity, str)
}
closure_schema = _read_json(
Path(__file__).resolve().parent.parent / "schemas" / "coverage.schema.json"
)["properties"]["resolvedDeferred"]
closures = {}
for observed in (current, previous):
for row in _resolved_deferred_rows(observed, closure_schema):
if row["id"] not in pending_ids:
closures.setdefault(row["id"], copy.deepcopy(row))
coverage.pop("resolvedDeferred", None)
if closures:
coverage["resolvedDeferred"] = list(closures.values())
if current.get("complete") is False or previous.get("complete") is False:
merged["complete"] = False
if (
coverage.get("deferred")
or merged.get("complete") is False
or (
isinstance(coverage.get("surfaces"), list)
and any(
isinstance(row, dict) and row.get("disposition") == "needs_follow_up"
for row in coverage["surfaces"]
)
)
or previous["coverage"].get("completeness") == "partial"
):
coverage["completeness"] = "partial"
return merged
def _saved_coverage_id(item: dict[str, Any]) -> str:
return item.get("candidateId") or f"saved-{_digest(item)[:16]}"
def _deferred_candidate_id(
row: dict[str, Any],
owner: str | None,
ambiguous_deferred: set[tuple[str | None, str]],
) -> str | None:
identity = row.get("candidateId") or row.get("id")
if not isinstance(identity, str):
return None
if (owner, identity) in ambiguous_deferred and not any(
key in row for key in ("candidateId", "candidate", "finding")
):
return None
return identity
def _generic_surface_updates(
sources: list[tuple[str, dict[str, Any], str | None]],
source_order: dict[str, tuple[int, int]],
closed_deferred: dict[tuple[str | None, str], tuple[tuple[int, int], dict[str, Any], str]],
active_deferred: dict[tuple[str | None, str], tuple[tuple[int, int], dict[str, Any], str]],
resolved_candidates: dict[tuple[str | None, str], str],
reopened_generic: set[tuple[str | None, str]],
surface_schema: dict[str, Any],
deferred_rows: dict[str, list[Any]],
ambiguous_deferred: set[tuple[str | None, str]],
) -> tuple[set[int], list[dict[str, Any]]]:
if not closed_deferred and not reopened_generic:
return set(), []
def linked(row: dict[str, Any], identity: str | None) -> bool:
surface_ids = row.get("surfaceIds", [])
return identity is not None and (
row.get("id") == identity or (isinstance(surface_ids, list) and identity in surface_ids)
)
saved_surfaces: dict[tuple[str | None, str], list[tuple[str, dict[str, Any]]]] = {}
for relative, draft, owner in sources:
rows = draft["coverage"].get("surfaces", [])
for row in rows if isinstance(rows, list) else []:
if isinstance(row, dict) and isinstance(row.get("id"), str):
saved_surfaces.setdefault((owner, row["id"]), []).append((relative, row))
replaced: set[int] = set()
updates: list[dict[str, Any]] = []
for relative, draft, owner in sources:
closed_ids = {
identity
for (saved_owner, identity), (_, _, source) in closed_deferred.items()
if saved_owner == owner and source == relative
}
reopened_ids = {
identity
for (saved_owner, identity), (_, _, source) in active_deferred.items()
if saved_owner == owner and source == relative and (owner, identity) in reopened_generic
}
if not closed_ids and not reopened_ids:
continue
current = draft["coverage"].get("surfaces", [])
for surface in current if isinstance(current, list) else []:
if not isinstance(surface, dict):
continue
reopening = surface.get("disposition") == "needs_follow_up"
work_ids = reopened_ids if reopening else closed_ids
if not work_ids:
continue
identity = surface.get("id")
if not isinstance(identity, str):
continue
matches = saved_surfaces[(owner, identity)]
# Older writers could assign one ID to distinct surfaces in a draft.
# A closure cannot identify which of those observations it replaces.
by_source = dict(matches)
if any(row != by_source[saved_path] for saved_path, row in matches):
continue
if any(
"candidateId" in row or "candidate" in row or "finding" in row for _, row in matches
):
continue
if any(
row is not surface
and source_order[saved_path] >= source_order[relative]
and row.get("disposition") != surface.get("disposition")
for saved_path, row in matches
):
continue
if not reopening and any(
saved_owner == owner
and (
not isinstance(row.get("id") or row.get("candidateId"), str)
or (
isinstance(row.get("id"), str)
and (owner, row["id"]) in ambiguous_deferred
and not any(key in row for key in ("candidateId", "candidate", "finding"))
)
)
and linked(row, identity)
for saved_path, _, saved_owner in sources
for row in deferred_rows[saved_path]
if isinstance(row, dict)
):
continue
if not reopening and any(
saved_owner == owner
and (owner, deferred_id) not in closed_deferred
and (
(candidate_id := _deferred_candidate_id(row, owner, ambiguous_deferred)) is None
or (owner, candidate_id) not in resolved_candidates
)
and linked(row, identity)
for (saved_owner, deferred_id), (_, row, _) in active_deferred.items()
):
continue
# An accepted checkpoint can update a saved surface by ID without
# optional surfaceIds links on its generic task.
latest_surface = max(matches, key=lambda match: source_order[match[0]])[1]
update = copy.deepcopy(latest_surface)
refs = update.setdefault("receiptRefs", [])
if isinstance(refs, list):
for _, row in matches:
previous_refs = row.get("receiptRefs", [])
for ref in previous_refs if isinstance(previous_refs, list) else []:
if ref not in refs:
refs.append(ref)
try:
_validate_schema_node(update, surface_schema, "coverage.surfaces")
except ContractError:
continue
replaced.update(
id(row)
for _, row in matches
if reopening
or row.get("disposition") in {"needs_follow_up", surface.get("disposition")}
)
if update not in updates:
updates.append(update)
return replaced, updates
@contextmanager
def preserve_parent_head_on_error(scan_dir: Path) -> Iterator[None]:
"""Keep rejected completion attempts from becoming accepted parent observations."""
head_path = scan_dir / "checkpoint-head.json"
previous = None
try:
head_path.lstat()
except FileNotFoundError:
pass
else:
descriptor = open_scan_local_file_descriptor(
scan_dir, "checkpoint-head.json", "Saved parent checkpoint head"
)
with os.fdopen(descriptor, "rb") as handle:
metadata = os.fstat(handle.fileno())
previous = (handle.read(), metadata)
directories = ("checkpoints", "checkpoint-heads")
previous_files = {
relative for directory in directories for relative in _checkpoint_paths(scan_dir, directory)
}
try:
yield
except ContractError:
if previous is None:
_remove_scan_local_file_if_exists(scan_dir, "checkpoint-head.json")
else:
payload, metadata = previous
write_scan_local_bytes(scan_dir, "checkpoint-head.json", payload)
os.utime(head_path, ns=(metadata.st_atime_ns, metadata.st_mtime_ns))
current_files = {
relative
for directory in directories
for relative in _checkpoint_paths(scan_dir, directory)
}
for relative in current_files - previous_files:
_remove_scan_local_file_if_exists(scan_dir, relative)
raise
def merge_saved_results(
scan_dir: Path,
scan_id: str,
binding: dict[str, Any],
workers: list[Any],
warnings: list[str],
*,
stopped: bool,
reason: str,
frozen_source_digests: dict[str, str] | None = None,
allow_frozen_legacy_parent: bool = False,
frozen_model_source: str | None = None,
selected_model_source: list[str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]] | None:
"""Read only bound parent/worker files; return an unsealed loss-preserving union."""
initial_warnings = set(warnings)
try:
source_times = _frozen_source_times(scan_dir, scan_id, frozen_source_digests or {})
except (ContractError, OSError, ValueError) as exc:
raise ContractError("Frozen stopped-scan checkpoint set is incomplete.") from exc
parent: dict[str, Any] | None = None
parent_manifest: dict[str, Any] | None = None
parent_is_canonical = False
parent_modified = 0
if frozen_source_digests is None or allow_frozen_legacy_parent:
try:
parent_manifest, parent = _read_saved_parent_result(scan_dir, scan_id)
# Without an accepted head, file-authored coverage is a full replacement.
parent_modified = (scan_dir / "coverage.json").lstat().st_mtime_ns
parent_is_canonical = True
except (ContractError, OSError, ValueError) as exc:
if not stopped:
raise
if (scan_dir / "scan-manifest.json").exists():
warnings.append(f"Could not read the saved parent draft: {exc}")
parent_manifest = None
parent = None
if parent_manifest is not None and parent is not None:
parent_scan = parent_manifest["scan"]
try:
previous_head, _, head_modified = _read_saved_result(
scan_dir, "checkpoint-head.json", scan_id
)
except (ContractError, OSError, ValueError):
head_modified = None
if head_modified is not None:
# A partial tool publication must not outrank its accepted head.
parent_modified = min(
(scan_dir / name).lstat().st_mtime_ns
for name in ("findings.json", "coverage.json", "scan-manifest.json")
)
if not parent_scan.get("sealedAt") or allow_frozen_legacy_parent:
head_path = scan_dir / "checkpoint-head.json"
tied_observations = False
if head_modified == parent_modified:
previous_parent, _, _ = _read_saved_result(
scan_dir, f"checkpoints/{previous_head['checkpoint']}", scan_id
)
if previous_parent != parent:
# Tied observations cannot decide which pending work came last.
parent = _merge_tied_parent_observations(parent, previous_parent)
parent_is_canonical = False
tied_observations = True
payload = _encoded(parent)
parent_digest = hashlib.sha256(payload).hexdigest()
parent_checkpoint = f"checkpoints/{parent_digest}.json"
checkpoint_path = scan_dir / parent_checkpoint
if not checkpoint_path.exists():
write_scan_local_bytes(scan_dir, parent_checkpoint, payload)
# A recovery copy must not appear newer than the review it copies.
os.utime(checkpoint_path, ns=(parent_modified, parent_modified))
if head_modified is None or head_modified < parent_modified or tied_observations:
write_scan_local_bytes(
scan_dir,
"checkpoint-head.json",
_encoded({"checkpoint": checkpoint_path.name}),
)
os.utime(head_path, ns=(parent_modified, parent_modified))
if frozen_source_digests is not None:
captured = _capture_saved_source(scan_dir, "checkpoint-head.json", scan_id)
frozen_source_digests = {
**frozen_source_digests,
parent_checkpoint: parent_digest,
**{path: value[0] for path, value in captured.items()},
}
sources: list[tuple[str, dict[str, Any], str | None]] = []
parent_preserved_sources: dict[str, str] = {}
source_digests: dict[str, str] = {}
source_order: dict[str, tuple[int, int]] = {}
worker_attempts: dict[str, tuple[str, int]] = {}
saved_heads: dict[str, str] = {}
if parent_manifest:
recorded = parent_manifest["scan"].get("preservedSources", {})
if isinstance(recorded, dict):
parent_preserved_sources = recorded
source_digests.update(parent_preserved_sources)
if frozen_source_digests is None:
source_times.update(_frozen_source_times(scan_dir, scan_id, recorded))
paths: dict[str, str | None] = {}
reducer_paths: set[str] = set()
current_results: set[str] = set()
reducer_outputs: list[tuple[Any, str, list[str], int]] = []
reducer = _latest_successful_reducer(workers)
latest_reducer: str | None = None
if reducer is not None:
try:
latest_reducer = Path(reducer["result_manifest_path"]).relative_to(scan_dir).as_posix()
paths[latest_reducer] = None
reducer_paths.add(latest_reducer)
except ValueError:
warnings.append("Skipped a reducer result outside the scan directory.")
def checkpoints(directory: str, worker_id: str | None, attempt: int = 0) -> None:
if worker_id is not None:
root = Path(directory).parent.as_posix()
worker_attempts[root] = (worker_id, attempt)
paths[f"{root}/checkpoint-head.json"] = worker_id
head_snapshots = (Path(directory).parent / "checkpoint-heads").as_posix()
for saved_directory in (directory, head_snapshots):
paths.update(dict.fromkeys(_checkpoint_paths(scan_dir, saved_directory), worker_id))
paths["checkpoint-head.json"] = None
checkpoints("checkpoints", None)
for worker in workers:
try:
outputs = _worker_outputs(scan_dir, worker)
except (TypeError, ValueError):
warnings.append("Skipped a worker checkpoint outside the scan directory.")
continue
if worker["kind"] == "dedup":
for directory, attempt in outputs:
checkpoint_paths = _checkpoint_paths(scan_dir, f"{directory}/checkpoints")
if not checkpoint_paths:
continue
result_path = f"{directory}/result.json"
retained_paths = [result_path, *checkpoint_paths]
paths.update(dict.fromkeys(retained_paths))
reducer_paths.update(retained_paths)
reducer_outputs.append((worker, result_path, checkpoint_paths, attempt))
continue
if worker["kind"] != "discovery":
continue
output, attempt = outputs[0]
paths[f"{output}/result.json"] = worker["id"]
current_results.add(f"{output}/result.json")
for archived, archived_attempt in outputs[1:]:
paths[f"{archived}/result.json"] = worker["id"]
checkpoints(f"{archived}/checkpoints", worker["id"], archived_attempt)
checkpoints(f"{output}/checkpoints", worker["id"], attempt)
if worker["result_manifest_path"]:
try:
current_path = Path(worker["result_manifest_path"]).relative_to(scan_dir).as_posix()
paths[current_path] = worker["id"]
current_results.add(current_path)
except ValueError:
warnings.append("Skipped a worker result outside the scan directory.")
if frozen_source_digests is None:
for relative, worker_id in list(paths.items()):
if Path(relative).name != "checkpoint-head.json":
continue
del paths[relative]
try:
captured = _capture_saved_source(scan_dir, relative, scan_id)
paths.update({path: worker_id for path in captured})
except (ContractError, OSError, ValueError) as exc:
if (scan_dir / relative).exists():
warnings.append(f"Preserved unreadable checkpoint {relative}: {exc}")
if frozen_source_digests is not None:
source_digests.update(
{
path: digest
for path, digest in frozen_source_digests.items()
if _is_source_order_snapshot(path)
}
)
paths = {
relative: worker_id
for relative, worker_id in paths.items()
if relative in frozen_source_digests
}
current_results.intersection_update(frozen_source_digests)
if latest_reducer not in frozen_source_digests:
latest_reducer = None
for relative, worker_id in paths.items():
try:
draft, digest, observed = _read_saved_result(
scan_dir, relative, scan_id, kind="dedup" if relative in reducer_paths else None
)
if frozen_source_digests is not None and frozen_source_digests[relative] != digest:
raise ContractError("checkpoint changed after the scan stopped")
source_digests[relative] = digest
source_times.setdefault(relative, observed)
source_order[relative] = (0, source_times[relative])
if _checkpoint_head_directory(relative) is not None:
saved_heads[relative] = draft["checkpoint"]
continue
# Recovery expects coverage, but reducer results only contain findings
# and context. Add an empty value after hashing the original result.
sources.append((relative, {"coverage": {}, **draft}, worker_id))
except (ContractError, OSError, ValueError) as exc:
if (scan_dir / relative).exists():
warnings.append(f"Preserved unreadable checkpoint {relative}: {exc}")
if frozen_source_digests is not None:
if frozen_source_digests.keys() - source_digests.keys():
raise ContractError("Frozen stopped-scan checkpoint set is incomplete.")
# Frozen observations retain checkpoint selection even if a worker moves its head.
headed_workers = {
worker_attempts[_checkpoint_head_directory(head).as_posix()][0]
for head in saved_heads
if _checkpoint_head_directory(head) != Path(".")
}
for relative, _, worker_id in sources:
if worker_id not in headed_workers:
continue
directory = Path(relative).parent
if directory.name == "checkpoints":
directory = directory.parent
_, attempt = worker_attempts.get(directory.as_posix(), (worker_id, 0))
source_order[relative] = (attempt, source_order[relative][1])
selected_observations: dict[str, tuple[int, int]] = {}
parent_heads: list[tuple[int, str]] = []
for head, checkpoint in saved_heads.items():
directory = _checkpoint_head_directory(head)
selected = (directory / "checkpoints" / checkpoint).as_posix()
if selected not in source_order:
raise ContractError("Checkpoint head is outside the saved source set.")
observed = source_order[head][1]
if directory == Path("."):
order = (0, max(source_order[selected][1], observed))
parent_heads.append((observed, selected))
else:
_, attempt = worker_attempts[directory.as_posix()]
order = (attempt, observed)
selected_observations[selected] = max(selected_observations.get(selected, order), order)
source_order.update(selected_observations)
drafts_by_path = {relative: draft for relative, draft, _ in sources}
latest_reducer_key = (
(reducer["completed_at"] or "", reducer["id"], int(reducer["attempt"] or 0))
if reducer is not None and latest_reducer in drafts_by_path
else None
)
if latest_reducer_key is None:
latest_reducer = None
for worker, result_path, checkpoint_paths, attempt in reducer_outputs:
result = drafts_by_path.get(result_path)
if result is None or not any(
drafts_by_path.get(checkpoint_path) == result for checkpoint_path in checkpoint_paths
):
continue
current_results.add(result_path)
candidate_key = (worker["completed_at"] or "", worker["id"], attempt)
if latest_reducer_key is None or candidate_key > latest_reducer_key:
latest_reducer_key = candidate_key
latest_reducer = result_path
if parent_heads:
latest_observation = max(observed for observed, _ in parent_heads)
for observed, parent_path in parent_heads:
if observed != latest_observation:
continue
draft = drafts_by_path[parent_path]
modified = source_order[parent_path][1]
if parent is None or modified > parent_modified:
parent = draft
parent_modified = modified
parent_is_canonical = False
elif modified == parent_modified and draft != parent:
parent = _merge_tied_parent_observations(parent, draft)
parent_is_canonical = False
if parent is None and latest_reducer is not None:
parent = drafts_by_path[latest_reducer]
all_sources = ([("parent", parent, None)] if parent else []) + sources
# Older checkpoints can omit IDs already assigned in their published output.
for field in ("deferred", "surfaces"):
named_rows: dict[str | None, list[dict[str, Any]]] = {}
for _, draft, owner in all_sources:
rows = draft["coverage"].get(field, [])
named_rows.setdefault(owner, []).extend(
row
for row in (rows if isinstance(rows, list) else [])
if isinstance(row, dict) and isinstance(row.get("id"), str)
)
for _, draft, owner in all_sources:
rows = draft["coverage"].get(field, [])
if not isinstance(rows, list):
continue
reserved = {
row["id"]
for row in rows
if isinstance(row, dict) and isinstance(row.get("id"), str)
}
for row in rows:
if not isinstance(row, dict) or "id" in row:
continue
if field == "deferred" and any(
key in row for key in ("candidateId", "candidate", "finding")
):
continue
content = {"receiptRefs": [], **row} if field == "surfaces" else row
identity = next(
(
named["id"]
for named in named_rows[owner]
if named["id"] not in reserved
and {
**({"receiptRefs": []} if field == "surfaces" else {}),
**{key: value for key, value in named.items() if key != "id"},
}
== content
),
None,
)
if identity is not None:
row["id"] = identity
reserved.add(identity)
if parent is None and not sources:
return None
if (
parent_manifest
and parent_manifest["scan"].get("sealedAt")
and parent_manifest["scan"].get("status") == binding["status"]
and parent_manifest["scan"].get("preservedSources") == source_digests
and all(warning in initial_warnings for warning in warnings)
):
return None
if (
frozen_source_digests is None
or any(_is_source_order_snapshot(path) for path in source_digests)
or allow_frozen_legacy_parent
):
_freeze_source_times(scan_dir, scan_id, source_digests, source_times)
target_kind = binding["allowedTargetKinds"][0]
if (
target_kind == "git_worktree"
and "snapshotDigest" not in binding["target"]
and "git_revision" in binding["allowedTargetKinds"]
):
target_kind = "git_revision"
target = {"kind": target_kind, **binding["target"]}
if target["kind"] == "git_diff" and "snapshotDigest" not in target:
diff_kind = {"commit": "commit", "branch_diff": "range"}[binding["coverageMode"]]
target["snapshotDigest"] = committed_diff_snapshot_digest(
diff_kind, target["baseRevision"], target["headRevision"]
)
manifest = (
copy.deepcopy(parent_manifest)
if parent_manifest
else {"scan": {"target": target, "scope": binding["scope"]}}
)
for key in ("sealedAt", "artifacts"):
manifest["scan"].pop(key, None)
manifest["scan"]["preservedSources"] = source_digests
# Completion follows the selected parent, including legacy terminal drafts
# that omit the optional marker, rather than an older canonical manifest.
if parent is not None:
manifest["scan"].pop("complete", None)
if "complete" in parent:
manifest["scan"]["complete"] = parent["complete"]
for key in ("scope", "threatModel"):
if isinstance(parent.get(key), dict):
manifest["scan"][key] = copy.deepcopy(parent[key])
if isinstance(manifest["scan"].get("scope"), dict):
manifest["scan"]["scope"].update(copy.deepcopy(binding["scope"]))
if frozen_model_source is not None:
# Publication retries retain the choice made with the frozen source set.
model = drafts_by_path.get(frozen_model_source, {}).get("threatModel")
if not isinstance(model, dict):
raise ContractError("Frozen stopped-scan model source is unavailable.")
manifest["scan"]["threatModel"] = copy.deepcopy(model)
if paths[frozen_model_source] is not None:
manifest["scan"]["threatModel"]["origin"] = "recovered"
if selected_model_source is not None:
selected_model_source[:] = [frozen_model_source]
coverage = (
copy.deepcopy(parent["coverage"])
if parent and parent["coverage"]
else {
"completeness": "partial",
"mode": binding["coverageMode"],
"inventoryStrategy": "diff"
if binding["coverageMode"] in {"commit", "branch_diff", "working_tree"}
else "scoped_path"
if binding["coverageMode"] == "scoped_path"
else "repository",
**binding["scope"],
"surfaces": [],
"explicitExclusions": [],
"deferred": [],
}
)
if isinstance(coverage.get("openQuestions"), list):
coverage["openQuestions"] = [
{"question": item.strip()} if isinstance(item, str) else item
for item in coverage["openQuestions"]
]
canonical_rows = (
{
id(item)
for field in ("surfaces", "explicitExclusions", "deferred")
for item in (coverage.get(field) if isinstance(coverage.get(field), list) else [])
}
if parent_is_canonical
else set()
)
findings: list[dict[str, Any]] = []
finding_positions: dict[str, int] = {}
represented: dict[str, str | None] = {}
represented_candidates: dict[tuple[str, str, Any, Any, Any], str | None] = {}
represented_history: dict[str, set[str]] = {}
represented_candidate_history: dict[tuple[str, str, Any, Any, Any], set[str]] = {}
rejected_history: dict[tuple[str, str], list[dict[str, Any]]] = {}
stopped_parent_seal = bool(
stopped and parent_manifest and parent_manifest["scan"].get("sealedAt")
)
def valid_finding(value: Any) -> bool:
# Use the finalizer's own per-record recovery before a draft can suppress
# an earlier checkpoint. Invalid latest records must not hide valid history.
document = {"scanId": scan_id, "findings": [copy.deepcopy(value)]}
if isinstance(document["findings"][0], dict):
_ensure_finding_identity(document["findings"][0], candidate_only=True)
_recover_unsealed_findings(
{"scan": {"id": scan_id, "target": binding["target"]}},
document,
Path(__file__).resolve().parent.parent / "schemas",
scan_dir,
[],
)
return bool(document["findings"])
source_order["parent"] = (0, parent_modified)
deferred_rows = {
relative: _deferred_rows(draft["coverage"]) for relative, draft, _ in all_sources
}
# A legacy source can contain independent tasks with the same explicit ID.
# Later rewrites cannot make an ID-only closure identify one of those tasks.
ambiguous_deferred: set[tuple[str | None, str]] = set()
for relative, _, owner in all_sources:
by_id: dict[str, dict[str, Any]] = {}
candidate_aliases = {
identity
for row in deferred_rows[relative]
if isinstance(row, dict)
and any(key in row for key in ("candidateId", "candidate", "finding"))
for identity in (row.get("id"), row.get("candidateId"))
if isinstance(identity, str)
}
for row in deferred_rows[relative]:
if not isinstance(row, dict) or not isinstance(identity := row.get("id"), str):
continue
if any(key in row for key in ("candidateId", "candidate", "finding")):
continue
if identity in candidate_aliases or (identity in by_id and row != by_id[identity]):
ambiguous_deferred.add((owner, identity))
by_id[identity] = row
current_drafts = ([("parent", parent, None)] if parent else []) + [
source for source in sources if source[0] in current_results | selected_observations.keys()
]
# Generic closures belong to one logical scan or worker, just like candidates.
# Keep them when recovering a terminal checkpoint without its canonical write.
closed_deferred: dict[tuple[str | None, str], tuple[tuple[int, int], dict[str, Any], str]] = {}
candidate_ids: set[tuple[str | None, str]] = set()
accepted_deferred_orders: dict[tuple[str | None, str], tuple[int, int]] = {}
active_deferred: dict[tuple[str | None, str], tuple[tuple[int, int], dict[str, Any], str]] = {}
coverage_schema = _read_json(
Path(__file__).resolve().parent.parent / "schemas" / "coverage.schema.json"
)["properties"]
for relative, draft, owner in all_sources:
order = source_order[relative]
for item in deferred_rows[relative]:
if not isinstance(item, dict):
continue
if isinstance(item.get("candidateId"), str) or "candidate" in item or "finding" in item:
candidate_ids.update(
(owner, identity)
for identity in (item.get("id"), item.get("candidateId"))
if isinstance(identity, str)
)
identity = item.get("id") or item.get("candidateId")
if isinstance(identity, str):
key = (owner, identity)
previous = active_deferred.get(key)
if previous is None or order > previous[0]:
active_deferred[key] = (order, item, relative)
if (
relative == "parent"
or relative in current_results
or relative in selected_observations
):
try:
# A malformed update cannot discard valid saved evidence.
_validate_schema_node(
item, coverage_schema["deferred"]["items"], "coverage.deferred"
)
except ContractError:
continue
accepted_deferred_orders[key] = max(
accepted_deferred_orders.get(key, order), order
)
for closure in _resolved_deferred_rows(draft, coverage_schema["resolvedDeferred"]):
key = (owner, closure["id"])
previous = closed_deferred.get(key)
if previous is None or order > previous[0]:
# The parent comes first, preserving its reason on equal timestamps.
closed_deferred[key] = (order, closure, relative)
reopened_rows: list[tuple[str | None, dict[str, Any]]] = []
reopened_generic: set[tuple[str | None, str]] = set()
for key, (order, _, _) in list(closed_deferred.items()):
# Equal timestamps cannot distinguish closure from reopened work.
reopened = [
active
for (owner, identity), active in active_deferred.items()
if owner == key[0]
and (identity == key[1] or active[1].get("candidateId") == key[1])
and active[0] >= order
]
if key in candidate_ids or key in ambiguous_deferred or reopened:
del closed_deferred[key]
for _, item, _ in reopened:
reopened_rows.append((key[0], item))
if key not in candidate_ids:
reopened_generic.add(key)
# Retain every distinct ambiguous task even when a terminal result supersedes
# its source. Existing output ID normalization keeps each row distinct.
for relative, _, owner in all_sources:
for row in deferred_rows[relative]:
if (
isinstance(row, dict)
and isinstance(identity := row.get("id"), str)
and (owner, identity) in ambiguous_deferred
and not any(key in row for key in ("candidateId", "candidate", "finding"))
and (owner, row) not in reopened_rows
):
reopened_rows.append((owner, row))
parent_closures = [
closure for (owner, _), (_, closure, _) in closed_deferred.items() if owner is None
]
coverage.pop("resolvedDeferred", None)
if parent_closures:
coverage["resolvedDeferred"] = copy.deepcopy(parent_closures)
if parent:
coverage["deferred"] = [
row
for row in deferred_rows["parent"]
if not isinstance(row, dict)
or not isinstance(row.get("id"), str)
or (None, row["id"]) not in closed_deferred
]
resolved: dict[tuple[str | None, str], str] = {}
ordered_candidates = {
(owner, identity)
for owner, row in reopened_rows
if (identity := _deferred_candidate_id(row, owner, ambiguous_deferred)) is not None
and (owner, identity) in candidate_ids
}
ordered_outcomes: dict[tuple[str | None, str], tuple[tuple[int, int], str]] = {}
outcomes: list[tuple[str, str | None, str, str]] = []
for relative, draft, owner in current_drafts:
for finding in draft["findings"]:
if (
isinstance(finding, dict)
and valid_finding(finding)
and (candidate_id := finding_candidate_id(finding))
):
outcomes.append((relative, owner, candidate_id, "reported"))
for field in ("surfaces", "explicitExclusions"):
items = draft["coverage"].get(field, [])
for item in items if isinstance(items, list) else []:
if (
isinstance(item, dict)
and isinstance(item.get("candidateId"), str)
and item.get("disposition") in {"rejected", "not_applicable"}
):
outcomes.append((relative, owner, item["candidateId"], item["disposition"]))
ordered_candidates.update(
(owner, candidate_id)
for relative, owner, candidate_id, _ in outcomes
if owner is not None and relative in selected_observations
)
# Reopened work and selected checkpoint outcomes follow the saved source order.
for relative, owner, candidate_id, disposition in outcomes:
key = (owner, candidate_id)
if key not in ordered_candidates:
if relative == "parent" or relative in current_results:
resolved.setdefault(key, disposition)
continue
order = source_order[relative]
if any(
saved_owner == owner
and _deferred_candidate_id(row, owner, ambiguous_deferred) == candidate_id
and modified >= order
for (saved_owner, _), (modified, row, _) in active_deferred.items()
):
continue
if key not in ordered_outcomes or order > ordered_outcomes[key][0]:
resolved[key] = disposition
ordered_outcomes[key] = (order, relative)
# Only the current parent may claim that another worker finding was absorbed.
# A superseded checkpoint must not suppress a newer independent result.
if parent:
for finding in parent["findings"]:
if valid_finding(finding):
canonical_key = _finding_key(finding)
for retained in _retained_findings(finding):
retained_key = _finding_key(retained)
if retained is not finding:
represented_history.setdefault(retained_key, set()).add(
_digest(_finding_content(retained))
)
previous_key = represented.get(retained_key)
if retained_key not in represented:
represented[retained_key] = canonical_key
elif previous_key != canonical_key:
# Ambiguous history cannot suppress an independent source.
represented[retained_key] = None
originals = finding["provenance"].get("sourceFindings", [])
for original in originals if isinstance(originals, list) else []:
if isinstance(original, dict) and isinstance(original.get("finding"), dict):
source_id = original.get("id")
candidate_id = finding_candidate_id(original["finding"])
if isinstance(source_id, str) and ":" in source_id and candidate_id:
candidate_key = _worker_candidate_key(
source_id.rsplit(":", 1)[0],
candidate_id,
original["finding"],
)
previous_key = represented_candidates.get(candidate_key)
if candidate_key not in represented_candidates:
represented_candidates[candidate_key] = canonical_key
elif previous_key != canonical_key:
# Candidate ids are only authoritative within one
# logical worker. Multiple canonical owners make
# that worker-local identity ambiguous.
represented_candidates[candidate_key] = None
represented_candidate_history.setdefault(candidate_key, set()).add(
_digest(_finding_content(original["finding"]))
)
replaced_surfaces, surface_updates = _generic_surface_updates(
all_sources,
source_order,
closed_deferred,
active_deferred,
resolved,
reopened_generic,
coverage_schema["surfaces"]["items"],
deferred_rows,
ambiguous_deferred,
)
replaced_rows = {
"surfaces": replaced_surfaces,
"deferred": {
id(row)
for relative, _, owner in all_sources
for row in deferred_rows[relative]
if isinstance(row, dict)
and isinstance(identity := row.get("id"), str)
and (key := (owner, identity)) not in candidate_ids
and key not in ambiguous_deferred
and (updated := accepted_deferred_orders.get(key)) is not None
and updated > source_order[relative]
},
}
for field, replaced in replaced_rows.items():
if parent and isinstance(coverage.get(field), list):
previous = parent["coverage"].get(field, [])
if isinstance(previous, list):
removed_parent = [row for row in previous if id(row) in replaced]
coverage[field] = [row for row in coverage[field] if row not in removed_parent]
if isinstance(coverage.get("surfaces"), list):
for surface in surface_updates:
if surface not in coverage["surfaces"]:
coverage["surfaces"].append(surface)
for surface in coverage["surfaces"]:
if isinstance(surface, dict):
surface.setdefault("receiptRefs", [])
# Reopened work survives a superseded checkpoint, but current candidate
# outcomes still apply. Parent closures cannot remove another worker's row.
for owner, item in reopened_rows:
if (identity := _deferred_candidate_id(item, owner, ambiguous_deferred)) is not None and (
owner,
identity,
) in resolved:
continue
pending = coverage.setdefault("deferred", [])
if isinstance(pending, list) and item not in pending:
pending.append(copy.deepcopy(item))
ambiguous_surface_ids = {
(owner, identity)
for owner, row in reopened_rows
if isinstance(row.get("id"), str)
and (owner, row["id"]) in ambiguous_deferred
and not any(key in row for key in ("candidateId", "candidate", "finding"))
for identity in [
row["id"],
*(row.get("surfaceIds", []) if isinstance(row.get("surfaceIds", []), list) else []),
]
if isinstance(identity, str)
}
for _, draft, owner in all_sources:
surfaces = draft["coverage"].get("surfaces", [])
by_id: dict[str, dict[str, Any]] = {}
for surface in surfaces if isinstance(surfaces, list) else []:
if not isinstance(surface, dict) or not isinstance(identity := surface.get("id"), str):
continue
if identity in by_id and surface != by_id[identity]:
ambiguous_surface_ids.add((owner, identity))
by_id[identity] = surface
for _, draft, owner in all_sources:
surfaces = draft["coverage"].get("surfaces", [])
for surface in surfaces if isinstance(surfaces, list) else []:
if (
isinstance(surface, dict)
and isinstance(surface.get("id"), str)
and (owner, surface["id"]) in ambiguous_surface_ids
and surface.get("disposition") == "needs_follow_up"
and not any(key in surface for key in ("candidateId", "candidate", "finding"))
and isinstance(coverage.get("surfaces"), list)
):
retained_surface = {**surface, "receiptRefs": surface.get("receiptRefs", [])}
if retained_surface not in coverage["surfaces"]:
coverage["surfaces"].append(copy.deepcopy(retained_surface))
selected_terminal_orders: dict[str, tuple[int, int]] = {}
for relative, draft, worker_id in sources:
if (
worker_id is not None
and relative in selected_observations
and draft.get("complete") is not False
):
order = source_order[relative]
selected_terminal_orders[worker_id] = max(
selected_terminal_orders.get(worker_id, order), order
)
terminal_worker_orders: dict[str | None, tuple[int, int]] = {}
for relative, draft, worker_id in sources:
if relative in current_results and draft.get("complete") is not False:
order = source_order[relative]
terminal_worker_orders[worker_id] = max(
terminal_worker_orders.get(worker_id, order), order
)
for relative, draft, worker_id in all_sources:
worker_result_order = terminal_worker_orders.get(worker_id)
selected_coverage_superseded = worker_id in selected_terminal_orders and (
source_order[relative] < selected_terminal_orders[worker_id]
or (
relative in current_results
and source_order[relative] <= selected_terminal_orders[worker_id]
)
)
selected_candidates = {
candidate_id
for (owner, candidate_id), (_, source) in ordered_outcomes.items()
if owner == worker_id and source == relative
}
superseded = (
worker_id is None
and parent is not None
and parent.get("complete") is not False
and relative != "parent"
and source_order[relative] <= (0, parent_modified)
and (not stopped_parent_seal or relative in parent_preserved_sources)
) or (
relative not in current_results
and worker_result_order is not None
and (
relative not in selected_observations
or source_order[relative] < worker_result_order
)
)
# A failed result write can leave pending work outside the accepted result.
accepted_order = (
(0, parent_modified)
if worker_id is None and parent is not None
else worker_result_order
)
retain_pending = (
superseded
and (worker_id in headed_workers or (worker_id is None and parent_heads))
and accepted_order is not None
and source_order[relative] >= accepted_order
and any(isinstance(row, dict) for row in deferred_rows[relative])
)
if (
(relative != "parent" or not parent_is_canonical)
and not superseded
and not selected_coverage_superseded
and (
draft.get("complete") is False
or draft["coverage"].get("completeness") != "complete"
)
and coverage.get("completeness") in {"complete", "unknown"}
):
coverage["completeness"] = "partial"
skip_superseded_findings = (
superseded
and not stopped
and all(valid_finding(finding) for finding in (parent["findings"] if parent else []))
)
if skip_superseded_findings and not selected_candidates and not retain_pending:
continue
if (
not skip_superseded_findings
and "threatModel" not in manifest["scan"]
and isinstance(draft.get("threatModel"), dict)
):
model = draft["threatModel"]
model_path = relative
checkpoint_dir = Path(relative).parent
prefer_worker_head = worker_id is not None and (
checkpoint_dir.name == "checkpoints" or draft.get("complete") is False
)
if worker_id is not None:
if checkpoint_dir.name != "checkpoints":
checkpoint_dir /= "checkpoints"
selected_models = [
path
for path in selected_observations
if Path(path).parent == checkpoint_dir
and isinstance(drafts_by_path[path].get("threatModel"), dict)
]
if selected_models:
head_path = max(selected_models, key=source_order.__getitem__)
current = drafts_by_path[head_path]
# A terminal checkpoint is committed before result.json is replaced.
# Use the admitted observation, including its frozen ordering on retries.
if not prefer_worker_head and current.get("complete") is not False:
prefer_worker_head = source_order[head_path] >= source_order[relative]
if prefer_worker_head:
model = current["threatModel"]
model_path = head_path
manifest["scan"]["threatModel"] = copy.deepcopy(model)
if worker_id is not None:
manifest["scan"]["threatModel"]["origin"] = "recovered"
if selected_model_source is not None and worker_id is not None:
selected_model_source[:] = [model_path]
for value in draft["findings"]:
if skip_superseded_findings and not (
isinstance(value, dict)
and (candidate_id := finding_candidate_id(value)) in selected_candidates
and resolved.get((worker_id, candidate_id)) == "reported"
):
continue
if relative == "parent" and parent_is_canonical:
finding = copy.deepcopy(value)
_ensure_finding_identity(finding, candidate_only=True)
provenance = finding.get("provenance") if isinstance(finding, dict) else None
owner = provenance.get("workerId") if isinstance(provenance, dict) else None
candidate_id = finding_candidate_id(finding) if isinstance(finding, dict) else None
if (
stopped_parent_seal
and isinstance(owner, str)
and candidate_id
and resolved.get((owner, candidate_id)) in {"rejected", "not_applicable"}
):
rejected_history.setdefault((owner, candidate_id), []).append(finding)
continue
if valid_finding(finding):
finding_positions.setdefault(_finding_key(finding), len(findings))
findings.append(finding)
continue
if relative != "parent" and parent and value in parent["findings"]:
continue
if not isinstance(value, dict):
warnings.append(f"Retained malformed finding evidence in {relative}.")
continue
finding = copy.deepcopy(value)
candidate_id = finding_candidate_id(finding)
if relative != "parent" and resolved.get((worker_id, candidate_id)) in {
"rejected",
"not_applicable",
}:
surfaces = coverage.get("surfaces")
for item in surfaces if isinstance(surfaces, list) else []:
if isinstance(item, dict) and item.get("candidateId") == candidate_id:
if not isinstance(item.get("previousFindings"), list):
item["previousFindings"] = []
history = item["previousFindings"]
if not any(
isinstance(previous, dict)
and _finding_key(previous) == _finding_key(finding)
and _finding_content(previous) == _finding_content(finding)
for previous in history
):
history.append(finding)
continue
for key in ("findingId", "occurrenceId", "fingerprints"):
finding.pop(key, None)
locations = finding.get("locations", [])
if not isinstance(locations, list) or not any(
isinstance(location, dict)
and isinstance(location.get("path"), str)
and any(
path_within_scope(location["path"], path)
for path in binding["scope"]["includePaths"]
)
for location in locations
):
warnings.append(f"Skipped out-of-scope finding from {relative}.")
coverage["completeness"] = "partial"
continue
provenance = finding.setdefault("provenance", {"source": "local_plugin"})
if not isinstance(provenance, dict):
findings.append(finding)
continue
if worker_id:
provenance.setdefault("workerId", worker_id)
_ensure_finding_identity(finding)
if not valid_finding(finding):
findings.append(finding)
continue
key = _finding_key(finding)
represented_by_parent = False
if relative != "parent":
if key in represented:
mapped_key = represented[key]
historical_contents = represented_history.get(key, set())
elif worker_id and candidate_id:
candidate_key = _worker_candidate_key(worker_id, candidate_id, finding)
if candidate_key not in represented_candidates:
represented_candidates[candidate_key] = key
mapped_key = represented_candidates[candidate_key]
historical_contents = represented_candidate_history.get(candidate_key, set())
else:
mapped_key = None
historical_contents = set()
if mapped_key is not None:
key = mapped_key
represented_by_parent = _digest(_finding_content(value)) in historical_contents
if key in finding_positions:
retained = findings[finding_positions[key]]
if finding != retained:
if not represented_by_parent and _finding_strength(finding) > _finding_strength(
retained
):
previous = copy.deepcopy(retained)
previous_history = previous["provenance"].pop("previousFindings", [])
retained = finding
findings[finding_positions[key]] = retained
else:
previous = copy.deepcopy(value)
previous_history = previous.get("provenance", {}).pop(
"previousFindings", []
)
retained_provenance = retained["provenance"]
retained_history = retained_provenance.get("previousFindings")
history = (
[item for item in retained_history if isinstance(item, dict)]
if isinstance(retained_history, list)
else []
)
retained_provenance["previousFindings"] = history
for original in [
*(previous_history if isinstance(previous_history, list) else []),
previous,
]:
if not isinstance(original, dict):
continue
source_key = _finding_key(original)
source_content = _finding_content(original)
already_retained = any(
source_key == _finding_key(historical)
and source_content == _finding_content(historical)
for historical in _retained_findings(retained)
)
if (
not already_retained
and original not in history
and original != retained
):
history.append(original)
continue
finding_positions[key] = len(findings)
findings.append(finding)
if superseded and not selected_candidates and not retain_pending:
continue
for field in ("surfaces", "explicitExclusions", "deferred", "openQuestions"):
if superseded and field not in {"surfaces", "explicitExclusions", "deferred"}:
continue
items = draft["coverage"].get(field, [])
if field == "deferred":
items = deferred_rows[relative]
if not isinstance(items, list):
continue
output = coverage.setdefault(field, [])
if not isinstance(output, list):
# Keep malformed canonical collections for the existing finalizer's
# recovery and warnings rather than silently changing its contract.
continue
for item in items:
# A selected terminal checkpoint replaces ordinary coverage, while
# candidate evidence and pending tasks retain their own reconciliation.
if (
selected_coverage_superseded
and field != "deferred"
and not (
isinstance(item, dict)
and any(key in item for key in ("candidateId", "candidate", "finding"))
)
):
continue
# A selected outcome must retain its evidence even if its result write failed.
if superseded and not (
isinstance(item, dict)
and (
(field == "deferred" and retain_pending)
or (
isinstance(item.get("candidateId"), str)
and item["candidateId"] in selected_candidates
and item.get("disposition")
== resolved.get((worker_id, item["candidateId"]))
)
)
):
continue
if id(item) in replaced_rows.get(field, ()):
continue
if (
field == "deferred"
and isinstance(item, dict)
and isinstance(item.get("id"), str)
and (worker_id, item["id"]) in closed_deferred
):
continue
if field == "openQuestions" and isinstance(item, str):
item = {"question": item.strip()}
if (
field == "surfaces"
and isinstance(item, dict)
and item.get("disposition") in {"rejected", "not_applicable"}
and isinstance(item.get("candidateId"), str)
and (history_findings := rejected_history.get((worker_id, item["candidateId"])))
):
item = copy.deepcopy(item)
if not isinstance(item.get("previousFindings"), list):
item["previousFindings"] = []
history = item["previousFindings"]
for finding in history_findings:
if not any(
isinstance(previous, dict)
and _finding_key(previous) == _finding_key(finding)
and _finding_content(previous) == _finding_content(finding)
for previous in history
):
history.append(copy.deepcopy(finding))
if (
isinstance(item, dict)
and isinstance(
identity := _deferred_candidate_id(item, worker_id, ambiguous_deferred)
if field == "deferred"
else item.get("candidateId"),
str,
)
and (worker_id, identity) in resolved
and (field == "deferred" or item.get("disposition") == "needs_follow_up")
):
continue
if field == "surfaces" and isinstance(item, dict):
item = {**item, "receiptRefs": item.get("receiptRefs", [])}
if isinstance(item, dict) and "id" not in item:
semantic_item = dict(item)
if any(
isinstance(existing, dict)
and {key: value for key, value in existing.items() if key != "id"}
== semantic_item
for existing in output
):
continue
if item not in output:
output.append(copy.deepcopy(item))
identities: dict[str, str] = {}
for finding in findings:
if not valid_finding(finding):
continue
identity = finding.get("identity")
if not isinstance(identity, dict):
continue
key = _encoded([finding.get("ruleId"), identity]).decode()
variant = _finding_key(finding)
if key in identities and identities[key] != variant:
finding.setdefault("provenance", {})["preservedIdentity"] = copy.deepcopy(identity)
identity["instance"] = f"{identity.get('instance', 'saved')}-{variant[:16]}"
identities[key] = variant
for field in ("surfaces", "explicitExclusions", "deferred"):
used: set[str] = set()
items = coverage.setdefault(field, [])
for item in items if isinstance(items, list) else []:
if not isinstance(item, dict):
continue
if id(item) in canonical_rows:
if isinstance(item.get("id"), str):
used.add(item["id"])
continue
item.setdefault("id", _saved_coverage_id(item))
if not isinstance(item["id"], str):
# Preserve malformed rows for per-record recovery, including frozen replay.
continue
if item["id"] in used:
item["id"] = f"{item['id']}-{_digest(item)[:16]}"
used.add(item["id"])
if field == "surfaces":
item.setdefault("receiptRefs", [])
if stopped or any(warning not in initial_warnings for warning in warnings):
coverage["completeness"] = "partial"
if stopped:
if not isinstance(coverage.get("deferred"), list):
coverage["deferred"] = []
item = {"id": "scan-stopped", "reason": reason}
if item not in coverage["deferred"]:
coverage["deferred"].append(item)
return manifest, {"findings": findings}, coverage
def coverage_for_comparison(db: Any, scan: Any) -> dict[str, Any]:
if scan["seal_manifest_digest"] is None:
raise SystemExit("Only sealed scans can be compared.")
scan_dir = db.require_canonical_scan_directory(Path(scan["scan_dir"]))
db.require_recorded_manifest_digest(scan, scan_dir)
try:
_, _, manifest, _, coverage, was_sealed, _ = _prepare_scan_finalization(scan_dir)
except ContractError as exc:
raise SystemExit(str(exc)) from exc
if not was_sealed or manifest["scan"]["id"] != scan["id"]:
raise SystemExit("Only sealed scans can be compared.")
return coverage
def _snapshot_published_outputs(scan_dir: Path) -> dict[str, bytes | None]:
snapshots: dict[str, bytes | None] = {}
for relative in _PUBLISHED_OUTPUTS:
descriptor = -1
try:
descriptor = open_scan_local_file_descriptor(
scan_dir, relative, "Published scan output"
)
with os.fdopen(descriptor, "rb") as handle:
descriptor = -1
snapshots[relative] = handle.read()
except ContractError:
path = scan_dir / relative
if path.exists() or path.is_symlink():
if relative == "threatmodel.md":
# This optional projection will not replace an unsafe destination.
continue
raise
snapshots[relative] = None
finally:
if descriptor >= 0:
os.close(descriptor)
return snapshots
def _restore_published_outputs(scan_dir: Path, snapshots: dict[str, bytes | None]) -> None:
for relative, contents in snapshots.items():
if contents is None:
path = scan_dir / relative
if path.exists() or path.is_symlink():
_remove_scan_local_file_if_exists(scan_dir, relative)
else:
write_scan_local_bytes(scan_dir, relative, contents)
def preserve_scan_results_locked(
db: Any,
connection: Any,
scan_id: str,
*,
recovery_source_digests: dict[str, str] | None = None,
include_parent_with_recovery: bool = False,
) -> bool:
"""Publish or verify retained terminal results through the workbench host."""
scan = db.require_scan(connection, scan_id)
if scan["status"] != "failed":
return False
frozen_source_digests: dict[str, str] | None = None
model_source: list[str] = []
saved_model_source: str | None = None
raw_frozen_sources = scan["retained_source_digests_json"]
if raw_frozen_sources is not None:
frozen_source_digests, saved_model_source = _retained_source_state(
json.loads(raw_frozen_sources)
)
if saved_model_source is not None:
model_source.append(saved_model_source)
if recovery_source_digests is not None:
frozen_source_digests = recovery_source_digests
scan_dir = db.require_canonical_scan_directory(Path(scan["scan_dir"]))
deep_run = connection.execute(
"SELECT status FROM deep_scan_runs WHERE scan_id = ?", (scan_id,)
).fetchone()
outcome = (
"canceled"
if scan["canceled_at"]
else "interrupted"
if deep_run and deep_run["status"] == "interrupted"
else "failed"
)
stored_warnings = json.loads(scan["completion_warnings_json"])
publication_follow_up_warnings = [
warning
for warning in stored_warnings
if isinstance(warning, str) and warning.startswith(_PUBLICATION_FOLLOW_UP_WARNING)
]
warnings = [
warning for warning in stored_warnings if warning not in publication_follow_up_warnings
]
def record_publication(manifest: dict[str, Any], findings: dict[str, Any]) -> None:
retained_sources = _source_digests(
manifest.get("scan", {}).get("preservedSources"),
"Stopped scan source digests could not be frozen.",
)
digest = db.published_manifest_digest(scan_dir, manifest)
timestamp = db.now()
with connection:
for kind, filename in db.ARTIFACTS.items():
path = db.artifact_path(scan_dir, filename, required=True)
connection.execute(
"INSERT OR REPLACE INTO scan_artifacts "
"(scan_id, kind, path, created_at) VALUES (?, ?, ?, ?)",
(scan_id, kind, str(path), scan["completed_at"]),
)
# Delete only vanished occurrences; stable IDs retain triage and remediation.
existing_ids = {
row["id"]
for row in connection.execute(
"SELECT id FROM finding_occurrences WHERE scan_id = ?", (scan_id,)
)
}
retained_ids = {finding["occurrenceId"] for finding in findings["findings"]}
connection.executemany(
"DELETE FROM finding_occurrences WHERE id = ? AND scan_id = ?",
((occurrence_id, scan_id) for occurrence_id in existing_ids - retained_ids),
)
db.index_findings(connection, scan_id, findings, scan["completed_at"])
connection.execute(
"UPDATE scans SET seal_manifest_digest = ?, retained_source_digests_json = ?, "
"completion_warnings_json = ?, "
"updated_at = ? WHERE id = ? AND status = 'failed'",
(
digest,
_encode_retained_sources(retained_sources, model_source),
json.dumps(list(dict.fromkeys(warnings))),
timestamp,
scan_id,
),
)
connection.execute(
"UPDATE scan_progress SET reportable_findings_count = ?, updated_at = ? "
"WHERE scan_id = ?",
(len(findings["findings"]), timestamp, scan_id),
)
existing_path = db.artifact_path(scan_dir, db.ARTIFACTS["manifest"], required=False)
existing_scan = db.read_json_object(existing_path).get("scan", {}) if existing_path else {}
existing = None
if scan["seal_manifest_digest"] is not None or (
isinstance(existing_scan, dict)
and (
existing_scan.get("sealedAt") is not None or existing_scan.get("artifacts") is not None
)
):
db.require_recorded_manifest_digest(scan, scan_dir)
existing, existing_findings, _ = finalize_scan(
scan_dir,
expected_coverage_mode=db.expected_coverage_mode(scan),
projection_warnings=warnings,
)
db.verify_manifest_binding(scan, existing)
if existing_scan.get("status") == outcome:
existing_sources = existing_scan.get("preservedSources")
if frozen_source_digests is None:
frozen_source_digests = _source_digests(
existing_sources, "Stopped scan source digests could not be frozen."
)
if existing_sources == frozen_source_digests:
if (
raw_frozen_sources is not None
and scan["seal_manifest_digest"] is not None
and not publication_follow_up_warnings
and warnings == stored_warnings
):
return True
record_publication(existing, existing_findings)
return True
if recovery_source_digests is None:
raise ContractError("Stopped scan sources changed after terminal publication.")
binding = {
**db.workbench_completion_binding(scan, scan["completed_at"], existing),
"status": outcome,
}
if recovery_source_digests is not None:
model_source.clear()
documents = merge_saved_results(
scan_dir,
scan_id,
binding,
connection.execute(
"SELECT * FROM deep_scan_workers WHERE scan_id = ? ORDER BY created_at, id",
(scan_id,),
).fetchall(),
warnings,
stopped=True,
reason=(
f"Scan {outcome}; saved findings and pending review were preserved. "
f"{scan['failure_message'] or ''}"
).strip(),
frozen_source_digests=frozen_source_digests,
frozen_model_source=model_source[0] if model_source else None,
selected_model_source=model_source,
allow_frozen_legacy_parent=(
include_parent_with_recovery
or (
recovery_source_digests is None
and frozen_source_digests == {}
and isinstance(existing_scan, dict)
and existing_scan.get("sealedAt") is not None
and "preservedSources" not in existing_scan
)
),
)
if documents is None:
unpublished_warnings = list(dict.fromkeys([*warnings, *publication_follow_up_warnings]))
if unpublished_warnings != stored_warnings:
with connection:
connection.execute(
"UPDATE scans SET completion_warnings_json = ?, updated_at = ? "
"WHERE id = ? AND status = 'failed'",
(json.dumps(unpublished_warnings), db.now(), scan_id),
)
return False
if frozen_source_digests is None or (
recovery_source_digests is None and saved_model_source is None and model_source
):
retained_sources = _source_digests(
documents[0].get("scan", {}).get("preservedSources"),
"Stopped scan source digests could not be frozen.",
)
with connection:
connection.execute(
"UPDATE scans SET retained_source_digests_json = ? "
"WHERE id = ? AND retained_source_digests_json IS ?",
(
_encode_retained_sources(retained_sources, model_source),
scan_id,
raw_frozen_sources,
),
)
prepared = _prepare_scan_finalization(
scan_dir,
expected_coverage_mode=db.expected_coverage_mode(scan),
completion_binding=binding,
completion_warnings=warnings,
draft_documents=documents,
)
snapshots = _snapshot_published_outputs(scan_dir)
try:
manifest, findings, _ = _write_prepared_scan_finalization(
prepared, projection_warnings=warnings
)
db.verify_manifest_binding(scan, manifest)
record_publication(manifest, findings)
except BaseException:
_restore_published_outputs(scan_dir, snapshots)
raise
return True
def recover_scan_results(db: Any, connection: Any, args: Any) -> dict[str, Any]:
scan_id = db.require_uuid(args.scan_id, "scan-id")
with db.scan_completion_lock(scan_id):
scan = db.require_scan(connection, scan_id)
if scan["status"] != "failed":
raise SystemExit("Only a stopped scan can recover terminal results.")
if scan["canceled_at"] is not None:
raise SystemExit("Canceled scans cannot recover terminal results.")
recovery_source_digests, include_parent = _recovery_source_digests(db, connection, scan)
if not preserve_scan_results_locked(
db,
connection,
scan_id,
recovery_source_digests=recovery_source_digests,
include_parent_with_recovery=include_parent,
):
raise SystemExit("No saved stopped-scan results were available to recover.")
db.deep_scan.clear_deep_scan_publication_failure(connection, scan_id)
return db.scan_context(connection, scan_id)
def preserve_scan_results(db: Any, connection: Any, args: Any) -> dict[str, Any]:
scan_id = db.require_uuid(args.scan_id, "scan-id")
with db.scan_completion_lock(scan_id):
scan = db.require_scan(connection, scan_id)
if scan["status"] != "failed":
raise SystemExit("Only a stopped scan can preserve terminal results.")
workspace = db.require_workspace(connection, scan["workspace_id"])
owner = (
scan["continuation_thread_id"]
or scan["deep_scan_owner_thread_id"]
or workspace["thread_id"]
)
if args.thread_id is not None and args.thread_id != owner:
raise SystemExit("Saved results can only be published from the owning Codex thread.")
if args.coordinator_generation is not None:
if args.thread_id is None:
raise SystemExit("A coordinator result refresh requires its owning thread.")
db.deep_scan.require_current_coordinator(
db.deep_scan.require_deep_scan_run(connection, scan_id), args
)
else:
db.handoff.require_current_continuation(
scan,
args.claim_token,
error_message="Saved results are owned by another continuation.",
)
published = preserve_scan_results_locked(db, connection, scan_id)
if not published and scan["canceled_at"] is not None:
raise SystemExit("Saved scan results could not be published or verified.")
if published:
db.deep_scan.clear_deep_scan_publication_failure(connection, scan_id)
return db.scan_context(connection, scan_id)
def read_or_save_artifact(args: Any) -> dict[str, Any]:
"""Read or publish supplemental bytes through verified filesystem handles."""
root = Path(args.artifact_root)
if args.command == "read-artifact":
descriptor = open_scan_local_file_descriptor(root, args.artifact_path, "Saved artifact")
with os.fdopen(descriptor, "rb") as source:
return {"content": base64.b64encode(source.read()).decode("ascii")}
write_scan_local_bytes(root, args.artifact_path, sys.stdin.buffer.read())
return {"path": str(root / args.artifact_path)}
def save_scan_artifact(db: Any, connection: Any, args: Any) -> dict[str, Any]:
"""Publish supplemental bytes under the same lock as finalization and recovery."""
scan_id = db.require_uuid(args.scan_id, "scan-id")
with db.scan_completion_lock(scan_id):
scan = db.require_scan(connection, scan_id)
db.handoff.require_current_continuation(
scan,
args.claim_token,
error_message="Scan artifacts are owned by another continuation.",
)
if scan["status"] != "running" or scan["seal_manifest_digest"] is not None:
raise SystemExit("The scan stopped; its artifacts cannot be modified.")
scan_dir = db.require_canonical_scan_directory(Path(scan["scan_dir"]))
manifest_path = db.artifact_path(scan_dir, "scan-manifest.json", required=False)
if manifest_path is not None:
manifest = db.read_json_object(manifest_path).get("scan", {})
if manifest.get("sealedAt") is not None or manifest.get("artifacts") is not None:
raise SystemExit("The scan is sealed; its artifacts cannot be modified.")
output = args.artifact_path
key = output.lower()
if not (
key.startswith(("artifacts/", "findings/", "hardening/"))
or key == "report_validation.md"
) or any(
key == reserved or key.startswith(reserved + "/")
for reserved in _RESERVED_ARTIFACT_PATHS
):
raise SystemExit("Use the typed scan tools for canonical artifacts and checkpoints.")
write_scan_local_bytes(scan_dir, output, sys.stdin.buffer.read())
return {"scanId": scan_id, "path": str(scan_dir / output)}
def write_scan_draft(db: Any, connection: Any, args: Any) -> dict[str, Any]:
scan_id = db.require_uuid(args.scan_id, "scan-id")
with db.scan_completion_lock(scan_id):
scan = db.require_scan(connection, scan_id)
db.handoff.require_current_continuation(
scan, args.claim_token, error_message="Scan draft is owned by another continuation."
)
if scan["status"] != "running" or scan["seal_manifest_digest"] is not None:
raise SystemExit(
"The scan stopped; its saved checkpoint was retained without replacing sealed results."
)
scan_dir = db.require_canonical_scan_directory(Path(scan["scan_dir"]))
if (
args.expected_draft_digest is not None
and args.expected_draft_digest != _scan_draft_digest(scan_dir)
):
raise SystemExit(
"scan_draft_conflict: canonical scan results changed; reconcile the saved checkpoint again."
)
try:
relative = Path(args.draft_path).relative_to(scan_dir).as_posix()
except ValueError as exc:
raise SystemExit(
"Scan draft must be inside the registered scan drafts directory."
) from exc
if not re.fullmatch(r"drafts/[0-9a-fA-F-]+\.json", relative):
raise SystemExit("Scan draft must be inside the registered scan drafts directory.")
draft = _read_scan_local_json(scan_dir, relative, "Staged scan draft")
manifest, findings, coverage = draft["manifest"], draft["findings"], draft["coverage"]
binding = db.workbench_completion_binding(scan, db.now())
# Save scan IDs without sealing the draft.
_populate_unsealed_manifest_envelope(manifest, manifest["scan"], binding)
_populate_unsealed_artifact_envelope(manifest, findings, coverage, binding)
_validate_completion_binding(manifest, findings, coverage, binding)
if args.checkpoint_path is not None:
try:
checkpoint_relative = Path(args.checkpoint_path).relative_to(scan_dir).as_posix()
except ValueError as exc:
raise SystemExit(
"Scan checkpoint must be inside the registered scan drafts directory."
) from exc
if not re.fullmatch(r"drafts/[0-9a-fA-F-]+\.checkpoint\.json", checkpoint_relative):
raise SystemExit(
"Scan checkpoint must be inside the registered scan drafts directory."
)
checkpoint, checkpoint_contents = _read_scan_local_json_bytes(
scan_dir, checkpoint_relative, "Staged scan checkpoint"
)
if checkpoint.get("scanId") != scan_id:
raise SystemExit("Staged scan checkpoint belongs to another scan.")
checkpoint_digest = hashlib.sha256(checkpoint_contents).hexdigest()
write_scan_local_bytes(
scan_dir,
f"checkpoints/{checkpoint_digest}.json",
checkpoint_contents,
)
checkpoint = _parent_scan_draft(scan_id, manifest["scan"], findings, coverage)
checkpoint_contents = _encoded(checkpoint)
checkpoint_name = f"{hashlib.sha256(checkpoint_contents).hexdigest()}.json"
checkpoint_relative = f"checkpoints/{checkpoint_name}"
if not (scan_dir / checkpoint_relative).exists():
write_scan_local_bytes(scan_dir, checkpoint_relative, checkpoint_contents)
write_scan_local_bytes(
scan_dir, "checkpoint-head.json", _encoded({"checkpoint": checkpoint_name})
)
for filename, document in (
("findings.json", findings),
("coverage.json", coverage),
("scan-manifest.json", manifest),
):
write_scan_local_bytes(
scan_dir,
filename,
(json.dumps(document, allow_nan=False, indent=2) + "\n").encode(),
)
model_warning = write_threat_model_projection_if_possible(scan_dir, manifest)
# Accepted Standard drafts are evidence of review or report assembly,
# even when the parent omitted its explicit progress call.
model_only_checkpoint = (
manifest["scan"].get("complete") is False
and isinstance(manifest["scan"].get("threatModel"), dict)
and not findings.get("findings")
and not coverage.get("surfaces")
and not coverage.get("deferred")
)
if scan["mode"] == "standard" and not model_only_checkpoint:
phase = "discovery" if manifest["scan"].get("complete") is False else "reporting"
earlier = PHASES[: PHASES.index(phase)]
placeholders = ",".join("?" for _ in earlier)
timestamp = db.now()
try:
with connection:
changed = connection.execute(
"UPDATE scans SET phase = ?, updated_at = ? "
f"WHERE id = ? AND status = 'running' AND phase IN ({placeholders})",
(phase, timestamp, scan_id, *earlier),
)
if changed.rowcount:
connection.execute(
"UPDATE scan_progress SET phase_items_total = 0, "
"phase_items_completed = 0, phase_progress_unit = NULL, updated_at = ? "
"WHERE scan_id = ?",
(timestamp, scan_id),
)
except sqlite3.Error as exc:
print(f"Could not save scan progress: {exc}", file=sys.stderr)
return {
"scanId": scan_id,
"status": "draft_written",
**({"warnings": [model_warning]} if model_warning else {}),
}
def _scan_draft_digest(scan_dir: Path) -> str:
digest = hashlib.sha256()
for filename in ("scan-manifest.json", "findings.json", "coverage.json"):
digest.update(filename.encode())
digest.update(b"\0")
try:
(scan_dir / filename).lstat()
except FileNotFoundError:
digest.update(b"missing\0")
continue
digest.update(b"present\0")
descriptor = open_scan_local_file_descriptor(scan_dir, filename, filename)
with os.fdopen(descriptor, "rb") as handle:
digest.update(handle.read())
digest.update(b"\0")
return digest.hexdigest()
def fail_scan(db: Any, connection: Any, args: Any) -> dict[str, Any]:
with db.scan_completion_lock(db.require_uuid(args.scan_id, "scan-id")):
return fail_scan_locked(db, connection, args)
def fail_scan_locked(db: Any, connection: Any, args: Any) -> dict[str, Any]:
scan_id = db.require_uuid(args.scan_id, "scan-id")
cost_json = db.parse_scan_cost(args.cost_json)
connection.execute("BEGIN IMMEDIATE")
try:
timestamp = db.now()
scan = db.require_scan(connection, scan_id)
if scan["status"] == "failed":
connection.commit()
return db.scan_context(connection, scan["id"])
if scan["status"] == "complete":
raise SystemExit("A completed scan cannot be marked failed.")
db.handoff.require_current_continuation(
scan,
args.claim_token,
error_message="Scan failure is owned by another continuation.",
)
message = db.optional_text(args.message, maximum=2400)
updated = connection.execute(
"""
UPDATE scans
SET status = 'failed', failure_message = ?, completed_at = ?, updated_at = ?,
cost_json = ?
WHERE id = ? AND status = 'running'
""",
(message, timestamp, timestamp, cost_json, scan["id"]),
)
if updated.rowcount != 1:
raise SystemExit("Only a running scan can be marked failed.")
db.deep_scan.fail_from_parent_scan(connection, scan["id"], message, timestamp)
progress_updated = connection.execute(
"UPDATE scan_progress SET updated_at = ? WHERE scan_id = ?",
(timestamp, scan["id"]),
)
if progress_updated.rowcount != 1:
raise SystemExit("Codex Security scan progress not found.")
connection.commit()
except BaseException:
connection.rollback()
raise
preserve_stopped_results_after_transition(db, connection, scan["id"])
return db.scan_context(connection, scan["id"])
def cancel_scan(db: Any, connection: Any, args: Any) -> dict[str, Any]:
with db.scan_completion_lock(db.require_uuid(args.scan_id, "scan-id")):
return cancel_scan_locked(db, connection, args)
def cancel_scan_locked(db: Any, connection: Any, args: Any) -> dict[str, Any]:
scan_id = db.require_uuid(args.scan_id, "scan-id")
thread_id = db.optional_text(args.thread_id, maximum=512)
connection.execute("BEGIN IMMEDIATE")
try:
timestamp = db.now()
scan = db.require_scan(connection, scan_id)
workspace = db.require_workspace(connection, scan["workspace_id"])
owning_thread_id = scan["continuation_thread_id"] or workspace["thread_id"]
if thread_id is not None and owning_thread_id != thread_id:
raise SystemExit("A scan can only be canceled from its owning Codex thread.")
if scan["canceled_at"] is not None:
connection.commit()
return db.workspace_state(connection, scan["workspace_id"])
if scan["status"] != "running":
raise SystemExit("Only a running scan can be canceled.")
updated = connection.execute(
"""
UPDATE scans
SET status = 'failed', canceled_at = ?, completed_at = ?, updated_at = ?
WHERE id = ? AND status = 'running'
""",
(timestamp, timestamp, timestamp, scan["id"]),
)
if updated.rowcount != 1:
raise SystemExit("Only a running scan can be canceled.")
db.deep_scan.cancel_from_parent_scan(connection, scan["id"], timestamp)
progress_updated = connection.execute(
"UPDATE scan_progress SET updated_at = ? WHERE scan_id = ?",
(timestamp, scan["id"]),
)
if progress_updated.rowcount != 1:
raise SystemExit("Codex Security scan progress not found.")
connection.commit()
except BaseException:
connection.rollback()
raise
preserve_stopped_results_after_transition(db, connection, scan["id"])
return db.workspace_state(connection, scan["workspace_id"])
def preserve_stopped_results_after_transition(db: Any, connection: Any, scan_id: str) -> None:
try:
published = preserve_scan_results_locked(db, connection, scan_id)
except (ContractError, OSError, SystemExit, ValueError) as exc:
scan = db.require_scan(connection, scan_id)
warnings = json.loads(scan["completion_warnings_json"])
warning = f"Saved scan evidence remains on disk; result publication needs follow-up: {exc}"
with connection:
connection.execute(
"UPDATE scans SET completion_warnings_json = ? WHERE id = ?",
(json.dumps(list(dict.fromkeys([*warnings, warning]))), scan_id),
)
return
if published:
db.deep_scan.clear_deep_scan_publication_failure(connection, scan_id)
if __name__ == "__main__":
argparse.ArgumentParser(description=__doc__).parse_args()
SHA-256: 436b23a4d169006437985bb352bfd792da9e465c4c7506dce31bfd5e78524c6d