diff --git a/.github/workflows/migration-guard.yml b/.github/workflows/migration-guard.yml new file mode 100644 index 00000000..3a7657ac --- /dev/null +++ b/.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 }}" diff --git a/infra/scripts/check-migrations.sh b/infra/scripts/check-migrations.sh new file mode 100755 index 00000000..838d54d6 --- /dev/null +++ b/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__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 diff --git a/services/knowledge-curator/src/curator/orchestrator/passes/inbox.py b/services/knowledge-curator/src/curator/orchestrator/passes/inbox.py index fc67a6d9..10df456b 100644 --- a/services/knowledge-curator/src/curator/orchestrator/passes/inbox.py +++ b/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 diff --git a/services/knowledge-curator/src/curator/store.py b/services/knowledge-curator/src/curator/store.py index 614a32a2..cb9856ed 100644 --- a/services/knowledge-curator/src/curator/store.py +++ b/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, *, diff --git a/services/knowledge-curator/tests/unit/test_orchestrator_passes.py b/services/knowledge-curator/tests/unit/test_orchestrator_passes.py index f1d33fc7..c5ef2e66 100644 --- a/services/knowledge-curator/tests/unit/test_orchestrator_passes.py +++ b/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: