feat: DuckDB como fuente de verdad (multi-valor, ownership selectivo, escritura, libretas)

F1 — migraciones: 002_multivalue (persons +telefonos/emails/direcciones/extra_fm JSON,
backfill desde singulares con to_json) + 003_addressbooks (tabla addressbooks + seed
idempotente de la libreta por defecto). Conteos intactos (697/1065/98).

F2 — ingest_vault selectivo (anti-pisado): personas que ya existen en DB solo actualizan
note_path + extra_fm vía duckdb_upsert(update_cols=...), NO pisan los campos OWNED por la
DB; personas nuevas = bootstrap completo. _link_contacts enlaza por listas telefonos[]/
emails[] además del singular. ingest_dav itera todas las libretas de la tabla addressbooks.

F3 — escritura estructurada (server/writes.py + endpoints en main.py): CRUD
/api/person|contact|event, /api/addressbook, /api/calendar, /api/person/{slug}/render
(DB→nota preservando la prosa del cuerpo), /api/push/dav (reconcilia DB→Xandikos). El push
DAV y el render ocurren fuera de la transacción de escritura para no bloquear la DB con
latencia de red. registry_bridge.py importa las funciones nuevas; app.md actualizado.

Verificado: 18 tests verdes; ownership probado sobre datos reales (un centinela DB-owned
sobrevivió a POST /api/ingest/vault sobre las 697 fichas); person CRUD + materialización
de la ficha .md en vivo, con cleanup sin residuo.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

server/ingest.py

@@ -1,9 +1,21 @@
 """Ingests del service osint_db: vault Obsidian y servidor DAV (Xandikos).
 
-Las tablas maestras se reconstruyen por reemplazo completo (DELETE + INSERT en
-una transacción): el vault y Xandikos son las fuentes de verdad, así que cada
-ingest deja la base exactamente como el origen. Tras cada ingest se re-enlazan
-los contactos con sus fichas y se reconstruyen las tablas derivadas.
+Inversión "DuckDB como fuente de verdad": las tablas de espejo puro (notes, y
+las entidades que solo guardan frontmatter JSON: organizations/domains/cases/
+places) se reconstruyen por reemplazo completo (DELETE + INSERT). Pero persons
+ya NO es un espejo del frontmatter: la DB es la dueña de sus campos estructurados
+(teléfonos, emails, direcciones, dni, ...), que se editan por la API y se
+materializan a la nota. Por eso el ingest del vault es SELECTIVO para persons:
+
+  - slug que YA existe en la DB -> solo refresca note_path + extra_fm (el
+    frontmatter no-owned); los campos OWNED de la DB se conservan.
+  - slug NUEVO -> INSERT completo (bootstrap): adopta el frontmatter como valor
+    inicial, poblando las listas multi-valor desde los singulares o las listas
+    del frontmatter.
+
+contacts y events siguen siendo espejo puro de Xandikos (DELETE + INSERT). Tras
+cada ingest se re-enlazan los contactos con sus fichas y se reconstruyen las
+derivadas. Todo bajo el lock single-writer del service.
 """
 
 from __future__ import annotations
@@ -19,12 +31,65 @@ from .db import write_conn
 from .derived import rebuild_derived
 from .registry_bridge import (
     dav_get_collection,
+    dav_list_addressbooks,
     dav_list_calendars,
+    duckdb_upsert,
     list_obsidian_notes,
     pass_get_secret,
     read_obsidian_note,
 )
 
+# Campos de persons de los que la DB es dueña: un re-ingest del vault NO los
+# pisa cuando la ficha ya existe. Incluye los singulares (derivados de las
+# listas) y fuente (origen de la ficha). dav_uid se deriva de fuente.
+PERSON_OWNED = (
+    "nombre",
+    "aliases",
+    "sexo",
+    "fecha_nacimiento",
+    "dni",
+    "telefono",
+    "email",
+    "direccion",
+    "telefonos",
+    "emails",
+    "direcciones",
+    "pais",
+    "contexto",
+    "fuente",
+    "dav_uid",
+    "tags",
+)
+
+# Claves de control del frontmatter que no van a extra_fm (ni son OWNED): las
+# gestiona el propio ingest o son metadata de la nota.
+PERSON_CONTROL = ("slug", "tipo", "note_path")
+
+# Orden de columnas de la tabla persons (debe casar con migrations 001 + 002).
+PERSON_COLUMNS = (
+    "slug",
+    "note_path",
+    "nombre",
+    "aliases",
+    "sexo",
+    "fecha_nacimiento",
+    "dni",
+    "telefono",
+    "email",
+    "direccion",
+    "pais",
+    "contexto",
+    "fuente",
+    "dav_uid",
+    "tags",
+    "updated_at",
+    "telefonos",
+    "emails",
+    "direcciones",
+    "extra_fm",
+)
+
+
 def _norm(value):
     """Normaliza 'null'/''/None del frontmatter a None real."""
     if value is None:
@@ -48,6 +113,22 @@ def _as_list(value) -> list:
     return v if isinstance(v, list) else [v]
 
 
+def _multi(fm: dict, plural: str, singular: str) -> list:
+    """Lista multi-valor desde el frontmatter: prioriza la clave plural.
+
+    Si el frontmatter trae la clave plural (telefonos/emails/direcciones) la
+    usa; si no, envuelve el singular (telefono/email/direccion) en lista. Cada
+    valor se normaliza a str y se descartan los vacíos.
+    """
+    raw = _as_list(fm.get(plural)) or _as_list(fm.get(singular))
+    out = []
+    for item in raw:
+        s = _as_str(item)
+        if s:
+            out.append(s)
+    return out
+
+
 def _json(value) -> str:
     """Serializa un valor a JSON compacto (sin escapar acentos)."""
     return json.dumps(value, ensure_ascii=False, default=str)
@@ -61,17 +142,75 @@ def _dav_uid_from_fuente(fuente) -> str | None:
     return m.group(1) if m else None
 
 
+def _extra_fm(fm: dict) -> str:
+    """JSON con las claves del frontmatter que NO son OWNED ni de control.
+
+    Permite que un re-ingest refresque el frontmatter no estructurado de la
+    ficha (campos libres que la DB no posee) sin pisar lo que la DB sí posee.
+    Las claves plurales multi-valor también se excluyen (son OWNED).
+    """
+    skip = set(PERSON_OWNED) | set(PERSON_CONTROL)
+    extra = {k: v for k, v in fm.items() if k not in skip}
+    return _json(extra)
+
+
+def _person_row_from_fm(slug: str, rel_path: str, fm: dict, mtime, base: str) -> dict:
+    """Construye la fila completa de persons desde el frontmatter (bootstrap).
+
+    Se usa SOLO para fichas cuyo slug aún no existe en la DB: adopta el
+    frontmatter como valor inicial de los campos OWNED, derivando las listas
+    multi-valor desde el plural o el singular del frontmatter y rellenando los
+    singulares con el primer elemento de cada lista.
+    """
+    telefonos = _multi(fm, "telefonos", "telefono")
+    emails = _multi(fm, "emails", "email")
+    direcciones = _multi(fm, "direcciones", "direccion")
+    return {
+        "slug": slug,
+        "note_path": rel_path,
+        "nombre": _as_str(fm.get("nombre")) or base,
+        "aliases": _json(_as_list(fm.get("aliases"))),
+        "sexo": _as_str(fm.get("sexo")),
+        "fecha_nacimiento": _as_str(fm.get("fecha_nacimiento")),
+        "dni": _as_str(fm.get("dni")),
+        "telefono": telefonos[0] if telefonos else None,
+        "email": emails[0] if emails else None,
+        "direccion": direcciones[0] if direcciones else None,
+        "pais": _as_str(fm.get("pais")),
+        "contexto": _as_str(fm.get("contexto")),
+        "fuente": _as_str(fm.get("fuente")),
+        "dav_uid": _dav_uid_from_fuente(fm.get("fuente")),
+        "tags": _json(_as_list(fm.get("tags"))),
+        "updated_at": mtime,
+        "telefonos": _json(telefonos),
+        "emails": _json(emails),
+        "direcciones": _json(direcciones),
+        "extra_fm": _extra_fm(fm),
+    }
+
+
 def ingest_vault(cfg: Config) -> dict:
     """Escanea el vault completo y reconstruye notes + tablas de entidades.
 
-    Devuelve {status:'ok', notes:N, persons:N, organizations:N, domains:N,
-    cases:N, places:N, skipped_unreadable:N, derived_rebuilt:[...]}.
+    notes y las entidades de espejo puro (organizations/domains/cases/places)
+    se reemplazan por completo. persons se ingesta de forma SELECTIVA: las
+    fichas existentes solo refrescan note_path + extra_fm (conservando los
+    campos OWNED de la DB), las nuevas se insertan completas (bootstrap).
+
+    Devuelve {status:'ok', notes:N, persons:N, persons_inserted:N,
+    persons_updated:N, organizations:N, domains:N, cases:N, places:N,
+    skipped_unreadable:N, derived_rebuilt:[...]}.
     """
     if not os.path.isdir(cfg.vault_dir):
         return {"status": "error", "error": f"vault no encontrado: {cfg.vault_dir}"}
 
     note_rows: list = []
-    entity_rows: dict = {table: [] for _, _, table in cfg.entity_folders}
+    # Filas de espejo puro (todas las entidades menos persons).
+    mirror_rows: dict = {
+        table: [] for _, _, table in cfg.entity_folders if table != "persons"
+    }
+    # Fichas de persona, deduplicadas por slug, como dicts {slug, fm, ...}.
+    person_fichas: dict = {}
     folder_to_table = {folder: table for folder, _, table in cfg.entity_folders}
     skipped = 0
 
@@ -105,28 +244,17 @@ def ingest_vault(cfg: Config) -> dict:
         if top_folder in folder_to_table and is_level1 and not base.startswith("_"):
             table = folder_to_table[top_folder]
             if table == "persons":
-                entity_rows[table].append(
-                    [
-                        slug,
-                        rel_path,
-                        _as_str(fm.get("nombre")) or base,
-                        _json(_as_list(fm.get("aliases"))),
-                        _as_str(fm.get("sexo")),
-                        _as_str(fm.get("fecha_nacimiento")),
-                        _as_str(fm.get("dni")),
-                        _as_str(fm.get("telefono")),
-                        _as_str(fm.get("email")),
-                        _as_str(fm.get("direccion")),
-                        _as_str(fm.get("pais")),
-                        _as_str(fm.get("contexto")),
-                        _as_str(fm.get("fuente")),
-                        _dav_uid_from_fuente(fm.get("fuente")),
-                        _json(_as_list(fm.get("tags"))),
-                        mtime,
-                    ]
-                )
+                # Gana la primera ficha vista con cada slug (respeta la PK).
+                if slug not in person_fichas:
+                    person_fichas[slug] = {
+                        "slug": slug,
+                        "rel_path": rel_path,
+                        "fm": fm,
+                        "mtime": mtime,
+                        "base": base,
+                    }
             else:
-                entity_rows[table].append(
+                mirror_rows[table].append(
                     [
                         slug,
                         rel_path,
@@ -137,7 +265,6 @@ def ingest_vault(cfg: Config) -> dict:
                     ]
                 )
 
-    derived_rebuilt: list = []
     with write_conn(cfg.db_path) as conn:
         conn.execute("BEGIN")
         try:
@@ -146,20 +273,27 @@ def ingest_vault(cfg: Config) -> dict:
                 conn.executemany(
                     "INSERT INTO notes VALUES (?, ?, ?, ?, ?, ?)", note_rows
                 )
-            conn.execute("DELETE FROM persons")
-            if entity_rows["persons"]:
-                conn.executemany(
-                    "INSERT INTO persons VALUES "
-                    "(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
-                    _dedup_by_slug(entity_rows["persons"]),
-                )
             for table in ("organizations", "domains", "cases", "places"):
                 conn.execute(f"DELETE FROM {table}")
-                if entity_rows[table]:
+                if mirror_rows[table]:
                     conn.executemany(
                         f"INSERT INTO {table} VALUES (?, ?, ?, ?, ?, ?)",
-                        _dedup_by_slug(entity_rows[table]),
+                        _dedup_by_slug(mirror_rows[table]),
                     )
+            conn.execute("COMMIT")
+        except Exception:
+            conn.execute("ROLLBACK")
+            raise
+
+    # persons: ingest selectivo via duckdb_upsert (ownership de campos OWNED).
+    # Se hace fuera de la transacción de espejo puro pero bajo el mismo lock de
+    # proceso (write_conn ya lo liberó): single-writer respetado.
+    p_inserted, p_updated = _ingest_persons_selective(cfg.db_path, person_fichas)
+
+    # Re-enlace de contactos + derivadas, de nuevo bajo el lock de escritura.
+    with write_conn(cfg.db_path) as conn:
+        conn.execute("BEGIN")
+        try:
             _link_contacts(conn)
             conn.execute("COMMIT")
         except Exception:
@@ -170,16 +304,89 @@ def ingest_vault(cfg: Config) -> dict:
     return {
         "status": "ok",
         "notes": len(note_rows),
-        "persons": len(_dedup_by_slug(entity_rows["persons"])),
-        "organizations": len(_dedup_by_slug(entity_rows["organizations"])),
-        "domains": len(_dedup_by_slug(entity_rows["domains"])),
-        "cases": len(_dedup_by_slug(entity_rows["cases"])),
-        "places": len(_dedup_by_slug(entity_rows["places"])),
+        "persons": len(person_fichas),
+        "persons_inserted": p_inserted,
+        "persons_updated": p_updated,
+        "organizations": len(_dedup_by_slug(mirror_rows["organizations"])),
+        "domains": len(_dedup_by_slug(mirror_rows["domains"])),
+        "cases": len(_dedup_by_slug(mirror_rows["cases"])),
+        "places": len(_dedup_by_slug(mirror_rows["places"])),
         "skipped_unreadable": skipped,
         "derived_rebuilt": derived_rebuilt,
     }
 
 
+def _ingest_persons_selective(db_path: str, person_fichas: dict) -> tuple:
+    """Upsert selectivo de persons: existentes solo note_path+extra_fm, nuevas full.
+
+    Lee qué slugs existen ya en la DB y reparte las fichas en dos lotes:
+      - existentes -> duckdb_upsert con update_cols=['note_path','extra_fm'],
+        de modo que los campos OWNED de la DB no se pisan.
+      - nuevas -> duckdb_upsert con update_cols=None (full insert de bootstrap).
+
+    Devuelve (inserted, updated). Bajo el lock single-writer del proceso:
+    duckdb_upsert abre su propia conexión, pero DuckDB comparte la instancia
+    de la base dentro del proceso, así que no hay conflicto de lock.
+
+    Raises:
+        RuntimeError: si algún upsert de persons devuelve status 'error'.
+    """
+    if not person_fichas:
+        return 0, 0
+
+    # Slugs ya presentes en la DB (lectura read-only, sin tocar el writer).
+    import duckdb
+
+    conn = duckdb.connect(db_path, read_only=True)
+    try:
+        existing = {row[0] for row in conn.execute("SELECT slug FROM persons").fetchall()}
+    finally:
+        conn.close()
+
+    existing_rows: list = []
+    new_rows: list = []
+    for slug, ficha in person_fichas.items():
+        full = _person_row_from_fm(
+            ficha["slug"], ficha["rel_path"], ficha["fm"], ficha["mtime"], ficha["base"]
+        )
+        if slug in existing:
+            # Solo los campos que el vault sigue gobernando para fichas vivas:
+            # dónde está la nota y el frontmatter no-owned.
+            existing_rows.append(
+                {
+                    "slug": full["slug"],
+                    "note_path": full["note_path"],
+                    "extra_fm": full["extra_fm"],
+                }
+            )
+        else:
+            new_rows.append(full)
+
+    inserted = updated = 0
+    if existing_rows:
+        res = duckdb_upsert(
+            db_path,
+            "persons",
+            existing_rows,
+            key_cols=["slug"],
+            update_cols=["note_path", "extra_fm"],
+        )
+        if res.get("status") != "ok":
+            raise RuntimeError(f"persons upsert (existentes): {res.get('error')}")
+        updated += res.get("updated", 0)
+        # Filas "existentes" que en realidad ya no estaban (carrera) se insertan.
+        inserted += res.get("inserted", 0)
+    if new_rows:
+        res = duckdb_upsert(
+            db_path, "persons", new_rows, key_cols=["slug"], update_cols=None
+        )
+        if res.get("status") != "ok":
+            raise RuntimeError(f"persons upsert (nuevas): {res.get('error')}")
+        inserted += res.get("inserted", 0)
+        updated += res.get("updated", 0)
+    return inserted, updated
+
+
 def _dedup_by_slug(rows: list) -> list:
     """Quita filas con slug repetido (gana la primera) para respetar la PK."""
     seen, out = set(), []
@@ -194,7 +401,10 @@ def _dedup_by_slug(rows: list) -> list:
 def ingest_dav(cfg: Config) -> dict:
     """Baja las colecciones de Xandikos y reconstruye contacts + events.
 
-    Devuelve {status:'ok', contacts:N, events:N, calendars:[...],
+    Itera TODAS las libretas CardDAV registradas en la tabla addressbooks (no
+    solo la colección fija): cada contacto guarda su collection real. Si la
+    tabla está vacía cae a la colección por defecto de la config. Devuelve
+    {status:'ok', contacts:N, events:N, addressbooks:[...], calendars:[...],
     contacts_linked:N, derived_rebuilt:[...]} o {status:'error', error}.
     """
     secret = pass_get_secret(cfg.pass_secret)
@@ -206,37 +416,42 @@ def ingest_dav(cfg: Config) -> dict:
         }
     pwd = secret["value"]  # sensible: nunca logear
 
-    coll = dav_get_collection(
-        cfg.dav_base, cfg.dav_user, pwd, cfg.dav_contacts_collection, "vcard"
-    )
-    if coll.get("status") != "ok":
-        return {
-            "status": "error",
-            "error": f"CardDAV: {coll.get('error')} (http {coll.get('http_status')})",
-        }
+    collections = _addressbook_collections(cfg)
 
     now = datetime.now(tz=timezone.utc)
     contact_rows: list = []
     seen_uids: set = set()
-    for res in coll.get("resources", []):
-        parsed = davparse.parse_vcard(res.get("data", ""))
-        uid = parsed["uid"] or os.path.splitext(os.path.basename(res["href"]))[0]
-        if uid in seen_uids:
-            continue
-        seen_uids.add(uid)
-        contact_rows.append(
-            [
-                uid,
-                cfg.dav_contacts_collection,
-                res.get("etag"),
-                parsed["fn"] or None,
-                _json(parsed["tels"]),
-                _json(parsed["emails"]),
-                res.get("data", ""),
-                None,  # note_path se rellena en el enlace posterior
-                now,
-            ]
+    used_addressbooks: list = []
+    for collection in collections:
+        coll = dav_get_collection(
+            cfg.dav_base, cfg.dav_user, pwd, collection, "vcard"
         )
+        if coll.get("status") != "ok":
+            return {
+                "status": "error",
+                "error": f"CardDAV {collection}: {coll.get('error')} "
+                f"(http {coll.get('http_status')})",
+            }
+        used_addressbooks.append(collection)
+        for res in coll.get("resources", []):
+            parsed = davparse.parse_vcard(res.get("data", ""))
+            uid = parsed["uid"] or os.path.splitext(os.path.basename(res["href"]))[0]
+            if uid in seen_uids:
+                continue
+            seen_uids.add(uid)
+            contact_rows.append(
+                [
+                    uid,
+                    collection,
+                    res.get("etag"),
+                    parsed["fn"] or None,
+                    _json(parsed["tels"]),
+                    _json(parsed["emails"]),
+                    res.get("data", ""),
+                    None,  # note_path se rellena en el enlace posterior
+                    now,
+                ]
+            )
 
     cals = dav_list_calendars(cfg.dav_base, cfg.dav_user, pwd, cfg.dav_calendar_home)
     if cals.get("status") != "ok":
@@ -308,33 +523,63 @@ def ingest_dav(cfg: Config) -> dict:
         "status": "ok",
         "contacts": len(contact_rows),
         "events": len(event_rows),
+        "addressbooks": used_addressbooks,
         "calendars": calendar_names,
         "contacts_linked": linked,
         "derived_rebuilt": derived_rebuilt,
     }
 
 
+def _addressbook_collections(cfg: Config) -> list:
+    """Colecciones CardDAV a recorrer en el ingest DAV.
+
+    Fuente de verdad: la tabla addressbooks de la DB. Si está vacía (o ilegible)
+    cae a la colección por defecto de la config para no romper el ingest.
+    """
+    try:
+        import duckdb
+
+        conn = duckdb.connect(cfg.db_path, read_only=True)
+        try:
+            rows = conn.execute(
+                "SELECT collection_path FROM addressbooks "
+                "WHERE collection_path IS NOT NULL ORDER BY slug"
+            ).fetchall()
+        finally:
+            conn.close()
+        collections = [r[0] for r in rows if r[0]]
+    except Exception:  # noqa: BLE001 — tabla ausente/ilegible -> default
+        collections = []
+    if not collections:
+        collections = [cfg.dav_contacts_collection]
+    return collections
+
+
 def _link_contacts(conn) -> int:
     """Enlaza contacts.note_path contra las fichas de persons.
 
     Orden de matching por fiabilidad: UID estilo osint-<slug> (creado por el
-    push del vault), dav_uid registrado en la ficha, teléfono normalizado y
-    por último email. Devuelve el número de contactos enlazados.
+    push del vault), dav_uid registrado en la ficha, teléfono normalizado
+    (singular o cualquiera de la lista telefonos[]) y por último email (singular
+    o cualquiera de emails[]). Devuelve el número de contactos enlazados.
     """
     persons = conn.execute(
-        "SELECT slug, note_path, telefono, email, dav_uid FROM persons"
+        "SELECT slug, note_path, telefono, email, dav_uid, telefonos, emails "
+        "FROM persons"
     ).fetchall()
     by_slug, by_dav_uid, by_phone, by_email = {}, {}, {}, {}
-    for slug, note_path, telefono, email, dav_uid in persons:
+    for slug, note_path, telefono, email, dav_uid, telefonos, emails in persons:
         by_slug[slug] = note_path
         if dav_uid:
             by_dav_uid.setdefault(dav_uid, note_path)
-        if telefono:
-            key = davparse.norm_phone(telefono)
+        # Teléfonos: singular + todos los de la lista multi-valor.
+        for tel in _coalesce_values(telefono, telefonos):
+            key = davparse.norm_phone(tel)
             if key:
                 by_phone.setdefault(key, note_path)
-        if email:
-            by_email.setdefault(str(email).strip().lower(), note_path)
+        # Emails: singular + todos los de la lista multi-valor.
+        for em in _coalesce_values(email, emails):
+            by_email.setdefault(str(em).strip().lower(), note_path)
 
     contacts = conn.execute("SELECT uid, tels, emails FROM contacts").fetchall()
     linked = 0
@@ -362,3 +607,24 @@ def _link_contacts(conn) -> int:
             )
             linked += 1
     return linked
+
+
+def _coalesce_values(singular, list_json) -> list:
+    """Une el valor singular con la lista JSON multi-valor (sin vacíos ni dups)."""
+    out, seen = [], set()
+    candidates = []
+    if singular:
+        candidates.append(singular)
+    try:
+        parsed = json.loads(list_json) if list_json else []
+    except (TypeError, ValueError):
+        parsed = []
+    if isinstance(parsed, list):
+        candidates.extend(parsed)
+    for v in candidates:
+        s = str(v).strip()
+        key = s.lower()
+        if s and key not in seen:
+            seen.add(key)
+            out.append(s)
+    return out