feat(pre-push-safety-net): establish shared push and engine contracts

This commit is contained in:
Marius Ciepluch
2026-09-12 19:21:31 +02:00
parent 051f3b0058
commit 14a2a60e52
18 changed files with 810 additions and 25 deletions
+5
View File
@@ -28,6 +28,11 @@ ROOTS = ("gate", "map_gate", "pair_gate", "cli", "__main__")
# Modules that stand alone: their own CLI subcommand or external protocol, deliberately off the scan path.
SPEC_OWNED = {
"contracts": "pre-push-safety-net Req 7.3: strict shared manifest serialization, independent of push",
"model.contracts": "pre-push-safety-net Req 4.1/7.3: generic deadline, change and exact scope contracts; integration pending",
"cpg.artifact": "pre-push-safety-net Req 4.3/7.3: backend graph identity and live lease contract; validation integration pending",
"push": "pre-push-safety-net Req 5.5/7.3: experimental push contract package; no hook installed",
"push.contracts": "pre-push-safety-net Req 5.5/7.3: immutable comparisons and independent result statuses; runner pending",
"model.specs": "model-grounded-detection Req 7.2-7.4: the security vocabularies the CPG queries are parameterised by",
"model.endpoint": "model-grounded-detection Req 8: the judge client, wired by model/judge.py in group 3",
"model.audit": "model-grounded-detection Req 3: this classifier; run by the maintainer, not by the scan",
+115
View File
@@ -0,0 +1,115 @@
"""Language-agnostic change evidence and exact ranker scope (pre-push Req 7.3).
These are integration contracts; constructing them neither selects nor executes work.
Revision identities are opaque to the engine. The snapshot adapter resolves Git objects.
"""
from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass
from typing import Literal
from openultrasast.contracts import Contract
@dataclass(frozen=True)
class ExecutionBudget(Contract):
"""One absolute monotonic deadline, with a distinct bounded cleanup allowance.
Process-local: do not carry the absolute clock value across process restarts.
Manifests should record elapsed stage costs and configured durations instead.
"""
deadline_monotonic: float
cancellation_allowance_seconds: float
def __post_init__(self) -> None:
super().__post_init__()
if self.deadline_monotonic < 0 or self.cancellation_allowance_seconds <= 0:
raise ValueError("deadline must be nonnegative and cancellation allowance positive")
@dataclass(frozen=True)
class QuestionIdentity(Contract):
unit: str
language: str
path: str
function: str | None
family: str
@property
def question_id(self) -> str:
encoded = json.dumps(self.to_payload(), sort_keys=True, separators=(",", ":")).encode()
return hashlib.sha256(encoded).hexdigest()
@dataclass(frozen=True)
class ChangedSpan(Contract):
path: str
start_line: int
end_line: int
side: Literal["base", "head"]
def __post_init__(self) -> None:
super().__post_init__()
if self.start_line < 1 or self.end_line < self.start_line:
raise ValueError("changed spans require positive inclusive ordered lines")
@dataclass(frozen=True)
class PathRename(Contract):
base_path: str
head_path: str
@dataclass(frozen=True)
class AffectedRelationship(Contract):
source: QuestionIdentity
target: QuestionIdentity
kind: str
evidence: tuple[str, ...]
@dataclass(frozen=True)
class ChangeContext(Contract):
base_revision: str | None
head_revision: str
changed_paths: tuple[str, ...]
deleted_paths: tuple[str, ...]
spans: tuple[ChangedSpan, ...]
renames: tuple[PathRename, ...]
declaration_paths: tuple[str, ...]
relationships: tuple[AffectedRelationship, ...]
unresolved_boundaries: tuple[str, ...]
@dataclass(frozen=True)
class RankedQuestion(Contract):
identity: QuestionIdentity
priority: float
evidence: tuple[str, ...]
@dataclass(frozen=True)
class DeferredQuestion(Contract):
identity: QuestionIdentity
reason: str
evidence: tuple[str, ...]
@dataclass(frozen=True)
class ScopeDecision(Contract):
ranker_version: str
ranking_mode: str
# Explicit census assertion: an empty list by itself is never proof of no work.
population_complete: bool
selected: tuple[RankedQuestion, ...]
deferred: tuple[DeferredQuestion, ...]
unresolved_boundaries: tuple[str, ...]
def __post_init__(self) -> None:
super().__post_init__()
identities = [item.identity for item in self.selected] + [item.identity for item in self.deferred]
if len(set(identities)) != len(identities):
raise ValueError("question identities must be unique across selected and deferred scope")