pre-push(17.5): say when the engine cannot finish, and offer it in the background

--engine inline|background|off (hook: OUSAST_PUSH_ENGINE). background returns
after the quick tier and starts one detached explicit replay of the same base
and head (push/background.py), one at a time per artifact directory; its
artifact lands under <artifact dir>/engine/<repo key>/<head>.json and the next
run prints "Engine result for <head> from the previous push" once. A deadline
miss says so in one line naming the deadline and the background option. No
engine speed work. Requirement 9.5.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo
This commit is contained in:
Marius Ciepluch
2026-10-02 22:31:14 +02:00
co-authored by Claude Opus 5.5
parent 511de4b75f
commit 66357ffea1
9 changed files with 392 additions and 10 deletions
+24 -1
View File
@@ -154,6 +154,19 @@ def _main(argv: list[str] | None) -> int:
replay_parser.add_argument("--max-regions", type=int, default=500)
replay_parser.add_argument("--mode", choices=("advisory", "blocking"), default="advisory")
replay_parser.add_argument("--incomplete-coverage", choices=("allow", "block"), default="allow")
replay_parser.add_argument(
"--engine",
choices=("inline", "background", "off"),
default="inline",
help="inline (default): run the Joern engine inside the deadline; background: return after the quick rules and "
"run the engine detached, showing its result on the next run; off: quick rules only",
)
replay_parser.add_argument(
"--background-deadline",
type=float,
default=900.0,
help="seconds the detached engine run of --engine background may take (default 900)",
)
replay_parser.add_argument(
"--experimental-declarations",
type=Path,
@@ -459,6 +472,8 @@ def _pre_push(args: argparse.Namespace, parser: argparse.ArgumentParser) -> int:
head=args.head,
record_vetoes=args.experimental_record_vetoes,
declarations=args.experimental_declarations,
engine=args.engine,
background_deadline=args.background_deadline,
**options,
)
else:
@@ -481,7 +496,15 @@ def _pre_push(args: argparse.Namespace, parser: argparse.ArgumentParser) -> int:
def updates() -> str:
return (prior_data if prior_data is not None else stream.read(1024 * 1024 + 1)).decode("utf-8")
delivery = push(args.path, updates=updates, remote_name=args.remote[0], remote_url=args.remote[1], **options)
delivery = push(
args.path,
updates=updates,
remote_name=args.remote[0],
remote_url=args.remote[1],
engine=args.engine,
background_deadline=args.background_deadline,
**options,
)
except ValueError as error:
parser.error(str(error))
print(delivery.text, end="", flush=True)
+167
View File
@@ -0,0 +1,167 @@
"""Run the engine after the push, and show its result on the next one (task 17.5, Requirement 9.5).
The engine cannot finish a small repository inside a 30 s hook deadline (the 2026-10-02 audit
measured 260 s on 26 files). Until graph reuse makes it fast, `--engine background` lets the hook
return after the quick tier and starts one detached explicit replay of the same base and head. That
replay writes an ordinary artifact under the artifact directory; the next run in the same repository
prints "engine result for <head> from the previous push" once, then marks it shown.
One background engine at a time per artifact directory: the host this was built on has 7 GB, and
two Joern JVMs is what this project has seen killed. No speed work happens here.
"""
from __future__ import annotations
import contextlib
import hashlib
import json
import os
import subprocess
import sys
from collections.abc import Mapping
from pathlib import Path
from openultrasast.push.report import CAPABILITY_VETOES
_LOCK = "running.json"
def results_dir(artifact: Path, repository: Path) -> Path:
"""Per-repository directory beside the artifact; the repository path is hashed, never stored raw."""
key = hashlib.sha256(str(repository.resolve()).encode()).hexdigest()[:16]
return artifact.parent / "engine" / key
def _alive(pid: object) -> bool:
if not isinstance(pid, int) or pid <= 0:
return False
try:
os.kill(pid, 0)
except ProcessLookupError:
return False
except PermissionError:
return True
return True
def start(
repository: Path,
*,
base: str,
head: str,
artifact: Path,
deadline_seconds: float,
max_regions: int = 500,
cache_dir: Path | None = None,
) -> dict[str, object]:
"""Start one detached engine replay of base..head; never waits for it and never raises."""
directory = results_dir(artifact, repository)
try:
directory.mkdir(parents=True, exist_ok=True)
lock = directory / _LOCK
try:
running = json.loads(lock.read_text())
except (OSError, ValueError):
running = {}
if isinstance(running, dict) and _alive(running.get("pid")):
return {"status": "busy", "head": str(running.get("head", "")), "directory": str(directory)}
target = directory / f"{head}.json"
log = directory / f"{head}.log"
command = [
sys.executable,
"-m",
"openultrasast.cli",
"pre-push",
str(repository.resolve()),
"--base",
base,
"--head",
head,
"--artifact",
str(target),
"--deadline",
f"{deadline_seconds:g}",
"--max-regions",
str(max_regions),
"--engine",
"inline",
]
if cache_dir is not None:
command.extend(("--cache-dir", str(cache_dir)))
with open(log, "wb") as output:
process = subprocess.Popen( # noqa: S603 -- our own CLI with immutable commit ids, no shell
command,
stdin=subprocess.DEVNULL,
stdout=output,
stderr=subprocess.STDOUT,
cwd=str(repository.resolve()),
start_new_session=True,
close_fds=True,
)
lock.write_text(json.dumps({"pid": process.pid, "head": head, "base": base}))
return {
"status": "started",
"head": head,
"deadline_seconds": deadline_seconds,
"artifact": str(target),
"log": str(log),
}
except Exception as error: # noqa: BLE001 -- a background start that fails is a stated gap, never a block
return {"status": "failed", "reason": type(error).__name__}
def _summary(data: Mapping[str, object]) -> dict[str, object]:
admission = data.get("admission")
result = data.get("result")
admission = admission if isinstance(admission, Mapping) else {}
result = result if isinstance(result, Mapping) else {}
defects = admission.get("defects")
reasons = admission.get("coverage_reasons")
reasons = [str(r) for r in reasons] if isinstance(reasons, list) else []
advisory = []
dispositions = admission.get("dispositions")
for item in dispositions if isinstance(dispositions, list) else []:
try:
delta = item["candidate"]["delta"]
op = delta["head_operation"] or {}
if (
not item["admitted"]
and delta["novelty"] in ("new", "worsened")
and set(item["reasons"]) <= CAPABILITY_VETOES
and op
and delta["witness"]
):
advisory.append({"family": delta["family"], "path": op["path"], "line": op["line"], "novelty": delta["novelty"]})
except (KeyError, TypeError):
continue
unique = list({(a["family"], a["path"], a["line"]): a for a in advisory}.values())
return {
"alerts": len(defects) if isinstance(defects, list) else 0,
"advisory": unique,
"coverage": result.get("coverage_status", "unknown"),
"finished": "deadline_exhausted" not in reasons,
"engine_ran": not any(r in ("cpg_build_failed", "cpg_unavailable") for r in reasons),
}
def previous(artifact: Path, repository: Path) -> list[dict[str, object]]:
"""Finished background results not yet shown for this repository; each is shown exactly once."""
directory = results_dir(artifact, repository)
shown: list[dict[str, object]] = []
try:
candidates = sorted(directory.glob("*.json"), key=lambda p: p.stat().st_mtime)
except OSError:
return shown
for path in candidates:
if path.name == _LOCK or path.with_suffix(".shown").exists():
continue
try:
data = json.loads(path.read_text())
except (OSError, ValueError):
continue # still being written, or unreadable: leave it for the next run
if not isinstance(data, dict):
continue
shown.append({"head": path.stem, "artifact": str(path), **_summary(data)})
with contextlib.suppress(OSError):
path.with_suffix(".shown").touch()
return shown
+1
View File
@@ -58,6 +58,7 @@ GLOSSARY: dict[str, str] = {
"regions_truncated": "only the highest-ranked regions were examined (--max-regions)",
"files_unparsed": "some files could not be parsed by the engine",
"engine_off": "engine checks were switched off for this run (--engine off)",
"engine_background": "engine checks run in the background after this push; the next push shows their result",
"cross_partition_semantics_unresolved": "flows between languages in this repository are not followed",
"resolution": "the pushed ref could not be resolved ({detail})",
"push_input_failed": "the push input could not be read ({detail})",
+47 -2
View File
@@ -47,6 +47,8 @@ class PushReport:
experimental: Mapping[str, object] = field(default_factory=dict)
# The quick-rule tier (task 17.3): one payload per comparison it ran for; advisory only.
quick: tuple[Mapping[str, object], ...] = ()
# The background engine (task 17.5): what this run started, and earlier results shown now.
background: Mapping[str, object] = field(default_factory=dict)
def __post_init__(self) -> None:
object.__setattr__(self, "model_assistance", MappingProxyType(dict(self.model_assistance)))
@@ -137,6 +139,7 @@ def _write_artifact(report: PushReport, target: Path, temporary: Path, connectio
"resolution": report.resolution.to_payload() if report.resolution else None,
"experimental": dict(report.experimental),
"quick_tier": [dict(item) for item in report.quick],
"engine_background": dict(report.background),
}
json.dump(payload, stream, ensure_ascii=True, allow_nan=False, indent=2)
stream.write("\n")
@@ -222,7 +225,7 @@ def _experimental_line(experimental: Mapping[str, object]) -> str:
# Vetoes that say only "this capability is not qualified yet". A candidate whose novelty is established
# and whose only vetoes are these is shown as an advisory engine finding (task 17.4, Requirement 9.4);
# any comparison, evidence or location veto keeps it out of the terminal.
_CAPABILITY_VETOES = frozenset(
CAPABILITY_VETOES = frozenset(
{
"capability_unavailable",
"capability_disabled",
@@ -243,7 +246,7 @@ def _advisory_engine_lines(report: PushReport) -> list[str]:
if (
disposition.admitted
or delta.novelty not in ("new", "worsened")
or not set(disposition.reasons) <= _CAPABILITY_VETOES
or not set(disposition.reasons) <= CAPABILITY_VETOES
or op is None
or delta.witness is None
or delta.defect_id in shown
@@ -263,6 +266,46 @@ def _advisory_engine_lines(report: PushReport) -> list[str]:
return lines
def _previous_engine_lines(background: Mapping[str, object]) -> list[str]:
"""Background engine results from an earlier push, shown once (task 17.5)."""
lines: list[str] = []
earlier = background.get("previous")
for item in earlier if isinstance(earlier, list) else []:
if not isinstance(item, Mapping):
continue
advisory = item.get("advisory")
found = [a for a in advisory if isinstance(a, Mapping)] if isinstance(advisory, list) else []
state = (
"the engine did not run (Joern unavailable)"
if not item.get("engine_ran")
else f"{item.get('alerts', 0)} alert(s), {len(found)} advisory engine finding(s), coverage {item.get('coverage')}"
+ ("" if item.get("finished") else ", did not finish within its deadline")
)
head = str(item.get("head", ""))[:12]
lines.append(f"Engine result for {_short(head)} from the previous push: {_short(state, 300)}.")
for entry in found[:3]:
lines.append(f"- {_short(str(entry.get('family')))} at {_short(str(entry.get('path')), 4096)}:{entry.get('line')}")
lines.append(" Details: " + _short(str(item.get("artifact", "")), 4096))
return lines
def _background_lines(background: Mapping[str, object]) -> list[str]:
status = background.get("status")
head = _short(str(background.get("head", ""))[:12])
if status == "started":
return [
f"Engine: running in the background for {head} (deadline {background.get('deadline_seconds')} s); "
"the next push shows its result."
]
if status == "busy":
return [f"Engine: a background engine run for {head} is still busy, so none was started for this push."]
if status == "failed":
return [f"Engine: the background run could not be started ({_short(str(background.get('reason')))})."]
if status == "not_started":
return [f"Engine: no background run was started ({_short(str(background.get('reason')))})."]
return []
# Quick-rule matches shown per comparison; the artifact keeps every one.
_QUICK_LINES = 5
@@ -329,6 +372,7 @@ def render_report(report: PushReport, *, artifact: Path | None, error: str | Non
)
if len(defects) > 3:
lines.append(f"{len(defects) - 3} more actionable defects retained in the detailed result.")
lines.extend(_previous_engine_lines(report.background))
lines.extend(_advisory_engine_lines(report))
for payload in report.quick:
lines.extend(_quick_lines(payload))
@@ -364,6 +408,7 @@ def render_report(report: PushReport, *, artifact: Path | None, error: str | Non
notices.append("Details could not be saved. Choose an artifact path outside the analyzed repository.")
elif error:
notices.append("Details could not be saved. Retry with a writable artifact path and available reporting time.")
lines.extend(_background_lines(report.background))
if notices:
lines.append("Notice: " + " ".join(notices))
if report.experimental:
+55 -3
View File
@@ -268,7 +268,7 @@ def _analyze(
resolved: PushComparison | None = None,
record_vetoes: bool = False,
declarations: Path | None = None,
engine: Literal["inline", "off"] = "inline",
engine: Literal["inline", "off", "background"] = "inline",
quick_basis: Basis = "comparison",
) -> PushReport:
"""Run a single explicit comparison; exit policy never stands for coverage.
@@ -390,8 +390,8 @@ def _analyze(
quick = QuickTierResult("failed", quick_basis, reason=type(error).__name__)
reasons.append("quick_rules_failed:" + type(error).__name__)
stage_started = _stage(timings, stage, stage_started)
if engine == "off":
reasons.append("engine_off")
if engine != "inline":
reasons.append("engine_" + engine)
if engine == "inline":
stage = "head"
head_scan = scan_repository(
@@ -523,6 +523,40 @@ def _last_commit_quick(
)
def _background(
repository: Path,
pending: Sequence[tuple[str | None, str]],
artifact: Path,
deadline: float,
max_regions: int,
cache_dir: Path | None,
) -> dict[str, object]:
"""Earlier background results to show, and at most one new background engine run (task 17.5)."""
from openultrasast.push import background
state: dict[str, object] = {"previous": background.previous(artifact, repository)}
ready = [(base, head) for base, head in pending if base is not None]
if ready:
base, head = ready[0]
assert base is not None
state.update(
background.start(
repository,
base=base,
head=head,
artifact=artifact,
deadline_seconds=deadline,
max_regions=max_regions,
cache_dir=cache_dir,
)
)
if len(ready) > 1:
state["not_started"] = [h for _, h in ready[1:]] # one engine at a time
elif pending:
state.update(status="not_started", reason="no base revision to compare against")
return state
def _deliver(repository: Path, report: PushReport, artifact: Path, budget: ExecutionBudget) -> ReportDelivery:
if artifact.resolve().is_relative_to(repository.resolve()):
error = "artifact_inside_repository"
@@ -543,6 +577,8 @@ def replay(
model_config: Path | None = None,
record_vetoes: bool = False,
declarations: Path | None = None,
engine: Literal["inline", "off", "background"] = "inline",
background_deadline: float = 900.0,
) -> ReportDelivery:
config = config or PushConfig()
if declarations is not None and not record_vetoes:
@@ -560,7 +596,10 @@ def replay(
execution_budget=budget,
record_vetoes=record_vetoes,
declarations=declarations,
engine=engine,
)
pending = [(base, head)] if engine == "background" else []
report = replace(report, background=_background(repository, pending, artifact, background_deadline, max_regions, cache_dir))
from openultrasast.push.assistance import assist
report = assist(report, config=model_config, budget=budget)
@@ -580,6 +619,8 @@ def push(
cache_dir: Path | None = None,
model_config: Path | None = None,
execution_budget: ExecutionBudget | None = None,
engine: Literal["inline", "off", "background"] = "inline",
background_deadline: float = 900.0,
) -> ReportDelivery:
"""Consume one Git transaction. Resolution, every comparison and reporting share a deadline."""
config = config or PushConfig()
@@ -623,6 +664,7 @@ def push(
cache_dir=cache_dir,
execution_budget=budget,
resolved=comparison,
engine=engine,
)
reports.append(report)
analyses.extend(report.result.analyses)
@@ -674,6 +716,16 @@ def push(
snapshots=tuple(s for r in reports for s in r.snapshots),
change_contexts=tuple(r.change_context for r in reports if r.change_context is not None),
quick=tuple(q for r in (*reports, *fallbacks) for q in r.quick),
background=_background(
repository,
[(r.change_context.base_revision, r.change_context.head_revision) for r in reports if r.change_context]
if engine == "background"
else [],
artifact,
background_deadline,
max_regions,
cache_dir,
),
)
from openultrasast.push.assistance import assist