This is an automated email from the ASF dual-hosted git repository.
terrymanu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new b2f81b2ff64 Improve review evidence validation and coverage checks
(#40169)
b2f81b2ff64 is described below
commit b2f81b2ff64ce3d283540dbbc590e2d1ca151823
Author: Liang Zhang <[email protected]>
AuthorDate: Tue Oct 13 13:50:41 2026 +0800
Improve review evidence validation and coverage checks (#40169)
---
.codex/harness/agents/cases.toml | 40 ++
.codex/skills/review-pr/SKILL.md | 28 ++
.../review-pr/references/review-corrections.md | 6 +-
.codex/skills/review-pr/scripts/review_ledger.py | 379 ++++++++++++++++--
.../skills/review-pr/tests/test_review_ledger.py | 432 +++++++++++++++++++--
5 files changed, 813 insertions(+), 72 deletions(-)
diff --git a/.codex/harness/agents/cases.toml b/.codex/harness/agents/cases.toml
index 7331a946dc4..8a14e86d4f1 100644
--- a/.codex/harness/agents/cases.toml
+++ b/.codex/harness/agents/cases.toml
@@ -3555,3 +3555,43 @@ max_summary_chars = 107
critical = true
policy_binding_profile = "gen-ut"
policy_assertions = [{source = "gen-ut", statements = ["Edit only
`<ResolvedTestFileSet>` under `src/test/java` or `src/test/resources`.", "Do
not edit production, generated output, unrelated tests, or another file type.",
"Capture `Task scope baseline` after resolving the file set and before
editing.", "Any candidate-relevant changed path outside that set is a failure,
including further mutation of a path already dirty at task start.", "Exclude
only untracked or ignored reproducible ver [...]
+
+[[cases]]
+id = "same_review_basis_has_same_assessment_across_candidate_types"
+group = "review-pr assessment basis"
+description = "Keeps equal local and public content under the same applicable
task requirement and its proven exception on one assessment."
+phase = "read-only review-basis assessment"
+ordinary_loop = false
+profile = "review-pr"
+semantic_policy_assertions = true
+prompt = "Answer this self-contained read-only review-policy question using
only the supplied facts and bound rules; do not inspect files, run checks, edit
anything or produce a Formal Review report. A and B review the same effective
patch from the same original baseline, with equal content, delta, authoritative
files, accepted requirements, Code Correctness focus and admissible evidence. A
is a local candidate with pending edits; B is the committed public candidate,
so their Git HEAD la [...]
+decision = "proceed"
+required_actions = []
+allowed_actions = []
+forbidden_actions = ["edit_code", "edit_non_code", "run_local_checks",
"measure_performance", "run_pre_handoff_review", "propose_commit_message",
"mutate_git", "mutate_remote"]
+required_reasons = ["read_only_request"]
+required_summary_terms = ["A=change-request", "B=change-request",
"C=mergeable", "D=mergeable"]
+forbidden_summary_terms = ["A=continue-review", "A=mergeable",
"A=review-incomplete", "A=needs-discussion", "B=continue-review",
"B=mergeable", "B=review-incomplete", "B=needs-discussion",
"C=continue-review", "C=review-incomplete", "C=change-request",
"C=needs-discussion", "D=continue-review", "D=review-incomplete",
"D=change-request", "D=needs-discussion"]
+critical = true
+policy_binding_profile = "review-pr"
+policy_assertions = [{source = "review-pr", statements = ["Record each
applicable requirement's source, content version, scope and valid exceptions in
readable basis source material; a task-only requirement applies only to its
authorized task candidate and must not become a universal repository or
public-PR rule.", "Keep effective content and delta identity separate from Git
revision lineage so equal local and public candidates retain one
code-correctness basis.", "Two reviews with the s [...]
+
+[[cases]]
+id = "unchanged_candidate_corrects_proven_prior_review_miss"
+group = "review-pr assessment corrections"
+description = "Corrects a proven earlier miss on unchanged inputs after a
complete independent current assessment."
+phase = "read-only review-correction assessment"
+ordinary_loop = false
+profile = "review-pr"
+semantic_policy_assertions = true
+prompt = "Answer this self-contained read-only review-correction question
using only the supplied facts and bound rules; do not inspect files, run
checks, edit anything or produce a Formal Review report. An earlier
complete-looking Formal Review returned Mergeable. The current effective
candidate content and delta, original baseline, applicable requirements and
exceptions, Code Correctness focus and admissible evidence are identical. No
commit, repair, new requirement or external fact in [...]
+decision = "proceed"
+required_actions = []
+allowed_actions = []
+forbidden_actions = ["edit_code", "edit_non_code", "run_local_checks",
"measure_performance", "run_pre_handoff_review", "propose_commit_message",
"mutate_git", "mutate_remote"]
+required_reasons = ["read_only_request"]
+required_summary_terms = ["origin=previous-review-miss",
"current=change-request", "prior-verdict=not-evidence"]
+forbidden_summary_terms = ["origin=introduced-by-latest-commits",
"origin=exposed-by-previous-fix", "origin=changed-review-basis",
"current=continue-review", "current=mergeable", "current=review-incomplete",
"current=needs-discussion", "prior-verdict=evidence"]
+critical = true
+policy_binding_profile = "review-pr"
+policy_assertions = [{source = "review-pr", statements = ["For a proven
previous-review miss, contradictory assessments on an unchanged basis, a
triggered high-risk review criterion, or an explicit user request for
independent review, obtain an independent read-only assessment through this
workflow and the Completion Gate using the current basis and complete effective
candidate without the previous verdict or an expected finding count.", "Never
use a previous Formal Review result as\nevi [...]
diff --git a/.codex/skills/review-pr/SKILL.md b/.codex/skills/review-pr/SKILL.md
index 082448f7ed6..37ec75e104e 100644
--- a/.codex/skills/review-pr/SKILL.md
+++ b/.codex/skills/review-pr/SKILL.md
@@ -44,6 +44,9 @@ Resolve one review basis before discovery: the effective
candidate, applicable r
confirmed findings consolidated by fix boundary, needs-discussion conditions,
incomplete-evidence gaps, and Completion Gate state.
+Record each applicable requirement's source, content version, scope and valid
exceptions in readable basis source material; a task-only requirement applies
only to its authorized task candidate and must not become a universal
repository or public-PR rule.
+Keep effective content and delta identity separate from Git revision lineage
so equal local and public candidates retain one code-correctness basis.
+
Output mode must not affect candidate discovery, proof, classification,
coverage, or convergence. Never use a previous Formal Review result as
evidence or as a conclusion to match. Treat previous public findings only as
@@ -174,6 +177,8 @@ Apply this workflow to the canonical review basis without
letting output mode or
Start every Formal Review round with fresh coverage of the complete effective
candidate and its affected behavior and contracts, preserving the original task
baseline and attributed scope.
Reuse only facts and verification results whose current-candidate validity is
established; do not inherit reviewed-file statuses, discovery passes, proof
conclusions, or convergence results.
+For a proven previous-review miss, contradictory assessments on an unchanged
basis, a triggered high-risk review criterion, or an explicit user request for
independent review, obtain an independent read-only assessment through this
workflow and the Completion Gate using the current basis and complete effective
candidate without the previous verdict or an expected finding count.
+
1. Complete `Repository Code Policy References`, then establish the
authoritative effective-candidate scope and applicable requirements.
2. Complete the Problem and Project Commitment Gate and establish the accepted
behavior boundary.
3. Confirm the selected review focus and admissible evidence.
@@ -202,10 +207,33 @@ Apply the Completion Gate to every Formal Review and to a
discussion reply that
- Use `scripts/build_review_inventory.py --format json` when local refs are
available; its Markdown is only a bounded summary.
- Use `scripts/review_ledger.py` for multi-file, high-risk, or omission-prone
review; it accounts for review state but does not judge correctness.
- Initialize a new ledger for each round that requires one, from the full
current inventory with pending file statuses and no inherited review passes.
+- Initialize with `--review-basis <JSON>` containing the selected focus and
readable sources identified by role and origin, so the ledger binds their
actual content versions to the effective candidate, original baseline and
authoritative files.
+- Record each substantive changed block's behavior, applicable risk axes and
requirement origins, strongest counter-evidence and current source anchors with
`mark-file --coverage <JSON>`, and attach finding anchors with `add-finding
--anchors <JSON>`.
+- Rebuild ledgers with an older schema or stale inputs without inheriting
their completion state.
+- Validate the current basis and coverage before selecting a result; a passing
ledger proves mechanical coverage and evidence freshness, while the reviewer
must still establish semantic proof under the existing gates.
- For a standalone local candidate with index or working-tree changes, provide
the same exact task-path file to both scripts with `--candidate-files <path>`.
The scripts resolve those paths from the computed merge-base through the
working tree, include matching untracked files, and fail when a listed path is
unchanged.
- Mutate one ledger sequentially; do not run ledger commands concurrently.
- Keep ledger data private and remove only the exact current-review ledger.
+Use these JSON shapes in local files passed to the arguments, replacing the
example values with current evidence and block IDs from the initialized ledger:
+Sources use `rule`, `requirement` or `evidence` roles; `--anchors` takes an
anchor-array file shaped like `anchors` below.
+
+```json
+{"focus": "Code Correctness Review", "sources": [{"role": "rule", "origin":
"CODE_OF_CONDUCT.md", "path": "CODE_OF_CONDUCT.md"}]}
+```
+
+```json
+[
+ {
+ "block": "<block ID>", "behavior": "<verified behavior>", "risk_axes":
["contracts"],
+ "requirements": ["CODE_OF_CONDUCT.md"], "counter_evidence": "<contrary
scenario checked and result>",
+ "anchors": [{"side": "candidate", "path": "<changed repository-relative
path>", "start_line": 1, "end_line": 2}]
+ }
+]
+```
+
+Use `base` for original content and `candidate` for effective content; text
blocks need an anchor overlapping their changed lines, whole-object binary,
mode or rename changes omit the line bounds, and `source` anchors require the
declared source's matching origin and path.
+
If the gate cannot pass because a gap satisfies the Review Incomplete Proof
Gate, return `Review Incomplete` in Formal Review and state the incomplete
evidence without a formal verdict in a discussion reply.
When confirmed blockers coexist with a proven gap that could hide more
blockers, list them as confirmed partial facts but do not present them as the
complete change-request set.
diff --git a/.codex/skills/review-pr/references/review-corrections.md
b/.codex/skills/review-pr/references/review-corrections.md
index 43f21f4c58d..30fc1dced44 100644
--- a/.codex/skills/review-pr/references/review-corrections.md
+++ b/.codex/skills/review-pr/references/review-corrections.md
@@ -25,6 +25,9 @@ Use public previous-round feedback only for public PR
conclusions.
For standalone local corrections, inherit the main Skill's task baseline,
attributed local delta, authorized evidence boundary and selected mode's output
contract.
Do not expose private review notes, local chat iterations, raw inventory, or
internal scratch state.
+Before changing an overall conclusion, compare the prior and current effective
candidates, requirement sources and versions, applicable exceptions, review
focus, evidence identities and finding sets; candidate-type or Git revision
labels and a different previous verdict do not establish a changed basis.
+Correct a proven earlier error rather than preserving an unsupported previous
verdict for consistency.
+
For every prior public finding, classify the latest head as:
- Fixed.
@@ -51,8 +54,9 @@ Classify every finding first reported after an earlier formal
review:
reachable or materially changed the evidence.
- `Missed in the previous review`: the problem was already present in the
previously reviewed head.
+- `Established by a changed review basis`: a proven applicable requirement,
selected focus or decisive fact changed and newly supports a finding that the
prior basis did not support.
-Prove the classification from the previous and current effective candidates.
+Prove the classification from the previous and current effective candidates
and their applicable requirements, focus and decisive facts.
Do not describe an earlier-candidate problem as newly introduced.
For a previous-review miss, acknowledge it directly, rerun the full Completion
Gate, and return the complete remaining blocker set rather than publishing only
the late finding.
diff --git a/.codex/skills/review-pr/scripts/review_ledger.py
b/.codex/skills/review-pr/scripts/review_ledger.py
index 6361cfa49be..913c70ba1cc 100755
--- a/.codex/skills/review-pr/scripts/review_ledger.py
+++ b/.codex/skills/review-pr/scripts/review_ledger.py
@@ -21,24 +21,28 @@
from __future__ import annotations
import argparse
+import difflib
+import hashlib
import json
import os
import re
import shutil
+import stat
+import subprocess
import sys
import tempfile
import time
from collections import Counter
-from pathlib import Path
+from pathlib import Path, PurePosixPath
from typing import Any, Iterable
from review_common import (
- categorize, compare_github_files, final_paths, get_repo_root,
parse_name_status, resolve_candidate_changes, run_git,
+ categorize, compare_github_files, get_repo_root, parse_name_status,
resolve_candidate_changes, run_git,
)
LEDGER_KIND = "review-pr-coverage-ledger"
-LEDGER_VERSION = 3
+LEDGER_VERSION = 4
LEDGER_FILE_NAME = "ledger.json"
FILE_STATUSES = frozenset({"pending", "reviewed", "churn-only",
"test-only-reviewed", "not-applicable", "blocked"})
FINAL_FILE_STATUSES = FILE_STATUSES - {"pending"}
@@ -46,6 +50,8 @@ SUBSTANTIVE_FILE_STATUSES = frozenset({"reviewed",
"test-only-reviewed"})
EXPLAINED_FILE_STATUSES = frozenset({"churn-only", "not-applicable"})
FINDING_STATUSES = frozenset({"candidate", "confirmed", "withdrawn",
"review-incomplete-gap", "non-blocking", "out-of-scope"})
PASS_FOCUSES = ("root-cause", "blast-radius", "tests-runtime", "convergence")
+REVIEW_FOCUSES = ("Code Correctness Review", "Mergeability Review", "CI
Review")
+SOURCE_ROLES = frozenset({"rule", "requirement", "evidence"})
def ledger_root() -> Path:
@@ -93,21 +99,38 @@ def validate_schema(ledger: object) -> None:
"findings": list,
"passes": list,
"review_revision": int,
+ "basis": dict,
}
if any(not isinstance(ledger.get(key), expected_type) for key,
expected_type in required_types.items()):
raise RuntimeError("Malformed review coverage ledger")
if not isinstance(ledger["scope"].get("github_files", {}), dict):
raise RuntimeError("Malformed review coverage ledger scope")
+ scope_types = {"repo_root": str, "base_sha": str, "head_sha": str,
"merge_base": str, "head_ref": str, "changed_file_count": int,
+ "candidate_files": dict, "review_basis_file": str}
+ if any(not isinstance(ledger["scope"].get(key), value) for key, value in
scope_types.items()):
+ raise RuntimeError("Malformed review coverage ledger scope")
+ basis_types = {"focus": str, "baseline": dict, "delta": list, "sources":
list}
+ if any(not isinstance(ledger["basis"].get(key), value) for key, value in
basis_types.items()):
+ raise RuntimeError("Malformed review coverage ledger basis")
+ candidate_files = ledger["scope"]["candidate_files"]
+ if not isinstance(candidate_files.get("provided"), bool) or not
isinstance(candidate_files.get("path"), str):
+ raise RuntimeError("Malformed review coverage ledger candidate scope")
+ if any(not has_text(ledger["basis"]["baseline"].get(key)) for key in
("commit", "tree")):
+ raise RuntimeError("Malformed review coverage ledger original
baseline")
+ for entry in ledger["basis"]["delta"]:
+ if not isinstance(entry, dict) or not has_text(entry.get("path")) or
any(not isinstance(entry.get(key), dict) for key in ("base", "candidate")):
+ raise RuntimeError("Malformed review coverage ledger delta")
entry_schemas = (
("file", ledger["files"], {
"path": str, "status": str, "clusters": list, "risk_axes": list,
"findings": list,
+ "base": dict, "candidate": dict, "blocks": list, "coverage": list,
}),
("finding", ledger["findings"], {
"id": str, "status": str, "origin": str, "fix_boundary": str,
"evidence": list, "full_path": list,
- "counter_evidence": list, "necessity": str, "scope_proof": str,
"files": list, "reason": str,
+ "counter_evidence": list, "necessity": str, "scope_proof": str,
"files": list, "reason": str, "anchors": list,
}),
("pass", ledger["passes"], {
- "focus": str, "new_findings": int, "review_revision": int,
+ "focus": str, "new_findings": int, "review_revision": int,
"basis_fingerprint": str,
}),
)
for entry_name, entries, schema in entry_schemas:
@@ -120,6 +143,23 @@ def validate_schema(ledger: object) -> None:
for entry in entries:
if any(not all(isinstance(value, str) for value in entry[field])
for field in fields):
raise RuntimeError("Malformed review coverage ledger list
entry")
+ for entry in ledger["files"]:
+ if "old_path" not in entry or entry["old_path"] is not None and not
isinstance(entry["old_path"], str):
+ raise RuntimeError("Malformed review coverage ledger file entry")
+ for block in entry["blocks"]:
+ fields = {"id": str, "kind": str, "base_path": str,
"candidate_path": str,
+ "base_start": int, "base_count": int, "candidate_start":
int, "candidate_count": int}
+ if not isinstance(block, dict) or any(not
isinstance(block.get(key), value) for key, value in fields.items()):
+ raise RuntimeError("Malformed review coverage ledger block")
+ for coverage in entry["coverage"]:
+ validate_coverage_shape(coverage, captured=True)
+ for finding in ledger["findings"]:
+ for anchor in finding["anchors"]:
+ validate_anchor_shape(anchor, captured=True)
+ for source in ledger["basis"]["sources"]:
+ validate_source_shape(source)
+ if any(not has_text(source.get(key)) for key in ("mode", "sha256")):
+ raise RuntimeError("Malformed review coverage ledger source
snapshot")
def read_ledger(value: str | Path) -> dict[str, Any]:
@@ -162,6 +202,238 @@ def has_text_entries(values: Iterable[object]) -> bool:
return bool(entries) and all(has_text(each) for each in entries)
+def digest(value: Any) -> str:
+ return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",",
":")).encode()).hexdigest()
+
+
+def validate_source_shape(source: Any) -> None:
+ if not isinstance(source, dict) or any(not has_text(source.get(key)) for
key in ("role", "origin", "path")) or source["role"] not in SOURCE_ROLES:
+ raise RuntimeError("Review basis sources require role, origin, and a
readable material path")
+
+
+def validate_anchor_shape(anchor: Any, captured: bool = False) -> None:
+ if not isinstance(anchor, dict) or anchor.get("side") not in {"base",
"candidate", "source"} or not has_text(anchor.get("path")):
+ raise RuntimeError("Source anchors require side and path")
+ if "source" == anchor["side"] and not has_text(anchor.get("origin")):
+ raise RuntimeError("Source anchors require an explicit declared
origin")
+ if ("start_line" in anchor) != ("end_line" in anchor):
+ raise RuntimeError("Source anchors require both line bounds or
neither")
+ for key in ("start_line", "end_line"):
+ if key in anchor and type(anchor[key]) is not int:
+ raise RuntimeError("Source anchor line bounds must be integers")
+ if captured and any(not has_text(anchor.get(key)) for key in ("sha256",
"mode", "range_sha256")):
+ raise RuntimeError("Captured source anchors require content
fingerprints")
+ if captured and any(key not in anchor for key in ("start_line",
"end_line")):
+ raise RuntimeError("Captured source anchors require normalized line
bounds")
+
+
+def validate_coverage_shape(coverage: Any, captured: bool = False) -> None:
+ if not isinstance(coverage, dict) or any(not has_text(coverage.get(key))
for key in ("block", "behavior", "counter_evidence")):
+ raise RuntimeError("Block coverage requires block, behavior, and
counter-evidence")
+ if any(not isinstance(coverage.get(key), list) or not
has_text_entries(coverage[key]) for key in ("risk_axes", "requirements")):
+ raise RuntimeError("Block coverage requires risk axes and applicable
requirement origins")
+ if not isinstance(coverage.get("anchors"), list) or not
coverage["anchors"]:
+ raise RuntimeError("Block coverage requires real source anchors")
+ for anchor in coverage["anchors"]:
+ validate_anchor_shape(anchor, captured=captured)
+
+
+def read_object(repo_root: Path, scope: dict[str, Any], side: str, path: str)
-> tuple[dict[str, Any], bytes | None]:
+ if "source" != side and (PurePosixPath(path).is_absolute() or ".." in
PurePosixPath(path).parts or "." == path):
+ raise RuntimeError(f"Invalid repository-relative source anchor path:
{path}")
+ if "base" == side or "candidate" == side and not
scope["candidate_files"]["provided"]:
+ ref = scope["merge_base"] if "base" == side else scope["head_ref"]
+ process = subprocess.run(["git", "ls-tree", "-z", ref, "--", path],
cwd=repo_root, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+ env={**os.environ, "GIT_LITERAL_PATHSPECS":
"1"}, check=False)
+ if process.returncode:
+ raise RuntimeError(f"Cannot read {side} object: {path}")
+ objects = [each.split(b"\t", 1) for each in
process.stdout.split(b"\0") if each]
+ entry = next((value for metadata, value in objects if value.decode()
== path), None)
+ if entry is None:
+ return {"mode": None, "sha256": None}, None
+ metadata = next(metadata for metadata, value in objects if value ==
entry).decode().split()
+ mode, object_type, object_id = metadata
+ if "tree" == object_type:
+ return {"mode": None, "sha256": None}, None
+ if "commit" == object_type:
+ content = object_id.encode()
+ else:
+ blob = subprocess.run(["git", "cat-file", "blob", object_id],
cwd=repo_root, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False)
+ if blob.returncode:
+ raise RuntimeError(f"Cannot read {side} object bytes: {path}")
+ content = blob.stdout
+ else:
+ source_path = Path(path) if Path(path).is_absolute() else repo_root /
path
+ try:
+ metadata = source_path.stat() if "source" == side else
source_path.lstat()
+ except (FileNotFoundError, NotADirectoryError):
+ return {"mode": None, "sha256": None}, None
+ if stat.S_ISLNK(metadata.st_mode):
+ mode, content = "120000", os.fsencode(os.readlink(source_path))
+ elif stat.S_ISDIR(metadata.st_mode) and "candidate" == side:
+ index = subprocess.run(["git", "ls-files", "--stage", "-z", "--",
path], cwd=repo_root, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+ env={**os.environ, "GIT_LITERAL_PATHSPECS":
"1"}, check=False)
+ if index.returncode:
+ raise RuntimeError(f"Cannot determine candidate Git type:
{path}")
+ entries = [each.split(b"\t", 1) for each in
index.stdout.split(b"\0") if each]
+ if not any(value.decode() == path and metadata.split()[0] ==
b"160000" for metadata, value in entries):
+ return {"mode": None, "sha256": None}, None
+ if get_repo_root(source_path).resolve() != source_path.resolve():
+ raise RuntimeError(f"Gitlink checkout must have its own
repository: {path}")
+ if run_git(["status", "--porcelain", "--untracked-files=no"],
source_path).strip():
+ raise RuntimeError(f"Gitlink checkout has staged or unstaged
tracked changes: {path}")
+ mode, content = "160000", run_git(["rev-parse", "HEAD"],
source_path).strip().encode()
+ else:
+ mode = "100755" if metadata.st_mode & stat.S_IXUSR else "100644"
+ content = source_path.read_bytes()
+ return {"mode": mode, "sha256": hashlib.sha256(content).hexdigest()},
content
+
+
+def changed_blocks(path: str, base_path: str, base: dict[str, Any], candidate:
dict[str, Any], old: bytes | None, new: bytes | None) -> list[dict[str, Any]]:
+ result = []
+ try:
+ old_lines, new_lines = (old or
b"").decode("utf-8").splitlines(keepends=True), (new or
b"").decode("utf-8").splitlines(keepends=True)
+ opaque = b"\0" in (old or b"") or b"\0" in (new or b"") or
base["mode"] in {"120000", "160000"} or candidate["mode"] in {"120000",
"160000"}
+ except UnicodeDecodeError:
+ old_lines, new_lines, opaque = [], [], True
+ if old != new and opaque:
+ result.append({"kind": "binary", "base_start": 0, "base_count": 0,
"candidate_start": 0, "candidate_count": 0})
+ elif old != new:
+ for tag, old_start, old_end, new_start, new_end in
difflib.SequenceMatcher(None, old_lines, new_lines).get_opcodes():
+ if "equal" != tag:
+ result.append({"kind": "text", "base_start": old_start + 1,
"base_count": old_end - old_start,
+ "candidate_start": new_start + 1,
"candidate_count": new_end - new_start})
+ if base_path != path:
+ result.append({"kind": "rename", "base_start": 0, "base_count": 0,
"candidate_start": 0, "candidate_count": 0})
+ if base["mode"] and candidate["mode"] and base["mode"] !=
candidate["mode"]:
+ result.append({"kind": "mode", "base_start": 0, "base_count": 0,
"candidate_start": 0, "candidate_count": 0})
+ if not result and base != candidate:
+ result.append({"kind": "object", "base_start": 0, "base_count": 0,
"candidate_start": 0, "candidate_count": 0})
+ for block in result:
+ block.update({"base_path": base_path, "candidate_path": path})
+ block["id"] = digest({**block, "base": base, "candidate": candidate})
+ return result
+
+
+def capture_basis(scope: dict[str, Any]) -> tuple[dict[str, Any],
list[dict[str, Any]]]:
+ repo_root = Path(scope["repo_root"])
+ configuration =
json.loads(Path(scope["review_basis_file"]).read_text(encoding="utf-8"))
+ if not isinstance(configuration, dict) or configuration.get("focus") not
in REVIEW_FOCUSES or not isinstance(configuration.get("sources"), list):
+ raise RuntimeError("Review basis requires a supported focus and
declared sources")
+ sources = []
+ for source in configuration["sources"]:
+ validate_source_shape(source)
+ snapshot, content = read_object(repo_root, scope, "source",
source["path"])
+ if content is None:
+ raise RuntimeError(f"Review basis source is unavailable:
{source['origin']}")
+ sources.append({**source, **snapshot})
+ if not any(each["role"] in {"rule", "requirement"} for each in sources) or
len(sources) != len({each["origin"] for each in sources}):
+ raise RuntimeError("Review basis requires unambiguous applicable rule
or requirement sources")
+ candidate_files = scope["candidate_files"]
+ changes = resolve_candidate_changes(repo_root, scope["merge_base"],
candidate_files["path"]) if candidate_files["provided"] else parse_name_status(
+ run_git(["diff", "--name-status",
f"{scope['merge_base']}..{scope['head_ref']}"], repo_root))
+ files = []
+ paths = sorted({path for change in changes for path in (change.old_path or
change.path, change.path)})
+ objects = {(side, path): read_object(repo_root, scope, side, path) for
path in paths for side in ("base", "candidate")}
+ for change in changes:
+ base_path = change.old_path or change.path
+ base, old = objects[("base", base_path)]
+ candidate, new = objects[("candidate", change.path)]
+ files.append({"path": change.path, "old_path": change.old_path,
"git_status": change.status, "base": base, "candidate": candidate,
+ "blocks": changed_blocks(change.path, base_path, base,
candidate, old, new)})
+ delta = [{"path": path, "base": objects[("base", path)][0], "candidate":
objects[("candidate", path)][0]} for path in paths]
+ return {"focus": configuration["focus"], "baseline": {"commit":
scope["merge_base"], "tree": run_git(["rev-parse",
f"{scope['merge_base']}^{{tree}}"], repo_root).strip()},
+ "sources": sources, "delta": delta}, files
+
+
+def capture_anchor(anchor: dict[str, Any], ledger: dict[str, Any]) ->
dict[str, Any]:
+ if "source" == anchor["side"]:
+ source = next((each for each in ledger["basis"]["sources"] if
each["origin"] == anchor["origin"] and each["path"] == anchor["path"]), None)
+ if source is None:
+ raise RuntimeError("Source anchor does not match a declared
material and origin")
+ snapshot, content = read_object(Path(ledger["scope"]["repo_root"]),
ledger["scope"], anchor["side"], anchor["path"])
+ if content is None:
+ raise RuntimeError(f"Source anchor is unavailable:
{anchor['side']}:{anchor['path']}")
+ start, end = anchor.get("start_line", 0), anchor.get("end_line", 0)
+ if start or end:
+ try:
+ lines = content.decode("utf-8").splitlines(keepends=True)
+ except UnicodeDecodeError as ex:
+ raise RuntimeError("Line anchors require UTF-8 text; use a
whole-object anchor for binary evidence") from ex
+ if start < 1 or end < start or end > len(lines):
+ raise RuntimeError("Source anchor line range is outside the actual
source")
+ selected = "".join(lines[start - 1:end]).encode("utf-8")
+ else:
+ selected = content
+ return {**anchor, **snapshot, "start_line": start, "end_line": end,
"range_sha256": hashlib.sha256(selected).hexdigest()}
+
+
+def anchor_covers_block(anchor: dict[str, Any], block: dict[str, Any]) -> bool:
+ side = anchor["side"]
+ if side not in {"base", "candidate"} or anchor["path"] !=
block[f"{side}_path"]:
+ return False
+ if "text" != block["kind"]:
+ return not anchor["start_line"] and not anchor["end_line"]
+ count, start = block[f"{side}_count"], block[f"{side}_start"]
+ return count > 0 and anchor["start_line"] > 0 and anchor["start_line"] <=
start + count - 1 and anchor["end_line"] >= start
+
+
+def coverage_errors(entry: dict[str, Any], requirement_origins: set[str]) ->
list[str]:
+ errors = []
+ block_ids = {each["id"] for each in entry["blocks"]}
+ covered_blocks = {each["block"] for each in entry["coverage"]}
+ if covered_blocks - block_ids:
+ errors.append(f"{entry['path']} has unknown changed-block coverage")
+ if entry["status"] in SUBSTANTIVE_FILE_STATUSES and block_ids -
covered_blocks:
+ errors.append(f"{entry['path']} has uncovered changed blocks:
{len(block_ids - covered_blocks)}")
+ for coverage in entry["coverage"]:
+ if set(coverage["requirements"]) - requirement_origins:
+ errors.append(f"{entry['path']} block coverage references unknown
applicable requirement origins")
+ block = next((each for each in entry["blocks"] if each["id"] ==
coverage["block"]), None)
+ if block is not None and not any(anchor_covers_block(each, block) for
each in coverage["anchors"]):
+ errors.append(f"{entry['path']} block coverage has no anchor at
the actual change")
+ return errors
+
+
+def all_anchors(ledger: dict[str, Any]) -> Iterable[dict[str, Any]]:
+ for entry in ledger["files"]:
+ for coverage in entry["coverage"]:
+ yield from coverage["anchors"]
+ for finding in ledger["findings"]:
+ yield from finding["anchors"]
+
+
+def basis_fingerprint(ledger: dict[str, Any]) -> str:
+ basis = ledger["basis"]
+ scope_paths = {each["path"] for each in basis["delta"]}
+ evidence = {(each["side"], each["path"], each["mode"], each["sha256"]) for
each in all_anchors(ledger)
+ if "source" != each["side"] and each["path"] not in
scope_paths}
+ return digest({"focus": basis["focus"], "baseline":
basis["baseline"]["tree"], "delta": basis["delta"],
+ "sources": sorted((each["role"], each["origin"],
each["mode"], each["sha256"]) for each in basis["sources"]),
+ "evidence": sorted(evidence)})
+
+
+def validate_current_basis(ledger: dict[str, Any]) -> list[str]:
+ current, files = capture_basis(ledger["scope"])
+ errors = []
+ if current != ledger["basis"]:
+ errors.append("Effective candidate, original baseline, focus, or
declared review sources changed; initialize a new ledger")
+ snapshots = [{key: each[key] for key in ("path", "old_path", "base",
"candidate", "blocks")} for each in ledger["files"]]
+ actual = [{key: each[key] for key in ("path", "old_path", "base",
"candidate", "blocks")} for each in files]
+ if snapshots != actual or ledger["scope"]["changed_file_count"] !=
len(files):
+ errors.append("Ledger files, changed-file count, or changed blocks do
not match the effective candidate")
+ if ledger["scope"].get("github_files_file"):
+ comparison = compare_github_files((each["path"] for each in files),
ledger["scope"]["github_files_file"])
+ if comparison != ledger["scope"]["github_files"]:
+ errors.append("Authoritative GitHub file material changed or no
longer matches the ledger")
+ if errors:
+ return errors
+ for anchor in all_anchors(ledger):
+ if capture_anchor(anchor, ledger) != anchor:
+ errors.append(f"Source anchor is stale:
{anchor['side']}:{anchor['path']}")
+ return errors
+
+
def sync_file_findings(ledger: dict[str, Any]) -> None:
for each in ledger["files"]:
each["findings"] = []
@@ -176,8 +448,21 @@ def cmd_init(args: argparse.Namespace) -> int:
head_sha = run_git(["rev-parse", args.head_ref], repo_root).strip()
merge_base = run_git(["merge-base", args.base_ref, args.head_ref],
repo_root).strip()
candidate_files = getattr(args, "candidate_files", None)
- changed_files = resolve_candidate_changes(repo_root, merge_base,
candidate_files) if candidate_files else parse_name_status(
- run_git(["diff", "--name-status", f"{merge_base}..{args.head_ref}"],
repo_root))
+ scope = {
+ "repo": repo_root.name,
+ "repo_root": str(repo_root),
+ "pr": args.pr,
+ "base_ref": args.base_ref,
+ "base_sha": base_sha,
+ "head_ref": args.head_ref,
+ "head_sha": head_sha,
+ "merge_base": merge_base,
+ "candidate_files": {"provided": bool(candidate_files), "path":
str(Path(candidate_files).resolve()) if candidate_files else ""},
+ "review_basis_file": str(Path(args.review_basis).resolve()),
+ "github_files_file": str(Path(args.github_files).resolve()) if
args.github_files else "",
+ }
+ basis, changed_files = capture_basis(scope)
+ scope.update({"changed_file_count": len(changed_files), "github_files":
compare_github_files((each["path"] for each in changed_files),
args.github_files)})
ledger_dir = create_ledger_dir(repo_root, args.pr, head_sha)
ledger_file = ledger_dir / LEDGER_FILE_NAME
now = int(time.time())
@@ -187,27 +472,16 @@ def cmd_init(args: argparse.Namespace) -> int:
"created_at": now,
"updated_at": now,
"review_revision": 0,
- "scope": {
- "repo": repo_root.name,
- "pr": args.pr,
- "base_ref": args.base_ref,
- "base_sha": base_sha,
- "head_ref": args.head_ref,
- "head_sha": head_sha,
- "merge_base": merge_base,
- "changed_file_count": len(changed_files),
- "candidate_files": {"provided": bool(candidate_files)},
- "github_files": compare_github_files(final_paths(changed_files),
args.github_files),
- },
+ "scope": scope,
+ "basis": basis,
"files": [{
- "path": each.path,
- "git_status": each.status,
- "old_path": each.old_path,
- "category": categorize(each.path),
+ **each,
+ "category": categorize(each["path"]),
"status": "pending",
"clusters": [],
"risk_axes": [],
"findings": [],
+ "coverage": [],
"reason": "",
} for each in changed_files],
"findings": [],
@@ -222,11 +496,25 @@ def cmd_init(args: argparse.Namespace) -> int:
def cmd_mark_file(args: argparse.Namespace) -> int:
ledger_file = resolve_ledger_file(args.ledger)
ledger = read_ledger(ledger_file)
+ errors = validate_current_basis(ledger)
+ if errors:
+ raise RuntimeError("; ".join(errors))
entry = find_file_entry(ledger, args.path)
+ if args.coverage:
+ coverage = json.loads(Path(args.coverage).read_text(encoding="utf-8"))
+ if not isinstance(coverage, list):
+ raise RuntimeError("File coverage material must be a list of
reviewed blocks")
+ for block_review in coverage:
+ validate_coverage_shape(block_review)
+ block_review["anchors"] = [capture_anchor(each, ledger) for each
in block_review["anchors"]]
+ entry["coverage"] = coverage
entry["status"] = args.status
unique_extend(entry["clusters"], args.cluster or [])
unique_extend(entry["risk_axes"], args.risk_axis or [])
entry["reason"] = args.reason or ""
+ errors = coverage_errors(entry, {each["origin"] for each in
ledger["basis"]["sources"] if each["role"] in {"rule", "requirement"}})
+ if errors:
+ raise RuntimeError("; ".join(errors))
ledger["review_revision"] += 1
write_ledger(ledger_file, ledger)
print(f"marked={args.path} status={args.status}")
@@ -236,6 +524,14 @@ def cmd_mark_file(args: argparse.Namespace) -> int:
def cmd_add_finding(args: argparse.Namespace) -> int:
ledger_file = resolve_ledger_file(args.ledger)
ledger = read_ledger(ledger_file)
+ errors = validate_current_basis(ledger)
+ if errors:
+ raise RuntimeError("; ".join(errors))
+ anchors = json.loads(Path(args.anchors).read_text(encoding="utf-8")) if
args.anchors else []
+ if not isinstance(anchors, list):
+ raise RuntimeError("Finding anchors material must be a list")
+ for anchor in anchors:
+ validate_anchor_shape(anchor)
finding = {
"id": args.id,
"status": args.status,
@@ -248,6 +544,7 @@ def cmd_add_finding(args: argparse.Namespace) -> int:
"scope_proof": args.scope_proof or "",
"files": args.file or [],
"reason": args.reason or "",
+ "anchors": [capture_anchor(each, ledger) for each in anchors],
}
existing = next((each for each in ledger["findings"] if args.id ==
each["id"]), None)
if existing:
@@ -264,6 +561,9 @@ def cmd_add_finding(args: argparse.Namespace) -> int:
def cmd_add_pass(args: argparse.Namespace) -> int:
ledger_file = resolve_ledger_file(args.ledger)
ledger = read_ledger(ledger_file)
+ errors = validate_current_basis(ledger)
+ if errors:
+ raise RuntimeError("; ".join(errors))
if 0 > args.new_findings:
raise RuntimeError("New finding count must not be negative")
ledger["passes"].append({
@@ -272,6 +572,7 @@ def cmd_add_pass(args: argparse.Namespace) -> int:
"review_revision": ledger["review_revision"],
"finding_count": len(ledger["findings"]),
"created_at": int(time.time()),
+ "basis_fingerprint": basis_fingerprint(ledger),
})
write_ledger(ledger_file, ledger)
print(f"pass={args.focus} new_findings={args.new_findings}")
@@ -280,7 +581,14 @@ def cmd_add_pass(args: argparse.Namespace) -> int:
def validate_ledger(ledger: dict[str, Any], incomplete_result: bool = False)
-> list[str]:
validate_schema(ledger)
- result: list[str] = []
+ return _validate_ledger(ledger, incomplete_result)
+
+
+def _validate_ledger(ledger: dict[str, Any], incomplete_result: bool = False)
-> list[str]:
+ result = validate_current_basis(ledger)
+ requirement_origins = {each["origin"] for each in
ledger["basis"]["sources"] if each["role"] in {"rule", "requirement"}}
+ for entry in ledger["files"]:
+ result.extend(coverage_errors(entry, requirement_origins))
github_files = ledger["scope"].get("github_files", {})
if github_files.get("provided") and not github_files.get("matched"):
result.append("GitHub file list does not match local triple-dot scope")
@@ -347,6 +655,8 @@ def validate_ledger(ledger: dict[str, Any],
incomplete_result: bool = False) ->
if unknown_files:
result.append(f"{each['id']} references files outside scope: {',
'.join(unknown_files)}")
if "confirmed" == each["status"]:
+ if not each["anchors"]:
+ result.append(f"{each['id']} confirmed finding is missing real
source anchors")
if not has_text(each["origin"]):
result.append(f"{each['id']} confirmed finding is missing
origin")
if not has_text(each["fix_boundary"]):
@@ -368,6 +678,8 @@ def validate_ledger(ledger: dict[str, Any],
incomplete_result: bool = False) ->
missing_fact = "the unavailable fact" if incomplete_result
else "a reason"
result.append(f"{each['id']} incomplete gap is missing
{missing_fact}")
if incomplete_result:
+ if not each["anchors"]:
+ result.append(f"{each['id']} incomplete gap is missing
real unavailability-proof anchors")
if not has_text(each["origin"]):
result.append(f"{each['id']} incomplete gap is missing the
fact source")
if not has_text_entries(each["evidence"]):
@@ -384,7 +696,8 @@ def validate_ledger(ledger: dict[str, Any],
incomplete_result: bool = False) ->
result.append(f"{each['id']} incomplete gap is missing
affected scope files")
else:
result.append(f"{each['id']} requires a Review Incomplete
result")
- pass_foci = {each.get("focus") for each in ledger["passes"]}
+ current_fingerprint = basis_fingerprint(ledger)
+ pass_foci = {each.get("focus") for each in ledger["passes"] if
each["basis_fingerprint"] == current_fingerprint}
missing_passes = [each for each in PASS_FOCUSES if each not in pass_foci]
if missing_passes:
result.append(f"Required review passes missing: {',
'.join(missing_passes)}")
@@ -396,28 +709,30 @@ def validate_ledger(ledger: dict[str, Any],
incomplete_result: bool = False) ->
result.append("Latest convergence pass did not finish with zero
new findings")
if ledger["review_revision"] != final_pass.get("review_revision"):
result.append("Latest convergence pass predates a file or finding
change")
+ if current_fingerprint != final_pass["basis_fingerprint"]:
+ result.append("Latest convergence pass uses a different review
basis")
return result
def cmd_validate(args: argparse.Namespace) -> int:
ledger = read_ledger(args.ledger)
- errors = validate_ledger(ledger)
+ errors = _validate_ledger(ledger)
if errors:
for each in errors:
print(f"ERROR: {each}")
return 1
- print("OK")
+ print("OK: mechanical coverage and evidence are current; correctness
requires Formal Review")
return 0
def cmd_validate_incomplete(args: argparse.Namespace) -> int:
ledger = read_ledger(args.ledger)
- errors = validate_ledger(ledger, incomplete_result=True)
+ errors = _validate_ledger(ledger, incomplete_result=True)
if errors:
for each in errors:
print(f"ERROR: {each}")
return 1
- print("OK")
+ print("OK: mechanical incomplete-evidence record is current;
classification requires Formal Review")
return 0
@@ -430,6 +745,9 @@ def cmd_status(args: argparse.Namespace) -> int:
print(f"findings={dict(sorted(finding_counts.items()))}")
print(f"passes={dict(sorted(pass_counts.items()))}")
print(f"review_revision={ledger['review_revision']}")
+ print(f"basis_fingerprint={basis_fingerprint(ledger)}")
+ for entry in ledger["files"]:
+ print(f"blocks={entry['path']} " + ",".join(each["id"] for each in
entry["blocks"]))
return 0
@@ -453,6 +771,7 @@ def build_parser() -> argparse.ArgumentParser:
init.add_argument("--head-ref", required=True, help="PR head ref")
init.add_argument("--github-files", help="File containing GitHub changed
paths")
init.add_argument("--candidate-files", help="Exact repository-relative
local candidate paths, one per line")
+ init.add_argument("--review-basis", required=True, help="JSON focus and
rule/requirement/evidence sources with role, origin, and material path")
init.set_defaults(func=cmd_init)
mark_file = subparsers.add_parser("mark-file", help="Set one authoritative
file's coverage state")
mark_file.add_argument("--ledger", required=True, help="Ledger file or
directory")
@@ -461,6 +780,7 @@ def build_parser() -> argparse.ArgumentParser:
mark_file.add_argument("--cluster", action="append", help="Behavior
cluster containing this file")
mark_file.add_argument("--risk-axis", action="append", help="Risk axis
covered for this file")
mark_file.add_argument("--reason", help="Reason for a non-substantive file
classification")
+ mark_file.add_argument("--coverage", help="JSON changed-block coverage
with behavior, risk axes, requirements, counter-evidence, and source anchors")
mark_file.set_defaults(func=cmd_mark_file)
finding = subparsers.add_parser("add-finding", help="Add or update one
finding classification")
finding.add_argument("--ledger", required=True, help="Ledger file or
directory")
@@ -475,6 +795,7 @@ def build_parser() -> argparse.ArgumentParser:
finding.add_argument("--scope-proof", help="Why the PR owns the finding")
finding.add_argument("--file", action="append", help="Repository-relative
scope file")
finding.add_argument("--reason", help="Reason for an incomplete or
non-blocking classification")
+ finding.add_argument("--anchors", help="JSON real source anchors
supporting this finding or unavailable-evidence proof")
finding.set_defaults(func=cmd_add_finding)
review_pass = subparsers.add_parser("add-pass", help="Record a required
review pass")
review_pass.add_argument("--ledger", required=True, help="Ledger file or
directory")
diff --git a/.codex/skills/review-pr/tests/test_review_ledger.py
b/.codex/skills/review-pr/tests/test_review_ledger.py
index 8e22d1b7d96..986431d3434 100644
--- a/.codex/skills/review-pr/tests/test_review_ledger.py
+++ b/.codex/skills/review-pr/tests/test_review_ledger.py
@@ -15,11 +15,14 @@
# limitations under the License.
#
+import json
+import os
+import subprocess
import sys
import tempfile
import unittest
from argparse import Namespace
-from contextlib import redirect_stdout
+from contextlib import redirect_stderr, redirect_stdout
from io import StringIO
from pathlib import Path
from unittest.mock import patch
@@ -29,12 +32,50 @@ SCRIPTS_DIR = Path(__file__).resolve().parents[1] /
"scripts"
if str(SCRIPTS_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPTS_DIR))
-import review_common as common
import review_ledger as ledger
class ReviewLedgerTest(unittest.TestCase):
+ def setUp(self):
+ temporary = tempfile.TemporaryDirectory()
+ self.addCleanup(temporary.cleanup)
+ self.private_root = Path(temporary.name)
+ self.repo = self.private_root / "repo"
+ self.repo.mkdir()
+ temporary_root = patch.object(ledger.tempfile, "gettempdir",
return_value=str(self.private_root))
+ temporary_root.start()
+ self.addCleanup(temporary_root.stop)
+ self.source_path = "src/main/java/Foo.java"
+ self.owner_path = "src/main/java/Owner.java"
+ (self.repo / self.source_path).parent.mkdir(parents=True)
+ (self.repo /
self.source_path).write_text("first\nold_first\nthird\nfourth\nfifth\nold_second\nlast\n",
encoding="utf-8")
+ (self.repo / self.owner_path).write_text("Supported owner
contract.\n", encoding="utf-8")
+ (self.repo / "Rules.md").write_text("Use the owning public
boundary.\n", encoding="utf-8")
+ (self.repo / "Requirements.md").write_text("Review this candidate
under the task scope and accepted exceptions.\n", encoding="utf-8")
+ self.git("init", "-q")
+ self.git("add", ".")
+ self.git("commit", "-qm", "Synthetic baseline")
+ self.base_sha = self.git("rev-parse", "HEAD").strip()
+ (self.repo /
self.source_path).write_text("first\nnew_first\nthird\nfourth\nfifth\nnew_second\nlast\n",
encoding="utf-8")
+ self.candidate_files = self.private_root / "candidate-files.txt"
+ self.candidate_files.write_text(self.source_path + "\n",
encoding="utf-8")
+ self.attempt_log = self.private_root / "attempt.log"
+ self.attempt_log.write_text("Synthetic evidence route: unavailable
target version; repository alternatives checked.\n", encoding="utf-8")
+ self.basis_file = self.private_root / "basis.json"
+ self.basis_file.write_text(json.dumps({"focus": "Code Correctness
Review", "sources": [
+ {"role": "rule", "origin": "repository:Rules.md", "path":
"Rules.md"},
+ {"role": "requirement", "origin": "user:synthetic-task", "path":
"Requirements.md"},
+ {"role": "evidence", "origin": "attempt:target-version", "path":
str(self.attempt_log)},
+ ]}), encoding="utf-8")
+
+ def git(self, *arguments):
+ environment = {**os.environ, "GIT_CONFIG_NOSYSTEM": "1",
"GIT_CONFIG_GLOBAL": os.devnull,
+ "GIT_AUTHOR_NAME": "Synthetic Review",
"GIT_COMMITTER_NAME": "Synthetic Review",
+ "GIT_AUTHOR_EMAIL": "[email protected]",
"GIT_COMMITTER_EMAIL": "[email protected]"}
+ return subprocess.run(["git", "-c", "core.hooksPath=" + os.devnull,
*arguments], cwd=self.repo, env=environment,
+ check=True, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, text=True).stdout
+
def test_complete_ledger_is_valid(self):
self.assertEqual([],
ledger.validate_ledger(self.create_valid_ledger()))
@@ -52,6 +93,7 @@ class ReviewLedgerTest(unittest.TestCase):
"scope_proof": " ",
"files": [""],
"reason": "",
+ "anchors": [],
}]
errors = ledger.validate_ledger(actual_ledger)
self.assertIn("F1 confirmed finding is missing origin", errors)
@@ -91,19 +133,11 @@ class ReviewLedgerTest(unittest.TestCase):
self.assertIn("GitHub file list does not match local triple-dot
scope", errors)
def test_init_uses_exact_candidate_files(self):
- changed_files = [common.ChangedFile(status="M", path="AGENTS.md")]
- args = Namespace(repo_root=".", pr="local-task", base_ref="base",
head_ref="head", github_files=None,
- candidate_files="candidate.txt")
- with patch.object(ledger, "get_repo_root",
return_value=Path("/repo")), \
- patch.object(ledger, "run_git", side_effect=["base-sha\n",
"head-sha\n", "merge-base\n"]), \
- patch.object(ledger, "resolve_candidate_changes",
return_value=changed_files) as resolve_candidate_changes, \
- patch.object(ledger, "create_ledger_dir",
return_value=Path("ledger")), \
- patch.object(ledger, "write_ledger") as write_ledger,
redirect_stdout(StringIO()):
- self.assertEqual(0, ledger.cmd_init(args))
- actual = write_ledger.call_args.args[1]
- self.assertEqual({"provided": True},
actual["scope"]["candidate_files"])
- self.assertEqual("AGENTS.md", actual["files"][0]["path"])
- resolve_candidate_changes.assert_called_once_with(Path("/repo"),
"merge-base", "candidate.txt")
+ actual = self.initialize_ledger()
+ self.assertEqual({"provided": True, "path":
str(self.candidate_files.resolve())}, actual["scope"]["candidate_files"])
+ self.assertEqual([self.source_path], [each["path"] for each in
actual["files"]])
+ self.assertEqual(2, len(actual["files"][0]["blocks"]))
+ self.assertEqual(self.base_sha, actual["basis"]["baseline"]["commit"])
def test_confirmed_findings_must_have_distinct_fix_boundaries(self):
actual_ledger = self.create_valid_ledger()
@@ -139,6 +173,7 @@ class ReviewLedgerTest(unittest.TestCase):
"scope_proof": " ",
"files": [""],
"reason": " ",
+ "anchors": [],
}]
errors = ledger.validate_ledger(actual_ledger, incomplete_result=True)
expected_errors = [
@@ -195,32 +230,343 @@ class ReviewLedgerTest(unittest.TestCase):
with self.assertRaisesRegex(RuntimeError, "outside the private
review ledger root"):
ledger.ensure_safe_ledger_file(outside_file)
- @staticmethod
- def create_valid_ledger():
- review_passes = [{
- "focus": focus,
- "new_findings": 0,
- "review_revision": 1,
- } for focus in ledger.PASS_FOCUSES]
- return {
- "kind": ledger.LEDGER_KIND,
- "version": ledger.LEDGER_VERSION,
- "scope": {"github_files": {"provided": False}},
- "files": [{
- "path": "src/main/java/Foo.java",
- "status": "reviewed",
- "clusters": ["core"],
- "risk_axes": ["correctness"],
- "findings": [],
- "reason": "",
- }],
- "findings": [],
- "passes": review_passes,
- "review_revision": 1,
- }
+ def test_old_ledger_version_requires_new_review(self):
+ actual = self.create_valid_ledger()
+ actual["version"] = 3
+ with self.assertRaisesRegex(RuntimeError, "Unsupported or invalid"):
+ ledger.validate_ledger(actual)
+
+ def test_scope_count_must_match_actual_candidate(self):
+ actual = self.create_valid_ledger()
+ actual["scope"]["changed_file_count"] = 2
+ self.assertIn("Ledger files, changed-file count, or changed blocks do
not match the effective candidate", ledger.validate_ledger(actual))
+
+ def test_persisted_nested_shapes_have_cli_diagnostics(self):
+ original = self.create_valid_ledger()
+ cases = [
+ ("scope", "candidate_files", {}, "candidate scope"),
+ ("basis", "baseline", {}, "original baseline"),
+ ("basis", "delta", [{}], "delta"),
+ ("scope", "base_sha", None, "scope"),
+ ("scope", "head_sha", None, "scope"),
+ ]
+ for section, key, value, diagnostic in cases:
+ with self.subTest(key=key):
+ actual = json.loads(json.dumps(original))
+ actual[section][key] = value
+ self.ledger_file.write_text(json.dumps(actual),
encoding="utf-8")
+ output = StringIO()
+ with patch.object(sys, "argv", ["review_ledger.py",
"validate", "--ledger", str(self.ledger_file)]), redirect_stderr(output):
+ self.assertEqual(1, ledger.main())
+ self.assertEqual(f"error: Malformed review coverage ledger
{diagnostic}\n", output.getvalue())
+
+ def test_every_changed_block_requires_coverage(self):
+ actual = self.create_valid_ledger()
+ actual["files"][0]["coverage"].pop()
+ self.assertIn(f"{self.source_path} has uncovered changed blocks: 1",
ledger.validate_ledger(actual))
+
+ def test_explained_file_status_preserves_non_substantive_boundary(self):
+ original = self.create_valid_ledger()
+ for status in ("churn-only", "not-applicable"):
+ with self.subTest(status=status):
+ actual = json.loads(json.dumps(original))
+ actual["files"][0].update({"status": status, "reason":
"Current evidence supports this non-substantive classification.", "coverage":
[]})
+ for review_pass in actual["passes"]:
+ review_pass["basis_fingerprint"] =
ledger.basis_fingerprint(actual)
+ self.assertEqual([], ledger.validate_ledger(actual))
+
+ def test_missing_changed_block_cannot_hide_unreviewed_code(self):
+ actual = self.create_valid_ledger()
+ actual["files"][0]["blocks"].pop()
+ self.assertIn("Ledger files, changed-file count, or changed blocks do
not match the effective candidate", ledger.validate_ledger(actual))
+
+ def test_unknown_changed_block_coverage_is_rejected(self):
+ actual = self.create_valid_ledger()
+ actual["files"][0]["coverage"][0]["block"] = "unknown-block"
+ self.assertIn(f"{self.source_path} has unknown changed-block
coverage", ledger.validate_ledger(actual))
+
+ def test_reviewed_block_requires_known_requirement_origin(self):
+ actual = self.create_valid_ledger()
+ actual["files"][0]["coverage"][0]["requirements"] = ["unrelated-task"]
+ self.assertIn(f"{self.source_path} block coverage references unknown
applicable requirement origins", ledger.validate_ledger(actual))
+
+ def test_block_anchor_must_cover_changed_lines(self):
+ actual = self.create_valid_ledger()
+ coverage = actual["files"][0]["coverage"]
+ coverage[0]["anchors"] = [{"side": "candidate", "path":
self.source_path, "start_line": 3, "end_line": 3}]
+ coverage_file = self.private_root / "incorrect-coverage.json"
+ coverage_file.write_text(json.dumps(coverage), encoding="utf-8")
+ args = ledger.build_parser().parse_args(["mark-file", "--ledger",
str(self.ledger_file), "--path", self.source_path, "--status", "reviewed",
+ "--coverage",
str(coverage_file)])
+ with self.assertRaisesRegex(RuntimeError, "no anchor at the actual
change"):
+ ledger.cmd_mark_file(args)
+ self.assertEqual(actual["files"][0]["coverage"][1],
ledger.read_ledger(self.ledger_file)["files"][0]["coverage"][1])
+
+ def test_source_anchor_rejects_wrong_origin_and_range(self):
+ self.create_valid_ledger()
+ invalid = [
+ {"side": "source", "path": str(self.attempt_log), "origin":
"unknown-origin"},
+ {"side": "source", "path": "Requirements.md", "origin":
"attempt:target-version"},
+ {"side": "candidate", "path": self.source_path, "start_line": 1,
"end_line": 80},
+ {"side": "candidate", "path": self.source_path, "start_line":
True, "end_line": 1},
+ ]
+ for anchor in invalid:
+ with self.subTest(anchor=anchor), self.assertRaises(RuntimeError):
+ anchors_file = self.private_root / "invalid-anchors.json"
+ anchors_file.write_text(json.dumps([anchor]), encoding="utf-8")
+ args = ledger.build_parser().parse_args(["add-finding",
"--ledger", str(self.ledger_file), "--id", "F1", "--status", "non-blocking",
+ "--anchors",
str(anchors_file)])
+ ledger.cmd_add_finding(args)
+
+ def test_changed_candidate_and_scope_invalidate_review(self):
+ actual = self.create_valid_ledger()
+ (self.repo / self.source_path).write_text("A new effective
candidate.\n", encoding="utf-8")
+ self.assertIn("Effective candidate, original baseline, focus, or
declared review sources changed; initialize a new ledger",
ledger.validate_ledger(actual))
+ (self.repo / "new.txt").write_text("Additional task change.\n",
encoding="utf-8")
+ self.candidate_files.write_text(self.source_path + "\nnew.txt\n",
encoding="utf-8")
+ self.assertIn("Ledger files, changed-file count, or changed blocks do
not match the effective candidate", ledger.validate_ledger(actual))
+
+ def test_changed_rules_requirements_and_focus_invalidate_review(self):
+ actual = self.create_valid_ledger()
+ original_basis = self.basis_file.read_text(encoding="utf-8")
+ for changed in ("Rules.md", "Requirements.md", "focus"):
+ with self.subTest(changed=changed):
+ if "focus" == changed:
+ configuration = json.loads(original_basis)
+ configuration["focus"] = "CI Review"
+ self.basis_file.write_text(json.dumps(configuration),
encoding="utf-8")
+ else:
+ path = self.repo / changed
+ original = path.read_text(encoding="utf-8")
+ path.write_text("Changed actual source commitment.\n",
encoding="utf-8")
+ self.assertIn("Effective candidate, original baseline, focus,
or declared review sources changed; initialize a new ledger",
ledger.validate_ledger(actual))
+ if "focus" == changed:
+ self.basis_file.write_text(original_basis,
encoding="utf-8")
+ else:
+ path.write_text(original, encoding="utf-8")
+
+ def test_source_and_range_fingerprints_are_recomputed(self):
+ actual = self.create_valid_ledger()
+ actual["files"][0]["coverage"][0]["anchors"][0]["range_sha256"] = "f"
* 64
+ self.assertIn(f"Source anchor is stale: candidate:{self.source_path}",
ledger.validate_ledger(actual))
+
+ def test_changed_unchanged_owner_evidence_invalidates_review(self):
+ actual = self.create_valid_ledger()
+ (self.repo / self.owner_path).write_text("Changed owner
commitment.\n", encoding="utf-8")
+ self.assertIn(f"Source anchor is stale: candidate:{self.owner_path}",
ledger.validate_ledger(actual))
+
+ def test_additional_source_evidence_invalidates_prior_passes(self):
+ actual = self.create_valid_ledger()
+ (self.repo / "new-owner.txt").write_text("Additional relevant owner
evidence.\n", encoding="utf-8")
+
actual["files"][0]["coverage"][0]["anchors"].append(ledger.capture_anchor({"side":
"candidate", "path": "new-owner.txt"}, actual))
+ errors = ledger.validate_ledger(actual)
+ self.assertIn("Required review passes missing: root-cause,
blast-radius, tests-runtime, convergence", errors)
+ self.assertIn("Latest convergence pass uses a different review basis",
errors)
+
+ def test_metadata_only_change_preserves_discovery_pass_basis(self):
+ actual = self.create_valid_ledger()
+ actual["files"][0]["reason"] = "More precise wording with unchanged
evidence."
+ actual["review_revision"] += 1
+ actual["passes"][-1]["review_revision"] = actual["review_revision"]
+ self.assertEqual([], ledger.validate_ledger(actual))
+
+ def test_add_pass_binds_basis_and_rejects_changed_candidate(self):
+ actual = self.create_valid_ledger()
+ args = Namespace(ledger=str(self.ledger_file), focus="convergence",
new_findings=0)
+ with redirect_stdout(StringIO()):
+ self.assertEqual(0, ledger.cmd_add_pass(args))
+ self.assertEqual(actual["passes"][-1]["basis_fingerprint"],
ledger.read_ledger(self.ledger_file)["passes"][-1]["basis_fingerprint"])
+ (self.repo / self.source_path).write_text("A changed candidate must
not inherit passes.\n", encoding="utf-8")
+ with self.assertRaisesRegex(RuntimeError, "initialize a new ledger"):
+ ledger.cmd_add_pass(args)
+
+ def
test_same_effective_contents_match_local_and_committed_review_basis(self):
+ local = self.create_valid_ledger()
+ self.git("add", self.source_path)
+ self.git("commit", "-qm", "Synthetic candidate")
+ public = self.create_valid_ledger(committed=True)
+ self.assertNotEqual(local["scope"]["head_sha"],
public["scope"]["head_sha"])
+ self.assertEqual(ledger.basis_fingerprint(local),
ledger.basis_fingerprint(public))
+ self.assertEqual([], ledger.validate_ledger(local))
+ self.assertEqual([], ledger.validate_ledger(public))
+
+ def test_rename_identity_does_not_depend_on_git_status_heuristics(self):
+ renamed_path = "src/main/java/Renamed.java"
+ (self.repo / self.source_path).rename(self.repo / renamed_path)
+ self.candidate_files.write_text(self.source_path + "\n" + renamed_path
+ "\n", encoding="utf-8")
+ local = self.create_valid_ledger()
+ self.git("add", self.source_path, renamed_path)
+ self.git("commit", "-qm", "Synthetic rename")
+ public = self.create_valid_ledger(committed=True)
+ self.assertEqual(ledger.basis_fingerprint(local),
ledger.basis_fingerprint(public))
+ self.assertEqual([], ledger.validate_ledger(public))
+
+ def
test_deleted_added_empty_binary_and_mode_changes_have_real_coverage(self):
+ cases = ("deleted", "empty", "binary", "mode")
+ for scenario in cases:
+ with self.subTest(scenario=scenario):
+ path = self.repo / self.source_path
+ if not path.exists():
+
path.write_text("first\nold_first\nthird\nfourth\nfifth\nold_second\nlast\n",
encoding="utf-8")
+ path.chmod(0o644)
+ if "deleted" == scenario:
+ path.unlink()
+ elif "empty" == scenario:
+ path = self.repo / "empty.bin"
+ path.write_bytes(b"")
+ elif "binary" == scenario:
+ path.write_bytes(b"\x00\xffchanged binary")
+ else:
+
path.write_text("first\nold_first\nthird\nfourth\nfifth\nold_second\nlast\n",
encoding="utf-8")
+ path.chmod(0o755)
+
self.candidate_files.write_text(path.relative_to(self.repo).as_posix() + "\n",
encoding="utf-8")
+ actual = self.create_valid_ledger()
+ self.assertEqual([], ledger.validate_ledger(actual))
+ expected_kind = {"deleted": "text", "empty": "object",
"binary": "binary", "mode": "mode"}[scenario]
+ self.assertIn(expected_kind, {each["kind"] for each in
actual["files"][0]["blocks"]})
+
+ def test_confirmed_prose_requires_real_source_anchors(self):
+ actual = self.create_valid_ledger()
+ finding = self.create_confirmed_finding("F1")
+ finding.update({"evidence": ["x"], "full_path": ["x"],
"counter_evidence": ["x"], "necessity": "x", "scope_proof": "x", "anchors": []})
+ actual["findings"] = [finding]
+ self.assertIn("F1 confirmed finding is missing real source anchors",
ledger.validate_ledger(actual))
+
+ def test_deleted_file_replaced_by_directory_is_not_a_gitlink(self):
+ path = self.repo / self.source_path
+ path.unlink()
+ path.mkdir()
+ actual = self.create_valid_ledger()
+ self.assertEqual({"mode": None, "sha256": None},
actual["files"][0]["candidate"])
+ self.assertEqual("D", actual["files"][0]["git_status"])
+ self.assertEqual([], ledger.validate_ledger(actual))
+ child = self.repo / "container/child.txt"
+ child.parent.mkdir()
+ child.write_text("Tracked child.\n", encoding="utf-8")
+ self.git("add", "container/child.txt")
+ self.git("commit", "-qm", "Synthetic child baseline")
+ self.base_sha = self.git("rev-parse", "HEAD").strip()
+ child.unlink()
+ child.parent.rmdir()
+ child.parent.write_text("A parent file replaces the directory.\n",
encoding="utf-8")
+ self.candidate_files.write_text("container/child.txt\ncontainer\n",
encoding="utf-8")
+ actual = self.create_valid_ledger()
+ self.assertEqual({"mode": None, "sha256": None},
actual["files"][0]["candidate"])
+ self.assertEqual("D", actual["files"][0]["git_status"])
+ self.assertEqual({"mode": None, "sha256": None},
actual["files"][1]["base"])
+ self.assertEqual("100644", actual["files"][1]["candidate"]["mode"])
+ self.assertEqual([], ledger.validate_ledger(actual))
+
+ def test_symlink_evidence_uses_link_payload(self):
+ path = self.repo / self.source_path
+ path.unlink()
+ path.symlink_to("../../missing-target")
+ actual = self.create_valid_ledger()
+ self.assertEqual("120000", actual["files"][0]["candidate"]["mode"])
+ self.assertEqual([], ledger.validate_ledger(actual))
+
+ def test_gitlink_head_change_invalidates_review_basis(self):
+ checkout = self.create_gitlink_candidate()
+ actual = self.create_valid_ledger()
+ (checkout / "contract.txt").write_text("Third gitlink commit.\n",
encoding="utf-8")
+ self.git("-C", str(checkout), "add", "contract.txt")
+ self.git("-C", str(checkout), "commit", "-qm", "Synthetic gitlink C")
+ self.assertIn("Effective candidate, original baseline, focus, or
declared review sources changed; initialize a new ledger",
ledger.validate_ledger(actual))
+
+ def test_same_gitlink_pointer_matches_local_and_committed_basis(self):
+ checkout = self.create_gitlink_candidate()
+ local = self.create_valid_ledger()
+ pointer = self.git("-C", str(checkout), "rev-parse", "HEAD").strip()
+ self.git("update-index", "--cacheinfo", "160000," + pointer +
",dependency")
+ self.git("commit", "-qm", "Synthetic gitlink candidate")
+ public = self.create_valid_ledger(committed=True)
+ self.assertEqual("160000", local["files"][0]["candidate"]["mode"])
+ self.assertEqual(local["files"][0]["candidate"],
public["files"][0]["candidate"])
+ self.assertEqual(ledger.basis_fingerprint(local),
ledger.basis_fingerprint(public))
+ self.assertEqual([], ledger.validate_ledger(local))
+ self.assertEqual([], ledger.validate_ledger(public))
+
+ def test_gitlink_cannot_read_parent_repository_head(self):
+ checkout = self.create_gitlink_candidate()
+ pointer = self.git("-C", str(checkout), "rev-parse", "HEAD").strip()
+ self.git("update-index", "--cacheinfo", "160000," + pointer +
",dependency")
+ (checkout / ".git").rename(self.private_root / "detached-child-git")
+ with self.assertRaisesRegex(RuntimeError, "Gitlink checkout must have
its own repository"):
+ self.initialize_ledger()
+
+ def test_dirty_gitlink_cannot_reuse_pointer_basis(self):
+ checkout = self.create_gitlink_candidate()
+ actual = self.create_valid_ledger()
+ (checkout / "contract.txt").write_text("Uncommitted tracked
changes.\n", encoding="utf-8")
+ for staged in (False, True):
+ with self.subTest(staged=staged):
+ if staged:
+ self.git("-C", str(checkout), "add", "contract.txt")
+ with self.assertRaisesRegex(RuntimeError, "Gitlink checkout
has staged or unstaged tracked changes"):
+ ledger.validate_ledger(actual)
+
+ def create_gitlink_candidate(self):
+ checkout = self.repo / "dependency"
+ checkout.mkdir()
+ self.git("-C", str(checkout), "init", "-q")
+ (checkout / "contract.txt").write_text("First gitlink commit.\n",
encoding="utf-8")
+ self.git("-C", str(checkout), "add", "contract.txt")
+ self.git("-C", str(checkout), "commit", "-qm", "Synthetic gitlink A")
+ pointer = self.git("-C", str(checkout), "rev-parse", "HEAD").strip()
+ self.git("update-index", "--add", "--cacheinfo", "160000," + pointer +
",dependency")
+ self.git("commit", "-qm", "Synthetic gitlink baseline")
+ self.base_sha = self.git("rev-parse", "HEAD").strip()
+ (checkout / "contract.txt").write_text("Second gitlink commit.\n",
encoding="utf-8")
+ self.git("-C", str(checkout), "add", "contract.txt")
+ self.git("-C", str(checkout), "commit", "-qm", "Synthetic gitlink B")
+ self.candidate_files.write_text("dependency\n", encoding="utf-8")
+ return checkout
+
+ def test_incomplete_gap_requires_real_attempt_proof(self):
+ actual = self.create_valid_ledger()
+ finding = self.create_incomplete_gap()
+ finding["anchors"] = []
+ actual["findings"] = [finding]
+ self.assertIn("G1 incomplete gap is missing real unavailability-proof
anchors", ledger.validate_ledger(actual, incomplete_result=True))
+
+ def initialize_ledger(self, committed=False):
+ arguments = ["init", "--repo-root", str(self.repo), "--pr",
"synthetic-review", "--base-ref", self.base_sha,
+ "--head-ref", "HEAD", "--review-basis",
str(self.basis_file)]
+ if not committed:
+ arguments.extend(["--candidate-files", str(self.candidate_files)])
+ output = StringIO()
+ with redirect_stdout(output):
+ ledger.cmd_init(ledger.build_parser().parse_args(arguments))
+ self.ledger_file =
Path(output.getvalue().splitlines()[0].removeprefix("ledger="))
+ return ledger.read_ledger(self.ledger_file)
+
+ def create_valid_ledger(self, committed=False):
+ actual = self.initialize_ledger(committed)
+ for entry in actual["files"]:
+ coverage = []
+ for block in entry["blocks"]:
+ side = "candidate" if entry["candidate"]["mode"] else "base"
+ if "text" == block["kind"] and not block[f"{side}_count"]:
+ side = "base"
+ anchor = {"side": side, "path": block[f"{side}_path"]}
+ if "text" == block["kind"]:
+ anchor.update({"start_line": block[f"{side}_start"],
"end_line": block[f"{side}_start"] + block[f"{side}_count"] - 1})
+ coverage.append({"block": block["id"], "behavior": "Review the
changed owning behavior.", "risk_axes": ["correctness"],
+ "requirements": ["repository:Rules.md",
"user:synthetic-task"], "counter_evidence": "Check the stable owner contract.",
+ "anchors": [anchor, {"side": "candidate",
"path": self.owner_path, "start_line": 1, "end_line": 1}]})
+ coverage_file = self.private_root / "coverage.json"
+ coverage_file.write_text(json.dumps(coverage), encoding="utf-8")
+ args = ledger.build_parser().parse_args(["mark-file", "--ledger",
str(self.ledger_file), "--path", entry["path"], "--status", "reviewed",
+ "--cluster", "core",
"--risk-axis", "correctness", "--coverage", str(coverage_file)])
+ with redirect_stdout(StringIO()):
+ ledger.cmd_mark_file(args)
+ actual = ledger.read_ledger(self.ledger_file)
+ actual["passes"] = [{"focus": focus, "new_findings": 0,
"review_revision": actual["review_revision"],
+ "basis_fingerprint":
ledger.basis_fingerprint(actual)} for focus in ledger.PASS_FOCUSES]
+ ledger.write_ledger(self.ledger_file, actual)
+ return actual
- @staticmethod
- def create_confirmed_finding(finding_id):
+ def create_confirmed_finding(self, finding_id):
return {
"id": finding_id,
"status": "confirmed",
@@ -233,10 +579,10 @@ class ReviewLedgerTest(unittest.TestCase):
"scope_proof": "introduced by reviewed change",
"files": ["src/main/java/Foo.java"],
"reason": "",
+ "anchors":
ledger.read_ledger(self.ledger_file)["files"][0]["coverage"][0]["anchors"],
}
- @staticmethod
- def create_incomplete_gap():
+ def create_incomplete_gap(self):
return {
"id": "G1",
"status": "review-incomplete-gap",
@@ -249,6 +595,8 @@ class ReviewLedgerTest(unittest.TestCase):
"scope_proof": "The reviewed file invokes the version-specific
behavior.",
"files": ["src/main/java/Foo.java"],
"reason": "The target tool version is unavailable.",
+ "anchors": [ledger.capture_anchor({"side": "source", "path":
str(self.attempt_log), "origin": "attempt:target-version", "start_line": 1,
"end_line": 1},
+
ledger.read_ledger(self.ledger_file))],
}
