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 ab503c6758c Add atomic batch updates to review ledgers (#40346)
ab503c6758c is described below
commit ab503c6758c8bf322a7571b317eb56e458af4eae
Author: Liang Zhang <[email protected]>
AuthorDate: Thu Oct 22 20:32:03 2026 +0800
Add atomic batch updates to review ledgers (#40346)
* Run initial blind PR reviews in parallel
* Update release notes
* Add atomic batch updates to review ledgers
---
.codex/harness/agents/cases.toml | 20 +++
.codex/skills/review-pr/SKILL.md | 4 +
.codex/skills/review-pr/scripts/review_ledger.py | 61 ++++++-
.../skills/review-pr/tests/test_review_ledger.py | 193 +++++++++++++++++++++
4 files changed, 271 insertions(+), 7 deletions(-)
diff --git a/.codex/harness/agents/cases.toml b/.codex/harness/agents/cases.toml
index 6d331da5f0c..e46fa68f4d4 100644
--- a/.codex/harness/agents/cases.toml
+++ b/.codex/harness/agents/cases.toml
@@ -4846,3 +4846,23 @@ forbidden_summary_terms =
["start=finish-A-before-starting-B", "collect=only-A-b
critical = true
policy_binding_profile = "review-pr"
policy_assertions = [{source = "review-pr-coordination", statements = ["For
every top-level Formal Review and every discussion reply that makes or changes
an overall readiness conclusion, a coordinator must first obtain two mutually
blind independent complete assessments of the same effective candidate,
original baseline, authoritative scope, requirements and rule versions, and
review focus.", "Launch the initial pair in parallel, starting both isolated
reviewer contexts before waiting f [...]
+
+[[cases]]
+id = "review_ledger_batch_preserves_validation_and_blindness"
+group = "review-pr batch ledger recording"
+description = "Batches ordered updates to one independent ledger without
weakening freshness, atomic failure, final validation or blind-review
boundaries."
+phase = "read-only batch-ledger contract assessment"
+ordinary_loop = false
+profile = "review-pr"
+semantic_policy_assertions = true
+prompt = "Answer this self-contained read-only policy question using only the
supplied facts and bound rules; do not inspect files, run commands, edit
anything or produce a Formal Review report. A reviewer has several existing
ledger mutations ready for batching. Choose order=preserve-operation-order or
order=move-all-passes-last. A later operation fails or a captured source
changes before the final freshness check; choose
failure=leave-original-unchanged or failure=keep-earlier-updates. [...]
+decision = "proceed"
+required_actions = []
+allowed_actions = []
+forbidden_actions = ["inspect_local", "edit_code", "edit_non_code",
"run_local_checks", "run_pre_handoff_review", "propose_commit_message",
"mutate_git", "mutate_remote"]
+required_reasons = ["read_only_request"]
+required_summary_terms = ["order=preserve-operation-order",
"failure=leave-original-unchanged",
"validation=final-ledger-and-session-checks-required",
"records=one-batch-per-reviewer-ledger", "sharing=current-raw-facts-only"]
+forbidden_summary_terms = ["order=move-all-passes-last",
"failure=keep-earlier-updates", "validation=batch-proves-review-complete",
"records=one-shared-ledger", "sharing=A-findings-and-result"]
+critical = true
+policy_binding_profile = "review-pr"
+policy_assertions = [{source = "review-pr", statements = ["Mutate one ledger
sequentially; do not run ledger commands concurrently.", "Each batch applies
the existing operations in order and writes once only after all per-operation
checks and both freshness checks pass.", "Batch freshness checks verify the
current basis and captured source anchors before applying operations and again
before writing; any failed operation or freshness check leaves the original
ledger unchanged.", "Batching [...]
diff --git a/.codex/skills/review-pr/SKILL.md b/.codex/skills/review-pr/SKILL.md
index 5e21e01578f..62d68d03c90 100644
--- a/.codex/skills/review-pr/SKILL.md
+++ b/.codex/skills/review-pr/SKILL.md
@@ -226,6 +226,10 @@ Apply the Completion Gate to every Formal Review and to a
discussion reply that
- 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.
+- Batch related updates to one reviewer ledger with `scripts/review_ledger.py
batch --ledger <ledger> --operations <JSON>`, where the JSON is a nonempty
ordered array of argument arrays for `mark-file`, `add-finding` and `add-pass`,
using their existing flags and input files without `--ledger`.
+- Each batch applies the existing operations in order and writes once only
after all per-operation checks and both freshness checks pass.
+- Batch freshness checks verify the current basis and captured source anchors
before applying operations and again before writing; any failed operation or
freshness check leaves the original ledger unchanged.
+- Batching does not replace final ledger or session validation.
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.
diff --git a/.codex/skills/review-pr/scripts/review_ledger.py
b/.codex/skills/review-pr/scripts/review_ledger.py
index a5de373d023..66e73bb6595 100755
--- a/.codex/skills/review-pr/scripts/review_ledger.py
+++ b/.codex/skills/review-pr/scripts/review_ledger.py
@@ -505,6 +505,13 @@ def cmd_mark_file(args: argparse.Namespace) -> int:
errors = validate_current_basis(ledger)
if errors:
raise RuntimeError("; ".join(errors))
+ apply_mark_file(ledger, args)
+ write_ledger(ledger_file, ledger)
+ print(f"marked={args.path} status={args.status}")
+ return 0
+
+
+def apply_mark_file(ledger: dict[str, Any], args: argparse.Namespace) -> None:
entry = find_file_entry(ledger, args.path)
if args.coverage:
coverage = json.loads(Path(args.coverage).read_text(encoding="utf-8"))
@@ -522,9 +529,6 @@ def cmd_mark_file(args: argparse.Namespace) -> int:
if errors:
raise RuntimeError("; ".join(errors))
ledger["review_revision"] += 1
- write_ledger(ledger_file, ledger)
- print(f"marked={args.path} status={args.status}")
- return 0
def cmd_add_finding(args: argparse.Namespace) -> int:
@@ -533,6 +537,13 @@ def cmd_add_finding(args: argparse.Namespace) -> int:
errors = validate_current_basis(ledger)
if errors:
raise RuntimeError("; ".join(errors))
+ apply_add_finding(ledger, args)
+ write_ledger(ledger_file, ledger)
+ print(f"finding={args.id} status={args.status}")
+ return 0
+
+
+def apply_add_finding(ledger: dict[str, Any], args: argparse.Namespace) ->
None:
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")
@@ -559,9 +570,6 @@ def cmd_add_finding(args: argparse.Namespace) -> int:
ledger["findings"].append(finding)
sync_file_findings(ledger)
ledger["review_revision"] += 1
- write_ledger(ledger_file, ledger)
- print(f"finding={args.id} status={args.status}")
- return 0
def cmd_add_pass(args: argparse.Namespace) -> int:
@@ -570,6 +578,13 @@ def cmd_add_pass(args: argparse.Namespace) -> int:
errors = validate_current_basis(ledger)
if errors:
raise RuntimeError("; ".join(errors))
+ apply_add_pass(ledger, args)
+ write_ledger(ledger_file, ledger)
+ print(f"pass={args.focus} new_findings={args.new_findings}")
+ return 0
+
+
+def apply_add_pass(ledger: dict[str, Any], args: argparse.Namespace) -> None:
if 0 > args.new_findings:
raise RuntimeError("New finding count must not be negative")
ledger["passes"].append({
@@ -580,8 +595,36 @@ def cmd_add_pass(args: argparse.Namespace) -> int:
"created_at": int(time.time()),
"basis_fingerprint": basis_fingerprint(ledger),
})
+
+
+def cmd_batch(args: argparse.Namespace) -> int:
+ operations = json.loads(Path(args.operations).read_text(encoding="utf-8"))
+ if not isinstance(operations, list) or not operations:
+ raise RuntimeError("Batch operations must be a nonempty list of
argument lists")
+ ledger_file = resolve_ledger_file(args.ledger)
+ ledger = read_ledger(ledger_file)
+ errors = validate_current_basis(ledger)
+ if errors:
+ raise RuntimeError("; ".join(errors))
+ parser = build_parser()
+ updates = {"mark-file": apply_mark_file, "add-finding": apply_add_finding,
"add-pass": apply_add_pass}
+ for operation in operations:
+ if not isinstance(operation, list) or not operation or not
all(isinstance(each, str) for each in operation):
+ raise RuntimeError("Each batch operation must be a nonempty list
of argument strings")
+ if operation[0] not in updates:
+ raise RuntimeError("Batch operations support only mark-file,
add-finding and add-pass")
+ try:
+ update = parser.parse_args([operation[0], "--ledger",
str(ledger_file), *operation[1:]])
+ except SystemExit as ex:
+ raise RuntimeError("Invalid batch operation arguments") from ex
+ if update.ledger != str(ledger_file):
+ raise RuntimeError("Every batch operation must use the outer
ledger")
+ updates[update.command](ledger, update)
+ errors = validate_current_basis(ledger)
+ if errors:
+ raise RuntimeError("; ".join(errors))
write_ledger(ledger_file, ledger)
- print(f"pass={args.focus} new_findings={args.new_findings}")
+ print(f"batched={len(operations)}")
return 0
@@ -1028,6 +1071,10 @@ def build_parser() -> argparse.ArgumentParser:
review_pass.add_argument("--focus", required=True, choices=PASS_FOCUSES,
help="Pass focus")
review_pass.add_argument("--new-findings", required=True, type=int,
help="New independent findings")
review_pass.set_defaults(func=cmd_add_pass)
+ batch = subparsers.add_parser("batch", help="Apply ordered updates to one
ledger with one atomic write")
+ batch.add_argument("--ledger", required=True, help="Ledger file or
directory")
+ batch.add_argument("--operations", required=True, help="JSON nonempty
ordered list of existing mutation argument lists without --ledger")
+ batch.set_defaults(func=cmd_batch)
validate = subparsers.add_parser("validate", help="Validate mechanical
coverage completion")
validate.add_argument("--ledger", required=True, help="Ledger file or
directory")
validate.set_defaults(func=cmd_validate)
diff --git a/.codex/skills/review-pr/tests/test_review_ledger.py
b/.codex/skills/review-pr/tests/test_review_ledger.py
index 75e0d6e8a76..4a39c91d3a0 100644
--- a/.codex/skills/review-pr/tests/test_review_ledger.py
+++ b/.codex/skills/review-pr/tests/test_review_ledger.py
@@ -80,6 +80,194 @@ class ReviewLedgerTest(unittest.TestCase):
def test_complete_ledger_is_valid(self):
self.assertEqual([],
ledger.validate_ledger(self.create_valid_ledger()))
+ def test_batch_matches_sequential_updates_with_less_io(self):
+ actual = self.create_valid_ledger()
+ finding = self.create_confirmed_finding("F1")
+ finding["status"] = "candidate"
+ actual["findings"].append(finding)
+ ledger.sync_file_findings(actual)
+ ledger.write_ledger(self.ledger_file, actual)
+ original_bytes = self.ledger_file.read_bytes()
+ operations = [
+ ["mark-file", "--path", self.source_path, "--status", "reviewed",
"--cluster", "batch", "--risk-axis", "compatibility",
+ "--coverage", str(self.private_root / "coverage.json")],
+ ["add-finding", "--id", "F1", "--status", "withdrawn", "--reason",
"The supported owner contract disproves the candidate.", "--file",
self.source_path],
+ *[["add-pass", "--focus", focus, "--new-findings", "0"] for focus
in ledger.PASS_FOCUSES],
+ ]
+ with patch.object(ledger.time, "time", return_value=12345),
patch.object(ledger, "capture_basis", wraps=ledger.capture_basis) as captures, \
+ patch.object(ledger, "write_ledger",
wraps=ledger.write_ledger) as writes, redirect_stdout(StringIO()):
+ for operation in operations:
+ arguments = ledger.build_parser().parse_args([operation[0],
"--ledger", str(self.ledger_file), *operation[1:]])
+ self.assertEqual(0, arguments.func(arguments))
+ self.assertEqual(6, captures.call_count)
+ self.assertEqual(6, writes.call_count)
+ expected = ledger.read_ledger(self.ledger_file)
+ self.ledger_file.write_bytes(original_bytes)
+ output = StringIO()
+ arguments = self.batch_arguments(operations)
+ with patch.object(ledger.time, "time", return_value=12345),
patch.object(ledger, "capture_basis", wraps=ledger.capture_basis) as captures, \
+ patch.object(ledger, "write_ledger",
wraps=ledger.write_ledger) as writes, redirect_stdout(output):
+ self.assertEqual(0, arguments.func(arguments))
+ self.assertEqual(2, captures.call_count)
+ self.assertEqual(1, writes.call_count)
+ self.assertEqual("batched=6\n", output.getvalue())
+ actual = ledger.read_ledger(self.ledger_file)
+ self.assertEqual(expected, actual)
+ self.assertEqual(3, actual["review_revision"])
+ self.assertEqual(["F1"], actual["files"][0]["findings"])
+ self.assertEqual("withdrawn", actual["findings"][0]["status"])
+ self.assertEqual([], ledger.validate_ledger(actual))
+
+ def test_batch_allows_in_progress_ledger_updates(self):
+ self.initialize_ledger()
+ arguments = self.batch_arguments([
+ ["mark-file", "--path", self.source_path, "--status", "pending",
"--cluster", "core"],
+ ["add-finding", "--id", "F1", "--status", "candidate", "--file",
self.source_path],
+ ])
+ with redirect_stdout(StringIO()):
+ self.assertEqual(0, ledger.cmd_batch(arguments))
+ actual = ledger.read_ledger(self.ledger_file)
+ self.assertEqual(2, actual["review_revision"])
+ self.assertEqual("pending", actual["files"][0]["status"])
+ self.assertEqual(["F1"], actual["files"][0]["findings"])
+ self.assertEqual("candidate", actual["findings"][0]["status"])
+ self.assertIn("Pending files remain: 1",
ledger.validate_ledger(actual))
+
+ def test_batch_failure_preserves_original_ledger_bytes(self):
+ actual = self.create_valid_ledger()
+ original = self.ledger_file.read_bytes()
+ coverage_file = self.private_root / "empty-coverage.json"
+ coverage_file.write_text("[]\n", encoding="utf-8")
+ anchors_file = self.private_root / "out-of-range-anchors.json"
+ anchors_file.write_text(json.dumps([{"side": "candidate", "path":
self.source_path, "start_line": 1, "end_line": 80}]), encoding="utf-8")
+ invalid = [
+ ("negative-count", ["add-pass", "--focus", "convergence",
"--new-findings", "-1"], "New finding count must not be negative"),
+ ("outside-scope", ["mark-file", "--path", "outside.java",
"--status", "reviewed"], "File is not in ledger scope: outside.java"),
+ ("empty-coverage", ["mark-file", "--path", self.source_path,
"--status", "reviewed", "--coverage", str(coverage_file)],
+ f"{self.source_path} has uncovered changed blocks:
{len(actual['files'][0]['blocks'])}"),
+ ("anchor-range", ["add-finding", "--id", "F2", "--status",
"non-blocking", "--file", self.source_path, "--anchors", str(anchors_file)],
+ "Source anchor line range is outside the actual source"),
+ ]
+ for scenario, operation, expected in invalid:
+ with self.subTest(scenario=scenario):
+ arguments = self.batch_arguments([
+ ["add-finding", "--id", "F1", "--status", "candidate",
"--file", self.source_path],
+ operation,
+ ])
+ output = StringIO()
+ with patch.object(ledger, "write_ledger",
wraps=ledger.write_ledger) as writes, redirect_stdout(output),
self.assertRaises(RuntimeError) as caught:
+ ledger.cmd_batch(arguments)
+ self.assertEqual(expected, str(caught.exception))
+ self.assertEqual(0, writes.call_count)
+ self.assertEqual(original, self.ledger_file.read_bytes())
+ self.assertEqual("", output.getvalue())
+
+ def test_batch_rejects_malformed_operations_and_cli_arguments(self):
+ self.create_valid_ledger()
+ original = self.ledger_file.read_bytes()
+ invalid = [
+ None, {}, [], "add-pass", 1, True, [[]], [{}], [[1]],
[["add-pass", 0]],
+ [["cleanup"]], [["init"]], [["batch"]], [["status"]],
+ [["add-pass", "--focus", "invalid", "--new-findings", "0"]],
+ [["add-pass", "--focus", "convergence", "--new-findings",
"invalid"]],
+ [["add-pass", "--focus", "convergence", "--new-findings", "0",
"--unknown"]],
+ [["mark-file", "--path", self.source_path, "--status", "invalid"]],
+ [["add-finding", "--id", "F1", "--status", "invalid"]],
+ [["add-pass", "--help"]], [["mark-file", "-h"]],
+ ]
+ for operations in invalid:
+ with self.subTest(operations=operations):
+ arguments = self.batch_arguments(operations)
+ with patch.object(ledger, "write_ledger",
wraps=ledger.write_ledger) as writes, redirect_stdout(StringIO()),
redirect_stderr(StringIO()), \
+ self.assertRaises(RuntimeError):
+ ledger.cmd_batch(arguments)
+ self.assertEqual(0, writes.call_count)
+ self.assertEqual(original, self.ledger_file.read_bytes())
+
+ def test_batch_rejects_alternate_ledger_targets(self):
+ self.create_valid_ledger()
+ original = self.ledger_file.read_bytes()
+ other = ledger.ledger_root() / "other-ledger" / ledger.LEDGER_FILE_NAME
+ other.parent.mkdir(mode=0o700)
+ ledger.write_ledger(other, ledger.read_ledger(self.ledger_file))
+ other_original = other.read_bytes()
+ for override in (["--ledger", str(other)], ["--led", str(other)],
["--ledger=" + str(other)], ["--led=" + str(other)]):
+ with self.subTest(override=override):
+ arguments = self.batch_arguments([["add-pass", "--focus",
"convergence", "--new-findings", "0", *override]])
+ with patch.object(ledger, "write_ledger",
wraps=ledger.write_ledger) as writes, redirect_stdout(StringIO()),
redirect_stderr(StringIO()), \
+ self.assertRaises(RuntimeError):
+ ledger.cmd_batch(arguments)
+ self.assertEqual(0, writes.call_count)
+ self.assertEqual(original, self.ledger_file.read_bytes())
+ self.assertEqual(other_original, other.read_bytes())
+
+ def test_batch_rejects_stale_candidate_rule_and_owner_evidence(self):
+ self.create_valid_ledger()
+ original = self.ledger_file.read_bytes()
+ for path in (self.source_path, "Rules.md", self.owner_path):
+ with self.subTest(path=path):
+ source = self.repo / path
+ source_original = source.read_bytes()
+ source.write_text("Changed source commitment.\n",
encoding="utf-8")
+ try:
+ arguments = self.batch_arguments([["add-pass", "--focus",
"convergence", "--new-findings", "0"]])
+ with patch.object(ledger, "write_ledger",
wraps=ledger.write_ledger) as writes, redirect_stdout(StringIO()),
self.assertRaises(RuntimeError):
+ ledger.cmd_batch(arguments)
+ self.assertEqual(0, writes.call_count)
+ self.assertEqual(original, self.ledger_file.read_bytes())
+ finally:
+ source.write_bytes(source_original)
+
+ def test_batch_rechecks_evidence_before_persisting(self):
+ self.create_valid_ledger()
+ original = self.ledger_file.read_bytes()
+ apply_mark_file = ledger.apply_mark_file
+
+ def change_source(actual, arguments):
+ apply_mark_file(actual, arguments)
+ (self.repo / self.owner_path).write_text("Changed after the
initial freshness check.\n", encoding="utf-8")
+
+ arguments = self.batch_arguments([["mark-file", "--path",
self.source_path, "--status", "reviewed", "--cluster", "batch"]])
+ output = StringIO()
+ with patch.object(ledger, "apply_mark_file",
side_effect=change_source), patch.object(ledger, "write_ledger",
wraps=ledger.write_ledger) as writes, \
+ redirect_stdout(output), self.assertRaisesRegex(RuntimeError,
"Source anchor is stale"):
+ ledger.cmd_batch(arguments)
+ self.assertEqual(0, writes.call_count)
+ self.assertEqual(original, self.ledger_file.read_bytes())
+ self.assertEqual("", output.getvalue())
+
+ def test_batch_pass_before_mutation_remains_stale(self):
+ self.create_valid_ledger()
+ arguments = self.batch_arguments([
+ ["add-pass", "--focus", "convergence", "--new-findings", "0"],
+ ["mark-file", "--path", self.source_path, "--status", "reviewed",
"--reason", "More precise reviewed metadata."],
+ ])
+ with redirect_stdout(StringIO()):
+ self.assertEqual(0, ledger.cmd_batch(arguments))
+ actual = ledger.read_ledger(self.ledger_file)
+ self.assertEqual(actual["review_revision"] - 1,
actual["passes"][-1]["review_revision"])
+ self.assertIn("Latest convergence pass predates a file or finding
change", ledger.validate_ledger(actual))
+
+ def test_batch_new_evidence_invalidates_earlier_passes(self):
+ actual = self.create_valid_ledger()
+ original_fingerprint = ledger.basis_fingerprint(actual)
+ (self.repo / "new-owner.txt").write_text("Additional owning
contract.\n", encoding="utf-8")
+ coverage = actual["files"][0]["coverage"]
+ coverage[0]["anchors"].append({"side": "candidate", "path":
"new-owner.txt"})
+ coverage_file = self.private_root / "new-evidence-coverage.json"
+ coverage_file.write_text(json.dumps(coverage), encoding="utf-8")
+ arguments = self.batch_arguments([
+ ["add-pass", "--focus", "convergence", "--new-findings", "0"],
+ ["mark-file", "--path", self.source_path, "--status", "reviewed",
"--coverage", str(coverage_file)],
+ ])
+ with redirect_stdout(StringIO()):
+ self.assertEqual(0, ledger.cmd_batch(arguments))
+ actual = ledger.read_ledger(self.ledger_file)
+ self.assertEqual(original_fingerprint,
actual["passes"][-1]["basis_fingerprint"])
+ 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_confirmed_finding_requires_non_blank_proof(self):
actual_ledger = self.create_valid_ledger()
finding = self.create_confirmed_finding("F1")
@@ -1171,6 +1359,11 @@ class ReviewLedgerTest(unittest.TestCase):
report["ledger_sha256"] =
hashlib.sha256(ledger_file.read_bytes()).hexdigest()
self.save_session_report(session, index, report)
+ def batch_arguments(self, operations):
+ operations_file = self.private_root / "operations.json"
+ operations_file.write_text(json.dumps(operations), encoding="utf-8")
+ return ledger.build_parser().parse_args(["batch", "--ledger",
str(self.ledger_file), "--operations", str(operations_file)])
+
def initialize_ledger(self, committed=False):
arguments = ["init", "--repo-root", str(self.repo), "--pr",
"synthetic-review", "--base-ref", self.base_sha,
"--head-ref", self.git("rev-parse", "HEAD").strip() if
committed else "HEAD", "--review-basis", str(self.basis_file)]
