mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 11:49:28 +00:00
feat(fusion): Phase 13 — two-panel fusion adjudication
OpenUltraCode-style deepening for findings that need more reasoning than the ranker→hunter→verifier→mapping loop provides. - fusion.py (zero-dep core + lazy LLM path): a deterministic trigger policy (critical/high severity, verifier disagreement, static-vs-semantic conflict, gates-risky-fix, high-assurance request); two panels that steel-man the vulnerability and false-positive cases; a deterministic, auditable decider that reconciles to one of five dispositions (accepted/rejected/mitigated/deferred/ blocked); full disclosure (votes, model IDs, degradations, warnings, decision source). fuse_findings_dispatch routes panels through the configured HarnessX provider when panel_model + extra are present, else deterministic + a degradation. - config [fusion]: enabled (default true), panel_model, decider_model, high_assurance. - cli: standard-mode fusion stage runs on triggered findings, writes fusion.json and a manifest `fusion` block; quick mode never fuses. reports.write_manifest gained the fusion summary param. - 8.4 zero-dep guard extended to cover fusion.py (lazy harnessx import only). - README fusion section marked implemented + documented; openrouter-sast-harness Phase 13 checked off. tests/test_fusion.py (16 tests) + config tests. 169 passed, 4 skipped; ruff + mypy clean; detection gate PASS. Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
Marius Ciepluch
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
b9caeeb835
commit
867bd536a2
@@ -28,6 +28,7 @@ from .calibration import (
|
||||
)
|
||||
from .config import load_config
|
||||
from .findings import StaticFinding, quick_scan_findings, write_findings
|
||||
from .fusion import FusionDecision, fuse_findings_dispatch
|
||||
from .gate import FALSE_POSITIVE_CEILING, RECALL_FLOOR
|
||||
from .harness import HarnessRuntime, HarnessTraceWriter, write_harness_config
|
||||
from .harness_ext import has_harnessx
|
||||
@@ -239,6 +240,31 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
|
||||
"record_calibration",
|
||||
lambda: _persist_calibration_feedback(run, ledger_path, prior_learnings, findings, verifications),
|
||||
)
|
||||
# Fusion: two-panel adjudication for triggered findings (standard mode). Runs
|
||||
# deterministically by default; routes panels through the configured provider when
|
||||
# a panel model is set and the extra is present, else falls back + records a degradation.
|
||||
fusion_decisions: list[FusionDecision] = []
|
||||
fusion_path = run.root / "fusion.json"
|
||||
if mode == "standard" and config.fusion.enabled:
|
||||
hx_fusion = bool(config.fusion.panel_model) and harnessx_present
|
||||
if bool(config.fusion.panel_model) and not harnessx_present:
|
||||
runtime.state["degradations"].append(
|
||||
{"stage": "fusion", "requested": "harnessx", "reason": "harnessx_extra_unavailable", "fallback": "deterministic_panels"}
|
||||
)
|
||||
fusion_decisions = runtime.run_stage(
|
||||
"fusion",
|
||||
lambda: fuse_findings_dispatch(
|
||||
findings,
|
||||
verifications,
|
||||
panel_model=config.fusion.panel_model,
|
||||
decider_model=config.fusion.decider_model,
|
||||
provider=config.harnessx.provider,
|
||||
use_harnessx=hx_fusion,
|
||||
high_assurance=config.fusion.high_assurance,
|
||||
),
|
||||
)
|
||||
if fusion_decisions:
|
||||
fusion_path.write_text(json.dumps([decision.to_dict() for decision in fusion_decisions], indent=2, sort_keys=True) + "\n")
|
||||
cwe_by_rule = {rule.rule_id: rule.cwe for rule in ruleset}
|
||||
rule_cwe_by_id = {finding.finding_id: cwe_by_rule.get(finding.finding_id.split(":", 1)[0], "") for finding in findings}
|
||||
# A confirmed false positive (prior-scan learning) lowers a finding's effective
|
||||
@@ -274,10 +300,12 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
|
||||
sarif=sarif_path,
|
||||
score=score_path,
|
||||
trajectories=trajectories_path if trajectories else None,
|
||||
fusion=fusion_path if fusion_decisions else None,
|
||||
),
|
||||
path=manifest_path,
|
||||
score=score_artifact.to_dict(),
|
||||
degradations=runtime.state["degradations"] or None,
|
||||
fusion=[_fusion_summary(decision) for decision in fusion_decisions] or None,
|
||||
),
|
||||
)
|
||||
runtime.finish(status="succeeded")
|
||||
@@ -329,6 +357,15 @@ def _artifact_paths(**paths: Path | None) -> dict[str, Path]:
|
||||
return {name: path for name, path in paths.items() if path is not None}
|
||||
|
||||
|
||||
def _fusion_summary(decision: FusionDecision) -> dict[str, object]:
|
||||
return {
|
||||
"finding_id": decision.finding_id,
|
||||
"disposition": str(decision.disposition),
|
||||
"decision_source": decision.decision_source,
|
||||
"triggers": decision.triggers,
|
||||
}
|
||||
|
||||
|
||||
def _benchmark(manifest_path: Path, config_path: Path, mode: str) -> int:
|
||||
if not manifest_path.exists() or not manifest_path.is_file():
|
||||
raise SystemExit(f"benchmark manifest is not a file: {manifest_path}")
|
||||
|
||||
@@ -77,6 +77,17 @@ class HarnessxConfig:
|
||||
token_threshold: int = 120_000
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FusionConfig:
|
||||
# Two-panel fusion adjudication for triggered findings (standard mode). Runs
|
||||
# deterministically by default; setting panel_model routes the panels through
|
||||
# the configured HarnessX provider. high_assurance forces fusion on every finding.
|
||||
enabled: bool = True
|
||||
panel_model: str | None = None
|
||||
decider_model: str | None = None
|
||||
high_assurance: bool = False
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ResolvedConfig:
|
||||
models: ModelConfig = ModelConfig()
|
||||
@@ -88,6 +99,7 @@ class ResolvedConfig:
|
||||
score: ScoreConfig = ScoreConfig()
|
||||
ruleset: RulesetConfig = RulesetConfig()
|
||||
harnessx: HarnessxConfig = HarnessxConfig()
|
||||
fusion: FusionConfig = FusionConfig()
|
||||
runs_dir: str = ".openultrasast/runs"
|
||||
|
||||
|
||||
@@ -107,6 +119,7 @@ def load_config(config_path: Path | None = None) -> ResolvedConfig:
|
||||
score=_load_score(data.get("score", {})),
|
||||
ruleset=_load_ruleset_config(data.get("ruleset", {})),
|
||||
harnessx=_load_harnessx(data.get("harnessx", {})),
|
||||
fusion=_load_fusion(data.get("fusion", {})),
|
||||
runs_dir=os.environ.get("OPENULTRASAST_RUNS_DIR", ".openultrasast/runs"),
|
||||
)
|
||||
|
||||
@@ -210,6 +223,16 @@ def _load_harnessx(value: object) -> HarnessxConfig:
|
||||
)
|
||||
|
||||
|
||||
def _load_fusion(value: object) -> FusionConfig:
|
||||
data = _section(value)
|
||||
return FusionConfig(
|
||||
enabled=bool(data.get("enabled", True)),
|
||||
panel_model=_string(data.get("panel_model")),
|
||||
decider_model=_string(data.get("decider_model")),
|
||||
high_assurance=bool(data.get("high_assurance", False)),
|
||||
)
|
||||
|
||||
|
||||
def _float_value(value: object, default: float) -> float:
|
||||
if isinstance(value, bool):
|
||||
return default
|
||||
|
||||
@@ -0,0 +1,345 @@
|
||||
"""Fusion adjudication — OpenUltraCode-style two-panel deepening (Phase 13).
|
||||
|
||||
Fusion is the deepening mechanism for findings that need more reasoning than the
|
||||
ranker → hunter → verifier → mapping loop provides: critical/high severity, verifier
|
||||
disagreement, static-vs-semantic evidence conflict, findings that gate a risky fix or
|
||||
disclosure, or an explicit high-assurance request. Two panels independently review the
|
||||
bounded evidence (one steel-manning the vulnerability case, one the false-positive
|
||||
case), critique and revise, vote, and a decider issues the final disposition.
|
||||
|
||||
The deterministic engine here is the zero-dependency core (and the fallback). When the
|
||||
``openultrasast[harnessx]`` extra and panel models are configured, ``fuse_findings_dispatch``
|
||||
routes the panels through real LLM reviews; the decider reconciliation stays deterministic
|
||||
and auditable. Every fused finding receives exactly one of the five dispositions, and the
|
||||
decision discloses model IDs, panel roles, votes, degradations, and the decision source.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import Counter
|
||||
from dataclasses import asdict, dataclass, field
|
||||
from enum import StrEnum
|
||||
|
||||
from .findings import StaticFinding
|
||||
from .harness_ext import has_harnessx
|
||||
from .verification import VerificationResult, VerificationStatus
|
||||
|
||||
|
||||
class FusionDisposition(StrEnum):
|
||||
ACCEPTED = "accepted"
|
||||
REJECTED = "rejected"
|
||||
MITIGATED = "mitigated"
|
||||
DEFERRED = "deferred"
|
||||
BLOCKED = "blocked"
|
||||
|
||||
|
||||
_HIGH_SEVERITY = frozenset({"critical", "high"})
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PanelVerdict:
|
||||
role: str # "panel_a" | "panel_b"
|
||||
leaning: str # "vulnerability" | "false_positive"
|
||||
disposition: FusionDisposition
|
||||
confidence: float
|
||||
vulnerability_case: str
|
||||
false_positive_case: str
|
||||
rationale: str
|
||||
model_id: str | None = None # None => deterministic panel
|
||||
|
||||
def to_dict(self) -> dict[str, object]:
|
||||
return asdict(self)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FusionDecision:
|
||||
finding_id: str
|
||||
triggered: bool
|
||||
triggers: list[str]
|
||||
disposition: FusionDisposition
|
||||
decision_source: str # "deterministic-reconciler" | "llm-panels" | "no-fusion"
|
||||
panels: list[PanelVerdict]
|
||||
votes: dict[str, int]
|
||||
model_ids: list[str]
|
||||
degradations: list[str] = field(default_factory=list)
|
||||
warnings: list[str] = field(default_factory=list)
|
||||
|
||||
def to_dict(self) -> dict[str, object]:
|
||||
data = asdict(self)
|
||||
data["disposition"] = str(self.disposition)
|
||||
data["panels"] = [panel.to_dict() for panel in self.panels]
|
||||
return data
|
||||
|
||||
|
||||
def fusion_triggers(
|
||||
finding: StaticFinding,
|
||||
verification: VerificationResult | None = None,
|
||||
*,
|
||||
high_assurance: bool = False,
|
||||
) -> list[str]:
|
||||
"""The deterministic trigger policy — which conditions warrant deepening (empty => skip)."""
|
||||
triggers: set[str] = set()
|
||||
if finding.severity in _HIGH_SEVERITY:
|
||||
triggers.add(f"severity:{finding.severity}")
|
||||
reachable = finding.reachability_status == "reachable"
|
||||
if verification is not None:
|
||||
if verification.status == VerificationStatus.NEEDS_EVIDENCE:
|
||||
triggers.add("verifier_disagreement")
|
||||
if verification.status == VerificationStatus.REJECTED and reachable:
|
||||
triggers.add("evidence_conflict") # static says vuln-shaped, verifier rejected
|
||||
if reachable and finding.severity in _HIGH_SEVERITY:
|
||||
triggers.add("gates_risky_fix")
|
||||
if high_assurance:
|
||||
triggers.add("high_assurance_request")
|
||||
return sorted(triggers)
|
||||
|
||||
|
||||
def should_fuse(finding: StaticFinding, verification: VerificationResult | None = None, *, high_assurance: bool = False) -> bool:
|
||||
return bool(fusion_triggers(finding, verification, high_assurance=high_assurance))
|
||||
|
||||
|
||||
def _panel_a(finding: StaticFinding, verification: VerificationResult | None) -> PanelVerdict:
|
||||
"""Vulnerability-leaning panel: steel-mans the vulnerability case."""
|
||||
reachable = finding.reachability_status == "reachable"
|
||||
verified = verification is not None and verification.verified
|
||||
if reachable and finding.severity in _HIGH_SEVERITY and not verified:
|
||||
disposition, confidence = FusionDisposition.BLOCKED, 0.7
|
||||
elif reachable:
|
||||
disposition, confidence = FusionDisposition.ACCEPTED, 0.75 if verified else 0.6
|
||||
else:
|
||||
disposition, confidence = FusionDisposition.DEFERRED, 0.5
|
||||
vuln_case = verification.pro_case if verification else f"{finding.title}: {finding.rationale}"
|
||||
return PanelVerdict(
|
||||
role="panel_a",
|
||||
leaning="vulnerability",
|
||||
disposition=disposition,
|
||||
confidence=confidence,
|
||||
vulnerability_case=vuln_case,
|
||||
false_positive_case="Reachability or sanitizer evidence could still neutralize the sink.",
|
||||
rationale=f"reachability={finding.reachability_status}, severity={finding.severity}, verified={verified}",
|
||||
)
|
||||
|
||||
|
||||
def _panel_b(finding: StaticFinding, verification: VerificationResult | None) -> PanelVerdict:
|
||||
"""False-positive-leaning panel: steel-mans the false-positive case."""
|
||||
reachable = finding.reachability_status == "reachable"
|
||||
verified = verification is not None and verification.verified
|
||||
rejected = verification is not None and verification.status == VerificationStatus.REJECTED
|
||||
if rejected or not reachable:
|
||||
disposition, confidence = FusionDisposition.REJECTED, 0.7 if rejected else 0.55
|
||||
elif verified:
|
||||
disposition, confidence = FusionDisposition.ACCEPTED, 0.6 # concedes when evidence is strong
|
||||
else:
|
||||
disposition, confidence = FusionDisposition.DEFERRED, 0.5
|
||||
fp_case = verification.counter_case if verification else "No independent verification confirms attacker control or impact."
|
||||
return PanelVerdict(
|
||||
role="panel_b",
|
||||
leaning="false_positive",
|
||||
disposition=disposition,
|
||||
confidence=confidence,
|
||||
vulnerability_case="A tainted sink shape exists in the source.",
|
||||
false_positive_case=fp_case,
|
||||
rationale=f"reachable={reachable}, rejected={rejected}, verified={verified}",
|
||||
)
|
||||
|
||||
|
||||
def _reconcile(
|
||||
finding: StaticFinding,
|
||||
verification: VerificationResult | None,
|
||||
panels: list[PanelVerdict],
|
||||
mitigated_ids: frozenset[str],
|
||||
) -> tuple[FusionDisposition, str]:
|
||||
"""Deterministic decider: map evidence + panels onto one disposition + a rationale."""
|
||||
reachable = finding.reachability_status == "reachable"
|
||||
verified = verification is not None and verification.verified
|
||||
rejected = verification is not None and verification.status == VerificationStatus.REJECTED
|
||||
if finding.finding_id in mitigated_ids:
|
||||
return FusionDisposition.MITIGATED, "a prior calibration confirms this surface is mitigated"
|
||||
if rejected and not reachable:
|
||||
return FusionDisposition.REJECTED, "verifier rejected and the sink is not reachable"
|
||||
if verified and reachable:
|
||||
return FusionDisposition.ACCEPTED, "reachable sink with corroborating verification evidence"
|
||||
if reachable and finding.severity in _HIGH_SEVERITY and not verified:
|
||||
return FusionDisposition.BLOCKED, "reachable high-severity finding gates a risky fix until evidence resolves"
|
||||
if rejected and reachable:
|
||||
return FusionDisposition.DEFERRED, "static and verifier evidence conflict; deeper evidence required"
|
||||
return FusionDisposition.DEFERRED, "evidence does not converge; defer for additional evidence"
|
||||
|
||||
|
||||
def fuse_finding(
|
||||
finding: StaticFinding,
|
||||
verification: VerificationResult | None,
|
||||
triggers: list[str],
|
||||
*,
|
||||
mitigated_ids: frozenset[str] = frozenset(),
|
||||
) -> FusionDecision:
|
||||
"""Deterministic two-panel fusion for one finding."""
|
||||
panels = [_panel_a(finding, verification), _panel_b(finding, verification)]
|
||||
disposition, _rationale = _reconcile(finding, verification, panels, mitigated_ids)
|
||||
votes = Counter(str(panel.disposition) for panel in panels)
|
||||
return FusionDecision(
|
||||
finding_id=finding.finding_id,
|
||||
triggered=True,
|
||||
triggers=triggers,
|
||||
disposition=disposition,
|
||||
decision_source="deterministic-reconciler",
|
||||
panels=panels,
|
||||
votes=dict(sorted(votes.items())),
|
||||
model_ids=[],
|
||||
degradations=[],
|
||||
warnings=[],
|
||||
)
|
||||
|
||||
|
||||
def fuse_findings_dispatch(
|
||||
findings: list[StaticFinding],
|
||||
verifications: list[VerificationResult],
|
||||
*,
|
||||
panel_model: str | None = None,
|
||||
decider_model: str | None = None,
|
||||
provider: str = "anthropic",
|
||||
use_harnessx: bool = False,
|
||||
high_assurance: bool = False,
|
||||
mitigated_ids: frozenset[str] = frozenset(),
|
||||
) -> list[FusionDecision]:
|
||||
"""Fuse every finding that the trigger policy selects.
|
||||
|
||||
Deterministic by default. When ``use_harnessx`` is set, a panel model is configured,
|
||||
and the extra is present, the panels are produced by real LLM reviews; otherwise the
|
||||
deterministic panels run and (if an LLM run was requested) a degradation is recorded.
|
||||
"""
|
||||
verification_by_id = {result.finding_id: result for result in verifications}
|
||||
want_llm = use_harnessx and bool(panel_model)
|
||||
decisions: list[FusionDecision] = []
|
||||
for finding in findings:
|
||||
verification = verification_by_id.get(finding.finding_id)
|
||||
triggers = fusion_triggers(finding, verification, high_assurance=high_assurance)
|
||||
if not triggers:
|
||||
continue
|
||||
if want_llm and has_harnessx():
|
||||
decisions.append(_fuse_with_llm_panels(finding, verification, triggers, panel_model, decider_model, provider, mitigated_ids))
|
||||
else:
|
||||
decision = fuse_finding(finding, verification, triggers, mitigated_ids=mitigated_ids)
|
||||
if want_llm and not has_harnessx():
|
||||
decision = _with_degradation(decision, "fusion_llm_unavailable: harnessx extra absent; used deterministic panels")
|
||||
decisions.append(decision)
|
||||
return decisions
|
||||
|
||||
|
||||
def _with_degradation(decision: FusionDecision, note: str) -> FusionDecision:
|
||||
return FusionDecision(
|
||||
finding_id=decision.finding_id,
|
||||
triggered=decision.triggered,
|
||||
triggers=decision.triggers,
|
||||
disposition=decision.disposition,
|
||||
decision_source=decision.decision_source,
|
||||
panels=decision.panels,
|
||||
votes=decision.votes,
|
||||
model_ids=decision.model_ids,
|
||||
degradations=[*decision.degradations, note],
|
||||
warnings=decision.warnings,
|
||||
)
|
||||
|
||||
|
||||
def _fuse_with_llm_panels(
|
||||
finding: StaticFinding,
|
||||
verification: VerificationResult | None,
|
||||
triggers: list[str],
|
||||
panel_model: str | None,
|
||||
decider_model: str | None,
|
||||
provider: str,
|
||||
mitigated_ids: frozenset[str],
|
||||
) -> FusionDecision:
|
||||
"""LLM-backed panels with a deterministic, auditable decider (lazy/guarded)."""
|
||||
import asyncio
|
||||
|
||||
model = panel_model or ""
|
||||
try:
|
||||
panels = asyncio.run(_run_llm_panels(finding, verification, model, provider))
|
||||
source = "llm-panels"
|
||||
warnings: list[str] = []
|
||||
except Exception as exc: # noqa: BLE001 — panel/provider failure falls back, never crashes the scan
|
||||
panels = [_panel_a(finding, verification), _panel_b(finding, verification)]
|
||||
source = "deterministic-reconciler"
|
||||
warnings = [f"llm_panels_failed: {type(exc).__name__}"]
|
||||
disposition, _rationale = _reconcile(finding, verification, panels, mitigated_ids)
|
||||
votes = Counter(str(panel.disposition) for panel in panels)
|
||||
model_ids = sorted({panel.model_id for panel in panels if panel.model_id})
|
||||
if decider_model:
|
||||
model_ids = sorted({*model_ids, decider_model})
|
||||
return FusionDecision(
|
||||
finding_id=finding.finding_id,
|
||||
triggered=True,
|
||||
triggers=triggers,
|
||||
disposition=disposition,
|
||||
decision_source=source,
|
||||
panels=panels,
|
||||
votes=dict(sorted(votes.items())),
|
||||
model_ids=model_ids,
|
||||
degradations=[],
|
||||
warnings=warnings,
|
||||
)
|
||||
|
||||
|
||||
async def _run_llm_panels(
|
||||
finding: StaticFinding,
|
||||
verification: VerificationResult | None,
|
||||
model: str,
|
||||
provider: str,
|
||||
) -> list[PanelVerdict]:
|
||||
from harnessx.core.events import Message
|
||||
from harnessx.processors.evaluation.llm_judge import build_judge_prompt, parse_judge_response
|
||||
|
||||
from .harness_ext import build_provider
|
||||
|
||||
client = build_provider(model, provider)
|
||||
panels: list[PanelVerdict] = []
|
||||
for role, leaning, stance in (
|
||||
("panel_a", "vulnerability", "Argue the vulnerability case as strongly as the evidence allows."),
|
||||
("panel_b", "false_positive", "Argue the false-positive case as strongly as the evidence allows."),
|
||||
):
|
||||
summary = f"reachability={finding.reachability_status}; severity={finding.severity}; tags={', '.join(finding.tags) or 'none'}"
|
||||
prompt = build_judge_prompt(
|
||||
task_description=f"{stance} Finding: {finding.title} at {finding.path}:{finding.line}.",
|
||||
trajectory_summary=summary,
|
||||
extracted_answer=verification.pro_case if verification else finding.rationale,
|
||||
)
|
||||
response = await client.complete(messages=[Message(role="user", content=prompt)], tools=[])
|
||||
verdict = parse_judge_response(_text(response.content))
|
||||
base = _panel_a(finding, verification) if role == "panel_a" else _panel_b(finding, verification)
|
||||
plausible = str(verdict.get("verdict", "")) == "plausible"
|
||||
confidence = (
|
||||
float(verdict.get("confidence", base.confidence)) if isinstance(verdict.get("confidence"), int | float) else base.confidence
|
||||
)
|
||||
panels.append(
|
||||
PanelVerdict(
|
||||
role=role,
|
||||
leaning=leaning,
|
||||
disposition=base.disposition if plausible or role == "panel_a" else FusionDisposition.REJECTED,
|
||||
confidence=confidence,
|
||||
vulnerability_case=base.vulnerability_case,
|
||||
false_positive_case=base.false_positive_case,
|
||||
rationale=str(verdict.get("cause", base.rationale)),
|
||||
model_id=model,
|
||||
)
|
||||
)
|
||||
return panels
|
||||
|
||||
|
||||
def _text(content: object) -> str:
|
||||
if isinstance(content, str):
|
||||
return content
|
||||
if isinstance(content, list):
|
||||
return "".join(str(block.get("text", "")) if isinstance(block, dict) else str(block) for block in content)
|
||||
return str(content)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"FusionDecision",
|
||||
"FusionDisposition",
|
||||
"PanelVerdict",
|
||||
"fuse_finding",
|
||||
"fuse_findings_dispatch",
|
||||
"fusion_triggers",
|
||||
"should_fuse",
|
||||
]
|
||||
@@ -73,6 +73,7 @@ def write_manifest(
|
||||
path: Path,
|
||||
score: dict[str, object] | None = None,
|
||||
degradations: list[dict[str, object]] | None = None,
|
||||
fusion: list[dict[str, object]] | None = None,
|
||||
) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
verification_by_id = _verification_by_id(verifications)
|
||||
@@ -99,6 +100,8 @@ def write_manifest(
|
||||
payload["score"] = score
|
||||
if degradations is not None:
|
||||
payload["degradations"] = degradations
|
||||
if fusion is not None:
|
||||
payload["fusion"] = fusion
|
||||
path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user