"""Destructive only to unique disposable test namespaces; run via entrypoint --qualify-registry. Exercises actual HTTP controllers, PostgreSQL catalogs, Redis, graph/vector clients, and concurrent registry inserts. Never targets a preexisting brain or external data. """ from concurrent.futures import ThreadPoolExecutor import json import os from pathlib import Path import sys from uuid import uuid4 sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import httpx from src.config import config from src.core.instances import cache_adapter, data_adapter, graph_adapter, vector_store_adapter from src.lib.postgresql._naming import brain_db_name from src.lib.postgresql import _provisioning as pg from src.services.brain_identity import BrainAlreadyExistsError, BrainNotFoundError def check(condition, message): if not condition: raise RuntimeError(message) def identities(api): response = api.get('/system/brains-list') response.raise_for_status() return sorted((b['name_key'], b['id']) for b in response.json()) def storage_snapshot(): with pg.borrow(pg.get_system_pool()) as conn, conn.cursor() as cursor: cursor.execute('SELECT id, name_key, document FROM data_brains ORDER BY name_key') registry = cursor.fetchall() cursor.execute('SELECT document FROM data_workspaces ORDER BY id') workspaces = cursor.fetchall() with pg.borrow(pg.get_pool(pg._maintenance_dbname())) as conn, conn.cursor() as cursor: cursor.execute('SELECT datname FROM pg_database ORDER BY datname') databases = cursor.fetchall() # Include values, not just key names. Unrelated Celery heartbeat bookkeeping # is outside application namespaces and excluded from this comparison. redis = cache_adapter.cache.client cache = {key: redis.dump(key) for pattern in ('system:brain*', 'codex*') for key in redis.scan_iter(match=pattern)} return registry, workspaces, databases, cache def read(api, endpoint, brain_id): if endpoint == 'entities': return api.get('/retrieve/entities', params={'brain_id': brain_id}) return api.post('/retrieve/' + endpoint, json={ 'brain_id': brain_id, 'query' if endpoint == 'search' else 'text': 'registry isolation probe', }) def qualify(): suffix = uuid4().hex[:12] unknown = 'codex_unknown_' + suffix names = ['codex-test-' + suffix, 'codex_test_' + suffix] orphan = 'codex_orphan_' + suffix race = 'codex_race_' + suffix # Cleanup is limited to namespaces proven absent before this test. temporary = names + [orphan, race] created = set() results = {} api_url = os.getenv('QUALIFY_API_URL', 'http://127.0.0.1:7860') with httpx.Client(base_url=api_url, headers={'BrainPAT': config.system_pat}, timeout=120, trust_env=False) as api: before_ids = identities(api) before = storage_snapshot() for name in temporary + [unknown]: check(data_adapter.get_brain(name) is None, 'Test namespace collision') check(brain_db_name(name) not in {r[0] for r in before[2]}, 'Test storage collision') try: for endpoint in ('entities', 'search', 'context'): response = read(api, endpoint, unknown) check(response.status_code == 404, 'Unknown ' + endpoint + ' did not return 404') check(response.json()['error']['code'] == 'BRAIN_NOT_FOUND', 'Wrong missing-brain code') check(storage_snapshot() == before, 'Unknown read changed registry/workspaces/databases/cache') check(identities(api) == before_ids, 'Unknown read changed public identities') results['unknown_' + endpoint] = '404 BRAIN_NOT_FOUND; storage/cache unchanged' # Defense in depth: bypass HTTP and hit each real storage read path. data = data_adapter.data graph = graph_adapter.graph._store vectors = vector_store_adapter.vector_store backend_before = (set(data._initialized_brains), set(graph._schema_ready), dict(graph._brains), set(vectors._initialized_stores), set(vectors._brain_extensions_ready)) for operation in ( lambda: data.get_text_chunks(brain_id=unknown), lambda: graph.get_brain(unknown), lambda: vectors.search_vectors([1.0] + [0.0] * 511, brain_id=unknown, store='data'), ): try: operation() except BrainNotFoundError: pass else: raise RuntimeError('Unknown direct backend read accepted') check(storage_snapshot() == before, 'Direct backend read changed persistent state') check(backend_before == (set(data._initialized_brains), set(graph._schema_ready), dict(graph._brains), set(vectors._initialized_stores), set(vectors._brain_extensions_ready)), 'Unknown backend read populated a cache') results['direct_backend_reads'] = 'data/graph/vector rejected; catalogs and caches unchanged' for name in names: created.add(name) response = api.post('/system/brains', json={'brain_id': name}) check(response.status_code == 200, 'Explicit create failed') identity = (response.json()['name_key'], response.json()['id']) duplicate = api.post('/system/brains', json={'brain_id': name}) check(duplicate.status_code == 409 and duplicate.json()['error']['code'] == 'BRAIN_ALREADY_EXISTS', 'Duplicate was not conflict') check(identity in identities(api) and identities(api) == identities(api), 'Identity changed') for endpoint in ('entities', 'search', 'context'): response = read(api, endpoint, name) check(response.status_code == 200, 'Known ' + endpoint + ' failed: ' + str(response.status_code)) clear = api.post('/system/brains/' + name + '/clear') check(clear.status_code == 200, 'Clear rejected supported identifier') check(identity in identities(api), 'Clear changed registry identity') response = api.delete('/system/brains/' + name) check(response.status_code == 200, 'Delete rejected supported identifier') check(data_adapter.get_brain(name) is None, 'Delete retained registry') check(brain_db_name(name) not in {r[0] for r in storage_snapshot()[2]}, 'Delete retained physical storage') results['known_reads_identifiers_duplicates'] = 'GET/search/context 200; stable IDs; duplicate 409; clear/delete hyphen and underscore passed' # No preflight lookup: exercise real concurrent storage uniqueness handling. created.add(race) def create_concurrently(_): try: data_adapter.create_brain(race) return 'created' except BrainAlreadyExistsError: return 'conflict' with ThreadPoolExecutor(max_workers=2) as executor: outcomes = sorted(executor.map(create_concurrently, range(2))) check(outcomes == ['conflict', 'created'], 'Storage uniqueness race was not a domain conflict') results['storage_duplicate_race'] = 'one create, one domain conflict' # Create only a physical database, without ever creating a registry row. created.add(orphan) pg.ensure_database_exists(brain_db_name(orphan)) check(data_adapter.get_brain(orphan) is None, 'Orphan unexpectedly registered') check(all(n != orphan for n, _ in identities(api)), 'Orphan exposed as registered brain') def orphan_schema(): with pg.borrow(pg.get_pool(brain_db_name(orphan))) as conn, conn.cursor() as cur: cur.execute("SELECT schemaname, tablename FROM pg_tables ORDER BY schemaname, tablename") return cur.fetchall() orphan_before = orphan_schema() for endpoint in ('entities', 'search', 'context'): response = read(api, endpoint, orphan) check(response.status_code == 404, 'Orphan storage authorized an unknown read') check(orphan_schema() == orphan_before, 'Orphan read created tables') response = api.delete('/system/brains/' + orphan) check(response.status_code == 200 and response.json()['existed'] is False, 'Orphan cleanup failed') check(brain_db_name(orphan) not in {r[0] for r in storage_snapshot()[2]}, 'Orphan database remains') check(data_adapter.get_brain(orphan) is None, 'Orphan cleanup created a registry row') results['orphan_cleanup'] = 'physical database removed without registry creation' finally: for name in sorted(created): response = api.delete('/system/brains/' + name) check(response.status_code == 200, 'Temporary brain cleanup failed') check(identities(api) == before_ids, 'Temporary qualification changed preexisting brain identities') results['result'] = 'passed' print(json.dumps(results, sort_keys=True)) if __name__ == '__main__': qualify()