v0.4: Durable local control plane — PostgreSQL schema, SQLite repo, drift detector, and CLI

docs/ADR/004-postgresql-durable-local-control-plane.md

@@ -0,0 +1,39 @@
+# ADR 004: PostgreSQL-backed durable local control plane
+
+## Status
+
+Accepted for v0.4.
+
+## Decision
+
+Ordinary PostgreSQL is the production reference state provider. Repository contracts remain
+provider-neutral and a SQLite DB-API implementation supports hermetic local operation and contract
+tests. PostgreSQL JSONB stores documented JSON-compatible representations; it is not a canonical
+world model. Explicit tables partition desired revisions, authoritative decisions, observations,
+plans, runs, operations, attempts, transitions, evidence, checkpoints, leases, reconciliation, and
+drift. Secret material is not accepted by checkpoints or ordinary evidence.
+
+Events and logs may notify or diagnose, but never replace current projections and append-only
+operation history. Desired declarations are not authoritative decisions; authoritative decisions
+are not observations; observations are not provider results.
+
+Provider calls never occur inside a database transaction. The executor first commits `running`,
+calls the provider, then commits its result and next transition. Recovery treats an interruption
+between those transactions as uncertain: it observes before policy permits retry. Durable scoped
+idempotency keys reject conflicting content and prevent a verified effect being applied twice.
+
+World, run, resource, and maintenance leases are explicit rows. Atomic acquisition, expiry,
+owner-checked release, and monotonically increasing fencing tokens allow safe stale-lock recovery
+and inspection. These are local-process coordination semantics, not distributed leadership.
+
+Checkpoints are immutable logical journal summaries made before execution, after verified batches,
+before compensation, and at finalisation. They are not infrastructure snapshots. Resume rejects a
+different executable-plan fingerprint or schema version.
+
+## Consequences
+
+The transaction gap is visible and recoverable rather than hidden. PostgreSQL backup and restore
+preserves the entire schema, while metadata export is diagnostic only and excludes secrets. v0.4 is
+strictly single-node: HA, consensus, leader election, remote agents, operational providers, and
+continuous reconciliation are deferred. v0.5 supplies the first authoritative naming provider
+slice; it must consume these seams rather than alter their domain meaning.

docs/durable-control-plane-v0.4.md

@@ -0,0 +1,30 @@
+# NetEngine v0.4 durable local control plane
+
+v0.4 turns the v0.3 provider-neutral runtime into a restart-safe, repeatedly operable single-node
+control plane. `controlplane.models` defines immutable desired, authoritative, observed, checkpoint,
+lease, reconciliation, and drift records. `ControlPlaneRepository` is independent of SQL dialect;
+the PostgreSQL migration is the durable production schema and SQLite provides a local/test adapter.
+
+Acceptance stores an immutable revision and changes the active pointer in one transaction. Rejected
+admission is retained but never activated. Observations are versioned independently from apply
+results. Idempotency is unique per world and content conflicts fail closed. Provider calls occur
+between transactions; recovery observes any uncertain operation before retry.
+
+Leases cover worlds, execution runs, resources, and maintenance. Expiry permits recovery with a
+higher fencing token; only the owner releases an active lease. Checkpoints are logical journal
+summaries, not infrastructure snapshots, and reject an incompatible plan fingerprint or schema.
+
+Drift comparison deterministically classifies missing, unexpected, changed, unhealthy,
+unverifiable, stale, unavailable, and conformant resources. Reconciliation is explicitly invoked
+and bounded: observe, compare, plan, compile, execute, observe, assess. A conformant repeat is a
+no-op; v0.4 has no daemon loop.
+
+Production uses `NETENGINE_CONTROL_PLANE_DATABASE_URL`; local CLI tests may use
+`NETENGINE_CONTROL_PLANE_PATH`. Back up with
+`pg_dump --schema=netengine_control --format=custom DATABASE > control-plane.dump` and restore with
+`pg_restore --dbname=DATABASE control-plane.dump`. This preserves the v0.4 schema but is neither
+world portability nor a secret or infrastructure backup.
+
+Legacy JSON phase state, queues, exports, and logs remain intact and are not live workflow state.
+Real DNS, trust, identity, routing, workload, and public providers are deferred to v0.5 and later,
+as are HA, distributed locks, leader election, remote agents, and polished backup orchestration.

docs/roadmap.md

@@ -1,8 +1,8 @@
 # NetSovereign roadmap
 
-Implementation status: v0.1, v0.2, and the bounded in-memory v0.3 runtime core are
-implemented and tested. v0.4 durability and v0.5 naming materialisation remain
-planned and are not implied by the v0.3 interfaces.
+Implementation status: v0.1 through v0.4 are implemented and tested. v0.4 supplies the
+bounded, single-node durable control-plane seams and PostgreSQL schema; v0.5 naming
+materialisation remains planned and is not implied by those interfaces.
 
 NetSovereign develops authority and intent before operational adapters. Versions v0.2 through v0.4
 are a single dependency chain: real DNS, PKI, identity, or gateway providers must not begin until

migrations/004_durable_control_plane.sql

@@ -0,0 +1,56 @@
+-- NetEngine v0.4 durable local control plane (ordinary PostgreSQL 15+).
+-- Each partition is explicit; JSONB is a representation, not the domain model.
+BEGIN;
+CREATE SCHEMA IF NOT EXISTS netengine_control;
+CREATE TABLE IF NOT EXISTS netengine_control.schema_version(
+  version integer PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now());
+DO $$
+BEGIN
+  IF EXISTS (SELECT 1 FROM netengine_control.schema_version WHERE version > 4) THEN
+    RAISE EXCEPTION 'incompatible control-plane schema version';
+  END IF;
+END $$;
+CREATE TABLE IF NOT EXISTS netengine_control.desired_revisions(
+  revision_id text PRIMARY KEY, world_id text NOT NULL, parent_revision_id text,
+  schema_version text NOT NULL, declaration_digest text NOT NULL,
+  declaration jsonb NOT NULL, admission jsonb NOT NULL, accepted_at timestamptz NOT NULL,
+  actor text, authority_manifest_digest text NOT NULL, boundary_policy_digest text,
+  compatibility jsonb NOT NULL DEFAULT '{}', status text NOT NULL,
+  FOREIGN KEY(parent_revision_id) REFERENCES netengine_control.desired_revisions(revision_id),
+  UNIQUE(world_id,declaration_digest));
+CREATE TABLE IF NOT EXISTS netengine_control.active_desired_revisions(
+  world_id text PRIMARY KEY, revision_id text NOT NULL REFERENCES netengine_control.desired_revisions);
+CREATE TABLE IF NOT EXISTS netengine_control.authoritative_records(
+  record_id text PRIMARY KEY, world_id text NOT NULL, kind text NOT NULL, authority_id text NOT NULL,
+  mandate_id text, desired_revision_id text NOT NULL REFERENCES netengine_control.desired_revisions,
+  decision_reference text NOT NULL, status text NOT NULL, version integer NOT NULL CHECK(version>0),
+  lifecycle jsonb NOT NULL, value jsonb NOT NULL, recorded_at timestamptz NOT NULL,
+  UNIQUE(world_id,kind,record_id,version));
+CREATE TABLE IF NOT EXISTS netengine_control.observations(
+  observation_id text PRIMARY KEY, world_id text NOT NULL, resource_id text NOT NULL,
+  provider_id text NOT NULL, binding_id text NOT NULL, capability text NOT NULL,
+  provider_resource_id text, observation_type text NOT NULL, representation jsonb,
+  digest text, health text NOT NULL, conformance text NOT NULL, observed_at timestamptz NOT NULL,
+  stale_after timestamptz, execution_run_id text, operation_id text);
+CREATE INDEX IF NOT EXISTS observations_latest ON netengine_control.observations(world_id,resource_id,observed_at DESC);
+CREATE TABLE IF NOT EXISTS netengine_control.execution_plans(plan_id text PRIMARY KEY, fingerprint text NOT NULL UNIQUE, payload jsonb NOT NULL);
+CREATE TABLE IF NOT EXISTS netengine_control.execution_runs(run_id text PRIMARY KEY, plan_id text NOT NULL REFERENCES netengine_control.execution_plans, world_id text NOT NULL, current_state text NOT NULL, payload jsonb NOT NULL);
+CREATE TABLE IF NOT EXISTS netengine_control.operations(operation_id text PRIMARY KEY, run_id text NOT NULL REFERENCES netengine_control.execution_runs, world_id text NOT NULL, idempotency_key text NOT NULL, content_digest text NOT NULL, current_state text NOT NULL, payload jsonb NOT NULL, UNIQUE(world_id,idempotency_key));
+CREATE TABLE IF NOT EXISTS netengine_control.attempts(attempt_id text PRIMARY KEY, operation_id text NOT NULL REFERENCES netengine_control.operations, number integer NOT NULL, payload jsonb NOT NULL, UNIQUE(operation_id,number));
+CREATE TABLE IF NOT EXISTS netengine_control.transitions(transition_id text PRIMARY KEY, operation_id text NOT NULL REFERENCES netengine_control.operations, sequence integer NOT NULL, at timestamptz NOT NULL, payload jsonb NOT NULL, UNIQUE(operation_id,sequence));
+CREATE TABLE IF NOT EXISTS netengine_control.evidence(evidence_id text PRIMARY KEY, run_id text NOT NULL REFERENCES netengine_control.execution_runs, operation_id text REFERENCES netengine_control.operations, created_at timestamptz NOT NULL, payload jsonb NOT NULL);
+CREATE TABLE IF NOT EXISTS netengine_control.checkpoints(checkpoint_id text PRIMARY KEY, run_id text NOT NULL REFERENCES netengine_control.execution_runs, world_id text NOT NULL, desired_revision_id text NOT NULL REFERENCES netengine_control.desired_revisions, previous_checkpoint_id text REFERENCES netengine_control.checkpoints, schema_version integer NOT NULL, plan_fingerprint text NOT NULL, created_at timestamptz NOT NULL, payload jsonb NOT NULL);
+CREATE TABLE IF NOT EXISTS netengine_control.leases(lock_key text PRIMARY KEY, owner text NOT NULL, acquired_at timestamptz NOT NULL, expires_at timestamptz NOT NULL, fencing_token bigint NOT NULL CHECK(fencing_token > 0), renewed_at timestamptz, released_at timestamptz, release_reason text, payload jsonb NOT NULL, CHECK(expires_at >= acquired_at));
+CREATE TABLE IF NOT EXISTS netengine_control.reconciliations(reconciliation_id text PRIMARY KEY, world_id text NOT NULL, desired_revision_id text NOT NULL REFERENCES netengine_control.desired_revisions, trigger text NOT NULL, started_at timestamptz NOT NULL, completed_at timestamptz, result text NOT NULL, payload jsonb NOT NULL);
+CREATE TABLE IF NOT EXISTS netengine_control.drift(drift_id text PRIMARY KEY, reconciliation_id text REFERENCES netengine_control.reconciliations, world_id text NOT NULL, resource_id text NOT NULL, classification text NOT NULL, status text NOT NULL, first_detected_at timestamptz NOT NULL, last_detected_at timestamptz NOT NULL, payload jsonb NOT NULL);
+DO $$
+BEGIN
+  IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'observations_execution_run_fk' AND connamespace = 'netengine_control'::regnamespace) THEN
+    ALTER TABLE netengine_control.observations ADD CONSTRAINT observations_execution_run_fk FOREIGN KEY(execution_run_id) REFERENCES netengine_control.execution_runs(run_id);
+  END IF;
+  IF NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conname = 'observations_operation_fk' AND connamespace = 'netengine_control'::regnamespace) THEN
+    ALTER TABLE netengine_control.observations ADD CONSTRAINT observations_operation_fk FOREIGN KEY(operation_id) REFERENCES netengine_control.operations(operation_id);
+  END IF;
+END $$;
+INSERT INTO netengine_control.schema_version(version) VALUES(4) ON CONFLICT DO NOTHING;
+COMMIT;

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "netsovereign"
-version = "0.3.0"
+version = "0.4.0"
 description = "Provider-neutral compiler and reconciliation runtime for sovereign worlds"
 readme = "README.md"
 requires-python = ">=3.12"

src/netsovereign/__init__.py

@@ -31,4 +31,4 @@
     "compare_worlds",
     "validate_spec",
 ]
-__version__ = "0.2.0"
+__version__ = "0.4.0"

src/netsovereign/cli.py

@@ -12,6 +12,7 @@
 import yaml
 from pydantic import BaseModel, ValidationError
 
+from .controlplane.repository import SQLiteControlPlaneRepository
 from .io import load_spec
 from .manifest import build_manifest, explain_manifest
 from .planning import (
@@ -30,6 +31,23 @@
 from .validation import has_errors, validate_spec
 
 app = typer.Typer(no_args_is_help=True, help="Validate and explain sovereign world intent.")
+control_plane_app = typer.Typer(help="Operate the durable local control plane.")
+app.add_typer(control_plane_app, name="control-plane")
+
+
+@control_plane_app.command("init")
+def control_plane_init(
+    database: Annotated[Path, typer.Option(envvar="NETENGINE_CONTROL_PLANE_PATH")] = Path(
+        ".netengine-control.db"
+    ),
+) -> None:
+    """Initialise or inspect a local durable control-plane schema."""
+    repository = SQLiteControlPlaneRepository(database)
+    row = repository.connection.execute(
+        "SELECT value FROM cp_metadata WHERE key='schema_version'"
+    ).fetchone()
+    repository.connection.close()
+    _emit({"database": str(database), "schema_version": int(row[0]), "status": "ready"})
 
 
 def _parse(path: Path) -> WorldSpec:

src/netsovereign/controlplane/__init__.py

@@ -0,0 +1,12 @@
+"""Durable, single-node control-plane contracts and services."""
+
+from .models import *  # noqa: F403
+from .repository import ControlPlaneRepository, SQLiteControlPlaneRepository
+from .service import ControlPlaneService, DriftDetector
+
+__all__ = [
+    "ControlPlaneRepository",
+    "SQLiteControlPlaneRepository",
+    "ControlPlaneService",
+    "DriftDetector",
+]

src/netsovereign/controlplane/models.py

@@ -0,0 +1,188 @@
+"""Provider-neutral durable control-plane records.
+
+These records intentionally do not import a database or provider implementation.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from enum import StrEnum
+from typing import Any
+
+from pydantic import Field, field_validator, model_validator
+
+from ..base import DomainModel
+
+CONTROL_PLANE_SCHEMA_VERSION = 4
+_SECRET_TERMS = ("password", "secret", "token", "private_key", "credential")
+
+
+def now_utc() -> datetime:
+    return datetime.now(UTC)
+
+
+def _contains_secret(value: Any) -> bool:
+    if isinstance(value, dict):
+        return any(
+            any(term in str(key).lower() for term in _SECRET_TERMS) or _contains_secret(item)
+            for key, item in value.items()
+        )
+    if isinstance(value, list):
+        return any(_contains_secret(item) for item in value)
+    return False
+
+
+def _require_aware(value: datetime | None) -> datetime | None:
+    if value is None:
+        return None
+    if value.tzinfo is None or value.utcoffset() is None:
+        raise ValueError("control-plane timestamps must be timezone-aware")
+    return value
+
+
+class DesiredRevision(DomainModel):
+    world_id: str
+    revision_id: str
+    parent_revision_id: str | None = None
+    schema_version: str
+    declaration_digest: str
+    declaration: dict[str, Any]
+    accepted_at: datetime
+    admission: dict[str, Any]
+    actor: str | None = None
+    authority_manifest_digest: str
+    boundary_policy_digest: str | None = None
+    compatibility: dict[str, Any] = Field(default_factory=dict)
+    status: str = "accepted"
+
+    _accepted_at_is_aware = field_validator("accepted_at")(_require_aware)
+
+
+class AuthoritativeRecord(DomainModel):
+    record_id: str
+    world_id: str
+    kind: str
+    authority_id: str
+    mandate_id: str | None = None
+    desired_revision_id: str
+    decision_reference: str
+    status: str
+    version: int = Field(ge=1)
+    lifecycle: dict[str, Any] = Field(default_factory=dict)
+    value: dict[str, Any] = Field(default_factory=dict)
+    recorded_at: datetime
+
+
+class ObservedRecord(DomainModel):
+    observation_id: str
+    world_id: str
+    resource_id: str
+    provider_id: str
+    binding_id: str
+    capability: str
+    provider_resource_id: str | None = None
+    observation_type: str
+    representation: Any = None
+    digest: str | None = None
+    health: str = "unknown"
+    conformance: str = "unknown"
+    observed_at: datetime
+    stale_after: datetime | None = None
+    execution_run_id: str | None = None
+    operation_id: str | None = None
+
+    _timestamps_are_aware = field_validator("observed_at", "stale_after")(_require_aware)
+
+
+class LockLease(DomainModel):
+    key: str
+    owner: str
+    acquired_at: datetime
+    expires_at: datetime
+    fencing_token: int
+    renewed_at: datetime | None = None
+    released_at: datetime | None = None
+    release_reason: str | None = None
+
+    _timestamps_are_aware = field_validator(
+        "acquired_at", "expires_at", "renewed_at", "released_at"
+    )(_require_aware)
+
+    @model_validator(mode="after")
+    def validate_interval(self) -> LockLease:
+        if self.expires_at < self.acquired_at:
+            raise ValueError("lease expiry cannot precede acquisition")
+        return self
+
+
+class Checkpoint(DomainModel):
+    checkpoint_id: str
+    world_id: str
+    desired_revision_id: str
+    plan_fingerprint: str
+    run_id: str
+    completed_operations: list[str]
+    incomplete_operations: list[str]
+    latest_observation_ids: list[str]
+    compensation: dict[str, Any] = Field(default_factory=dict)
+    created_at: datetime
+    reason: str
+    schema_version: int = CONTROL_PLANE_SCHEMA_VERSION
+    previous_checkpoint_id: str | None = None
+
+    _created_at_is_aware = field_validator("created_at")(_require_aware)
+
+    @model_validator(mode="after")
+    def reject_secrets(self) -> Checkpoint:
+        if _contains_secret(self.compensation):
+            raise ValueError("checkpoint compensation state must not contain secrets")
+        return self
+
+
+class ReconciliationRecord(DomainModel):
+    reconciliation_id: str
+    world_id: str
+    trigger: str
+    desired_revision_id: str
+    observed_snapshot_id: str | None = None
+    admission: dict[str, Any] = Field(default_factory=dict)
+    plan: dict[str, Any] = Field(default_factory=dict)
+    executable_plan_id: str | None = None
+    execution_run_id: str | None = None
+    started_at: datetime
+    completed_at: datetime | None = None
+    result: str = "running"
+    drift_summary: dict[str, int] = Field(default_factory=dict)
+    conformance_summary: dict[str, Any] = Field(default_factory=dict)
+
+    _timestamps_are_aware = field_validator("started_at", "completed_at")(_require_aware)
+
+
+class DriftClassification(StrEnum):
+    MISSING = "missing_resource"
+    UNEXPECTED = "unexpected_resource"
+    CHANGED = "changed_resource"
+    UNHEALTHY = "unhealthy_resource"
+    UNVERIFIABLE = "unverifiable_resource"
+    STALE = "stale_observation"
+    PROVIDER_UNAVAILABLE = "provider_unavailable"
+    CONFORMANT = "conformant_resource"
+
+
+class DriftRecord(DomainModel):
+    drift_id: str
+    world_id: str
+    resource_id: str
+    desired_revision_id: str
+    expected_digest: str | None
+    observed_digest: str | None
+    provider_id: str
+    capability: str
+    classification: DriftClassification
+    severity: str
+    first_detected_at: datetime
+    last_detected_at: datetime
+    status: str = "open"
+    reconciliation_id: str | None = None
+
+    _timestamps_are_aware = field_validator("first_detected_at", "last_detected_at")(_require_aware)