Compare commits

..
Author SHA1 Message Date
Owner Coordinator aad6284fb7 P22 preserved finalization exact tree probe 2026-10-13 19:20:08 +09:00
7 changed files with 1355 additions and 129 deletions
@@ -2,11 +2,17 @@
原30パッケージ/88成果・部分点なしを維持する。614 identityは360施設、176人物、2状態群、76その他であり、614画像ではない。495 candidate、119 missing、19 sourceを全て保持する。選定済完成資産は0。D08/D09の最終内容・数量はunknown。
asset-contracts.jsonは公開JSON共有参照(d/l/s/r)で、accepted_inputの全フィールドとcurrent契約の全フィールドを損失なく保持する。各identityの命名はid、候補名はcandidate_refの実path、形式と現raw SHAはcandidate_pool、historical適用性はcandidate_refs、creatorと不足は各契約にある。差替鎖は歴史的representation→現candidate→未選定replacement。missingをcode/file/fallbackで充足しない。最終の解像度・frame/layout・命名・差替参照は未確定として扱う。
asset-contracts.jsonはp22-reference-contract-v2形式。immutable accepted input/current/evidenceをpathとraw SHAで参照し、全フィールドを損失なく再構成する。accepted inputのd/l/s/rをdecodeし、歴史的契約と現selections/source_pins/unit_proofsを分離する。missingをcode/file/fallbackで充足しない。最終の解像度・frame/layout・命名・差替参照は未確定。
Kenney Tiny Town/Tiny Dungeon/Interface Soundsのpublisher/member/license/manifest証拠はevidence_pinsの文書を完全再利用する。配布元URLとCC0記録はcandidate_poolおよび既存ライセンス証拠を参照。内部BGM合成recipeは内部由来であり、第三者CC0や最終聴感・権利受入へ昇格しない。自己制作・別許諾を排除しない。未知creator/rightsを推測しない。
checkerの通常0終了は記録整合性のみ。--require-readyは原条件未達なら2終了。READYには空でない完成資産、全exact useの選定・実利用・権利、未制作/開示不足の解消、必要な画面/聴感、Root静的検証と意味のあるnegative、独立Opusレビュー、統合・最終受入が必要。workerは正式試験も受入も主張しない。
checkerの通常0終了は記録整合性のみ。validation_readyには全exact useの使用可能選定資産・権利・必要な画面/聴感証拠とmandatory_scope/disclosure証拠が必要。独立レビュー・統合・最終受入を技術的readinessの事前ゲートにはしない。原成果受入には別途Root正式検証・意味のあるnegative・独立Opusレビュー・統合・finalが必要。
ローカル限定確認: python3 -Bで13テスト合格、ordinary exit 0、require-ready exit 2。614 pending、0 selected、19 candidate sources、disclosure/mandatory_scope不足。record_integrity=true、validation_ready=false、acceptance_claimed=false。正式試験・原成果受入は主張しない。
契約spec SHAはstdout.stripのUTF-8に対するf252de3c5cff7ebfa62322ae2f2df09fb80a44bc59eb286b76d4acb4c12ca875。raw SHA c26c0b12cf7e7a7c46e671acde2e00762904732c1cd6381a53836e3d7b3f9824との差は末尾改行のみ。
完全パケットの最終実測は未確認。指定patchと計測evidence/proposalがこのworktreeに存在せず、通常制作ディレクトリはアクセス拒否。487283はprospective値であり実測値として扱わない。500000未満の確認は未充足。
## 次の製品変更提案(この実装では適用しない)
@@ -0,0 +1,11 @@
# Root actual preflight and final bounded repair
{
"action": "request-repair",
"task_id": "P22-02",
"reason": "The Root preflight of c6ccc5a6 found three required defects (recorded in checkpoints/p22-02-slot-contract-r02/actual-checker-and-packet-preflight-r02.json). This uses the last ordinary repair (1 used, limit 2). Fix only within allowed_paths: execution/p22-02/** and tools/check_p22_asset_contracts.py. Do not change the spec, inputs, Root checkpoint, cap, contracts, registered validation argv, permissions, or Core.\n\n(1) The checker exits 2 on the exact registered argv `--verify-only --accepted-input docs/spec/inputs/P22-02-accepted-usage.json --require-ready`. Add argparse support for `--verify-only` and `--accepted-input` without changing the registration. Keep it read-only.\n\n(2) Readiness is hardcoded to always fail, which the spec forbids. The check forces selected_assets==[], asserts all 8 original_conditions are False before computing ready with all(values), and forces ledger.current to equal the frozen all-unmet snapshot. Remove these. Keep every field of the frozen accepted input, all 614 original identities and the source observations. Verify the new current, selection and actual evidence separately, and derive ready from that real data. Reject fake booleans, file-presence-only proof and promoted fallbacks. The current ledger is genuinely short of 614, so the real P22 run must still return a nonzero readiness exit.\n\nKeep validation-ready separate from original-completed. Formal validate covers the technical readiness of function, assets, rights and the required usability. Original acceptance comes later, through a separate review, then integration, then final. Remove the cycle that puts independent review or integration completion as a gate before validation. Do not record original acceptance ahead of time.\n\nProve reachability with pure functions: one isolated positive readiness case, plus negatives for wrong proof, wrong quantity and wrong pin. Apply meaningful negatives to: missing, duplicate and stale sources; dimension, layout, license and rights mismatches; counterfeit final; quantity; codec bool-as-index; cycles; duplicate keys; NaN/Inf; and path escape. Don't reject only sample mutations, and don't use a gate that requires the whole checkpoint to equal a mirror.\n\n(3) The exact unchanged Runner packet is actually 1,350,690 bytes; the worker's 472,408 figure was wrong. It must be under 500,000 bytes, counting the full diff, all sources, the registered validation_sources (the checker plus the 217 KiB accepted input) and the tests. Don't re-embed the whole accepted_input or the roughly 1.5 MB current data in the ledger. Rebuild losslessly from the existing immutable inputs, referenced by raw SHA, decoded semantic hash, or file plus identity. Keep all 19 candidate sources, each of the 614 selections, and every gap and required field. Cut the output down with shared profiles, text and contracts. Don't extend the Core's P13-only dedup to P22.\n\nRun python -B with ast.parse only; no engine or full regression. Return submitted with the exact list of incomplete original-unit conditions. Do not claim P22-02 acceptance. No duplicate worker and no new registration.",
"codex_model": "gpt-6-astra",
"codex_model_reason": "This is the last ordinary repair, so a failure leaves no fallback. The work is hard: three defects at once, redesigning the readiness logic so it is reachable but still fails nonzero on the real data, untangling the validate/review/integration cycle, and a lossless decoder rebuild that shrinks the packet from 1.35 MB to under 500 KB. It also needs many meaningful negatives. The previous worker misreported the packet size and shipped always-fail logic, which suggests the default model underestimates this task. Astra's quality is worth its extra time and usage."
}
No spec/input/core/cap changes. Preserve every accepted input field through immutable references, not a second embedded mirror. Exact Root packet body is in artifacts as diagnostic only; normal Runner packet uses full diff and source with validation sources, no P22 added-doc dedup. Current real readiness remains nonzero, but valid future proof must be able to reach readiness without assuming prior final integration. Root will check the exact body again before any reviewer launch.
File diff suppressed because one or more lines are too long
@@ -1,4 +1,8 @@
"""Build only the authorized ledger, preserving complete JSON fields."""
"""Initialize a compact ledger. Not invoked by read-only verification.
All historic fields are retained by raw-SHA references. This initializer
replaces selection work; never use it to refresh pins after a source change.
"""
import hashlib
import json
from pathlib import Path
@@ -9,67 +13,21 @@ INPUT = 'docs/spec/inputs/P22-02-accepted-usage.json'
CURRENT = 'docs/plans/original-completion-2026-10-02/checkpoints/p22-02-slot-contract-r02/all-slot-current-candidate-contracts.json'
EVIDENCE = 'docs/plans/original-completion-2026-10-02/checkpoints/p22-02-current-contract-r01'
def build():
keys, strings, pool = [], [], []
key_ids, string_ids, object_ids = {}, {}, {}
def key_id(key):
if key not in key_ids:
key_ids[key] = len(keys)
keys.append(key)
return key_ids[key]
def encode(value):
if isinstance(value, str):
if value not in string_ids:
string_ids[value] = len(strings)
strings.append(value)
return ['s', string_ids[value]]
if isinstance(value, (dict, list)):
signature = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':'))
if signature in object_ids:
return ['r', object_ids[signature]]
if isinstance(value, dict):
node = ['d']
for key, item in value.items():
node.extend([key_id(key), encode(item)])
else:
node = ['l'] + [encode(item) for item in value]
index = len(pool)
pool.append(node)
object_ids[signature] = index
return ['r', index]
return value
pins = {}
for path in [ROOT / CURRENT, *sorted((ROOT / EVIDENCE).iterdir())]:
if path.is_file():
pins[str(path.relative_to(ROOT))] = hashlib.sha256(path.read_bytes()).hexdigest()
paths = [INPUT, CURRENT] + [str(p.relative_to(ROOT)) for p in sorted((ROOT / EVIDENCE).iterdir()) if p.is_file()]
pins = {p: hashlib.sha256((ROOT / p).read_bytes()).hexdigest() for p in paths}
current = json.loads((ROOT / CURRENT).read_text())
ledger = {
'accepted_input': json.loads((ROOT / INPUT).read_text()),
'current': json.loads((ROOT / CURRENT).read_text()),
'evidence_pins': pins,
'original_conditions': {
'all_exact_uses_selected_and_usable': False,
'all_creator_and_rights_evidence_accepted': False,
'all_naming_resolution_layout_replacement_references_accepted': False,
'all_mandatory_unmade_and_disclosure_gaps_resolved': False,
'applicable_display_and_hearing_evidence': False,
'independent_opus_review': False,
'root_static_and_meaningful_negative_validation': False,
'integration_and_final_original_30_88_acceptance': False,
},
'selection_contract': {
'selected_assets': [],
'candidate_names': 'Exact candidate_pool path names; no rename or final assignment',
'replacement_chain': 'Accepted historical representation -> current candidate_ref (prior applicability retained) -> selected replacement UNKNOWN',
'format': 'candidate_pool observed_format for physical media; representation_kind and exact raw source for code. Final per-use format/layout UNKNOWN.',
'rights': 'candidate_pool rights_provenance and complete pinned publisher/license/manifest evidence; internal recipe is origin only, not final rights or hearing acceptance',
'quantity': 'D08/D09 UNKNOWN. 2 state groups are not images. Never infer final file/frame/variant counts from 614 identities.',
},
'schema': 'p22-reference-contract-v2',
'references': pins,
'reconstruction': 'All accepted codec fields and snapshot fields retained by raw SHA; selections inherit entire snapshot contracts by identity, including every unmade condition.',
'selections': {c['id']: ('candidate' if c['candidate_refs'] else 'missing') for c in current['contracts']},
'source_pins': {p: f['current_raw_sha256'] for p, f in current['candidate_pool'].items()},
'unit_proofs': {},
}
files = encode(ledger)
packet = {'keys': keys, 'strings': strings, 'pool': pool, 'files': files, 'encoding': 'd=flat key-id/value; l=list; s=string index; r=complete object/list index'}
raw = json.dumps(packet, ensure_ascii=False, separators=(',', ':')).encode() + b'\n'
(HERE / 'asset-contracts.json').write_bytes(raw)
print(json.dumps({'ledger_bytes': len(raw), 'formal_tests': 'not_run'}))
(HERE / 'asset-contracts.json').write_text(json.dumps(ledger, ensure_ascii=False, indent=2) + '\n')
if __name__ == '__main__':
build()
@@ -0,0 +1,195 @@
"""Runner-owned tests. No file writes, product execution, or real PASS from fixtures."""
import copy
import importlib.util
import io
import json
from pathlib import Path
import sys
import unittest
from unittest import mock
sys.dont_write_bytecode = True
ROOT = Path(__file__).resolve().parents[5]
spec = importlib.util.spec_from_file_location('p22_checker', ROOT / 'tools/check_p22_asset_contracts.py')
m = importlib.util.module_from_spec(spec)
spec.loader.exec_module(m)
class ContractTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.ledger = m.read(m.OUT + 'asset-contracts.json')
cls.accepted = m.read(m.INPUT)
_, cls.records = m.accepted_records(cls.accepted)
cls.snapshot = m.read(m.CURRENT)
def rejected(self, function, *args):
with self.assertRaises((ValueError, OSError, KeyError, TypeError)):
function(*args)
def proof(self, kind, subject):
return {'kind': kind, 'subject': copy.deepcopy(subject), 'result': 'accepted',
'issuer': {'role': 'root', 'name': 'isolated fixture; never P22 evidence'},
'observations': [{'explanation': 'Synthetic pure-function observation only',
'anchors': [{'path': 'synthetic.txt', 'sha256': 'a' * 64,
'line': 1, 'text': 'synthetic'}]}]}
def fixture(self):
row = {'slot_id': 'isolated', 'usage_condition': 'isolated pure function only',
'required_file_count': 1}
fmt = {'png_width': 16, 'png_height': 16}
sources = {'synthetic.png': {'sha256': 'a' * 64, 'format': fmt}}
asset = {'path': 'synthetic.png', 'sha256': 'a' * 64, 'format': copy.deepcopy(fmt),
'region': [0, 0, 16, 16], 'code_anchor': None, 'kind': 'image'}
return row, sources, asset
def test_isolated_reachable_and_wrong_proof_quantity_pin(self):
row, sources, asset = self.fixture()
m.asset_matches(asset, row, sources)
quantities = dict.fromkeys(m.QUANTITIES)
quantities['required_file_count'] = 1
m.quantities_valid(quantities, row)
subject = {'asset': asset, 'quantities': quantities, 'record_sha': m.semantic(row)}
fit = self.proof('fit', subject)
selected = {'isolated': m.proof_matches(fit, 'fit', subject)}
unit = {kind: self.proof(kind, subject) for kind in ('mandatory_scope', 'disclosure')}
self.assertTrue(m.technical_readiness(['isolated'], selected, unit, subject))
# Never run check() against these fixtures or count them as real assets.
bad = copy.deepcopy(fit)
bad['subject']['asset']['sha256'] = 'b' * 64
self.rejected(m.proof_matches, bad, 'fit', subject)
bad_quantity = dict(quantities, required_file_count=2)
self.rejected(m.quantities_valid, bad_quantity, row)
bad_asset = dict(asset, sha256='b' * 64)
self.rejected(m.asset_matches, bad_asset, row, sources)
self.rejected(m.proof_matches, True, 'fit', subject)
def test_nonvacuous_and_unmade_denominator(self):
self.rejected(m.technical_readiness, [], {}, {}, {})
self.rejected(m.technical_readiness, ['made', 'unmade'], {'made': True}, {}, {})
self.assertFalse(m.technical_readiness(['made', 'unmade'], {'made': True, 'unmade': False}, {}, {}))
self.rejected(m.technical_readiness, ['x'], {'x': 'true'}, {}, {})
def test_format_layout_state_and_quantity(self):
row, sources, asset = self.fixture()
for bad in (dict(asset, format={'png_width': 17, 'png_height': 16}),
dict(asset, region=[0, 0, 17, 16]),
dict(asset, region=[True, 0, 16, 16]),
dict(asset, kind='placeholder_media')):
self.rejected(m.asset_matches, bad, row, sources)
self.rejected(m.asset_matches, asset, {'group_id': 'state'}, sources)
self.rejected(m.quantities_valid, dict.fromkeys(m.QUANTITIES, True), row)
def test_codec_and_strict_json(self):
doc = {'keys': ['x'], 'strings': ['s'], 'pool': [], 'files': ['d', 0, 1]}
self.assertEqual(m.decode(doc), {'x': 1})
for files in (['s', True], ['s', -1], ['d', 0, 1, 0, 2],
['unknown'], ['r', 0], float('nan'), float('inf')):
self.rejected(m.decode, dict(doc, files=files))
self.rejected(m.decode, dict(doc, pool=[['l', ['r', 0]]], files=['r', 0]))
self.rejected(m.decode, dict(doc, pool=[['s', 0]], files=['r', 0]))
for text in ('{"x":1,"x":2}', '{"x":NaN}', '{"x":Infinity}', '{"x":1e999}'):
self.rejected(m.loads, text)
def test_path_escape(self):
for path in ('../escape', '/tmp/escape', 'assets/../../escape'):
self.rejected(m.safe_path, path)
def test_real_coverage_and_source_omission(self):
for action in ('omit_identity', 'duplicate_identity', 'omit_source', 'duplicate_source'):
d = copy.deepcopy(self.snapshot)
if action == 'omit_identity':
d['contracts'].pop()
elif action == 'duplicate_identity':
d['contracts'].append(d['contracts'][0])
elif action == 'omit_source':
d['candidate_pool'].pop(next(iter(d['candidate_pool'])))
else:
d['contracts'][0]['candidate_refs'].append(d['contracts'][0]['candidate_refs'][0])
self.rejected(m.reference_contracts, d, self.records)
def test_real_source_missing_and_stale(self):
path, digest = next(iter(self.ledger['source_pins'].items()))
self.rejected(m.pinned, path, '0' * 64)
self.rejected(m.pinned, m.OUT + 'nonexistent-proof.json', digest)
def test_real_png_crc_and_layout(self):
path = 'assets/kenney/tiny_town/Tilemap/tilemap_packed.png'
data = m.raw(path)
fmt = m.png_format(data)
self.assertEqual((fmt['png_width'], fmt['png_height']), (192, 176))
damaged = bytearray(data)
damaged[20] ^= 1
self.rejected(m.png_format, bytes(damaged))
sources = {path: {'sha256': m.sha(data), 'format': fmt}}
asset = {'path': path, 'sha256': m.sha(data), 'format': fmt,
'region': [191, 0, 16, 16], 'code_anchor': None, 'kind': 'image'}
self.rejected(m.asset_matches, asset, {}, sources)
def test_real_reference_mismatch(self):
a = copy.deepcopy(self.snapshot['contracts'][0]['source_anchors'][0])
a['text'] += 'not the actual reference'
self.rejected(m.anchor, a)
def test_real_license_bytes_and_rights_subject(self):
pack = m.read(m.EVIDENCE + 'publisher-archive-byte-verification.json')['packs'][0]
license_row = next(x for x in pack['members'] if x['archive_member'] == 'License.txt')
path, digest = license_row['repo_path'], license_row['raw_sha256']
original_raw = m.raw
with mock.patch.object(m, 'raw', side_effect=lambda p: b'fake license' if p == path else original_raw(p)):
self.rejected(m.pinned, path, digest)
subject = {'source': next(iter(self.ledger['source_pins'])),
'rights': {'creator': 'unknown', 'license': 'unknown'}}
proof = self.proof('rights', subject)
changed = copy.deepcopy(subject)
changed['rights']['license'] = 'CC0'
self.rejected(m.proof_matches, proof, 'rights', changed)
def test_real_false_finished_and_fake_booleans(self):
c = self.snapshot['contracts'][0]
row = self.records[c['id']][1]
for value in (True, False, [], {'state': 'finished'}, 'selected'):
self.rejected(m.selection_ready, c['id'], row, c, value, {})
self.rejected(m.receipt, {'path': m.OUT + 'validation.json', 'sha256': '0' * 64, 'index': 0}, 'fit', {})
self.rejected(m.receipt, True, 'fit', {})
def test_real_selected_contract_unknown_rights_and_forged_proof(self):
c = self.snapshot['contracts'][0]
row = self.records[c['id']][1]
path = 'assets/kenney/tiny_town/Tilemap/tilemap_packed.png'
data = m.raw(path)
sources = {path: {'sha256': m.sha(data), 'format': m.png_format(data)}}
product = 'scripts/ui/pixel_tower_art.gd'
product_data = m.raw(product)
ref = {'path': product, 'sha256': m.sha(product_data), 'line': 1,
'text': product_data.decode().splitlines()[0]}
rights_path = 'assets/kenney/tiny_town/License.txt'
s = {'asset': {'path': path, 'sha256': m.sha(data), 'format': sources[path]['format'],
'region': [0, 0, 16, 16], 'code_anchor': None, 'kind': 'image'},
'creator': 'Fixture declaration, not a real attribution',
'rights': {'license': 'unknown', 'origin': 'fixture only',
'documents': [{'path': rights_path, 'sha256': m.sha(m.raw(rights_path))}]},
'name': Path(path).name, 'replacement_chain': [c['id'], path], 'references': [ref],
'quantities': dict.fromkeys(m.QUANTITIES), 'quantity_basis': 'Fixture, no final count',
'usability': {'display': 'Requires actual exact-use display', 'hearing': 'Fixture image'}}
choice = {'state': 'selected', 'contract': s, 'proofs': dict.fromkeys(m.PROOF_KINDS, True)}
with self.assertRaisesRegex(ValueError, 'unknown license'):
m.selection_ready(c['id'], row, c, choice, sources)
s['rights']['license'] = 'CC0'
with self.assertRaisesRegex(ValueError, 'proof reference schema'):
m.selection_ready(c['id'], row, c, choice, sources)
def test_registered_cli_real_data_is_not_ready(self):
argv = ['checker', '--verify-only', '--accepted-input', m.INPUT, '--require-ready']
with mock.patch.object(sys, 'argv', argv), mock.patch('sys.stdout', new_callable=io.StringIO) as output:
code = m.main()
result = json.loads(output.getvalue())
self.assertEqual(code, 2, result)
self.assertTrue(result['record_integrity'])
self.assertFalse(result['validation_ready'])
self.assertEqual(result['pending_identity_count'], 614)
self.assertFalse(result['acceptance_claimed'])
if __name__ == '__main__':
unittest.main()
@@ -3,15 +3,60 @@
"formal_tests": "not_run; Runner owns execution",
"original_unit_ready": false,
"acceptance_claimed": false,
"spec_comparison": "UTF8 decode.strip.encode SHA only for contract spec",
"spec_comparison": "SHA256 of UTF-8 stdout.strip; raw file differs only by trailing newline",
"spec_sha256": "f252de3c5cff7ebfa62322ae2f2df09fb80a44bc59eb286b76d4acb4c12ca875",
"spec_raw_sha256": "c26c0b12cf7e7a7c46e671acde2e00762904732c1cd6381a53836e3d7b3f9824",
"accepted_input_raw_sha256": "45fa52acf0c8bd4dd58c3b40b681cb92066f05d617df157fc18370a5e8816281",
"ledger_bytes": 453678,
"packet_bytes_before_this_validation_metadata": 472408,
"ledger_bytes": 35441,
"complete_packet_bytes": null,
"complete_packet_verified_under_limit": false,
"packet_measurement_blocker": "Required patch and packet evidence/proposal absent from this worktree; normal project reference access denied. 487283 is prospective, not measured.",
"syntax_ast_parse": "passed; not formal validation",
"complete_packet_limit": 500000,
"runner_required": ["exact static validation", "meaningful negatives", "packet byte total", "applicable display and hearing", "independent Opus review", "integration and final original acceptance"],
"expected_exit_codes": {"record_integrity_if_valid": 0, "require_ready_with_current_unmet_conditions": 2, "integrity_error": 1},
"original_incomplete": ["614 exact uses lack completed selected usable assets; 495 candidates and 119 missing are not completion", "final per-use naming/resolution/frame/layout/replacement references", "creator and rights gaps including 20 media; internal BGM origin is not final approval", "mandatory unmade/disclosure gaps; D08/D09 content and quantities unknown", "applicable display/hearing evidence", "Root static validation and meaningful negatives", "independent Opus review", "integration and final original 30 packages/88 delivery units acceptance"]
"runner_required": [
"exact static validation",
"meaningful negatives",
"packet byte total",
"applicable display and hearing",
"independent Opus review",
"integration and final original acceptance"
],
"expected_exit_codes": {
"record_integrity_if_valid": 0,
"require_ready_with_current_unmet_conditions": 2,
"integrity_error": 1
},
"original_incomplete": [
"614 exact uses lack completed selected usable assets; 495 candidates and 119 missing are not completion",
"final per-use naming/resolution/frame/layout/replacement references",
"creator and rights gaps including 20 media; internal BGM origin is not final approval",
"mandatory unmade/disclosure gaps; D08/D09 content and quantities unknown",
"applicable display/hearing evidence",
"Root static validation and meaningful negatives",
"independent Opus review",
"integration and final original 30 packages/88 delivery units acceptance"
],
"local_bounded_checks": {
"interpreter": "python3 -B (python unavailable)",
"ast_parse": "passed: checker, build_contract.py, test_contract.py",
"test_count": 13,
"test_exit_code": 0,
"ordinary_exit_code": 0,
"require_ready_exit_code": 2,
"formal_runner_execution": false
},
"checker_output": {
"acceptance_claimed": false,
"accepted_identity_count": 614,
"candidate_source_count": 19,
"original_completion": "not_claimed; requires formal validate, independent review, integration, final",
"pending_identity_count": 614,
"record_integrity": true,
"selected_usable_count": 0,
"unit_proofs_missing": [
"disclosure",
"mandatory_scope"
],
"validation_ready": false
}
}
+419 -61
View File
@@ -1,8 +1,12 @@
#!/usr/bin/env python3
"""Read-only P22 record integrity; exit 2 for unmet original readiness."""
"""Read-only P22 record integrity; exit 2 for unmet technical readiness."""
import argparse
import hashlib
import json
import io
import math
import wave
import zlib
from pathlib import Path
import struct
import sys
@@ -22,7 +26,7 @@ def sha(raw):
return hashlib.sha256(raw).hexdigest()
def semantic(value):
return sha(json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':')).encode())
return sha(json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':'), allow_nan=False).encode())
def pairs(items):
result = {}
@@ -31,12 +35,27 @@ def pairs(items):
result[key] = value
return result
def invalid_constant(value):
raise ValueError('nonfinite JSON number: ' + value)
def finite_float(value):
number = float(value)
need(math.isfinite(number), 'nonfinite JSON number')
return number
def loads(value):
return json.loads(value, object_pairs_hook=pairs, parse_constant=invalid_constant, parse_float=finite_float)
def read(path):
return json.loads((ROOT / path).read_text(), object_pairs_hook=pairs)
return loads(raw(path))
def decode(doc, node=None):
need(type(doc) is dict and all(type(doc.get(k)) is list for k in ('keys', 'strings', 'pool')), 'wrong codec tables')
need(all(type(k) is str for k in doc['keys']) and len(set(doc['keys'])) == len(doc['keys']), 'duplicate/wrong key table')
need(all(type(s) is str for s in doc['strings']), 'wrong string table')
active = set()
def index(values, i):
need(type(i) is int and 0 <= i < len(values), 'bad reference index')
@@ -44,6 +63,7 @@ def decode(doc, node=None):
def walk(n):
if not isinstance(n, list):
need(n is None or type(n) in (str, int, float, bool), 'wrong scalar type')
need(type(n) is not float or math.isfinite(n), 'nonfinite codec scalar')
return n
need(bool(n) and type(n[0]) is str, 'wrong node type')
tag = n[0]
@@ -64,7 +84,7 @@ def decode(doc, node=None):
return value
i = n[1]
value = index(doc['pool'], i)
need(i not in active and isinstance(value, list) and value[0] in ('d', 'l'), 'cycle or wrong shared object type')
need(i not in active and isinstance(value, list) and value and value[0] in ('d', 'l'), 'cycle or wrong shared object type')
active.add(i)
result = walk(value)
active.remove(i)
@@ -74,82 +94,420 @@ def decode(doc, node=None):
walk(['r', i])
return result
def check(path):
spec = (ROOT / 'docs/spec/acceptance/P22-02.json').read_bytes()
need(sha(spec.decode().strip().encode()) == SPEC_SHA, 'contract spec SHA mismatch')
need(sha((ROOT / INPUT).read_bytes()) == INPUT_SHA, 'accepted input raw SHA mismatch')
accepted = read(INPUT)
# A pinned historical observation is retained, never treated as current readiness.
EVIDENCE = 'docs/plans/original-completion-2026-10-02/checkpoints/p22-02-current-contract-r01/'
CHECKPOINTS = 'docs/plans/original-completion-2026-10-02/checkpoints/'
CURRENT_SHA = '5cf93b6b3fd9d0cc060075878804f65195c0d90711c969cc36e2f58f3a1917fd'
QUANTITIES = ('required_file_count', 'required_frame_count', 'required_variant_count')
PROOF_KINDS = ('fit', 'rights', 'layout', 'quantity', 'gaps', 'usability')
def safe_path(path):
need(type(path) is str and path and not Path(path).is_absolute(), 'invalid relative path')
need('..' not in Path(path).parts, 'path escape')
resolved = (ROOT / path).resolve()
need(resolved != ROOT and ROOT in resolved.parents, 'path escape or symlink escape')
return resolved
def raw(path):
return safe_path(path).read_bytes()
def pinned(path, digest):
need(type(digest) is str and len(digest) == 64 and all(c in '0123456789abcdef' for c in digest), 'invalid SHA')
value = raw(path)
need(sha(value) == digest, 'raw SHA mismatch: ' + path)
return value
def exact(a, b):
# Python equality would otherwise equate True and 1.
return semantic(a) == semantic(b)
def anchor(a):
need(type(a) is dict and set(a) == {'path', 'sha256', 'line', 'text'}, 'anchor schema')
lines = pinned(a['path'], a['sha256']).decode('utf-8').splitlines()
need(type(a['line']) is int and 1 <= a['line'] <= len(lines), 'anchor line')
need(type(a['text']) is str and a['text'].strip() and lines[a['line'] - 1] == a['text'], 'reference mismatch')
def png_format(data):
need(data[:8] == b'\x89PNG\r\n\x1a\n', 'PNG signature')
pos, chunks, payload, palette, transparent = 8, [], bytearray(), False, False
width = height = depth = color = interlace = None
while pos < len(data):
need(pos + 12 <= len(data), 'PNG truncated chunk')
size = struct.unpack('>I', data[pos:pos + 4])[0]
kind = data[pos + 4:pos + 8]
end = pos + size + 12
need(end <= len(data), 'PNG chunk length')
body = data[pos + 8:end - 4]
crc = struct.unpack('>I', data[end - 4:end])[0]
need(zlib.crc32(kind + body) & 0xffffffff == crc, 'PNG CRC')
if not chunks:
need(kind == b'IHDR' and size == 13, 'PNG IHDR')
width, height, depth, color, comp, filt, interlace = struct.unpack('>IIBBBBB', body)
need(width > 0 and height > 0 and comp == filt == 0 and interlace in (0, 1), 'PNG dimensions/encoding')
depths = {0: (1, 2, 4, 8, 16), 2: (8, 16), 3: (1, 2, 4, 8), 4: (8, 16), 6: (8, 16)}
need(color in depths and depth in depths[color], 'PNG color/depth')
else:
need(kind != b'IHDR', 'duplicate PNG IHDR')
if kind == b'PLTE':
need(not payload and size > 0 and size % 3 == 0 and size <= 768, 'PNG palette')
palette = True
if kind == b'tRNS':
transparent = True
if kind == b'IDAT':
need(not payload or chunks[-1] == b'IDAT', 'PNG noncontiguous image data')
payload.extend(body)
chunks.append(kind)
pos = end
if kind == b'IEND':
need(size == 0 and pos == len(data), 'PNG trailing bytes')
break
need(chunks[-1:] == [b'IEND'] and payload and (color != 3 or palette), 'PNG missing image/palette/end')
# Bound decoded length by IHDR; validate scanline sizes and filter tags.
channels = {0: 1, 2: 3, 3: 1, 4: 2, 6: 4}[color]
passes = [(0, 0, 1, 1)] if interlace == 0 else [(0, 0, 8, 8), (4, 0, 8, 8), (0, 4, 4, 8), (2, 0, 4, 4), (0, 2, 2, 4), (1, 0, 2, 2), (0, 1, 1, 2)]
rows = []
for x, y, dx, dy in passes:
w, h = max(0, (width - x + dx - 1) // dx), max(0, (height - y + dy - 1) // dy)
if w and h:
rows.extend([1 + (w * channels * depth + 7) // 8] * h)
expected = sum(rows)
need(expected <= 268435456, 'PNG exceeds verification memory bound')
decoder = zlib.decompressobj()
pixels = decoder.decompress(bytes(payload), expected + 1)
need(decoder.eof and not decoder.unused_data and len(pixels) == expected, 'PNG scanline length')
offset = 0
for length in rows:
need(pixels[offset] <= 4, 'PNG scanline filter')
offset += length
return {'png_width': width, 'png_height': height, 'png_color_type': color,
'png_crc_verified': True, 'png_alpha_encoding_present': color in (4, 6) or transparent}
def observe(path, data):
if path.endswith('.png'):
return png_format(data)
if path.endswith('.wav'):
with wave.open(io.BytesIO(data), 'rb') as w:
need(w.getcomptype() == 'NONE', 'WAV compression')
return {'channels': w.getnchannels(), 'sample_rate': w.getframerate(),
'sample_width_bytes': w.getsampwidth(), 'frames': w.getnframes()}
if path.endswith('.ogg'):
need(data[:4] == b'OggS' and b'\x01vorbis' in data[:128], 'Ogg/Vorbis representation')
return {'encoding': 'ogg-vorbis', 'bytes': len(data)}
data.decode('utf-8')
return {'representation': 'code', 'bytes': len(data)}
def accepted_records(accepted):
original = decode(accepted)
need(set(original) == set(accepted['source_files']), 'accepted file coverage')
records = {}
for name, rows in original.items():
need(type(rows) is list, 'decoded file must be a list')
need(type(rows) is list, 'decoded file must be list')
pin = accepted['source_files'][name]
need(len(rows) == pin['records'] and semantic(rows) == pin['semantic_sha256'], 'decoded semantic file mismatch: ' + name)
need(type(pin['records']) is int and len(rows) == pin['records'] and semantic(rows) == pin['semantic_sha256'], 'decoded semantic mismatch: ' + name)
for row in rows:
need(type(row) is dict, 'accepted row type')
identity = row.get('slot_id', row.get('group_id'))
need(type(identity) is str and identity not in records, 'duplicate or invalid identity')
need(type(identity) is str and identity not in records, 'duplicate/invalid identity')
records[identity] = (name, row)
need(len(records) == 614, 'accepted identity coverage')
packed = read(path)
ledger = decode(packed)
need(ledger['accepted_input'] == accepted, 'lossless accepted input mismatch')
current = read(CURRENT)
need(ledger['current'] == current, 'current contract fields changed/omitted or false promotion')
evidence_dir = ROOT / 'docs/plans/original-completion-2026-10-02/checkpoints/p22-02-current-contract-r01'
expected_evidence = {CURRENT} | {str(p.relative_to(ROOT)) for p in evidence_dir.iterdir() if p.is_file()}
need(set(ledger['evidence_pins']) == expected_evidence, 'evidence omission/extra')
selection = ledger['selection_contract']
need(selection['selected_assets'] == [], 'unsupported selected asset promotion')
need('D08/D09 UNKNOWN' in selection['quantity'] and '2 state groups are not images' in selection['quantity'], 'invented quantities/state image promotion')
condition_keys = {'all_exact_uses_selected_and_usable', 'all_creator_and_rights_evidence_accepted', 'all_naming_resolution_layout_replacement_references_accepted', 'all_mandatory_unmade_and_disclosure_gaps_resolved', 'applicable_display_and_hearing_evidence', 'independent_opus_review', 'root_static_and_meaningful_negative_validation', 'integration_and_final_original_30_88_acceptance'}
need(set(ledger['original_conditions']) == condition_keys and all(v is False for v in ledger['original_conditions'].values()), 'unsupported original evidence promotion')
for evidence, digest in ledger['evidence_pins'].items():
need(sha((ROOT / evidence).read_bytes()) == digest, 'evidence raw SHA mismatch: ' + evidence)
seen = set()
used = set()
for contract in current['contracts']:
identity = contract['id']
need(identity in records and identity not in seen, 'contract identity coverage/duplicate')
seen.add(identity)
name, record = records[identity]
need(contract['source_file'] == name and semantic(record) == contract['accepted_record_semantic_sha256'], 'accepted record mismatch')
for anchor in contract['source_anchors']:
raw = (ROOT / anchor['path']).read_bytes()
need(sha(raw) == anchor['sha256'], 'anchor raw SHA')
need(raw.decode().splitlines()[anchor['line'] - 1] == anchor['text'], 'anchor line/reference mismatch')
for ref in contract['candidate_refs']:
need(len(records) == 614 and accepted['decoded_record_count'] == 614, 'accepted coverage')
need(sum('group_id' in row for _, row in records.values()) == 2, 'state groups')
return original, records
def proof_matches(proof, kind, subject):
"""Pure proof predicate; callers supply a pinned Root/owner receipt, not booleans."""
need(type(proof) is dict and set(proof) == {'kind', 'subject', 'result', 'issuer', 'observations'}, 'proof schema')
need(proof['kind'] == kind and exact(proof['subject'], subject), 'wrong proof binding')
need(proof['result'] == 'accepted', 'proof is not accepted')
issuer = proof['issuer']
need(type(issuer) is dict and set(issuer) == {'role', 'name'}, 'proof issuer schema')
need(issuer['role'] in ('root', 'owner') and type(issuer['name']) is str and issuer['name'].strip(), 'proof authority')
need(type(proof['observations']) is list and proof['observations'], 'proof has no observations')
for observation in proof['observations']:
need(type(observation) is dict and set(observation) == {'explanation', 'anchors'}, 'observation schema')
need(type(observation['explanation']) is str and observation['explanation'].strip(), 'empty observation')
need(type(observation['anchors']) is list and observation['anchors'], 'no observation anchors')
return True
def receipt(ref, kind, subject):
# Workers cannot manufacture approval inside their own output directory.
need(type(ref) is dict and set(ref) == {'path', 'sha256', 'index'}, 'proof reference schema')
need(ref['path'].startswith(CHECKPOINTS), 'proof must be a Root/owner checkpoint')
resolved = safe_path(ref['path'])
need((ROOT / CHECKPOINTS).resolve() in resolved.parents, 'proof symlink outside checkpoints')
document = loads(pinned(ref['path'], ref['sha256']))
need(type(document) is dict and document.get('schema') == 'p22-evidence-v1', 'receipt schema')
proofs = document['proofs']
need(type(proofs) is list and type(ref['index']) is int and 0 <= ref['index'] < len(proofs), 'proof index')
proof = proofs[ref['index']]
proof_matches(proof, kind, subject)
for observation in proof['observations']:
for a in observation['anchors']:
anchor(a)
return proof
def reference_contracts(snapshot, records):
contracts, used = {}, set()
for c in snapshot['contracts']:
identity = c['id']
need(identity in records and identity not in contracts, 'contract coverage/duplicate')
name, row = records[identity]
need(c['source_file'] == name and c['accepted_record_semantic_sha256'] == semantic(row), 'contract accepted record')
need(exact(c['source_anchors'], row['source_anchors']) and exact(c['usage_condition'], row['usage_condition']), 'accepted anchors/usage omitted')
for a in c['source_anchors']:
anchor(a)
refs = c['candidate_refs']
need(len({r['candidate_ref'] for r in refs}) == len(refs), 'duplicate candidate')
need({r['candidate_ref'] for r in refs} == {r['path'] for r in row['representations']}, 'source omission')
historical = {r['path']: r for r in row['representations']}
for ref in refs:
source = ref['candidate_ref']
need(source in current['candidate_pool'], 'omitted candidate source')
need(source in snapshot['candidate_pool'], 'source absent from pool')
used.add(source)
actual = current['candidate_pool'][source]['current_raw_sha256']
need(ref['prior_source_still_equal'] == (ref['prior_raw_sha256'] == actual), 'historical applicability mismatch')
need(seen == set(records), 'omitted identities')
need(used == set(current['candidate_pool']) and len(used) == 19, 'source coverage')
for source, fact in current['candidate_pool'].items():
raw = (ROOT / source).read_bytes()
need(sha(raw) == fact['current_raw_sha256'], 'current source raw SHA: ' + source)
fmt = fact['observed_format']
if source.endswith('.png'):
need(raw[:8] == b'\x89PNG\r\n\x1a\n', 'PNG signature')
width, height = struct.unpack('>II', raw[16:24])
need((width, height) == (fmt['png_width'], fmt['png_height']), 'dimension mismatch')
pending = [c['id'] for c in current['contracts'] if not c['original_asset_completion'] or c['missing_conditions'] or c['finished_asset'] is None]
# Nonvacuous: all accepted uses plus external original acceptance conditions.
ready = bool(seen) and not pending and all(ledger['original_conditions'].values())
return {'record_integrity': True, 'original_unit_ready': ready, 'pending_identity_count': len(pending), 'original_conditions': ledger['original_conditions'], 'acceptance_claimed': False}
need(ref['prior_raw_sha256'] == historical[source]['sha256'], 'historical pin mismatch')
need(type(ref['prior_source_still_equal']) is bool and ref['prior_source_still_equal'] == (ref['prior_raw_sha256'] == snapshot['candidate_pool'][source]['current_raw_sha256']), 'historical applicability')
contracts[identity] = c
need(set(contracts) == set(records), 'missing accepted use')
need(used == set(snapshot['candidate_pool']) and len(used) == 19, '19-source coverage')
return contracts
def quantities_valid(values, row):
need(type(values) is dict and set(values) == set(QUANTITIES), 'quantity schema')
for key, value in values.items():
need(value is None or type(value) is int and value > 0, 'invalid quantity')
# Unknown may only be resolved by the separate bound quantity receipt.
if row.get(key) is not None:
need(exact(value, row[key]), 'required quantity mismatch')
def selection_subject(identity, row, c, selection, pins):
return {'id': identity, 'accepted_record_semantic_sha256': semantic(row),
'usage_condition': row['usage_condition'], 'source_pins': pins,
'selection': selection,
# Includes every original unmade/provisional/disclosure field, not
# just c.missing_conditions. No empty-gap denominator is possible.
'original_record': row, 'missing_conditions': c['missing_conditions']}
def asset_matches(a, row, sources):
"""Pure source/format/layout binding, shared by real and isolated tests."""
need(type(a) is dict and set(a) == {'path', 'sha256', 'format', 'region', 'code_anchor', 'kind'}, 'asset schema')
need(a['path'] in sources and a['sha256'] == sources[a['path']]['sha256'], 'selected source pin')
need(exact(a['format'], sources[a['path']]['format']), 'selected dimension/representation mismatch')
need(a['kind'] in ('image', 'audio', 'code', 'state_group'), 'fallback/provisional kind')
if 'group_id' in row:
need(a['kind'] == 'state_group', 'state group promoted to image')
if a['path'].endswith('.png'):
need(a['kind'] in ('image', 'state_group'), 'image kind mismatch')
region = a['region']
need(type(region) is list and len(region) == 4 and all(type(x) is int for x in region), 'region schema')
x, y, w, h = region
need(x >= 0 and y >= 0 and w > 0 and h > 0 and x + w <= a['format']['png_width'] and y + h <= a['format']['png_height'], 'layout region outside source')
need(a['code_anchor'] is None, 'image code representation')
elif a['path'].endswith(('.wav', '.ogg')):
need(a['kind'] == 'audio' and a['region'] is None and a['code_anchor'] is None, 'audio representation')
else:
need(a['kind'] in ('code', 'state_group') and a['region'] is None, 'code representation')
need(type(a['code_anchor']) is dict, 'missing code anchor')
need(a['code_anchor']['path'] == a['path'] and a['code_anchor']['sha256'] == a['sha256'], 'wrong code anchor')
def selection_ready(identity, row, c, choice, sources):
if type(choice) is str:
need(choice == ('candidate' if c['candidate_refs'] else 'missing'), 'candidate/missing status mismatch')
return False
need(type(choice) is dict and set(choice) == {'state', 'contract', 'proofs'}, 'selection schema')
need(choice['state'] == 'selected', 'counterfeit final state')
s = choice['contract']
fields = {'asset', 'creator', 'rights', 'name', 'replacement_chain', 'references', 'quantities', 'quantity_basis', 'usability'}
need(type(s) is dict and set(s) == fields, 'selected contract fields')
a = s['asset']
asset_matches(a, row, sources)
if a['code_anchor'] is not None:
anchor(a['code_anchor'])
need(type(s['creator']) is str and s['creator'].strip().lower() not in ('', 'unknown', 'unassigned'), 'unknown creator')
need(type(s['name']) is str and s['name'] == Path(a['path']).name, 'naming mismatch')
need(type(s['quantity_basis']) is str and s['quantity_basis'].strip(), 'quantity basis missing')
quantities_valid(s['quantities'], row)
chain = s['replacement_chain']
need(type(chain) is list and len(chain) >= 2 and chain[-1] == a['path'], 'replacement chain')
need(chain[0] == identity and all(type(x) is str and x for x in chain), 'replacement identity')
need(type(s['references']) is list and s['references'], 'missing product reference')
pins = {a['path']: a['sha256']}
for ref in s['references']:
anchor(ref)
need(ref['path'].startswith(('scripts/', 'scenes/', 'assets/')), 'product reference must be in product')
pins[ref['path']] = ref['sha256']
rights = s['rights']
need(type(rights) is dict and set(rights) == {'license', 'origin', 'documents'}, 'rights schema')
need(type(rights['license']) is str and rights['license'].strip().lower() not in ('', 'unknown', 'unassigned'), 'unknown license')
need(type(rights['origin']) is str and rights['origin'].strip(), 'rights origin')
need(type(rights['documents']) is list and rights['documents'], 'missing rights evidence')
for doc in rights['documents']:
need(type(doc) is dict and set(doc) == {'path', 'sha256'}, 'rights document schema')
pinned(doc['path'], doc['sha256'])
pins[doc['path']] = doc['sha256']
need(type(s['usability']) is dict and set(s['usability']) == {'display', 'hearing'}, 'usability schema')
for mode, explanation in s['usability'].items():
need(type(explanation) is str and explanation.strip(), 'applicability needs explanation, not boolean')
need(type(choice['proofs']) is dict and set(choice['proofs']) == set(PROOF_KINDS), 'missing required proof')
subject = selection_subject(identity, row, c, s, pins)
for kind in PROOF_KINDS:
receipt(choice['proofs'][kind], kind, subject)
return True
def technical_readiness(required, selected, unit_proofs, subject):
"""Pure reducer. No review/integration/final, no global all-false gate."""
need(type(required) is list and required and len(set(required)) == len(required), 'nonempty unique denominator')
need(type(selected) is dict and set(selected) == set(required), 'readiness exact coverage')
need(all(type(v) is bool for v in selected.values()), 'readiness booleans must be derived')
need(type(unit_proofs) is dict and set(unit_proofs) <= {'mandatory_scope', 'disclosure'}, 'unit proof schema')
for kind, proof in unit_proofs.items():
proof_matches(proof, kind, subject)
return all(selected.values()) and set(unit_proofs) == {'mandatory_scope', 'disclosure'}
def reconstruct(ledger):
"""Lossless public expansion; references are data, never omitted fields.
The caller can inspect every input/provenance field and every historical
contract. Selection is a separate plane and cannot overwrite that history.
This function performs no writes and confers no readiness.
"""
documents = {}
for source, digest in ledger['references'].items():
data = pinned(source, digest)
documents[source] = loads(data) if source.endswith('.json') else data.decode('utf-8')
return {'reference_documents': documents,
'accepted_files': decode(documents[INPUT]),
'historical_contracts': documents[CURRENT],
'selections': ledger['selections'], 'source_pins': ledger['source_pins'],
'unit_proofs': ledger['unit_proofs']}
def check(path, accepted_input=INPUT):
need(accepted_input == INPUT, 'accepted input path must match fixed contract')
spec = raw('docs/spec/acceptance/P22-02.json')
need(sha(spec.decode().strip().encode()) == SPEC_SHA, 'contract strip spec SHA')
need(sha(spec) == 'c26c0b12cf7e7a7c46e671acde2e00762904732c1cd6381a53836e3d7b3f9824', 'raw spec SHA')
accepted = loads(pinned(INPUT, INPUT_SHA))
original, records = accepted_records(accepted)
ledger = read(path)
need(type(ledger) is dict and set(ledger) == {'schema', 'references', 'reconstruction', 'selections', 'source_pins', 'unit_proofs'}, 'ledger schema')
need(ledger['schema'] == 'p22-reference-contract-v2', 'ledger version')
expected = {INPUT, CURRENT} | {str(p.relative_to(ROOT)) for p in safe_path(EVIDENCE).iterdir() if p.is_file()}
need(set(ledger['references']) == expected, 'reference omission/extra')
need(ledger['references'][INPUT] == INPUT_SHA and ledger['references'][CURRENT] == CURRENT_SHA, 'immutable reference pin')
expanded = reconstruct(ledger)
snapshot = expanded['historical_contracts']
contracts = reference_contracts(snapshot, records)
selections = ledger['selections']
need(type(selections) is dict and set(selections) == set(records), 'selection exact 614 coverage')
required_sources = set(snapshot['candidate_pool'])
for choice in selections.values():
if type(choice) is dict:
required_sources.add(choice['contract']['asset']['path'])
need(type(ledger['source_pins']) is dict and set(ledger['source_pins']) == required_sources, 'current source omission/extra')
sources = {}
for source, digest in ledger['source_pins'].items():
data = pinned(source, digest)
sources[source] = {'sha256': digest, 'format': observe(source, data)}
historic = snapshot['candidate_pool'].get(source)
if historic:
need(historic['path'] == source, 'source path mismatch')
if digest == historic['current_raw_sha256'] and source.endswith('.png'):
need(exact(sources[source]['format'], historic['observed_format']), 'observed dimension mismatch')
verify_existing_evidence(sources)
selected = {i: selection_ready(i, records[i][1], contracts[i], selections[i], sources) for i in records}
subject = {'accepted_input_raw_sha256': INPUT_SHA, 'source_pins': ledger['source_pins'],
'selection_sha256': semantic(selections), 'required_ids': sorted(records),
'original_denominator': {'packages': 30, 'delivery_units': 88, 'no_partial_credit': True},
'unresolved_scope_requires_owner_evidence': ['D08', 'D09']}
need(type(ledger['unit_proofs']) is dict, 'unit proofs type')
unit_proofs = {kind: receipt(ref, kind, subject) for kind, ref in ledger['unit_proofs'].items()}
ready = technical_readiness(list(records), selected, unit_proofs, subject)
return {'record_integrity': True, 'validation_ready': ready,
'pending_identity_count': sum(not v for v in selected.values()),
'selected_usable_count': sum(selected.values()), 'accepted_identity_count': len(records),
'candidate_source_count': len(snapshot['candidate_pool']),
'unit_proofs_missing': sorted({'mandatory_scope', 'disclosure'} - set(unit_proofs)),
'original_completion': 'not_claimed; requires formal validate, independent review, integration, final',
'acceptance_claimed': False}
def verify_existing_evidence(sources):
"""Rebind existing publisher/member/license/manifest and layout to bytes."""
packs = read(EVIDENCE + 'publisher-archive-byte-verification.json')
manifests = read('assets/kenney/sources.json')
need(sha(raw('assets/kenney/sources.json')) == packs['source_manifest_sha256'], 'Kenney manifest pin')
for pack in packs['packs']:
matching = [m for m in manifests if m['slug'] == pack['slug']]
need(len(matching) == 1, 'publisher manifest coverage')
m = matching[0]
need(m['page'] == pack['official_page'] and m['zip_sha256'] == pack['archive_sha256'] and m['download_url'] == pack['official_archive_url'], 'publisher mismatch')
for member in pack['members']:
# Publisher facts only apply to exactly equal bytes. A changed
# selection needs its own new rights evidence, never inherited CC0.
if member['repo_path'] in sources and sources[member['repo_path']]['sha256'] == member['raw_sha256']:
pinned(member['repo_path'], member['raw_sha256'])
licenses = [x for x in pack['members'] if x['archive_member'].lower() == 'license.txt']
need(len(licenses) == 1, 'publisher license member')
pinned(licenses[0]['repo_path'], licenses[0]['raw_sha256'])
ui = read(EVIDENCE + 'ui-audio-publisher-verification.json')
manifest = loads(pinned('assets/audio/ui/source_manifest.json',
'f981c2ce12ed364010217dee996908f037d2abfcc3b21b8cf501dd591f7744c9'))
need(manifest['official_page'] == ui['official_page'] and manifest['official_archive_sha256'] == ui['archive_sha256'], 'UI publisher mismatch')
pinned(manifest['license']['path'], manifest['license']['sha256'])
need(manifest['license']['name'] == 'Creative Commons Zero, CC0', 'UI license mismatch')
need(len(manifest['assets']) == len(ui['assets']) == 4, 'UI asset coverage')
for member in ui['assets']:
matches = [a for a in manifest['assets'] if a['path'] == member['repo_path']]
need(len(matches) == 1 and matches[0]['sha256'] == member['raw_sha256'] and matches[0]['bytes'] == member['bytes'], 'UI member manifest mismatch')
observed = read(EVIDENCE + 'observed-existing-asset-contracts.json')
# These layout/reference observations are valid only with their exact
# dependency pins. Changes require a refreshed external observation.
for source, digest in observed['source_pins'].items():
pinned(source, digest)
atlas = observed['pixel_atlas']
need(atlas['columns'] * atlas['cell'][0] == atlas['physical_dimensions'][0] and atlas['rows'] * atlas['cell'][1] == atlas['physical_dimensions'][1] and atlas['gap'] == 0, 'atlas layout')
for source in atlas['vendor_paths_preserved']:
fmt = sources[source]['format']
need([fmt['png_width'], fmt['png_height']] == atlas['physical_dimensions'], 'atlas dimensions/layout mismatch')
internal = read(EVIDENCE + 'internal-origin-observations.json')
for bgm in internal['bgm']:
source = sources[bgm['path']]
if source['sha256'] == bgm['raw_sha256']:
need(exact(source['format'], bgm['format_observed']), 'BGM representation mismatch')
# Origin declarations remain historical facts. They never create proof.
def main():
parser = argparse.ArgumentParser()
parser.add_argument('--ledger', default=OUT + 'asset-contracts.json')
parser.add_argument('--verify-only', action='store_true', help='Read-only verification (also the default)')
parser.add_argument('--accepted-input', default=INPUT)
parser.add_argument('--require-ready', action='store_true')
args = parser.parse_args()
try:
result = check(args.ledger)
result = check(args.ledger, args.accepted_input)
print(json.dumps(result, ensure_ascii=False, sort_keys=True))
return 2 if args.require_ready and not result['original_unit_ready'] else 0
except (ValueError, KeyError, TypeError, IndexError, OSError, RecursionError) as exc:
return 2 if args.require_ready and not result['validation_ready'] else 0
except (ValueError, KeyError, TypeError, IndexError, OSError, RecursionError, struct.error, zlib.error, wave.Error, UnicodeError) as exc:
print(json.dumps({'record_integrity': False, 'error': str(exc), 'acceptance_claimed': False}))
return 1
if __name__ == '__main__':
sys.exit(main())