P04-01: 同時発生の更新規約

This commit is contained in:
Agent Control Runner
2026-10-18 12:27:51 +09:00
parent 8e25213e91
commit befa0ef1f6
7 changed files with 289 additions and 20 deletions
@@ -25,4 +25,4 @@ derived 184 suites / 329 parse; truthful index; Root exact-candidate qualificati
Root order is py_compile → wrapper fixtures → UID controls with explicit scratch Root order is py_compile → wrapper fixtures → UID controls with explicit scratch
→ fresh check-only plan → exactly one new-name diagnostic. The failed r35 → fresh check-only plan → exactly one new-name diagnostic. The failed r35
diagnostic is preserved. All 7 domains + EV, formal 16/88, original 30/88, diagnostic is preserved. All 7 domains + EV, formal 16/88, original 30/88,
P04 credit 0, all six D05 approvals pending, M c0ea009 and allowed_paths stay fixed. P04 credit 0, all six D05 approved in the actual human record; adoption verification pending, M c0ea009 and allowed_paths stay fixed.
@@ -0,0 +1,23 @@
{
"checks": [
{
"path": "s5/s5_power_adapter.py",
"sha256": "448f7a99cd08ead1d4e2b91907a66e8805261c9d24c9f55eed9a047fac48b57c",
"py_compile": "pass",
"ast_parse": "pass"
},
{
"path": "s5/s5_power_adapter_controls.py",
"sha256": "32d71db80994052b905829e6a2da38448ae789825c44570a901e67fe2d07fd33",
"py_compile": "pass",
"ast_parse": "pass"
}
],
"executed": [
"py_compile",
"ast.parse"
],
"functional_controls_executed": false,
"native_executed": false,
"accepted": false
}
@@ -0,0 +1,15 @@
# REPAIR37 — bounded preparation submitted; no acceptance
Two probe corrections preserve recovery inputs/save/restore: typed String snapshot staff versus StringName live staff, and rejected duplicate enqueue bound to the completed resource command. The completion hook still requires success; duplicate rejection must not emit command_completed. All payment, reservation, receipt, ledger, phase roster and Reader assertions remain unchanged.
The adapter loads the existing series parser by spec_from_file_location and records its SHA256. Typed staff and duplicate checks bind both roles to their own completed commands. New pure controls verify every INDEX member sha/bytes before scratch creation, use the actual q36 pair, and require ValueError for the specified mutations including JSON-equal typed changes. Controls remain unexecuted.
Read dependencies: local q36-writer-contract INDEX, TYPED_AUDIT, SAFETY_REAUDIT, command_queue.gd.reference, facility_instance.gd.reference, both before/resource-replay JSON records; series Node/variant and checker loader pattern. Raw fixture decoding and controls are reserved to Root. A shell heredoc failed before Python execution because shell temporary-file creation was denied; subsequent edits used python -c with no temporary files.
Only py_compile and AST parsing are run, as recorded in REPAIR37_STATIC_CHECKS.json. No engine, Git, Runner, native run, other agent or /tmp operation was performed. No functional pass is claimed. Source basis remains 8e25213; counters remain 37, original 30/88, formal 16/88, P04 credit 0 and M c0ea009 frozen. p04-s5-DIAG-r36 remains FAILED without retrospective pass. All six D05 decisions are approved in the actual human record; adoption verification remains pending.
Root order: py_compile → s5_uid_controls → adapter controls with explicit fresh BUILD scratch → fresh check-only plan → at most one fresh source-pinned diagnostic under a new name. UID sources/settings are unchanged. No full184 until the whole batch is ready.
Context gap: whether current runtime submit_command emits command_completed on enqueue rejection is grounded only by the command_queue reference and TYPED_AUDIT raw. Root must observe this in the diagnostic; any emission remains a failure.
Pending: S5 destruction/midnight/exit; S1–S4/collector fresh Q; full 184/329; whole independent Opus review; normal prepare/exact candidate/integration/final/Root join/ledger. No acceptance or completion is claimed.
@@ -31,12 +31,12 @@ After the final submission, Root alone prepares Q using these steps (not perform
checkout carrier of Q, never normal Runner C. R34 and Root normal_c_join checkout carrier of Q, never normal Runner C. R34 and Root normal_c_join
remain unchanged. remain unchanged.
3. Stage the s5 directory under Root's own B, the registered BUILD directory. 3. Stage the s5 directory under Root's own B, the registered BUILD directory.
4. Follow the REPAIR36 order below, with fresh output and plan paths: 4. Follow the REPAIR37 order below, with fresh output and plan paths:
```sh ```sh
BUILD_OUTPUT_DIR="$B" python3 "$B/.../run_s5_native.py" \ BUILD_OUTPUT_DIR="$B" python3 "$B/.../run_s5_native.py" \
--project-root "$wt" --main "$M" --submission "$S" --tested-tree "$Qt" \ --project-root "$wt" --main "$M" --submission "$S" --tested-tree "$Qt" \
--check-only --plan "$B/p04-s5-plan-r36.json" --output "$B/p04-s5-DIAG-r36" --check-only --plan "$B/p04-s5-plan-r37.json" --output "$B/p04-s5-DIAG-r37"
``` ```
5. Pure fixtures and controls follow the order below. Their JSON has accepted=false. 5. Pure fixtures and controls follow the order below. Their JSON has accepted=false.
@@ -69,16 +69,24 @@ replace declared_reports while preserving the execute_phase code object.
All four phase UID receipts and the post-cleanup seal are required. All four phase UID receipts and the post-cleanup seal are required.
Historical mode has no UID instrumentation and remains read-only. Historical mode has no UID instrumentation and remains read-only.
Root order, all unexecuted by this worker: Root order for REPAIR37, all unexecuted by this worker (supersedes the earlier r36 order):
1. py_compile the wrapper and both pure control files. 1. py_compile the wrapper, adapter and pure controls.
2. Run s5_wrapper_fixtures.py. 2. Run s5_uid_controls with explicit fresh BUILD scratch; UID sources/settings remain unchanged.
3. Run `python3 "$B/.../s5_uid_controls.py" --project-root "$wt" --scratch "$B/p04-s5-uid-controls-r36"`. 3. Run the adapter controls against the read-only q36-writer-contract bundle:
Scratch must be absolute, absent, outside Q, with an existing parent in BUILD. `python3 "$B/.../s5_power_adapter_controls.py" --fixture-dir "$fixture" --scratch "$B/p04-s5-adapter-controls-r37"`.
4. Produce a fresh check-only plan with the command above. Scratch must be explicit, absolute, absent, outside project/fixture, with existing parent inside registered BUILD.
5. Perform exactly one fresh diagnostic with the identical command and paths, Verify accepted=false and root_only=true; this grants no acceptance.
removing only --check-only. Use a new output name; preserve 4. Produce a fresh source-pinned check-only plan with the command above.
`p04-s5-DIAG-r35` as FAILED. This is a bounded diagnostic, not full admission. 5. Perform at most one fresh source-pinned diagnostic under a new name, removing only --check-only.
Preserve p04-s5-DIAG-r36 as FAILED without retrospective pass. No full184 run.
Observe whether submit_command enqueue rejection emits command_completed: the probe's existing success expectation is unchanged, so such emission remains a failure.
Current grounding for this behavior is limited to command_queue.gd.reference and TYPED_AUDIT raw.
All six D05 decisions are approved in the actual human record; adoption verification is pending.
Counters stay 37; M c0ea009, original 30/88, formal 16/88 and P04 credit 0 remain frozen.
Whole independent Opus review and normal prepare/exact candidate/integration/final/Root join/ledger remain pending.
No full184 until the whole batch is ready.
Root must confirm the actual Q run_all_tests.py matches the supplied reference Root must confirm the actual Q run_all_tests.py matches the supplied reference
(sha256 prefix 320ca668); the worker read only the mirror for Q API context. (sha256 prefix 320ca668); the worker read only the mirror for Q API context.
@@ -19,6 +19,7 @@ var _price: int = 0
var _duration: int = 0 var _duration: int = 0
var _patience: int = 0 var _patience: int = 0
var _reserve: int = 0 var _reserve: int = 0
var _resource_command: Dictionary = {}
var _recovery: Dictionary = {} var _recovery: Dictionary = {}
var _commands: Array = [] var _commands: Array = []
var _results: Array = [] var _results: Array = []
@@ -58,6 +59,7 @@ func _fresh() -> void:
_duration = 0 _duration = 0
_patience = 0 _patience = 0
_reserve = 0 _reserve = 0
_resource_command = {}
_recovery = {} _recovery = {}
_replay_reference = {} _replay_reference = {}
_commands = [] _commands = []
@@ -208,6 +210,7 @@ func _advance(label: String) -> void:
elif label == "recovered": elif label == "recovered":
_expect(_clock.absolute_minute == _boundary + 1, "recover only after held") _expect(_clock.absolute_minute == _boundary + 1, "recover only after held")
_cmd(&"change_resources", {&"deltas": {&"mana": _reserve}}) _cmd(&"change_resources", {&"deltas": {&"mana": _reserve}})
_resource_command = {&"key": _commands.back()[&"idempotency_key"], &"command_id": int(_commands.back()[&"result"][&"command_id"])}
_recovery[&"key"] = _commands.back()[&"idempotency_key"] _recovery[&"key"] = _commands.back()[&"idempotency_key"]
while _clock.absolute_minute < _boundary + _duration: while _clock.absolute_minute < _boundary + _duration:
_clock.advance_one_minute() _clock.advance_one_minute()
@@ -221,8 +224,27 @@ func _advance(label: String) -> void:
var before_replay: Dictionary = _runtime.create_snapshot() var before_replay: Dictionary = _runtime.create_snapshot()
var replay: CommandResult = _runtime.submit_command(&"change_resources", {&"deltas": {&"mana": _reserve}}, &"player", &"", 10, -1, _recovery[&"key"]) var replay: CommandResult = _runtime.submit_command(&"change_resources", {&"deltas": {&"mana": _reserve}}, &"player", &"", 10, -1, _recovery[&"key"])
var after_replay: Dictionary = _runtime.create_snapshot() var after_replay: Dictionary = _runtime.create_snapshot()
_expect(replay.is_success() and replay.is_replay and var_to_bytes(before_replay) == var_to_bytes(after_replay), "same key replay changes no Runtime field") var key: StringName = _resource_command[&"key"]
_replay_reference = _write("resource-replay", {&"before": before_replay, &"after": after_replay, &"result": replay.to_dictionary()}) var command_id: int = _resource_command[&"command_id"]
var originals: Array[Dictionary] = []
for row: Dictionary in _results:
if int(row[&"result"][&"command_id"]) == command_id: originals.append(row[&"result"])
_expect(originals.size() == 1, "one completed original resource command")
if originals.size() != 1: return
var original: Dictionary = originals[0]
var result: Dictionary = replay.to_dictionary()
var details: Dictionary = result[&"details"]
_expect(result[&"code"] == "DUPLICATE_IDEMPOTENCY_KEY" and not replay.is_replay, "duplicate enqueue rejected without replay")
_expect(details["processed"] == true and details["idempotency_key"] == String(key) and int(details["original_command_id"]) == command_id and var_to_bytes(details["original_result"]) == var_to_bytes(original), "duplicate details bind completed original")
var queue: Dictionary = before_replay[&"systems"][&"runtime"][&"command_queue"]
_expect(queue["idempotency"][String(key)] == command_id, "typed idempotency binding")
var processed: Array[Dictionary] = []
for row: Dictionary in queue["processed"]:
if int(row["command_id"]) == command_id: processed.append(row)
_expect(processed.size() == 1, "one typed processed original")
if processed.size() == 1: _expect(var_to_bytes(processed[0]) == var_to_bytes(original), "typed processed original bytes")
_expect(var_to_bytes(before_replay) == var_to_bytes(after_replay), "same key replay changes no Runtime field")
_replay_reference = _write("resource-replay", {&"before": before_replay, &"after": after_replay, &"result": result, &"recorded_command": {&"key": key, &"command_id": command_id, &"original": original}})
_clock.advance_one_minute() _clock.advance_one_minute()
else: else:
_clock.advance_one_minute() _clock.advance_one_minute()
@@ -250,10 +272,16 @@ func _assert_before(point: Dictionary) -> void:
_expect(point[&"mana"] > 0 and point[&"facility"][&"power_available"] and point[&"related_facility_operating"], "initial power and dependency") _expect(point[&"mana"] > 0 and point[&"facility"][&"power_available"] and point[&"related_facility_operating"], "initial power and dependency")
_expect(point[&"service"][&"state"] == "ACTIVE" and point[&"service"][&"elapsed_minutes"] == _duration - 1, "natural pre-completion point") _expect(point[&"service"][&"state"] == "ACTIVE" and point[&"service"][&"elapsed_minutes"] == _duration - 1, "natural pre-completion point")
func _snapshot_staff_ids() -> Array[String]:
var result: Array[String] = []
for id: StringName in _recovery[&"staff_ids"]: result.append(String(id))
return result
func _assert_unconfounded(point: Dictionary) -> void: func _assert_unconfounded(point: Dictionary) -> void:
_expect(not point[&"operations_enabled"] and point[&"actual_coordinator_absent"], "isolated fixture") _expect(not point[&"operations_enabled"] and point[&"actual_coordinator_absent"], "isolated fixture")
_expect(point[&"facility"][&"operator_open"] and point[&"facility"][&"durability"] > 0 and point[&"event"].is_empty(), "no closure or destruction") _expect(point[&"facility"][&"operator_open"] and point[&"facility"][&"durability"] > 0 and point[&"event"].is_empty(), "no closure or destruction")
_expect(point[&"facility"][&"active_staff_ids"] == _recovery[&"staff_ids"], "no shift confound") var ids: Variant = point[&"facility"][&"active_staff_ids"]
_expect(ids is Array and ids.is_typed() and ids.get_typed_builtin() == TYPE_STRING and ids.size() == _recovery[&"staff_ids"].size() and ids == _snapshot_staff_ids() and var_to_bytes(ids) == var_to_bytes(_snapshot_staff_ids()) and _runtime.facility_service.get_instance(_facility_id).active_staff_ids == _recovery[&"staff_ids"], "no shift confound")
_expect(point[&"graph_invariants"][&"valid"], "graph invariants") _expect(point[&"graph_invariants"][&"valid"], "graph invariants")
if point[&"service"][&"state"] != "COMPLETED": if point[&"service"][&"state"] != "COMPLETED":
_expect(point[&"guests"][String(GUEST)][&"state"] == "RECEIVING_SERVICE" and point[&"guests"][String(GUEST)][&"planned_exit_at"] > point[&"minute"], "guest ownership and no exit") _expect(point[&"guests"][String(GUEST)][&"state"] == "RECEIVING_SERVICE" and point[&"guests"][String(GUEST)][&"planned_exit_at"] > point[&"minute"], "guest ownership and no exit")
@@ -6,9 +6,17 @@ Its output is preparation only and cannot replace the registered evidence gate.
""" """
import argparse import argparse
import hashlib import hashlib
import importlib.util
import struct
import json import json
from pathlib import Path from pathlib import Path
SERIES_PATH = Path(__file__).resolve().parents[1] / 'p04_native_series.py'
SERIES_SHA256 = hashlib.sha256(SERIES_PATH.read_bytes()).hexdigest()
_spec = importlib.util.spec_from_file_location('s5_power_series', SERIES_PATH)
series = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(series)
CASES = ("S5A_PURE", "S5B_DEPENDENCY") CASES = ("S5A_PURE", "S5B_DEPENDENCY")
GUEST = "p04.s5.guest.a" GUEST = "p04.s5.guest.a"
REQUEST = "p04.s5.request.a" REQUEST = "p04.s5.request.a"
@@ -46,6 +54,46 @@ class Raw:
self.pin(reference["var"]["path"], reference["var"]["sha256"]) self.pin(reference["var"]["path"], reference["var"]["sha256"])
return json.loads(self.pin(reference["json"]["path"], reference["json"]["sha256"]), object_pairs_hook=unique) return json.loads(self.pin(reference["json"]["path"], reference["json"]["sha256"]), object_pairs_hook=unique)
def typed(self, reference):
require(set(reference) == {"var", "json"}, "typed and JSON raw references required")
data = self.pin(reference["var"]["path"], reference["var"]["sha256"])
self.pin(reference["json"]["path"], reference["json"]["sha256"])
return series.variant(data)
def staff_check(point_node, recovery_node):
ids = point_node.at("facility").at("active_staff_ids")
expected = recovery_node.at("staff_ids")
for node, element in ((ids, 4), (expected, 21)):
require(node.kind == 28 and node.raw[:8] == struct.pack("<II", 28 | (1 << 16), element)
and all(member.kind == element for member in node.value), "typed staff representation")
require(len(ids.value) > 0 and ids.plain() == expected.plain(), "staff values/order/count")
def command_queue(snapshot):
return snapshot.at("systems").at("runtime").at("command_queue")
def duplicate_check(replay_node, recorded):
before, after = replay_node.at("before"), replay_node.at("after")
require(before.raw == after.raw, "resource replay changes no typed Runtime field")
result = replay_node.at("result")
require(result.at("code").value == "DUPLICATE_IDEMPOTENCY_KEY", "duplicate rejection code")
flag = result.at("is_replay")
require(flag.kind == 1 and flag.value is False, "duplicate is not replay")
details = result.at("details")
processed = details.at("processed")
require(processed.kind == 1 and processed.value is True, "original processed")
key, ident, original = recorded.at("key"), recorded.at("command_id"), recorded.at("original")
require(key.kind in (4, 21) and ident.kind == 2, "recorded command types")
require(details.at("idempotency_key").value == key.value
and details.at("original_command_id").value == ident.value
and details.at("original_result").raw == original.raw, "duplicate original binding")
queue = command_queue(before)
require(queue.at("idempotency").at(key.value).value == ident.value, "typed idempotency binding")
rows = [row for row in queue.at("processed").value if row.at("command_id").value == ident.value]
require(len(rows) == 1 and rows[0].raw == original.raw, "typed processed original")
def derive(root, writer_sha, reader_sha): def derive(root, writer_sha, reader_sha):
raw = Raw(root) raw = Raw(root)
@@ -58,12 +106,22 @@ def derive(root, writer_sha, reader_sha):
branch = raw.reference(writer["branches"][case]) branch = raw.reference(writer["branches"][case])
replay = raw.reference(reader["branches"][case]) replay = raw.reference(reader["branches"][case])
raw.reference(branch["commands"]) raw.reference(branch["commands"])
for role in (branch, replay): for role, reference in ((branch, writer["branches"][case]), (replay, reader["branches"][case])):
guard = raw.reference(role["resource_replay"]) typed_role = raw.typed(reference)
require(guard["result"]["is_replay"] is True and guard["result"]["code"] == "OK" guard = raw.typed(role["resource_replay"])
and guard["before"] == guard["after"], "resource replay changes no Runtime field") recorded = guard.at("recorded_command")
commands = (raw.typed(branch["commands"]).at("completed_including_fixture")
if role is branch else typed_role.at("replay_completed_commands"))
rows = [row.at("result") for row in commands.value
if row.at("result").at("command_id").value == recorded.at("command_id").value]
require(len(rows) == 1 and rows[0].raw == recorded.at("original").raw,
"recorded original cross-check against typed commands")
duplicate_check(guard, recorded)
points = {name: raw.reference(ref) for name, ref in branch["points"].items()} points = {name: raw.reference(ref) for name, ref in branch["points"].items()}
require(set(points) == {"before", "boundary", "held", "recovered", "duplicate_guard"}, "full point roster") require(set(points) == {"before", "boundary", "held", "recovered", "duplicate_guard"}, "full point roster")
for reference in branch["points"].values():
point = raw.typed(reference)
staff_check(point, point.at("inputs").at("recovery"))
before = points["before"] before = points["before"]
price, duration = before["inputs"]["price"], before["inputs"]["duration"] price, duration = before["inputs"]["price"], before["inputs"]["duration"]
require(price > 0 and duration > 1 and before["mana"] > 0, "paid natural service") require(price > 0 and duration > 1 and before["mana"] > 0, "paid natural service")
@@ -123,7 +181,7 @@ def derive(root, writer_sha, reader_sha):
{"dependency_stop": {"charges": 0, "progress_delta": 0, "reservation_ids_preserved": True, "service_id_preserved": True}}), {"dependency_stop": {"charges": 0, "progress_delta": 0, "reservation_ids_preserved": True, "service_id_preserved": True}}),
"recovery_completion": {"completions": 1, "charges": 1}}, "recovery_completion": {"completions": 1, "charges": 1}},
reason_sequence=reasons, resume_minute=resume) reason_sequence=reasons, resume_minute=resume)
return dict(cases=output, raw_sha256=raw.hashes, native_provenance_verified=False, return dict(cases=output, series_sha256=SERIES_SHA256, raw_sha256=raw.hashes, native_provenance_verified=False,
whole_goal_accepted=False, unit_accepted=False) whole_goal_accepted=False, unit_accepted=False)
@@ -0,0 +1,137 @@
#!/usr/bin/env python3
"""Pure actual-fixture controls. Root runs this with a fresh BUILD scratch."""
import argparse
import hashlib
import importlib.util
import json
from pathlib import Path
import struct
import sys
sys.dont_write_bytecode = True
spec = importlib.util.spec_from_file_location("s5_power_adapter", Path(__file__).with_name("s5_power_adapter.py"))
a = importlib.util.module_from_spec(spec)
spec.loader.exec_module(a)
def scalar(kind, value):
if kind in (4, 21):
data = value.encode()
return struct.pack("<II", kind, len(data)) + data + b"\0" * (-len(data) % 4)
return struct.pack("<Ii", kind, value)
def replace(node, target, replacement):
"""Re-encode only ancestors of the targeted typed node, preserving other bytes."""
if node is target:
return replacement
if node.kind == 27:
return node.raw[:8] + b"".join(replace(k, target, replacement) + replace(v, target, replacement)
for k, v in node.value)
if node.kind == 28:
start = 12 if node.raw[:4] == struct.pack("<I", 28 | (1 << 16)) else 8
return node.raw[:start] + b"".join(replace(v, target, replacement) for v in node.value)
return node.raw
def array(element, members):
return struct.pack("<III", 28 | (1 << 16), element, len(members)) + b"".join(members)
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--fixture-dir", required=True, type=Path)
parser.add_argument("--scratch", required=True, type=Path)
args = parser.parse_args()
fixture = args.fixture_dir.resolve(strict=True)
projects = {parent for origin in (Path(__file__).resolve(), fixture)
for parent in origin.parents if (parent / "project.godot").is_file()}
a.require(bool(projects), "project ancestry required for scratch exclusion")
index = json.loads((fixture / "INDEX.json").read_text(), object_pairs_hook=a.unique)
for name, row in index["files"].items():
path = fixture / name
a.require(path.is_file() and not path.is_symlink() and path.resolve().is_relative_to(fixture), "fixture path")
data = path.read_bytes()
a.require(len(data) == row["bytes"] and hashlib.sha256(data).hexdigest() == row["sha256"], "INDEX sha/bytes")
scratch = args.scratch
a.require(scratch.is_absolute() and not scratch.exists() and not scratch.is_symlink(), "explicit absolute fresh scratch")
parent = scratch.parent.resolve(strict=True)
# BUILD is the registered allocation, never inferred from fixture provenance.
build = Path("/Users/korizatou114/Projects/TheTree-managed-worktrees/p04-01/.runtime/P04-01/build").resolve(strict=True)
scratch = parent / scratch.name
a.require(parent.is_dir() and parent.is_relative_to(build)
and all(not scratch.is_relative_to(project) for project in projects) and not scratch.is_relative_to(fixture), "scratch inside BUILD outside project/fixture")
scratch.mkdir()
audit = json.loads((fixture / "TYPED_AUDIT.json").read_text())
results = []
def negative(label, data, check):
path = scratch / (label + ".var")
path.write_bytes(data)
try:
check(a.series.variant(path.read_bytes()))
except ValueError:
results.append(label)
return
raise ValueError("negative accepted: " + label)
for case in a.CASES:
point = a.series.variant((fixture / ("writer-" + case + "-before.var")).read_bytes())
replay = a.series.variant((fixture / ("writer-" + case + "-resource-replay.var")).read_bytes())
recovery = point.at("inputs").at("recovery")
a.staff_check(point, recovery)
queue = a.command_queue(replay.at("before"))
key = audit["cases"][case]["replay_result"]["details"]["idempotency_key"]
ident = queue.at("idempotency").at(key)
rows = [row for row in queue.at("processed").value if row.at("command_id").value == ident.value]
a.require(len(rows) == 1, "actual processed binding")
original = rows[0]
# Derived exclusively from actual typed queue, not duplicate result details.
recorded_bytes = struct.pack("<II", 27, 3) + scalar(4, "key") + scalar(4, key) + scalar(4, "command_id") + ident.raw + scalar(4, "original") + original.raw
recorded = a.series.variant(recorded_bytes)
a.duplicate_check(replay, recorded)
results.append(case + "-positive")
ids = point.at("facility").at("active_staff_ids")
members = [v.raw for v in ids.value]
staff_bad = {
"wrong-id": array(4, [scalar(4, "wrong")] + members[1:]),
"stringname": array(21, [scalar(21, v.value) for v in ids.value]),
"missing": array(4, members[:-1]),
"extra": array(4, members + [scalar(4, "extra")]),
}
if case == "S5B_DEPENDENCY":
staff_bad["swapped"] = array(4, members[::-1])
for label, data in staff_bad.items():
negative(case + "-staff-" + label, replace(point, ids, data),
lambda n: a.staff_check(n, recovery))
result = replay.at("result")
details = result.at("details")
mutations = [
("key", details.at("idempotency_key"), scalar(4, "wrong")),
("id", details.at("original_command_id"), scalar(2, -1)),
("original", details.at("original_result"), struct.pack("<II", 27, 0)),
("processed-false", details.at("processed"), scalar(1, 0)),
("code", result.at("code"), scalar(4, "OK")),
("replay", result.at("is_replay"), scalar(1, 1)),
]
for label, target, data in mutations:
negative(case + "-duplicate-" + label, replace(replay, target, data),
lambda n: a.duplicate_check(n, recorded))
pairs = [(k, v) for k, v in details.value if k.value != "processed"]
missing = struct.pack("<II", 27, len(pairs)) + b"".join(k.raw + v.raw for k, v in pairs)
negative(case + "-processed-missing", replace(replay, details, missing),
lambda n: a.duplicate_check(n, recorded))
after = replay.at("after")
target = a.command_queue(after).at("idempotency")
value = target.at(key)
changed = replace(replay, value, struct.pack("<If", 3, float(value.value)))
changed_node = a.series.variant(changed)
a.require(changed_node.plain() == replay.plain() and changed_node.at("after").raw != after.raw,
"typed-only mutation retains JSON plain value")
negative(case + "-typed-after", changed, lambda n: a.duplicate_check(n, recorded))
print(json.dumps(dict(accepted=False, root_only=True, checks=results,
series_sha256=a.SERIES_SHA256), sort_keys=True))
if __name__ == "__main__":
main()