# P3-WP04 Edge Gateway Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Build an enterprise-edge engineering baseline that executes approved data operations inside the enterprise boundary and sends only policy-approved, minimized events to the control plane.

**Architecture:** Reuse the existing standalone Runner for governed execution and its single-use task-token and durable-ledger semantics. Add a focused `app/core/edge_gateway/` boundary for contracts, egress/retention policy, an offline SQLite queue, control-plane PostgreSQL state, and gateway lifecycle services; expose management and machine pull/event endpoints through the existing data-source blueprint. The application records certificate fingerprints and rotation state, while private keys remain at the edge and real mTLS termination, enterprise network topology, and security/legal sign-off remain external UAT gates.

**Tech Stack:** Python 3.11, Flask, SQLAlchemy/PostgreSQL, SQLite WAL, PyJWT/cryptography, Alembic, Docker Compose, pytest, OpenAPI 3.1.

---

## Engineering and acceptance boundary

- Engineering completion may reach `ENGINEERING_COMPLETE_ENTERPRISE_EDGE_UAT_BLOCKED`.
- `enterprise_source_1`, `enterprise_source_2`, `network`, and `security_legal` remain `TBD_EXTERNAL` until enterprise evidence is supplied.
- The control plane must never persist raw rows, recent device detail, credentials, private keys, SQL text, or unrestricted task parameters from edge outbound events.
- The edge may execute `collect`, `profile`, `quality`, `lineage`, and `controlled_query`; the control plane may receive only `desensitized_metadata`, `statistics`, `lineage`, `evidence`, and bounded health/diagnostic summaries.
- Existing user-owned untracked architecture assets and `canvas/` are outside this work package.
- Use one final P3-WP04 commit after all tasks and reviews; do not create task-slice commits.

### Task 1: Edge contracts, data boundary policy, and durable offline queue

**Files:**
- Create: `app/core/edge_gateway/__init__.py`
- Create: `app/core/edge_gateway/contracts.py`
- Create: `app/core/edge_gateway/policy.py`
- Create: `app/core/edge_gateway/queue.py`
- Mirror: matching paths under `deployment/app/core/edge_gateway/`
- Test: `tests/test_phase3_wp04_edge_gateway.py`

- [x] **Step 1: Write failing contract and policy tests**

```python
def test_raw_and_recent_detail_never_cross_the_control_plane_boundary():
    policy = EdgeEgressPolicy(allowed_control_hosts={"control.example.com"})
    for classification in ("raw", "recent_detail", "restricted"):
        with pytest.raises(EdgePolicyError):
            policy.approve_event({"classification": classification, "payload": {"rows": [{"secret": "x"}]}})

def test_only_supported_task_types_and_bounded_approved_results_are_valid():
    task = EdgeTaskContract.from_mapping(APPROVED_TASK)
    assert task.task_type == "profile"
    with pytest.raises(EdgeContractError):
        EdgeTaskContract.from_mapping({**APPROVED_TASK, "task_type": "shell"})
```

- [x] **Step 2: Run the new unit tests and verify RED**

Run: `PYTHONPATH=. .venv/bin/pytest -q tests/test_phase3_wp04_edge_gateway.py`

Expected: collection/import failure because the edge gateway package does not exist.

- [x] **Step 3: Implement immutable contracts and fail-closed policy**

Implement canonical SHA-256 digests, stable event IDs, strict enums, recursive sensitive-key rejection, per-channel byte limits, HTTPS host allowlisting, high-sensitivity default deny, and explicit retention classes:

```python
EDGE_ONLY = {"raw", "recent_detail", "restricted"}
CONTROL_PLANE_ALLOWED = {
    "desensitized_metadata",
    "statistics",
    "lineage",
    "evidence",
    "health_summary",
    "diagnostic_summary",
}
RETENTION_DAYS = {"raw": 0, "recent_detail": 365, "metadata": 1095, "evidence": 2190}
```

- [x] **Step 4: Implement `SqliteEdgeQueue` with WAL and idempotent event storage**

The queue must atomically enqueue/claim/complete/cancel tasks, persist outbound events before delivery, retry with bounded exponential backoff, reject duplicate IDs with different digests, and return an existing terminal acknowledgement for exact replays.

- [x] **Step 5: Verify GREEN and focused queue concurrency**

Run: `PYTHONPATH=. .venv/bin/pytest -q tests/test_phase3_wp04_edge_gateway.py -k 'contract or policy or queue'`

Expected: all selected tests pass, including two independent queue connections claiming one task only.

### Task 2: Control-plane persistence, identity lifecycle, and machine pull/event API

**Files:**
- Create: `app/core/edge_gateway/repository.py`
- Create: `app/core/edge_gateway/service.py`
- Create: `app/api/data_source/edge_routes.py`
- Modify: `app/api/data_source/__init__.py`
- Modify: `app/core/system/permissions.py`
- Create: `app/models/edge_gateway.py`
- Modify: `app/models/__init__.py`
- Create: `migrations/versions/20260809_477_edge_gateway_control_plane.py`
- Create: `migrations/versions/20260809_478_edge_gateway_security_cleanup.py`
- Create: `migrations/versions/20260809_479_edge_gateway_signed_authority.py`
- Mirror: matching paths under `deployment/app/` and `deployment/migrations/`
- Test: `tests/integration/test_phase3_wp04_edge_gateway_postgres.py`
- Test: `tests/test_permission_matrix.py`

- [x] **Step 1: Write failing PostgreSQL tests for registration and event idempotency**

```python
def test_gateway_registration_requires_one_time_enrollment_and_exact_certificate_binding(pg_session):
    enrollment = service.issue_enrollment(APPROVED_GATEWAY)
    gateway = service.register(enrollment.secret, certificate_sha256=CERT_SHA256)
    assert gateway["status"] == "active"
    with pytest.raises(EdgeAuthenticationError):
        service.register(enrollment.secret, certificate_sha256=CERT_SHA256)

def test_same_event_id_with_different_digest_is_rejected(pg_service):
    pg_service.accept_event(GATEWAY_ID, APPROVED_EVENT)
    with pytest.raises(EdgeConflictError):
        pg_service.accept_event(GATEWAY_ID, {**APPROVED_EVENT, "payload": {"count": 2}})
```

- [x] **Step 2: Run the PostgreSQL file and verify RED**

Run: `TEST_DATABASE_URL=postgresql://dataops:dataops-test-password@127.0.0.1:15432/dataops PYTHONPATH=. .venv/bin/pytest -q tests/integration/test_phase3_wp04_edge_gateway_postgres.py`

Expected: failure because migration 477, repository, and service are absent.

- [x] **Step 3: Add migrations 477-479 and repository CAS operations**

Create governed tables for gateways, one-time enrollments, credentials/certificate generations, tasks, outbound events, releases, and audit events. Add exact unique keys, digest conflict checks, status constraints, lease/cancel fields, and database triggers that prevent active gateways without an active credential and certificate fingerprint.

- [x] **Step 4: Implement service lifecycle**

Support enrollment, register, heartbeat, credential/certificate rotate and revoke, task issue/pull/cancel, event accept/ack, reconnect reconciliation, safe diagnostic summary, release offer/ack/rollback, and stable audit evidence. Authenticate the complete gateway/environment/network-zone/certificate-generation binding before consuming one-time credentials.

- [x] **Step 5: Add human and machine routes with deny-by-default permissions**

Human management routes require `edge-gateways:read|operate|manage`; machine register, heartbeat, pull, event, and release acknowledgement routes are public only at the platform bearer layer and must require the dedicated one-time enrollment or edge credential plus exact certificate fingerprint.

- [x] **Step 6: Verify GREEN, migration head, and app/deployment parity**

Run the PostgreSQL file, permission matrix nodes, `alembic current`, and recursive parity checks. Actual expected head: `20260809_479`.

### Task 3: Edge agent, local execution boundary, reconnect, and release rollback

**Files:**
- Create: `app/edge_gateway/__init__.py`
- Create: `app/edge_gateway/agent.py`
- Create: `app/edge_gateway/transport.py`
- Create: `app/edge_gateway/bootstrap.py`
- Modify: `app/runner/bootstrap.py`
- Modify: `app/runner/api.py`
- Mirror: matching paths under `deployment/app/`
- Test: `tests/security/test_phase3_wp04_data_boundary.py`
- Test: `tests/integration/test_phase3_wp04_edge_agent.py`

- [x] **Step 1: Write failing offline/reconnect and malicious-task tests**

```python
def test_offline_execution_reconnect_delivers_each_approved_event_once(tmp_path):
    agent = build_agent(tmp_path, transport=FlakyPullTransport(disconnect_after=1))
    agent.run_once()
    agent.run_once()
    assert agent.pending_events == 1
    agent.transport.reconnect()
    agent.flush_events()
    agent.flush_events()
    assert agent.transport.accepted_event_ids == {STABLE_EVENT_ID}

def test_malicious_task_cannot_request_raw_export_or_unapproved_host(agent):
    with pytest.raises(EdgePolicyError):
        agent.accept_task(MALICIOUS_RAW_EXPORT_TASK)
```

- [x] **Step 2: Verify RED**

Run: `PYTHONPATH=. .venv/bin/pytest -q tests/security/test_phase3_wp04_data_boundary.py tests/integration/test_phase3_wp04_edge_agent.py`

- [x] **Step 3: Implement pull-only agent and controlled Runner adapter**

The edge initiates all control traffic, verifies signed task contracts, stages tasks before execution, dispatches only allowlisted Runner node types, and converts results to approved summaries before enqueueing outbound events. Raw artifacts and recent details remain referenced by local digest only.

- [x] **Step 4: Implement reconnect/cancel/diagnostic and release state machine**

Use bounded backoff with jitter, exact event replay, task cancellation probes, safe diagnostics without log bodies, signed release manifest verification, candidate activation, health confirmation, and rollback to the previous locally verified release.

- [x] **Step 5: Verify GREEN and attack-path coverage**

Expected tests include malicious task type, raw payload, secret-like fields, oversized summary, event replay with changed digest, revoked certificate, expired credential, unapproved host/proxy, disconnected control plane, and failed candidate rollback.

### Task 4: Installation/rotation tools, OpenAPI, runbook, and delivery evidence

**Files:**
- Create: `scripts/edge_gateway_admin.py`
- Create: `app/edge_gateway/runtime.py`
- Create: `app/edge_gateway/__main__.py`
- Create: `app/edge_gateway/healthcheck.py`
- Create: `deploy/edge/Dockerfile.edge`
- Create: `deploy/edge/docker-compose.edge.yml`
- Create: `deploy/edge/edge.env.example`
- Create: `deploy/edge/edge.config.example.json`
- Create: `deploy/edge/edge.config.schema.json`
- Create: `deploy/edge/README.md`
- Create: `docs/runbooks/EDGE_GATEWAY_OPERATIONS.md`
- Create: `docs/phase3/P3_WP04_EDGE_GATEWAY.md`
- Create: `docs/validation/P3_WP04_EDGE_GATEWAY_EVIDENCE.md`
- Create: `docs/validation/P3_WP04_FAILURE_INJECTION.json`
- Modify: `scripts/generate_openapi.py`
- Modify: `docs/architecture/OPENAPI.yaml`
- Modify: `docs/DATAOPS_PHASE3_6_MONTH_DEVELOPMENT_PLAN_20260802.md`
- Modify: `docs/phase3/P3_WP00_REQUIREMENTS.json`
- Test: `tests/test_phase3_wp04_delivery_contract.py`
- Test: `tests/integration/test_phase3_wp04_edge_runtime.py`

- [x] **Step 1: Write failing delivery-contract tests**

Assert that the Compose topology is pull-only, mounts key/certificate paths read-only, uses a non-root/read-only container, defines no inbound control-plane port, requires explicit host/proxy allowlists, and exposes health without sensitive fields. Assert that the admin tool creates a private key and CSR locally with restrictive permissions and never prints or uploads private-key material.

- [x] **Step 2: Verify RED**

Run: `PYTHONPATH=. .venv/bin/pytest -q tests/test_phase3_wp04_delivery_contract.py`

- [x] **Step 3: Implement tools and deployment contract**

Provide `init`, `csr`, `register-manifest`, `rotate`, `revoke-status`, `release-check`, `rollback`, and read-only `status` operations. Require a dedicated absolute root, O_NOFOLLOW/fcntl mutation locking, same-file-descriptor artifact verification/copy, atomic files, full Ed25519 release authority, bounded backups, and operator confirmation for rollback; never embed credentials in Compose or documentation. The Edge image runs the actual pull-only loop, verifies a fresh server CRL with `CRL_CHECK_LEAF`, publishes only a safe health file, degrades safely during disconnect, and exposes no inbound listener.

- [x] **Step 4: Generate OpenAPI and write evidence-bound documentation**

Document architecture, allowed data classes, default-deny decisions, retention/legal hold/delete receipts, queue v1-to-v2 copy migration, incident thresholds and diagnostics, database role order, client/server CRL updates, install/rotate/revoke/upgrade/rollback procedures, and external UAT inputs. Package the full OpenAPI, failure-injection ledger, edge runtime/config schema and full-file SHA-256 closure. Mark only locally verified engineering work complete.

- [x] **Step 5: Run final P3-WP04 targeted gates**

Run only P3-WP04 unit, security, PostgreSQL integration, edge-agent, Runner directly affected tests, permission/OpenAPI contracts, Ruff, `py_compile`, JSON/YAML parsing, Compose config validation, migration head, `app/deployment` parity, `git diff --check`, and secret scan. Do not run the repository-wide regression suite.

- [x] **Step 6: Two-stage review and one final commit**

Require specification PASS and quality/security approval. Then stage only P3-WP04 files, preserve unrelated untracked assets, and create one local commit without pushing or deploying.