feat(mutation): route serial add mutations through coordinator (#156)

* Add serial add mutation coordinator

* Route serial add paths through coordinator

* fix(cloud-import): resolve final doc name under ingest lock

* test(add): cover coordinator conflict edge cases

* fix(indexer): roll back the PageIndex blob when indexing fails after col.add

Since the add mutation stopped eagerly snapshotting .openkb/files (it registers the new blob via track_new only on success), a blob written by col.add() leaks if index_long_document raises afterward: pageindex.db is rolled back by the snapshot but the blob file is not, and nothing reclaims it. Delete the doc on the failure path so the indexer owns cleanup of a half-applied add, restoring the .openkb/files rollback-surface guarantee.

* fix(cloud-import): surface real error instead of mislabeling snapshot prep

run_add_mutation handles snapshot/body failures itself and returns False, so the broad except only ever catches pre-mutation errors (name resolution, registry read, plan construction). Echo the real exception instead of the old 'Failed to prepare mutation snapshot' label, which hid the cause at DEBUG level.

* style: wrap long lines for ruff E501 (CI gate from #159)

* style: apply ruff format to add_coordinator.py and test_add_command.py

* refactor(cli): drop duplicate _cleanup_staging, reuse coordinator helper

cli._cleanup_staging duplicated add_coordinator._cleanup_staging_dirs (both guard None then shutil.rmtree with ignore_errors). Drop the cli copy and route its two pre-coordinator call sites through the coordinator helper.

* refactor(add_coordinator): drop unused rollback_error_message field

AddMutationPlan.rollback_error_message had zero callers — every construction site (cli.py x2, tests x6) used the default. Remove the field and inline the literal at its single read site.

Author: gwokhou

openkb/add_coordinator.py

@@ -0,0 +1,134 @@
+from __future__ import annotations
+
+import logging
+import shutil
+from collections.abc import Callable, Sequence
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import click
+
+from openkb.locks import kb_ingest_lock_held
+from openkb.mutation import MutationSnapshot, snapshot_paths
+
+logger = logging.getLogger(__name__)
+
+MutationBody = Callable[[MutationSnapshot], None]
+PostCommitHook = Callable[[], None]
+
+
+class DirtyRollbackError(RuntimeError):
+    """A mutation's rollback failed, leaving an active journal on disk.
+
+    The KB may be in a partially-applied state that the retained journal will
+    attempt to roll back on the next exclusive-lock acquisition. Batch owners
+    (the parallel/serial ``add`` loops) MUST stop committing further mutations
+    on top of this dirty state instead of continuing — otherwise the next
+    recovery rolls this journal back over the shared paths it recorded
+    (``hashes.json``, ``index.md``, ``concepts/``, ``entities/``) and silently
+    clobbers the later commits. Single-mutation callers should let it propagate
+    so the command fails loudly; rerunning recovers via the drain.
+    """
+
+    def __init__(self, operation: str, journal_path: Path) -> None:
+        super().__init__(
+            f"Dirty rollback for {operation}; journal retained at {journal_path}. "
+            f"Rerun the command to recover."
+        )
+        self.operation = operation
+        self.journal_path = journal_path
+
+
+@dataclass(slots=True)
+class AddMutationPlan:
+    operation: str
+    details: dict
+    touched_paths: Sequence[Path]
+    body: MutationBody
+    post_commit_hooks: Sequence[PostCommitHook] = field(default_factory=tuple)
+    hardlink_dirs: set[Path] = field(default_factory=set)
+    staging_dirs: Sequence[Path | None] = field(default_factory=tuple)
+
+
+def _cleanup_staging_dirs(staging_dirs: Sequence[Path | None]) -> None:
+    for staging_dir in staging_dirs:
+        if staging_dir is not None:
+            shutil.rmtree(staging_dir, ignore_errors=True)
+
+
+def _rollback_snapshot(plan: AddMutationPlan, snapshot) -> Path | None:
+    """Best-effort rollback; returns the retained journal path on dirty failure.
+
+    Returns ``snapshot.journal_path`` when the snapshot existed but rollback
+    FAILED (the active journal is retained for next-run recovery), otherwise
+    ``None`` — covering both "snapshot is None" (nothing was applied; the
+    failure happened during snapshot setup before the body ran) and a clean
+    rollback that discarded its journal.
+    """
+    if snapshot is None:
+        _cleanup_staging_dirs(plan.staging_dirs)
+        return None
+    rollback_error = snapshot.rollback_best_effort()
+    if rollback_error is None:
+        snapshot.discard_best_effort()
+    else:
+        click.echo(
+            "  [ERROR] Rollback failed; mutation journal retained for recovery: "
+            f"{snapshot.journal_path}"
+        )
+    _cleanup_staging_dirs(plan.staging_dirs)
+    return snapshot.journal_path if rollback_error is not None else None
+
+
+def _failure_target(details: dict) -> str:
+    for key in ("name", "doc_name", "doc_id"):
+        value = details.get(key)
+        if value:
+            return f" for {value}"
+    return ""
+
+
+def run_add_mutation(kb_dir: Path, plan: AddMutationPlan) -> bool:
+    if not kb_ingest_lock_held(kb_dir / ".openkb"):
+        raise RuntimeError("run_add_mutation requires the caller to hold kb_ingest_lock")
+    snapshot = None
+    try:
+        snapshot = snapshot_paths(
+            kb_dir,
+            list(plan.touched_paths),
+            operation=plan.operation,
+            details=plan.details,
+            hardlink_dirs=plan.hardlink_dirs,
+        )
+        plan.body(snapshot)
+        snapshot.mark_committed()
+    except Exception as exc:
+        dirty_journal = _rollback_snapshot(plan, snapshot)
+        if dirty_journal is not None:
+            # Rollback failed and left an active journal. Stop the batch rather
+            # than committing more docs on top of dirty state that the next
+            # recovery would roll back over.
+            raise DirtyRollbackError(plan.operation, dirty_journal)
+        click.echo(f"  [ERROR] {plan.operation} failed{_failure_target(plan.details)}: {exc}")
+        logger.debug("%s mutation failed:", plan.operation, exc_info=True)
+        return False
+    except BaseException:
+        # Interrupt (KeyboardInterrupt / SystemExit): best-effort rollback for
+        # its side-effects only. Do NOT raise DirtyRollbackError — propagate the
+        # interrupt so the user's abort is honored. Any retained journal or
+        # orphaned staging is recovered next run by the drain + reaper.
+        _rollback_snapshot(plan, snapshot)
+        raise
+    finally:
+        _cleanup_staging_dirs(plan.staging_dirs)
+
+    for hook in plan.post_commit_hooks:
+        try:
+            hook()
+        except Exception as exc:
+            logger.warning("Post-commit hook failed for %s: %s", plan.operation, exc)
+
+    cleanup_error = snapshot.discard_best_effort()
+    if cleanup_error is not None:
+        click.echo(f"  [WARN] mutation journal cleanup failed: {cleanup_error}")
+    return True