DataOps-platform/docs/phase3/P3_WP04_EDGE_GATEWAY_IMPLEMENTATION_PLAN.md

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

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"})

• 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.

• 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:

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}

• 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.

• 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

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}})

• 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.

• 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.

• 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.

• 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.

• 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

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)

• 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

• 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.

• 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.

• 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.

• Step 2: Verify RED

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

• 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.

• 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.

• 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.

• 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.