DataOps-platform/migrations/versions/20260720_100_llamaindex_lightrag_knowledge.py

"""Add the governance knowledge retrieval and dynamic-update control plane."""

from alembic import op

revision = "20260720_100"
down_revision = "20260719_90"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(
        """
        CREATE EXTENSION IF NOT EXISTS vector;
        CREATE EXTENSION IF NOT EXISTS pg_trgm;

        CREATE TABLE IF NOT EXISTS public.user_business_domain_grants (
            user_id UUID NOT NULL REFERENCES public.users(id) ON DELETE CASCADE,
            business_domain_uid UUID NOT NULL,
            grant_type VARCHAR(20) NOT NULL CHECK (grant_type IN ('read','manage')),
            granted_by UUID REFERENCES public.users(id) ON DELETE SET NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            PRIMARY KEY (user_id, business_domain_uid, grant_type)
        );

        CREATE TABLE IF NOT EXISTS public.knowledge_embedding_profiles (
            id UUID PRIMARY KEY,
            provider VARCHAR(30) NOT NULL,
            model VARCHAR(120) NOT NULL,
            dimension INTEGER NOT NULL CHECK (dimension > 0),
            distance VARCHAR(20) NOT NULL DEFAULT 'cosine',
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('building','active','retired','failed')
            ),
            config_hash CHAR(64) NOT NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            activated_at TIMESTAMPTZ
        );
        CREATE UNIQUE INDEX IF NOT EXISTS uq_knowledge_embedding_profile_active
            ON public.knowledge_embedding_profiles ((status))
            WHERE status = 'active';

        CREATE TABLE IF NOT EXISTS public.knowledge_points (
            id UUID PRIMARY KEY,
            point_key VARCHAR(500) NOT NULL,
            point_revision BIGINT NOT NULL CHECK (point_revision > 0),
            source_type VARCHAR(50) NOT NULL,
            source_uid UUID NOT NULL,
            source_revision BIGINT NOT NULL CHECK (source_revision > 0),
            semantic_path VARCHAR(500) NOT NULL,
            content TEXT NOT NULL,
            content_hash CHAR(64) NOT NULL,
            metadata JSONB NOT NULL DEFAULT '{}'::jsonb,
            metadata_hash CHAR(64) NOT NULL,
            permission_scope JSONB NOT NULL DEFAULT '{}'::jsonb,
            permission_hash CHAR(64) NOT NULL,
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('building','active','superseded','deleted','failed')
            ),
            valid_from TIMESTAMPTZ,
            valid_to TIMESTAMPTZ,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            activated_at TIMESTAMPTZ,
            UNIQUE (point_key, point_revision)
        );
        CREATE UNIQUE INDEX IF NOT EXISTS uq_knowledge_point_active
            ON public.knowledge_points(point_key) WHERE status = 'active';
        CREATE INDEX IF NOT EXISTS idx_knowledge_point_source_revision
            ON public.knowledge_points(source_type, source_uid, source_revision);
        CREATE INDEX IF NOT EXISTS idx_knowledge_point_content_hash
            ON public.knowledge_points(content_hash);

        CREATE TABLE IF NOT EXISTS public.knowledge_point_dependencies (
            from_point_key VARCHAR(500) NOT NULL,
            to_point_key VARCHAR(500) NOT NULL,
            relation_type VARCHAR(50) NOT NULL,
            source VARCHAR(30) NOT NULL CHECK (
                source IN ('governance','deterministic_rule')
            ),
            generation BIGINT NOT NULL,
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('active','superseded','deleted')
            ),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            PRIMARY KEY (
                from_point_key, to_point_key, relation_type, source, generation
            )
        );
        CREATE INDEX IF NOT EXISTS idx_knowledge_dependency_reverse
            ON public.knowledge_point_dependencies(to_point_key, status);

        CREATE TABLE IF NOT EXISTS public.knowledge_change_sets (
            id UUID PRIMARY KEY,
            correlation_id UUID NOT NULL,
            source_type VARCHAR(50) NOT NULL,
            source_uid UUID NOT NULL,
            source_revision BIGINT NOT NULL,
            change_type VARCHAR(20) NOT NULL CHECK (
                change_type IN ('create','update','delete','permission','rebuild')
            ),
            source_snapshot_hash CHAR(64),
            added_count INTEGER NOT NULL DEFAULT 0,
            modified_count INTEGER NOT NULL DEFAULT 0,
            deleted_count INTEGER NOT NULL DEFAULT 0,
            impacted_point_count INTEGER NOT NULL DEFAULT 0,
            impact_truncated BOOLEAN NOT NULL DEFAULT FALSE,
            status VARCHAR(30) NOT NULL CHECK (status IN (
                'pending','diffed','building','validating','canonical_active',
                'projecting','complete','degraded','failed','rolled_back'
            )),
            target_generation BIGINT NOT NULL,
            last_error VARCHAR(1000),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            activated_at TIMESTAMPTZ,
            UNIQUE (source_type, source_uid, source_revision, target_generation)
        );
        CREATE INDEX IF NOT EXISTS idx_knowledge_change_set_status
            ON public.knowledge_change_sets(status, created_at);

        ALTER TABLE public.governance_documents
            ADD COLUMN IF NOT EXISTS point_set_hash CHAR(64),
            ADD COLUMN IF NOT EXISTS active_generation BIGINT,
            ADD COLUMN IF NOT EXISTS change_set_id UUID
                REFERENCES public.knowledge_change_sets(id);

        CREATE TABLE IF NOT EXISTS public.knowledge_change_items (
            change_set_id UUID NOT NULL REFERENCES public.knowledge_change_sets(id)
                ON DELETE CASCADE,
            point_key VARCHAR(500) NOT NULL,
            change_kind VARCHAR(20) NOT NULL CHECK (
                change_kind IN ('added','modified','deleted','impacted')
            ),
            old_point_revision BIGINT,
            new_point_revision BIGINT,
            old_content_hash CHAR(64),
            new_content_hash CHAR(64),
            caused_by_point_key VARCHAR(500),
            propagation_hop INTEGER,
            canonical_status VARCHAR(20) NOT NULL DEFAULT 'pending',
            embedding_status VARCHAR(20) NOT NULL DEFAULT 'pending',
            cache_status VARCHAR(20) NOT NULL DEFAULT 'pending',
            lightrag_status VARCHAR(20) NOT NULL DEFAULT 'pending',
            attempts INTEGER NOT NULL DEFAULT 0,
            last_error VARCHAR(1000),
            PRIMARY KEY (change_set_id, point_key)
        );

        ALTER TABLE public.governance_chunks
            ADD COLUMN IF NOT EXISTS chunk_kind VARCHAR(30),
            ADD COLUMN IF NOT EXISTS section_path VARCHAR(500),
            ADD COLUMN IF NOT EXISTS metadata JSONB NOT NULL DEFAULT '{}'::jsonb,
            ADD COLUMN IF NOT EXISTS token_count INTEGER,
            ADD COLUMN IF NOT EXISTS lexical_text TEXT,
            ADD COLUMN IF NOT EXISTS search_vector TSVECTOR,
            ADD COLUMN IF NOT EXISTS source_locator JSONB NOT NULL DEFAULT '{}'::jsonb,
            ADD COLUMN IF NOT EXISTS primary_point_key VARCHAR(500),
            ADD COLUMN IF NOT EXISTS point_keys JSONB NOT NULL DEFAULT '[]'::jsonb,
            ADD COLUMN IF NOT EXISTS point_set_hash CHAR(64),
            ADD COLUMN IF NOT EXISTS change_set_id UUID REFERENCES public.knowledge_change_sets(id);
        CREATE INDEX IF NOT EXISTS idx_governance_chunks_search_vector
            ON public.governance_chunks USING GIN(search_vector);
        CREATE INDEX IF NOT EXISTS idx_governance_chunks_metadata
            ON public.governance_chunks USING GIN(metadata jsonb_path_ops);
        CREATE INDEX IF NOT EXISTS idx_governance_chunks_lexical_trgm
            ON public.governance_chunks USING GIN(lexical_text gin_trgm_ops);

        CREATE TABLE IF NOT EXISTS public.knowledge_chunk_embeddings (
            id UUID PRIMARY KEY,
            chunk_id UUID NOT NULL REFERENCES public.governance_chunks(id)
                ON DELETE CASCADE,
            profile_id UUID NOT NULL REFERENCES public.knowledge_embedding_profiles(id),
            embedding vector(1024) NOT NULL,
            embedding_hash CHAR(64) NOT NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (chunk_id, profile_id)
        );
        CREATE INDEX IF NOT EXISTS idx_knowledge_chunk_embedding_hnsw
            ON public.knowledge_chunk_embeddings
            USING hnsw (embedding vector_cosine_ops);

        CREATE TABLE IF NOT EXISTS public.knowledge_index_projections (
            id UUID PRIMARY KEY,
            document_id UUID NOT NULL REFERENCES public.governance_documents(id)
                ON DELETE CASCADE,
            change_set_id UUID REFERENCES public.knowledge_change_sets(id)
                ON DELETE SET NULL,
            engine VARCHAR(30) NOT NULL CHECK (
                engine IN ('canonical_vector','lightrag')
            ),
            generation BIGINT NOT NULL,
            workspace VARCHAR(300) NOT NULL,
            external_document_id VARCHAR(700) NOT NULL,
            content_hash CHAR(64) NOT NULL,
            status VARCHAR(20) NOT NULL CHECK (status IN (
                'pending','processing','ready','failed','deleting','deleted','unverified'
            )),
            attempts INTEGER NOT NULL DEFAULT 0,
            external_track_id VARCHAR(300),
            available_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            last_error VARCHAR(1000),
            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (document_id, engine, generation, workspace)
        );

        CREATE TABLE IF NOT EXISTS public.knowledge_cache_dependencies (
            cache_key_hash CHAR(64) NOT NULL,
            point_key VARCHAR(500) NOT NULL,
            point_revision BIGINT NOT NULL,
            generation BIGINT NOT NULL,
            business_domain_uid UUID,
            expires_at TIMESTAMPTZ NOT NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            PRIMARY KEY (cache_key_hash, point_key, point_revision, generation)
        );
        CREATE INDEX IF NOT EXISTS idx_knowledge_cache_point
            ON public.knowledge_cache_dependencies(point_key, expires_at);

        CREATE TABLE IF NOT EXISTS public.knowledge_query_audits (
            id UUID PRIMARY KEY,
            query_hash CHAR(64) NOT NULL,
            user_id UUID REFERENCES public.users(id) ON DELETE SET NULL,
            roles JSONB NOT NULL DEFAULT '[]'::jsonb,
            business_domain_uids JSONB NOT NULL DEFAULT '[]'::jsonb,
            mode VARCHAR(30) NOT NULL,
            retriever_counts JSONB NOT NULL DEFAULT '{}'::jsonb,
            cited_points JSONB NOT NULL DEFAULT '[]'::jsonb,
            degraded_components JSONB NOT NULL DEFAULT '[]'::jsonb,
            correlation_id UUID NOT NULL,
            latency_ms INTEGER,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
        );

        CREATE TABLE IF NOT EXISTS public.knowledge_evaluation_sets (
            id UUID PRIMARY KEY,
            name VARCHAR(200) NOT NULL,
            description TEXT,
            status VARCHAR(20) NOT NULL DEFAULT 'draft',
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
        );
        CREATE TABLE IF NOT EXISTS public.knowledge_evaluation_cases (
            id UUID PRIMARY KEY,
            evaluation_set_id UUID NOT NULL REFERENCES public.knowledge_evaluation_sets(id)
                ON DELETE CASCADE,
            case_type VARCHAR(40) NOT NULL,
            query TEXT NOT NULL,
            allowed_business_domains JSONB NOT NULL DEFAULT '[]'::jsonb,
            expected_sources JSONB NOT NULL DEFAULT '[]'::jsonb,
            expected_answer_points JSONB NOT NULL DEFAULT '[]'::jsonb,
            must_refuse BOOLEAN NOT NULL DEFAULT FALSE,
            mutation JSONB,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
        );
        CREATE TABLE IF NOT EXISTS public.knowledge_evaluation_runs (
            id UUID PRIMARY KEY,
            evaluation_set_id UUID NOT NULL REFERENCES public.knowledge_evaluation_sets(id),
            configuration JSONB NOT NULL,
            status VARCHAR(20) NOT NULL,
            started_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            finished_at TIMESTAMPTZ
        );
        CREATE TABLE IF NOT EXISTS public.knowledge_evaluation_results (
            run_id UUID NOT NULL REFERENCES public.knowledge_evaluation_runs(id)
                ON DELETE CASCADE,
            case_id UUID NOT NULL REFERENCES public.knowledge_evaluation_cases(id)
                ON DELETE CASCADE,
            metrics JSONB NOT NULL DEFAULT '{}'::jsonb,
            citations JSONB NOT NULL DEFAULT '[]'::jsonb,
            passed BOOLEAN NOT NULL,
            failure_reason VARCHAR(1000),
            PRIMARY KEY (run_id, case_id)
        );
        """
    )


def downgrade() -> None:
    # Dynamic knowledge data is intentionally preserved during application rollback.
    pass