Spaces:
Running
Running
| """ | |
| Handoff receipts between layers. A layer's job ends with produce(); the next | |
| layer's job starts with receive(). Neither layer knows anything else about the | |
| other — and Layer 1 can see any report that was produced but never received. | |
| """ | |
| import json | |
| import sqlite3 | |
| import uuid | |
| from intel.db import now_iso | |
| STALE_AWARENESS = "stale_awareness" | |
| DATA_STREAM = "data_stream" | |
| INVESTIGATION = "investigation" | |
| INTERFACE = "interface" | |
| SEARCH = "search" | |
| # Layers that exist and run. A report addressed to a layer that isn't built yet | |
| # is queued, not overdue — Layer 1 lists it separately instead of raising it. | |
| ACTIVE_LAYERS = {STALE_AWARENESS, DATA_STREAM} | |
| def produce(con: sqlite3.Connection, run_id: str, from_layer: str, to_layer: str, report: dict) -> str: | |
| hid = uuid.uuid4().hex | |
| con.execute( | |
| "insert into handoffs (id, run_id, from_layer, to_layer, report, produced_at) values (?, ?, ?, ?, ?, ?)", | |
| (hid, run_id, from_layer, to_layer, json.dumps(report, default=str), now_iso()), | |
| ) | |
| con.commit() | |
| return hid | |
| def receive(con: sqlite3.Connection, handoff_id: str, by: str) -> bool: | |
| """Acknowledges a report. Returns False if it was already received (or | |
| doesn't exist), so a receiver can tell a duplicate from a first delivery.""" | |
| cur = con.execute( | |
| "update handoffs set received_at = ?, received_by = ? where id = ? and received_at is null", | |
| (now_iso(), by, handoff_id), | |
| ) | |
| con.commit() | |
| return cur.rowcount == 1 | |
| def pending(con: sqlite3.Connection, to_layer: str) -> list[dict]: | |
| rows = con.execute( | |
| "select * from handoffs where to_layer = ? and received_at is null order by produced_at", (to_layer,) | |
| ).fetchall() | |
| return [{**dict(r), "report": json.loads(r["report"])} for r in rows] | |