forked from chiki2bum2/SniperGold_ML
Chiki Bum
0d5ca3c197
Implements the frozen P3_DATA_ENGINE_V1_SPEC (SHA 84bf0f217ffba51197112a6bbacbcc297058e04b5ca47f0459028fe33e0321e5). Components: engine/ producer (certify, chunkmap, parse, canonical, worker, dispatcher, aggregate, storage, journal, checkpoint, lock, evidence, manifest, run_complete, dataset_builder, cli) + engine/verify independent verifier (vparse, vaggregate, vinvariants, vcompare); headless CLI sniper-data; golden corpus G01-G17; unit/property/adversarial/mutation/legacy-diff/CLI suites. Qualification verdict: QUALIFIED (all 9 mandatory gates pass; independent verifier accepted). Spec, governance record, and legacy checkpoint untouched. G-14 NOT AUTHORIZED honored; no real-data processing, no pilot, no workload 46, no chunk 760 access.
89 lines
No EOL
2.7 KiB
Python
89 lines
No EOL
2.7 KiB
Python
"""Commit journal (spec 13.1): append-only, hash-chained, fail-closed.
|
|
|
|
Each line is ``{"seq", "chunk", "content_id", "file_sha256", "prev",
|
|
"ts_utc"}``; ``prev`` is the hash of the previous line's exact UTF-8 text
|
|
(SHA-256 of the canonical JSON text of the previous line, no trailing
|
|
newline). seq 0 uses SHA-256 of the empty string as prev. ``journal_tail``
|
|
recorded in the checkpoint is (seq, hash of last line).
|
|
"""
|
|
|
|
import os
|
|
|
|
from .util import canonical_json, sha256_bytes, atomic_write_bytes
|
|
|
|
EMPTY_HASH = sha256_bytes(b"")
|
|
|
|
|
|
class JournalError(Exception):
|
|
pass
|
|
|
|
|
|
def _line_for(record):
|
|
return canonical_json(record)
|
|
|
|
|
|
def _line_hash(line_text):
|
|
return sha256_bytes(line_text)
|
|
|
|
|
|
def append_commit(journal_path, seq, chunk_index, content_id, file_sha256,
|
|
ts_utc, prev_hash):
|
|
"""Append one commit line atomically (fail-closed: fsync)."""
|
|
record = {
|
|
"seq": seq,
|
|
"chunk": chunk_index,
|
|
"content_id": content_id,
|
|
"file_sha256": file_sha256,
|
|
"prev": prev_hash,
|
|
"ts_utc": ts_utc,
|
|
}
|
|
line = _line_for(record)
|
|
os.makedirs(os.path.dirname(os.path.abspath(journal_path)), exist_ok=True)
|
|
with open(journal_path, "ab") as fh:
|
|
fh.write(line.encode("utf-8"))
|
|
fh.flush()
|
|
os.fsync(fh.fileno())
|
|
# Line hash is defined over the line text WITHOUT the trailing newline
|
|
# (read_journal recomputes it the same way).
|
|
return _line_hash(line.rstrip("\n"))
|
|
|
|
|
|
def read_journal(journal_path):
|
|
"""Read and validate the chain. Returns (records, tail_hash).
|
|
Raises JournalError on chain break or malformed line."""
|
|
if not os.path.exists(journal_path):
|
|
return [], EMPTY_HASH
|
|
records = []
|
|
prev = EMPTY_HASH
|
|
tail = EMPTY_HASH
|
|
with open(journal_path, "rb") as fh:
|
|
for raw in fh:
|
|
text = raw.rstrip(b"\r\n")
|
|
if not text:
|
|
continue
|
|
if raw.endswith(b"\r\n"):
|
|
line = raw[:-2]
|
|
elif raw.endswith(b"\n"):
|
|
line = raw[:-1]
|
|
else:
|
|
line = raw
|
|
try:
|
|
import json
|
|
record = json.loads(line.decode("utf-8"))
|
|
except Exception as e:
|
|
raise JournalError("journal line unparseable: %s" % e)
|
|
if record.get("prev") != prev:
|
|
raise JournalError("journal chain broken at seq %r"
|
|
% record.get("seq"))
|
|
lh = _line_hash(line.decode("utf-8"))
|
|
records.append(record)
|
|
prev = lh
|
|
tail = lh
|
|
return records, tail
|
|
|
|
|
|
def seq_after(records):
|
|
"""Next seq = last seq + 1 (0 when journal is empty)."""
|
|
if not records:
|
|
return 0
|
|
return records[-1].get("seq", -1) + 1 |