from __future__ import annotations from sqlalchemy import text def knowledge_status(session) -> dict: change_sets = session.execute( text( """ SELECT status, COUNT(*) FROM public.knowledge_change_sets GROUP BY status """ ) ).all() projections = session.execute( text( """ SELECT engine, status, COUNT(*) FROM public.knowledge_index_projections GROUP BY engine, status """ ) ).all() freshness = session.execute( text( """ SELECT COUNT(*) FILTER (WHERE status = 'active') AS active_documents, MAX(source_updated_at) FILTER (WHERE status = 'active') AS latest_source FROM public.governance_documents """ ) ).one() return { "change_sets": {row[0]: row[1] for row in change_sets}, "projections": {f"{row[0]}:{row[1]}": row[2] for row in projections}, "active_documents": freshness[0], "latest_source_updated_at": ( freshness[1].isoformat() if freshness[1] else None ), } def list_change_sets(session, *, limit: int = 50) -> list[dict]: rows = session.execute( text( """ SELECT id::text, source_type, source_uid::text, source_revision, change_type, added_count, modified_count, deleted_count, impacted_point_count, impact_truncated, status, target_generation, last_error, created_at, activated_at FROM public.knowledge_change_sets ORDER BY created_at DESC LIMIT :limit """ ), {"limit": limit}, ).mappings() return [dict(row) for row in rows] def change_set_detail(session, change_set_id: str) -> dict | None: change_set = ( session.execute( text( """ SELECT id::text, source_type, source_uid::text, source_revision, change_type, source_snapshot_hash, added_count, modified_count, deleted_count, impacted_point_count, impact_truncated, status, target_generation, last_error, created_at, updated_at, activated_at FROM public.knowledge_change_sets WHERE id = CAST(:id AS uuid) """ ), {"id": change_set_id}, ) .mappings() .one_or_none() ) if change_set is None: return None items = session.execute( text( """ SELECT point_key, change_kind, old_point_revision, new_point_revision, old_content_hash, new_content_hash, caused_by_point_key, propagation_hop, canonical_status, embedding_status, cache_status, lightrag_status, attempts, last_error FROM public.knowledge_change_items WHERE change_set_id = CAST(:id AS uuid) ORDER BY point_key """ ), {"id": change_set_id}, ).mappings() return {**dict(change_set), "items": [dict(row) for row in items]} def retry_change_set(session, change_set_id: str) -> bool: updated = session.execute( text( """ UPDATE public.knowledge_change_sets SET status = 'pending', last_error = NULL, updated_at = CURRENT_TIMESTAMP WHERE id = CAST(:id AS uuid) AND status IN ('failed','degraded') """ ), {"id": change_set_id}, ).rowcount if updated: session.execute( text( """ UPDATE public.knowledge_index_projections SET status = 'pending', last_error = NULL, updated_at = CURRENT_TIMESTAMP WHERE change_set_id = CAST(:id AS uuid) AND status IN ('failed','unverified') """ ), {"id": change_set_id}, ) return updated == 1 def rollback_change_set(session, change_set_id: str) -> bool: current = session.execute( text( """ SELECT d.id::text, d.object_uid::text, d.object_version FROM public.governance_documents d JOIN public.knowledge_change_sets cs ON cs.id = d.change_set_id WHERE cs.id = CAST(:id AS uuid) AND cs.status = 'canonical_active' AND d.status = 'active' FOR UPDATE """ ), {"id": change_set_id}, ).one_or_none() if current is None: return False previous = session.execute( text( """ SELECT id::text, object_version FROM public.governance_documents WHERE object_uid = CAST(:uid AS uuid) AND object_version < :version ORDER BY object_version DESC LIMIT 1 FOR UPDATE """ ), {"uid": current[1], "version": current[2]}, ).one_or_none() if previous is None: return False session.execute( text( """ UPDATE public.governance_documents SET status = 'superseded' WHERE id = CAST(:id AS uuid); UPDATE public.governance_documents SET status = 'active' WHERE id = CAST(:previous_id AS uuid); UPDATE public.knowledge_points SET status = 'superseded', valid_to = CURRENT_TIMESTAMP WHERE source_uid = CAST(:uid AS uuid) AND status = 'active'; WITH restore AS ( SELECT DISTINCT ON (point_key) id FROM public.knowledge_points WHERE source_uid = CAST(:uid AS uuid) AND source_revision <= :previous_version AND status <> 'deleted' ORDER BY point_key, point_revision DESC ) UPDATE public.knowledge_points p SET status = 'active', valid_to = NULL, activated_at = CURRENT_TIMESTAMP FROM restore WHERE p.id = restore.id; UPDATE public.knowledge_change_sets SET status = 'rolled_back', updated_at = CURRENT_TIMESTAMP WHERE id = CAST(:change_set_id AS uuid); UPDATE public.knowledge_index_projections SET status = 'deleting', updated_at = CURRENT_TIMESTAMP WHERE document_id = CAST(:id AS uuid) AND status <> 'deleted'; """ ), { "id": current[0], "previous_id": previous[0], "uid": current[1], "previous_version": previous[1], "change_set_id": change_set_id, }, ) return True