feat(scripts): edge-ingest demo — the $100 Agent story, signed and fed to the platform

CHANGELOG.md

@@ -9,6 +9,8 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 ## [Unreleased]
 
 ### Added
+- **Edge-ingest demo — the $100 Agent story, signed end-to-end.** `scripts/demo_edge_ingest.py` generates the four mandate-draw records of the spend storyline as chained, DSSE-signed OCSF events (the emitter's exact shape) and plays four acts against `POST /api/v1/audit/ingest`: the story verifies with chain positions, the replay dedupes, a record edited after signing quarantines with `fingerprint mismatch`, and a record after a lost epoch verifies with the stream gap surfaced. Offline mode (`--out`) writes the signed NDJSON + demo public key, so the same artifact feeds the immutable-ledger cpex adapter — one generator, both sinks of the joint demo. An end-to-end test drives all four acts through the real endpoint so the demo and the verifier can't silently drift apart. Script + test only — no product impact. (#485)
+
 - **OCSF edge ingest — the platform now accepts signed audit records from enforcement points it doesn't host.** Component A of the edge-enforcement spec (#483): org admins register an edge deployment (`POST /api/v1/edges` — globally-unique `chain_uid`, validated ECDSA P-256 verification key, show-once `aid_edge_` credential hashed at rest), and the edge streams NDJSON OCSF records to `POST /api/v1/audit/ingest`, where every record is verified **on arrival** — covered bytes reconstructed exactly per the emitter's `sign::signing_input` (RFC 8785 minus the attestation envelope), fingerprint recomputed, DER ECDSA-P256 signature checked over the DSSE PAE, chain continuity checked against the last verified head, `stream_seq` density per stream. Integrity failures quarantine the record (stored verbatim with the reason, never dropped, never advancing chain state — a forged record can't re-root the chain); continuity observations on a crypto-valid record (gap, discontinuity after a crashed edge epoch) land as `anomalies` on a verified row and the chain re-anchors, so one lost epoch can't cascade-quarantine everything after it. Replays dedupe on `metadata.uid` (stream-stamp fallback) and report `duplicate`. Two new tables via forward-only migration `g4b5c6d7e8f9`; no new dependencies. 19 new tests sign real chained records the way the Rust emitter does and cover authz, tamper/wrong-key/unsigned quarantine, gap/re-anchor, replay, caps, and cross-org isolation. (#484)
 
 - **Edge-enforcement spec drafted — the on-prem/VPC roadmap item gets its concrete shape.** `docs/specs/praxis-edge-enforcement.md` designs the split the CPEX/Praxis collaboration makes possible: enforcement runs in the customer's cluster (Praxis + PPE with two AI Identity plugins), while authority and evidence stay with the platform — raw payloads never leave the cluster, only signed decision records carrying content hashes plus draw-settlement metadata. Three gated components: an org-scoped OCSF ingest endpoint that verifies fingerprints, DSSE, chain continuity, and `stream_seq` density on arrival (gaps quarantined and surfaced, never laundered — a gap is itself evidence); a `cpex-plugin-mandate` crate that verifies Biscuits offline via the published root public key (the token's embedded ceiling checks make forged caps fail cryptographically) but settles draws against the mandate service, keeping cumulative spend single-writer and the Ed25519 root key out of the edge entirely — fail-closed on every path; and a reference deployment smoke-tested by the committed $100-mandate scenario. Build is explicitly gated behind the praxis#11 port verification per `PRAXIS-PORT-PLAN.md`'s sequencing rule, and September stays committed to v0.5.0 — components target the v0.6.0 window alongside agent spend control. Design only — no product impact. (#483)

api/tests/test_edge_ingest_demo.py

@@ -0,0 +1,78 @@
+"""The demo assets feed the real ingest endpoint end-to-end.
+
+Loads scripts/demo_edge_ingest.py and drives its four acts through the
+actual API: the signed $100-agent story verifies, the replay dedupes,
+the tampered record quarantines, and the lost-epoch record verifies with
+the gap surfaced. If the demo script and the verifier ever drift apart,
+this is the test that says so before a live demo does.
+"""
+
+import importlib.util
+import pathlib
+
+import pytest
+
+from common.models import EdgeAuditEvent, OrgMembership
+
+_SCRIPT = pathlib.Path(__file__).resolve().parents[2] / "scripts" / "demo_edge_ingest.py"
+_spec = importlib.util.spec_from_file_location("demo_edge_ingest", _SCRIPT)
+demo = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(demo)
+
+
+@pytest.fixture
+def demo_edge(client, db_session, test_user, auth_headers):
+    """An edge registered with the demo script's chain identity and key."""
+    db_session.add(OrgMembership(org_id=test_user.org_id, user_id=test_user.id, role="owner"))
+    db_session.commit()
+    resp = client.post(
+        "/api/v1/edges",
+        headers=auth_headers,
+        json={
+            "name": "Edge demo — $100 Agent",
+            "chain_uid": demo.CHAIN_UID,
+            "verify_key_pem": demo.demo_public_key_pem(),
+            "key_id": "edge-demo-key-1",
+        },
+    )
+    assert resp.status_code == 201, resp.text
+    return {"Authorization": f"Bearer {resp.json()['ingest_key']}"}
+
+
+def test_demo_story_plays_all_four_acts(client, db_session, demo_edge):
+    story = demo.build_story()
+
+    # Act 1 — the $100 agent: three draws land, the fourth is denied
+    act1 = client.post("/api/v1/audit/ingest", headers=demo_edge, content=demo.ndjson(story))
+    assert act1.status_code == 202, act1.text
+    data = act1.json()
+    assert data["verified"] == 4 and data["quarantined"] == 0 and data["anomalies"] == 0
+    assert [r["chain_position"] for r in data["results"]] == [1, 2, 3, 4]
+    denied = db_session.query(EdgeAuditEvent).filter_by(metadata_uid="draw-004").one()
+    assert denied.event["action"] == "Denied"
+    assert denied.event["status_code"] == "mandate_exceeded"
+    # The draw-receipt join key rides inside the verified bytes
+    assert denied.event["unmapped"]["cmf.request.request_id"] == "corr-d4e55b37"
+
+    # Act 2 — replay: at-least-once, nothing re-stored
+    act2 = client.post("/api/v1/audit/ingest", headers=demo_edge, content=demo.ndjson(story)).json()
+    assert act2["duplicates"] == 4 and act2["verified"] == 0
+    assert db_session.query(EdgeAuditEvent).count() == 4
+
+    # Act 3 — a record edited after signing: quarantined, stored verbatim
+    tampered = demo.build_tampered(story)
+    act3 = client.post(
+        "/api/v1/audit/ingest", headers=demo_edge, content=demo.ndjson([tampered])
+    ).json()
+    assert act3["quarantined"] == 1
+    assert "fingerprint mismatch" in act3["results"][0]["reason"]
+    row = db_session.query(EdgeAuditEvent).filter_by(metadata_uid="draw-evil").one()
+    assert row.verification_status == "quarantined"
+
+    # Act 4 — after a lost epoch: verified WITH the gap surfaced, chain re-anchored
+    gap = demo.build_gap_record(story)
+    act4 = client.post("/api/v1/audit/ingest", headers=demo_edge, content=demo.ndjson([gap])).json()
+    assert act4["verified"] == 1 and act4["anomalies"] == 1
+    result = act4["results"][0]
+    assert result["status"] == "verified" and result["chain_position"] == 5
+    assert "stream_seq gap" in result["anomalies"]