DataOps-platform/migrations/versions/20260818_553_governed_plugin_platform.py

"""Create the closed, fixture-only P3-WP13 plugin control plane."""

from alembic import op

revision = "20260818_553"
down_revision = "20260818_552"
branch_labels = None
depends_on = None

_TABLES = (
    "plugin_registry_versions",
    "plugin_approvals",
    "plugin_runtime_leases",
    "plugin_runs",
    "plugin_dead_letters",
    "plugin_audit_outbox",
)


def upgrade() -> None:
    op.get_bind().exec_driver_sql(r'''
    DO $roles$ BEGIN
      IF NOT EXISTS(SELECT 1 FROM pg_roles WHERE rolname='dataops_plugin_platform_owner' AND NOT rolcanlogin AND NOT rolsuper AND NOT rolcreaterole)
        OR NOT EXISTS(SELECT 1 FROM pg_roles WHERE rolname='dataops_plugin_platform_control' AND NOT rolcanlogin)
        OR NOT EXISTS(SELECT 1 FROM pg_roles WHERE rolname='dataops_app_runtime') THEN
        RAISE EXCEPTION 'WP13 roles must be provisioned by role-init';
      END IF;
    END $roles$;
    CREATE TABLE public.plugin_registry_versions (
      plugin_uid text NOT NULL CHECK(plugin_uid ~ '^[a-z][a-z0-9-]{2,62}$'),
      version text NOT NULL CHECK(version ~ '^[0-9]+[.][0-9]+[.][0-9]+([-.][A-Za-z0-9.]+)?$'),
      plugin_type text NOT NULL CHECK(plugin_type IN ('connector','parser','quality','notification','approval','agent_mcp')),
      manifest_digest char(64) NOT NULL CHECK(manifest_digest ~ '^[0-9a-f]{64}$'),
      artifact_digest char(64) NOT NULL CHECK(artifact_digest ~ '^[0-9a-f]{64}$'),
      trust_store_key_id text NOT NULL CHECK(trust_store_key_id='local-fixture-key-v1'),
      signature_digest char(64) NOT NULL CHECK(signature_digest ~ '^[0-9a-f]{64}$'),
      sbom_digest char(64) NOT NULL CHECK(sbom_digest ~ '^[0-9a-f]{64}$'),
      license_digest char(64) NOT NULL CHECK(license_digest ~ '^[0-9a-f]{64}$'),
      vulnerability_digest char(64) NOT NULL CHECK(vulnerability_digest ~ '^[0-9a-f]{64}$'),
      provenance_digest char(64) NOT NULL CHECK(provenance_digest ~ '^[0-9a-f]{64}$'),
      fixture_id text NOT NULL CHECK(fixture_id='ENGINEERING_EVIDENCE_ONLY'),
      permissions jsonb NOT NULL CHECK(permissions='{"file": false, "network": false, "secret": false, "child_process": false}'::jsonb),
      state text NOT NULL DEFAULT 'draft' CHECK(state IN ('draft','reviewed','approved','canary','active','paused','revoked','rolled_back','recovery')),
      review_actor text, reviewed_at timestamptz, activated_at timestamptz,
      created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
      PRIMARY KEY(plugin_uid,version), UNIQUE(manifest_digest), UNIQUE(artifact_digest)
    );
    CREATE UNIQUE INDEX plugin_registry_one_active ON public.plugin_registry_versions(plugin_uid) WHERE state='active';
    CREATE TABLE public.plugin_approvals (
      approval_uid uuid PRIMARY KEY DEFAULT gen_random_uuid(), plugin_uid text NOT NULL, version text NOT NULL,
      action_name text NOT NULL CHECK(action_name IN ('approve','canary','activate','pause','rollback','revoke','recover')),
      actor_ref text NOT NULL CHECK(actor_ref ~ '^[A-Za-z0-9_.:-]{1,120}$'),
      reviewer_ref text NOT NULL CHECK(reviewer_ref ~ '^[A-Za-z0-9_.:-]{1,120}$' AND reviewer_ref<>actor_ref),
      tenant_ref text NOT NULL CHECK(tenant_ref ~ '^[a-z][a-z0-9-]{0,62}$'), domain_ref text NOT NULL CHECK(domain_ref ~ '^[a-z][a-z0-9-]{0,62}$'),
      manifest_digest char(64) NOT NULL CHECK(manifest_digest ~ '^[0-9a-f]{64}$'), expires_at timestamptz NOT NULL,
      consumed_at timestamptz, created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
      FOREIGN KEY(plugin_uid,version) REFERENCES public.plugin_registry_versions(plugin_uid,version) ON DELETE RESTRICT,
      UNIQUE(plugin_uid,version,action_name,actor_ref,tenant_ref,domain_ref,manifest_digest)
    );
    CREATE TABLE public.plugin_runtime_leases (
      plugin_uid text NOT NULL, version text NOT NULL, tenant_ref text NOT NULL,
      lease_owner text, lease_token uuid, lease_fence bigint NOT NULL DEFAULT 0 CHECK(lease_fence>=0), lease_expires_at timestamptz,
      PRIMARY KEY(plugin_uid,version,tenant_ref), FOREIGN KEY(plugin_uid,version) REFERENCES public.plugin_registry_versions(plugin_uid,version) ON DELETE RESTRICT
    );
    CREATE TABLE public.plugin_runs (
      run_uid uuid PRIMARY KEY DEFAULT gen_random_uuid(), plugin_uid text NOT NULL, version text NOT NULL,
      tenant_ref text NOT NULL, domain_ref text NOT NULL, principal_ref text NOT NULL, operation_name text NOT NULL,
      idempotency_key text NOT NULL CHECK(idempotency_key ~ '^[A-Za-z0-9_.:-]{1,120}$'),
      input_digest char(64) NOT NULL CHECK(input_digest ~ '^[0-9a-f]{64}$'), output_digest char(64),
      request_digest char(64) NOT NULL CHECK(request_digest ~ '^[0-9a-f]{64}$'), lease_fence bigint NOT NULL CHECK(lease_fence>=0),
      state text NOT NULL CHECK(state IN ('running','succeeded','failed','dead_letter')),
      created_at timestamptz NOT NULL DEFAULT clock_timestamp(), completed_at timestamptz,
      FOREIGN KEY(plugin_uid,version) REFERENCES public.plugin_registry_versions(plugin_uid,version) ON DELETE RESTRICT,
      UNIQUE(plugin_uid,version,tenant_ref,idempotency_key)
    );
    CREATE TABLE public.plugin_dead_letters (
      dead_letter_uid uuid PRIMARY KEY DEFAULT gen_random_uuid(), run_uid uuid NOT NULL UNIQUE REFERENCES public.plugin_runs(run_uid) ON DELETE RESTRICT,
      failure_digest char(64) NOT NULL CHECK(failure_digest ~ '^[0-9a-f]{64}$'), incident_ref text, created_at timestamptz NOT NULL DEFAULT clock_timestamp()
    );
    CREATE TABLE public.plugin_audit_outbox (
      audit_uid uuid PRIMARY KEY DEFAULT gen_random_uuid(), plugin_uid text NOT NULL, version text NOT NULL, tenant_ref text NOT NULL,
      event_type text NOT NULL CHECK(event_type IN ('registered','reviewed','approved','canary','active','paused','rolled_back','revoked','recovery','invoked','dead_letter')),
      actor_ref text NOT NULL, payload_digest char(64) NOT NULL CHECK(payload_digest ~ '^[0-9a-f]{64}$'), created_at timestamptz NOT NULL DEFAULT clock_timestamp()
    );
    DO $owner$ DECLARE t text; BEGIN
      FOREACH t IN ARRAY ARRAY['plugin_registry_versions','plugin_approvals','plugin_runtime_leases','plugin_runs','plugin_dead_letters','plugin_audit_outbox'] LOOP
        EXECUTE 'ALTER TABLE public.'||quote_ident(t)||' OWNER TO dataops_plugin_platform_owner';
        EXECUTE 'REVOKE ALL ON TABLE public.'||quote_ident(t)||' FROM PUBLIC,dataops_app,dataops_app_runtime';
      END LOOP;
    END $owner$;
    CREATE FUNCTION public.plugin_platform_runtime_execute(p jsonb) RETURNS jsonb LANGUAGE plpgsql SECURITY DEFINER SET search_path=pg_catalog,public AS $run$
    DECLARE r record; digest text; out_digest text;
    BEGIN
      IF NOT pg_has_role(session_user,'dataops_app_runtime','MEMBER') OR jsonb_typeof(p)<>'object'
        OR p-ARRAY['plugin_uid','version','tenant_ref','domain_ref','principal_ref','operation_name','idempotency_key','input_digest','lease_token','lease_fence']<>'{}'::jsonb
        OR p->>'plugin_uid' !~ '^[a-z][a-z0-9-]{2,62}$' OR p->>'input_digest' !~ '^[0-9a-f]{64}$' THEN RAISE EXCEPTION 'plugin_runtime_denied'; END IF;
      SELECT * INTO r FROM public.plugin_registry_versions WHERE plugin_uid=p->>'plugin_uid' AND version=p->>'version' AND state='active' AND fixture_id='ENGINEERING_EVIDENCE_ONLY' AND permissions='{"file": false, "network": false, "secret": false, "child_process": false}'::jsonb;
      IF NOT FOUND THEN RAISE EXCEPTION 'plugin_not_active'; END IF;
      digest:=encode(sha256(convert_to((p-ARRAY['lease_token'])::text,'utf8')),'hex');
      SELECT encode(sha256(convert_to('ENGINEERING_EVIDENCE_ONLY|'||r.manifest_digest||'|'||digest,'utf8')),'hex') INTO out_digest;
      INSERT INTO public.plugin_runs(plugin_uid,version,tenant_ref,domain_ref,principal_ref,operation_name,idempotency_key,input_digest,output_digest,request_digest,lease_fence,state,completed_at)
        VALUES(r.plugin_uid,r.version,p->>'tenant_ref',p->>'domain_ref',p->>'principal_ref',p->>'operation_name',p->>'idempotency_key',p->>'input_digest',out_digest,digest,(p->>'lease_fence')::bigint,'succeeded',clock_timestamp())
        ON CONFLICT(plugin_uid,version,tenant_ref,idempotency_key) DO NOTHING;
      RETURN jsonb_build_object('plugin_uid',r.plugin_uid,'version',r.version,'output_digest',out_digest,'fixture_id','ENGINEERING_EVIDENCE_ONLY');
    END $run$;
    ALTER FUNCTION public.plugin_platform_runtime_execute(jsonb) OWNER TO dataops_plugin_platform_owner;
    REVOKE ALL ON FUNCTION public.plugin_platform_runtime_execute(jsonb) FROM PUBLIC,dataops_app;
    GRANT EXECUTE ON FUNCTION public.plugin_platform_runtime_execute(jsonb) TO dataops_app_runtime;
    ''')


def downgrade() -> None:
    bind = op.get_bind()
    checks = " OR ".join(f"EXISTS(SELECT 1 FROM public.{table} LIMIT 1)" for table in _TABLES)
    if bind.exec_driver_sql(f"SELECT {checks}").scalar():
        raise RuntimeError("downgrade refused: WP13 plugin facts are nonempty")
    bind.exec_driver_sql("""
      REVOKE ALL ON FUNCTION public.plugin_platform_runtime_execute(jsonb) FROM PUBLIC,dataops_app,dataops_app_runtime;
      DROP FUNCTION public.plugin_platform_runtime_execute(jsonb);
      DROP TABLE public.plugin_audit_outbox,public.plugin_dead_letters,public.plugin_runs,public.plugin_runtime_leases,public.plugin_approvals,public.plugin_registry_versions;
    """)