113 lines
4.2 KiB
Python
113 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
||
"""Deterministic lane-selection fallback for the review gauntlet.
|
||
|
||
Reached ONLY via select_lanes' `fallback:` route, when the additive LLM
|
||
judgment dies after retries. Emits `forced_lanes` — code-review + adversary,
|
||
plus probe when the deterministic signals flagged a consumer-facing surface
|
||
AND a local-run recipe exists (build_items gates probe the same way — a
|
||
probe without a recipe is INCONCLUSIVE by construction),
|
||
plus security when the attack-surface signals (auth/deps/exec) or a hardened
|
||
posture demand it, unioned with any lanes the caller forced (canonicalized
|
||
with the same aliases build_items uses) — which build_items honors
|
||
ADDITIVELY, so selection degrades WIDER, never narrower. Unavailable diff
|
||
signals also widen: unknown surface adds security (and probe when a
|
||
local-run recipe exists). The degradation note goes into
|
||
`lanes_degraded`; build_items folds it into lanes_summary so the final
|
||
report names the degradation.
|
||
|
||
Fail-closed: if the fallback itself crashes it emits the widest safe
|
||
selection (defaults ∪ security) plus a PIPELINE-FAULT in signals_error so
|
||
verdict_gate blocks — a crashed fallback never narrows selection silently.
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
|
||
MAX_DETAIL_CHARS = 500
|
||
|
||
# Keep in sync with build_items.ALIASES — a caller-forced lane must
|
||
# canonicalize the same way here as it would on the non-degraded path.
|
||
ALIASES = {
|
||
"code-review": "code-review",
|
||
"code_review": "code-review",
|
||
"code-reviewer": "code-review",
|
||
"adversary": "adversary",
|
||
"adversarial": "adversary",
|
||
"adversarial-review": "adversary",
|
||
"security": "security",
|
||
"security-review": "security",
|
||
"security-reviewer": "security",
|
||
"probe": "probe",
|
||
"usage": "probe",
|
||
"usage-pattern": "probe",
|
||
"usage-pattern-testing": "probe",
|
||
}
|
||
|
||
# Stable emission order keeps the fallback deterministic.
|
||
LANE_ORDER = ["code-review", "adversary", "security", "probe"]
|
||
|
||
|
||
def load_state():
|
||
if path := os.environ.get("GRAPH_STATE_FILE"):
|
||
with open(path) as f:
|
||
return json.load(f)
|
||
return json.loads(os.environ.get("GRAPH_STATE", "{}"))
|
||
|
||
|
||
def main():
|
||
state = load_state()
|
||
lanes = {"code-review", "adversary"}
|
||
probe_context = state.get("probe_context")
|
||
if (
|
||
state.get("consumer_surface")
|
||
and isinstance(probe_context, str)
|
||
and probe_context.strip()
|
||
):
|
||
lanes.add("probe")
|
||
# The deterministic security signals are already in state when this
|
||
# fallback runs; a degraded selection must never drop the security lane
|
||
# build_items' hard rules would have selected.
|
||
if (
|
||
state.get("touches_auth")
|
||
or state.get("touches_deps")
|
||
or state.get("touches_exec")
|
||
or state.get("security_posture") == "hardened"
|
||
):
|
||
lanes.add("security")
|
||
# Unavailable signals = surface UNKNOWN → widen (mirrors build_items).
|
||
if (state.get("signals_summary") or "") == "signals unavailable":
|
||
lanes.add("security")
|
||
if isinstance(probe_context, str) and probe_context.strip():
|
||
lanes.add("probe")
|
||
# Caller-forced lanes survive degradation — union, never overwrite.
|
||
for lane in state.get("forced_lanes") or []:
|
||
if canon := ALIASES.get(str(lane).strip().lower()):
|
||
lanes.add(canon)
|
||
forced = [lane for lane in LANE_ORDER if lane in lanes]
|
||
raw = state.get("select_lanes_failure")
|
||
detail = raw.strip().replace("\n", " ") if isinstance(raw, str) else ""
|
||
if len(detail) > MAX_DETAIL_CHARS:
|
||
detail = detail[: MAX_DETAIL_CHARS - 1] + "…"
|
||
note = "lane selection degraded to deterministic defaults"
|
||
if detail:
|
||
note += f" — {detail}"
|
||
print(json.dumps({"forced_lanes": forced, "lanes_degraded": note}))
|
||
|
||
|
||
try:
|
||
main()
|
||
except Exception as e: # noqa: BLE001 — the fallback itself must never crash
|
||
print(
|
||
json.dumps(
|
||
{
|
||
"forced_lanes": ["code-review", "adversary", "security"],
|
||
"lanes_degraded": "lane selection degraded to deterministic "
|
||
f"defaults (fallback script error: {e})",
|
||
"signals_error": (
|
||
f"PIPELINE-FAULT: lane-selection fallback crashed — {e}; "
|
||
"lanes widened to defaults ∪ security"
|
||
),
|
||
}
|
||
)
|
||
)
|