knowledge-curator: recover inbox notes missing from vault clone via DB fallback

.github/workflows/migration-guard.yml

@@ -0,0 +1,29 @@
+# Blocks the failure mode that crashlooped assistant-api: an already-applied
+# Flyway migration was edited, flipping its checksum so `validate` hard-fails
+# on boot. Enforces that committed migrations are immutable and that versions
+# only ever increment. Fast (seconds) and unconditional so it is safe to mark
+# as a required status check — a PR that touches no migrations still reports a
+# green "Migration Guard".
+name: Migration Guard
+
+on:
+  pull_request:
+
+jobs:
+  guard:
+    name: Migration Guard
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v6.0.2
+        with:
+          # Full history so the base branch and merge-base are available to
+          # diff against.
+          fetch-depth: 0
+      - name: Check migrations are immutable and incrementing
+        env:
+          # A reviewed, deliberate correction (e.g. restoring a migration that
+          # was wrongly edited) can label the PR `allow-migration-change` to
+          # downgrade the immutability error to a warning. Versioning checks
+          # always apply.
+          ALLOW_MIGRATION_CHANGE: ${{ contains(github.event.pull_request.labels.*.name, 'allow-migration-change') }}
+        run: bash infra/scripts/check-migrations.sh "origin/${{ github.base_ref }}"

infra/scripts/check-migrations.sh

@@ -0,0 +1,83 @@
+#!/usr/bin/env bash
+# Guard for Flyway-style SQL migrations. Enforces two invariants that, if
+# broken, crashloop a service in production:
+#
+#   1. IMMUTABILITY — a migration that already exists on the base branch must
+#      never be modified, renamed, or deleted. Flyway checksums each file in
+#      full (comments included), so editing an applied migration breaks
+#      `validate` on the next boot. Add a NEW migration instead. Override for
+#      a deliberate, reviewed correction with ALLOW_MIGRATION_CHANGE=true.
+#
+#   2. VERSIONING — within each service, migration versions are unique, and
+#      every newly added migration's version is strictly greater than the
+#      highest version already on the base branch (incrementing; no reuse, no
+#      backfilling a lower/middle version).
+#
+# Usage: check-migrations.sh [base-ref]   (default base: origin/main)
+set -euo pipefail
+
+base="${1:-origin/main}"
+allow_change="${ALLOW_MIGRATION_CHANGE:-false}"
+
+# V<version>__name.sql under any service's migration tree(s). knowledge-api
+# splits versions across migration/ and migration-pg/, so both are in scope
+# and share one version namespace per service.
+mig_re='services/[^/]+/src/main/resources/db/migration(-pg)?/V[0-9][^/]*\.sql$'
+
+fail=0
+
+# Extract the numeric version token from a path: V12__foo.sql -> 12,
+# V1_1__foo.sql -> 1.1 (dotted form sorts correctly under `sort -V`).
+verkey() { basename "$1" | sed -E 's/^V([0-9]+(_[0-9]+)*)__.*/\1/; s/_/./g'; }
+
+# Return 0 iff $1 > $2 as version strings.
+ver_gt() { [[ "$1" != "$2" && "$(printf '%s\n%s\n' "$1" "$2" | sort -V | tail -1)" == "$1" ]]; }
+
+# --- Invariant 1: immutability of existing migrations ---
+changed="$(git diff --diff-filter=MDR --name-only "${base}...HEAD" | { grep -E "$mig_re" || true; })"
+if [[ -n "$changed" ]]; then
+  if [[ "$allow_change" == "true" ]]; then
+    echo "::warning::Existing migrations changed — allowed via ALLOW_MIGRATION_CHANGE override:"
+    echo "$changed" | sed 's/^/  - /'
+  else
+    echo "::error::Applied migrations are immutable. Do not modify, rename, or delete a committed migration — add a NEW migration with a higher version. Offending files:"
+    echo "$changed" | sed 's/^/  - /'
+    fail=1
+  fi
+fi
+
+# --- Invariant 2: unique + incrementing versions, per service ---
+head_files="$(git ls-files | { grep -E "$mig_re" || true; })"
+services="$(printf '%s\n' "$head_files" | sed -E 's#(services/[^/]+)/.*#\1#' | sort -u)"
+base_all="$(git ls-tree -r --name-only "$base" | { grep -E "$mig_re" || true; })"
+
+for svc in $services; do
+  [[ -z "$svc" ]] && continue
+  svc_re="^${svc}/src/main/resources/db/migration(-pg)?/V[0-9][^/]*\.sql$"
+  cur="$(printf '%s\n' "$head_files" | { grep -E "$svc_re" || true; })"
+  base_svc="$(printf '%s\n' "$base_all" | { grep -E "$svc_re" || true; })"
+
+  # duplicate versions within the service (across both migration dirs)
+  dups="$(for f in $cur; do verkey "$f"; done | sort | uniq -d)"
+  if [[ -n "$dups" ]]; then
+    echo "::error::[${svc##*/}] duplicate migration version(s): $(printf '%s ' $dups)"
+    fail=1
+  fi
+
+  # newly added migrations must exceed the highest version already on base
+  base_max="$(for f in $base_svc; do [[ -n "$f" ]] && verkey "$f"; done | sort -V | tail -1)"
+  [[ -z "$base_max" ]] && continue
+  for f in $cur; do
+    printf '%s\n' "$base_svc" | grep -qxF "$f" && continue   # not newly added
+    v="$(verkey "$f")"
+    if ! ver_gt "$v" "$base_max"; then
+      echo "::error::[${svc##*/}] new migration $(basename "$f") (version $v) must be greater than the highest existing version ($base_max)"
+      fail=1
+    fi
+  done
+done
+
+if [[ $fail -eq 0 ]]; then
+  echo "Migration guard: OK"
+fi
+exit $fail

services/knowledge-curator/src/curator/orchestrator/passes/inbox.py

@@ -71,18 +71,20 @@ def __init__(
         self._regenerate_indexes = regenerate_indexes
 
     def has_work(self, state: PassState) -> bool:
-        # Cheapest possible probe: short-circuit on the first
-        # promotable `.md` file we see. The filesystem itself is the
-        # watermark — successful promotions move files out, so an
-        # empty inbox implies no work.
+        # Fast path: scan the local vault clone. Successful promotions
+        # git-mv files out, so a non-empty inbox always means work.
         inbox_root = self._vault_clone_dir / INBOX_DIRNAME
-        if not inbox_root.exists():
-            return False
-        for path in inbox_root.rglob("*.md"):
-            rel = path.relative_to(self._vault_clone_dir).as_posix()
-            if not rel.startswith(NEEDS_REVIEW_PREFIX):
-                return True
-        return False
+        if inbox_root.exists():
+            for path in inbox_root.rglob("*.md"):
+                rel = path.relative_to(self._vault_clone_dir).as_posix()
+                if not rel.startswith(NEEDS_REVIEW_PREFIX):
+                    return True
+        # Filesystem found nothing. Fall back to the DB to catch
+        # DB/git desync — a note in kb_notes whose file was never
+        # committed to (or was lost from) the local vault clone.
+        # run() calls vault.sync() first, so if the file exists on
+        # the remote it will be pulled before the inbox scan runs.
+        return self._store.has_inbox_notes()
 
     def run(self, state: PassState) -> PassOutcome:
         # Pull + rebase so concurrent captures from the ingest-worker

services/knowledge-curator/src/curator/store.py

@@ -44,6 +44,14 @@ class OrchestratorPassRow:
 class CuratorStore(Protocol):
     def existing_ids(self, ids: Iterable[str]) -> set[str]: ...
 
+    def has_inbox_notes(self) -> bool:
+        """Return True if any note with scope='_inbox' exists in kb_notes,
+        excluding files already routed to ``_inbox/_needs-review/``.
+        Fast existence check used by InboxPass.has_work as a DB fallback
+        when the vault clone appears empty (DB/git desync recovery).
+        """
+        ...
+
     def promote_note(
         self,
         *,
@@ -254,6 +262,19 @@ def existing_ids(self, ids: Iterable[str]) -> set[str]:
             )
             return {row[0] for row in cur.fetchall()}
 
+    def has_inbox_notes(self) -> bool:
+        with self._pool.connection() as conn, conn.cursor() as cur:
+            cur.execute(
+                sql.SQL(
+                    "SELECT 1 FROM kb_notes "
+                    "WHERE scope = '_inbox' "
+                    "  AND vault_path IS NOT NULL "
+                    "  AND vault_path NOT LIKE '_inbox/_needs-review/%' "
+                    "LIMIT 1"
+                )
+            )
+            return cur.fetchone() is not None
+
     def promote_note(
         self,
         *,
@@ -599,10 +620,20 @@ def __init__(self, existing: Iterable[str] = ()) -> None:
         # returns every queued row that lacks a see_also edge in
         # `self.relations`.
         self.relation_enrichment_queue: list[tuple[str, str, str, str]] = []
+        # Simulated kb_notes rows with scope='_inbox' for has_inbox_notes()
+        # tests. Populate with vault_path strings — mirrors the Postgres
+        # query which excludes '_inbox/_needs-review/%' paths.
+        self.inbox_vault_paths: list[str] = []
 
     def existing_ids(self, ids: Iterable[str]) -> set[str]:
         return {i for i in ids if i in self._existing}
 
+    def has_inbox_notes(self) -> bool:
+        return any(
+            p is not None and not p.startswith("_inbox/_needs-review/")
+            for p in self.inbox_vault_paths
+        )
+
     def promote_note(
         self,
         *,

services/knowledge-curator/tests/unit/test_orchestrator_passes.py

@@ -110,6 +110,37 @@ def test_inbox_pass_has_work_returns_true_when_promotable_file_present(tmp_path:
     assert pass_.has_work(_state("inbox")) is True
 
 
+def test_inbox_pass_has_work_falls_back_to_db_when_filesystem_empty(tmp_path: Path) -> None:
+    # Filesystem is empty but the DB has an _inbox note — the case
+    # that caused the May-2026 stuck note (captured to kb_notes and
+    # pushed to the remote but absent from the local vault clone).
+    store = InMemoryCuratorStore()
+    store.inbox_vault_paths = ["_inbox/2026-05-18/200949-some-note--abcdef01.md"]
+    pass_ = InboxPass(
+        vault_clone_dir=tmp_path,
+        promoter=_StubPromoter(),  # type: ignore[arg-type]
+        vault=_StubVault(),  # type: ignore[arg-type]
+        store=store,
+        regenerate_indexes=lambda: None,
+    )
+    assert pass_.has_work(_state("inbox")) is True
+
+
+def test_inbox_pass_has_work_ignores_needs_review_paths_in_db_fallback(tmp_path: Path) -> None:
+    # Only _needs-review paths in the DB — not work for the inbox pass;
+    # those are owned by NeedsReviewDrainPass.
+    store = InMemoryCuratorStore()
+    store.inbox_vault_paths = ["_inbox/_needs-review/stale.md"]
+    pass_ = InboxPass(
+        vault_clone_dir=tmp_path,
+        promoter=_StubPromoter(),  # type: ignore[arg-type]
+        vault=_StubVault(),  # type: ignore[arg-type]
+        store=store,
+        regenerate_indexes=lambda: None,
+    )
+    assert pass_.has_work(_state("inbox")) is False
+
+
 def test_inbox_pass_run_processes_files_and_calls_regenerate_when_promotions_happen(
     tmp_path: Path,
 ) -> None: