diff --git a/scripts/build.py b/scripts/build.py
index e2fb518..6905791 100644
--- a/scripts/build.py
+++ b/scripts/build.py
@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Conservative SQLite index from CVEProject, NVD and OSV raw JSON.
+"""Conservative SQLite index from CVEProject, NVD, OSV and Red Hat CSAF raw JSON.
 Do not mistake this index for a complete vulnerability matching engine.
 """
 import argparse
@@ -11,25 +11,24 @@ import shutil
 import sqlite3
 import subprocess
 import tempfile
+import sys
 from datetime import datetime, timezone
 from pathlib import Path
 
+# Make the top-level importers package importable when invoked as scripts/build.py.
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from importers import IMPORTERS
+
 CVE = re.compile(r'^CVE-\d{4}-\d{4,}$', re.I)
-FILES = ('vulnerabilities','aliases','affected_packages','cpe_matches')
+FILES = ('vulnerabilities','aliases','affected_packages','cpe_nodes','cpe_matches',
+         'redhat_advisories','redhat_advisory_cves','sources','meta')
+SCHEMA_VERSION = 3
 
 
 def canonical(obj):
     return json.dumps(obj, ensure_ascii=False, sort_keys=True, separators=(',', ':'))
 
 
-def sha(s):
-    return hashlib.sha256(s.encode('utf-8')).hexdigest()[:32]
-
-
-def date(value):
-    return value if isinstance(value, str) and value else None
-
-
 def get_git_revision(path):
     return subprocess.check_output(['git', '-C', str(path), 'rev-parse', 'HEAD'], text=True).strip()
 
@@ -44,168 +43,90 @@ def input_files(repo):
         yield from sorted(d.rglob('CVE-*.json'))
         return
 
+    if repo.name == 'vuln-redhat':
+        yield from sorted((d / 'csaf').rglob('*.json'))
+        return
+
     # Andere Quellen unverändert verarbeiten
     yield from sorted(d.rglob('*.json'))
 
 
-def upsert_vuln(db, vid, published=None, modified=None, state=None, summary=None):
-    if not vid or not (CVE.match(vid) or vid.startswith(('GHSA-','OSV-','PYSEC-','RUSTSEC-','GO-','DSA-','USN-'))):
-        return
-    db.execute('''INSERT INTO vulnerabilities(id,published,modified,state,summary) VALUES(?,?,?,?,?)
-      ON CONFLICT(id) DO UPDATE SET
-      published=COALESCE(excluded.published,vulnerabilities.published),
-      modified=COALESCE(excluded.modified,vulnerabilities.modified),
-      state=COALESCE(excluded.state,vulnerabilities.state),
-      summary=COALESCE(excluded.summary,vulnerabilities.summary)''',
-      (vid, published, modified, state, summary[:500] if isinstance(summary,str) else None))
-
-
-def process_cve(db, obj):
-    if not isinstance(obj, dict):
-      raise ValueError(
-        f"Unexpected CVE JSON type: {type(obj).__name__}; "
-        f"sample={str(obj)[:300]}"
-    )
-    m = obj.get('cveMetadata')
-    if not isinstance(m,dict):
-        return False
-    vid=m.get('cveId')
-    if not isinstance(vid,str) or not CVE.match(vid):
-        return False
-    upsert_vuln(db,vid,m.get('datePublished'),m.get('dateUpdated'),m.get('state'))
-    return True
-
-
-def walk_cpe(node, allow_simple=True):
-    """Only emit independent leaf matches for simple non-negated OR nodes.
-    Complex AND/NOT expressions must remain unsupported rather than cause false positives.
-    """
-    if not isinstance(node,dict) or node.get('negate'):
-        return
-    if node.get('operator','OR').upper() != 'OR':
-        return
-    matches=node.get('cpeMatch') or []
-    if isinstance(matches,list):
-        for match in matches:
-            if isinstance(match,dict):
-                yield match
-    for child in node.get('children') or []:
-        yield from walk_cpe(child)
-
-
-def process_nvd(db,obj):
-    # NVD mirror stores the CVE object directly; API responses wrap it in {'cve': ...}.
-    if not isinstance(obj,dict):
-        return False
-    cve=obj.get('cve') if isinstance(obj.get('cve'),dict) else obj
-    vid=cve.get('id')
-    if not isinstance(vid,str) or not CVE.match(vid):
-        return False
-    upsert_vuln(db,vid,cve.get('published'),cve.get('lastModified'),cve.get('vulnStatus'))
-    db.execute('DELETE FROM cpe_matches WHERE vulnerability_id=? AND source=?',(vid,'vuln-nvd'))
-    for node in cve.get('configurations') or []:
-        # NVD root configurations may contain nodes; a configuration with several
-        # nodes can be an AND and is not safe to flatten.
-        if not isinstance(node,dict) or node.get('operator','OR').upper() != 'OR' or node.get('negate'):
-            continue
-        for n in node.get('nodes') or []:
-            for match in walk_cpe(n):
-                criteria=match.get('criteria')
-                if not isinstance(criteria,str):
-                    continue
-                vals=(criteria, int(bool(match.get('vulnerable',False))),
-                      match.get('versionStartIncluding'), match.get('versionStartExcluding'),
-                      match.get('versionEndIncluding'),match.get('versionEndExcluding'))
-                key=sha(canonical((vid,vals)))
-                db.execute('''INSERT OR REPLACE INTO cpe_matches
-                  (id,vulnerability_id,criteria,vulnerable,version_start_including,version_start_excluding,version_end_including,version_end_excluding,source)
-                  VALUES(?,?,?,?,?,?,?,?,?)''',(key,vid,*vals,'vuln-nvd'))
-    return True
-
-
-def process_osv(db,obj):
-    vid=obj.get('id')
-    if not isinstance(vid,str) or not vid:
-        return False
-    # Only capture OSV ids with a known prefix or valid CVE.
-    if not (CVE.match(vid) or vid.startswith(('GHSA-','OSV-','PYSEC-','RUSTSEC-','GO-','DSA-','USN-'))):
-        return False
-    upsert_vuln(db,vid,date(obj.get('published')),date(obj.get('modified')),
-                'WITHDRAWN' if obj.get('withdrawn') else None,obj.get('summary'))
-    db.execute('DELETE FROM affected_packages WHERE vulnerability_id=? AND source=?',(vid,'vuln-osv'))
-    db.execute('DELETE FROM aliases WHERE vulnerability_id=?',(vid,))
-    for a in obj.get('aliases') or []:
-        if isinstance(a,str) and a:
-            db.execute('INSERT OR IGNORE INTO aliases(alias,vulnerability_id) VALUES(?,?)',(a,vid))
-    for i,a in enumerate(obj.get('affected') or []):
-        if not isinstance(a,dict):
-            continue
-        pkg=a.get('package') or {}
-        if not isinstance(pkg,dict):
-            continue
-        eco=pkg.get('ecosystem'); name=pkg.get('name')
-        ranges=a.get('ranges') or []
-        versions=a.get('versions') or []
-        if not isinstance(ranges,list) or not isinstance(versions,list):
-            continue
-        if not isinstance(eco,str) or not isinstance(name,str):
-            continue
-        # Preserve version/range event semantics; no lexicographic comparisons.
-        key=sha(canonical((vid,i,eco,name,ranges,versions)))
-        db.execute('INSERT OR REPLACE INTO affected_packages VALUES(?,?,?,?,?,?,?)',
-                   (key,vid,eco,name,canonical(ranges),canonical(versions),'vuln-osv'))
-    return True
-
-
-def stream_source(db,name,path):
-    count=0; bad=0
+def stream_source(db, name, path):
+    importer = IMPORTERS[name]
+    count = 0
+    bad = 0
+    scanned = 0
     for file in input_files(path):
+        scanned += 1
         try:
-            with file.open('r',encoding='utf-8') as f:
-                obj=json.load(f)
-            if name=='vuln-cve':
-                accepted=process_cve(db,obj)
-            elif name=='vuln-nvd':
-                accepted=process_nvd(db,obj)
-            elif name=='vuln-osv':
-                accepted=process_osv(db,obj)
-            else:
-                accepted=False
-            count+=int(accepted)
-        except (ValueError,UnicodeError,TypeError,sqlite3.Error) as exc:
-            bad+=1
-            if bad<=5:
-                print(f'INVALID {name} {file}: {exc}',flush=True)
+            with file.open('r', encoding='utf-8') as stream:
+                obj = json.load(stream)
+            count += int(bool(importer(db, obj)))
+        except (ValueError, UnicodeError, TypeError, KeyError, sqlite3.Error) as exc:
+            bad += 1
+            if bad <= 5:
+                print(f'INVALID {name} {file}: {exc}', flush=True)
     if bad:
         raise RuntimeError(f'{name}: {bad} invalid records; database not published')
     if name == 'vuln-nvd' and count == 0:
         raise RuntimeError('vuln-nvd: no usable documents; check source data and JSON shape')
+    if name in ('vuln-cve', 'vuln-osv', 'vuln-redhat') and count == 0:
+        raise RuntimeError(f'{name}: no usable documents; refusing to publish')
+    print(f'{name}: scanned={scanned} accepted={count}', flush=True)
     return count
 
 
 def dump_table(db,table):
     # memory efficient cursor; a canonical row object for deltas
+    # Tables added by new importers do not exist in the previous SQLite master.
+    if db.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,)).fetchone() is None:
+        return
     cols=[r[1] for r in db.execute(f'PRAGMA table_info({table})')]
     for row in db.execute(f'SELECT * FROM {table} ORDER BY 1'):
         yield row[0],dict(zip(cols,row))
 
 
-def make_delta(old_path,new_db,path):
-    old=sqlite3.connect(old_path)
-    with path.open('w',encoding='utf-8') as f:
-        for table in FILES:
-            prev=dump_table(old,table); curr=dump_table(new_db,table)
-            a=next(prev,None); b=next(curr,None)
-            while a is not None or b is not None:
-                if a is not None and (b is None or a[0]<b[0]):
-                    f.write(canonical({'table':table,'op':'delete','key':a[0]})+'\n'); a=next(prev,None)
-                elif b is not None and (a is None or b[0]<a[0]):
-                    f.write(canonical({'table':table,'op':'upsert','row':b[1]})+'\n'); b=next(curr,None)
-                else:
-                    if a[1]!=b[1]:
-                        f.write(canonical({'table':table,'op':'upsert','row':b[1]})+'\n')
-                    a=next(prev,None);b=next(curr,None)
-    old.close()
+def make_delta(old_path, new_db, path):
+    """Stream row differences with referentially safe table ordering.
+
+    Updaters must apply entire delta in ONE transaction with
+    PRAGMA defer_foreign_keys=ON, as CPE nodes may reference a later node.
+    """
+    old = sqlite3.connect(old_path)
+    try:
+        with tempfile.TemporaryDirectory(prefix='delta-parts-', dir=path.parent) as td:
+            temp_dir = Path(td)
+            deletion_files = {}
+            insertion_files = {}
+            for table in FILES:
+                deletions = temp_dir / f'{table}.del.jsonl'
+                insertions = temp_dir / f'{table}.upsert.jsonl'
+                previous = dump_table(old, table)
+                current = dump_table(new_db, table)
+                a, b = next(previous, None), next(current, None)
+                with deletions.open('w', encoding='utf-8') as d, insertions.open('w', encoding='utf-8') as u:
+                    while a is not None or b is not None:
+                        if a is not None and (b is None or a[0] < b[0]):
+                            d.write(canonical({'table': table, 'op': 'delete', 'key': a[0]}) + '\n')
+                            a = next(previous, None)
+                        elif b is not None and (a is None or b[0] < a[0]):
+                            u.write(canonical({'table': table, 'op': 'upsert', 'row': b[1]}) + '\n')
+                            b = next(current, None)
+                        else:
+                            if a[1] != b[1]:
+                                u.write(canonical({'table': table, 'op': 'upsert', 'row': b[1]}) + '\n')
+                            a, b = next(previous, None), next(current, None)
+                deletion_files[table] = deletions
+                insertion_files[table] = insertions
+            with path.open('wb') as output:
+                for table in reversed(FILES):
+                    with deletion_files[table].open('rb') as f:
+                        shutil.copyfileobj(f, output)
+                for table in FILES:
+                    with insertion_files[table].open('rb') as f:
+                        shutil.copyfileobj(f, output)
+    finally:
+        old.close()
 
 
 def compress(src,out):
@@ -213,17 +134,17 @@ def compress(src,out):
 
 
 def build_signature(source_revs, schema_path):
-    """Fingerprint inputs that can change the generated data, not wall-clock time."""
-    scripts_dir = Path(__file__).resolve().parent
+    """Hash source revisions plus ALL executable importer logic and schema."""
+    root = Path(__file__).resolve().parents[1]
     h = hashlib.sha256()
-    # Include every Python importer/helper alongside the builder, as well as SQL schema.
-    inputs = sorted(scripts_dir.glob('*.py'))
-    inputs.append(schema_path.resolve())
+    inputs = sorted((root / 'scripts').glob('*.py'))
+    inputs += sorted((root / 'importers').glob('*.py'))
+    inputs += [schema_path.resolve()]
     for path in inputs:
         if not path.is_file():
             raise RuntimeError(f'Missing build input: {path}')
-        logical_name = 'schema.sql' if path == schema_path.resolve() else f'scripts/{path.name}'
-        h.update(logical_name.encode('utf-8') + b'\x00')
+        logical_name = path.relative_to(root).as_posix() if path.is_relative_to(root) else 'sql/schema.sql'
+        h.update(logical_name.encode('utf-8') + b'\0')
         h.update(hashlib.sha256(path.read_bytes()).digest())
     h.update(canonical(source_revs).encode('utf-8'))
     return h.hexdigest()
@@ -238,7 +159,13 @@ def existing_release_ok(output, manifest):
     with packed.open('rb') as stream:
         for block in iter(lambda: stream.read(1024 * 1024), b''):
             h.update(block)
-    return h.hexdigest() == manifest['master_sha256']
+    if h.hexdigest() != manifest['master_sha256']:
+        return False
+    try:
+        with sqlite3.connect(f'file:{dbfile}?mode=ro', uri=True) as con:
+            return con.execute('PRAGMA quick_check').fetchone()[0] == 'ok'
+    except sqlite3.Error:
+        return False
 
 
 def main():
@@ -253,7 +180,7 @@ def main():
     if not active:
         raise SystemExit('No source repositories checked out; nothing published')
     # Fail closed if expected core sources are missing.
-    for name in ('vuln-cve','vuln-nvd','vuln-osv'):
+    for name in IMPORTERS:
         if name not in [n for n,_ in active]:
             raise SystemExit(f'Missing core source: {name}')
     source_revs={name:get_git_revision(path) for name,path in active}
@@ -277,39 +204,71 @@ def main():
         tmp=Path(td); new=tmp/'master.db'; db=sqlite3.connect(new)
         db.executescript(args.schema.read_text(encoding='utf-8'))
         stats={}
+        build_time = datetime.now(timezone.utc).isoformat(timespec='seconds').replace('+00:00','Z')
+        # sync_sources.py saves this file only AFTER all repository updates succeed.
+        source_manifest = args.sources.parent / 'source-manifest.json'
+        sync_info = {}
+        if source_manifest.is_file():
+            try:
+                for entry in json.loads(source_manifest.read_text(encoding='utf-8')):
+                    if isinstance(entry, dict) and isinstance(entry.get('repo'), str):
+                        sync_info[entry['repo']] = entry
+            except (ValueError, UnicodeError, TypeError) as exc:
+                raise RuntimeError(f'Invalid source manifest: {exc}') from exc
         for name,path in active:
-            if name in ('vuln-cve','vuln-nvd','vuln-osv'):
+            if name in IMPORTERS:
                 stats[name]=stream_source(db,name,path)
             else:
                 stats[name]=0
-            db.execute('INSERT INTO sources VALUES(?,?,?,?)',(name,source_revs[name],int(name in ('vuln-cve','vuln-nvd','vuln-osv')),stats[name]))
+            supported = name in IMPORTERS
+            # Timestamp of successful synchronization, recorded by sync_sources.py.
+            item = sync_info.get(name, {})
+            synced_at = item.get('synced_at') if item.get('commit') == source_revs[name] else None
+            status = 'success' if supported else 'unsupported'
+            db.execute('''INSERT INTO sources
+                (name,revision,indexed,documents,last_sync,last_import,import_status,import_errors)
+                VALUES (?,?,?,?,?,?,?,?)''',
+                (name,source_revs[name],int(supported),stats[name],synced_at,
+                 build_time if supported else None,status,0))
             print(f'{name}: indexed={stats[name]} revision={source_revs[name][:12]}',flush=True)
-        if not any(stats.get(n) for n in ('vuln-cve','vuln-nvd','vuln-osv')):
+        if not any(stats.get(n) for n in IMPORTERS):
             raise RuntimeError('Core source indexes are empty; refusing to publish')
         old_version=None
+        old_schema_version=None
         if old_db.is_file():
             with sqlite3.connect(old_db) as o:
                 row=o.execute("SELECT value FROM meta WHERE key='version'").fetchone()
                 old_version=row[0] if row else None
+                schema_row=o.execute("SELECT value FROM meta WHERE key='schema_version'").fetchone()
+                old_schema_version=schema_row[0] if schema_row else None
+        # Existing clients cannot apply changed DDL with a row-only delta.
+        delta_allowed = bool(old_version and old_schema_version == str(SCHEMA_VERSION))
         version=datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%S%fZ')
-        for k,v in (('version',version),('schema_version','1')):
-            db.execute('INSERT INTO meta VALUES(?,?)',(k,v))
+        for k,v in (('version',version),('schema_version',str(SCHEMA_VERSION))):
+            db.execute('''INSERT INTO meta(key,value) VALUES(?,?)
+              ON CONFLICT(key) DO UPDATE SET value=excluded.value''',(k,v))
         db.commit(); db.execute('PRAGMA optimize'); db.execute('VACUUM'); db.commit()
         assert db.execute('PRAGMA integrity_check').fetchone()[0]=='ok'
         db.close()
-        manifest={'version':version,'schema_version':1,'previous':old_version,'source_commits':source_revs,
+        manifest={'version':version,'schema_version':SCHEMA_VERSION,'previous':old_version,'source_commits':source_revs,
                   'build_signature':signature,
+                  'requires_full_master': bool(old_version and not delta_allowed),
+                  'delta_available': delta_allowed,
                   'indexed_documents':stats,'format':'sqlite+zstd','delta_format':'jsonl+zstd',
-                  'note':'Only CVE/NVD/OSV indexed in v1; no complete version-matching engine'}
+                  'note':'CVE/NVD/OSV/Red Hat; NVD CPE trees and CSAF product statuses preserved; importer coverage is partial',
+                  'active_importers':sorted(IMPORTERS),
+                  'source_status':{name: ('success' if name in IMPORTERS else 'unsupported') for name in source_revs},
+                  'delta_tables':list(FILES),
+                  'delta_requires_deferred_foreign_keys':True}
         compress(new,tmp/'master.db.zst')
-        if old_version:
+        if delta_allowed:
             with sqlite3.connect(new) as nd:
                 make_delta(old_db,nd,tmp/'delta.jsonl')
             compress(tmp/'delta.jsonl',tmp/f'delta-{old_version}-to-{version}.jsonl.zst')
         # Atomically replace individual artifacts; manifest written last as commit marker.
         os.replace(new,old_db)
         os.replace(tmp/'master.db.zst',args.output/'master.db.zst')
-        if old_version:
+        if delta_allowed:
             os.replace(tmp/f'delta-{old_version}-to-{version}.jsonl.zst',args.output/f'delta-{old_version}-to-{version}.jsonl.zst')
         with (args.output/'master.db.zst').open('rb') as hf:
             h=hashlib.sha256()
diff --git a/scripts/run.sh b/scripts/run.sh
index 9063783..8fbd6e5 100644
--- a/scripts/run.sh
+++ b/scripts/run.sh
@@ -3,11 +3,10 @@
 set -Eeuo pipefail
 umask 077
 ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
-CACHE_ROOT="${CACHE_ROOT:-/opt/gitea-builder/cache/vuln-builder}"
+CACHE_ROOT="${CACHE_ROOT:-/vuln-cache/vuln-builder}"
 export CACHE_ROOT
 mkdir -p "$CACHE_ROOT" "$CACHE_ROOT/releases"
 exec 9>"$CACHE_ROOT/.builder.lock"
 flock -w 600 9 || { echo 'Another builder is running' >&2; exit 1; }
 python3 "$ROOT/scripts/sync_sources.py"
 python3 "$ROOT/scripts/build.py" --sources "$CACHE_ROOT/sources" --output "$CACHE_ROOT/releases" "$@"
-
diff --git a/scripts/sync_sources.py b/scripts/sync_sources.py
index 5d2213a..726a524 100644
--- a/scripts/sync_sources.py
+++ b/scripts/sync_sources.py
@@ -2,6 +2,7 @@
 #!/usr/bin/env python3
 
 import json
+from datetime import datetime, timezone
 import os
 import re
 import subprocess
@@ -84,7 +85,8 @@ def sync_repository(owner, repo, branch):
     return {
         "repo": repo,
         "commit": result,
-        "path": str(path)
+        "path": str(path),
+        "synced_at": datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")
     }