test(persistence): cover historical run-change repair and rollback (#5518)

* fix(persistence): repair run-change clock schema skipped by the 0023 insertion

0023_run_change_seq was chained ahead of the already-shipped
0023_user_preferences revision, so databases stamped at that revision or
later treat it as an applied ancestor and never execute it: the
run_change_clock table and runs.change_seq column are permanently missing
and the first thread deletion fails with 'no such table:
run_change_clock' (#5516). 0025_repair_run_change_seq re-applies the same
guarded DDL on upgrade and no-ops on healthy shapes. RunChangeClockRow and
UserPreferenceRow are also registered in the ORM model registry.

Fixes #5516

* fix(persistence): preserve run-change schema when rolling back repair

---------

Co-authored-by: 1553126902 <1553126902@qq.com>

CHANGELOG.md

@@ -952,7 +952,9 @@ This release closes that milestone with **765 merged pull requests**.
   upgrade and no-ops on healthy shapes. `RunChangeClockRow` and
   `UserPreferenceRow` are also registered in the ORM model registry so
   `create_all` and autogenerate see every table through explicit imports
-  instead of module side effects.
+  instead of module side effects. Rolling back the repair to
+  `0024_project_documents` intentionally leaves the ancestor-owned schema and
+  existing change positions intact; the repair downgrade is a no-op.
 - **nginx:** Extend the 600-second read timeout to the two remaining locations
   whose routes wait on the Gateway, both left on nginx's 60-second default by
   the thread-route fix. Behind the `/api/` catch-all, the stateless

README.md

@@ -347,6 +347,11 @@ For persistent deployments, configure `database.backend` as `sqlite` or
 LangGraph Store, and DeerFlow application data. The deprecated `checkpointer`
 section, when present, overrides the first two for backward compatibility.
 
+Gateway startup automatically repairs the missing run-change schema affecting
+some existing databases (#5516). The repair preserves run history and existing
+change positions; downgrading the repair to its predecessor also retains the
+schema and positions required by that version.
+
 For lightweight single-process event persistence, `run_events.backend: jsonl`
 keeps Unicode message content intact, including line and paragraph separators.
 Existing valid JSONL records remain readable without rewriting the files.

backend/packages/harness/deerflow/persistence/migrations/AGENTS.md

@@ -161,7 +161,7 @@ on installs that never enabled it. The convention is:
 - `migrations/versions/0022_scheduled_occurrence_seq.py` — adds the per-task `last_occurrence_seq` high-water mark, nullable occurrence `occurrence_seq` and `launch_accounted`, and a unique `(task_id, occurrence_seq)` index. New occurrences allocate their sequence under the existing parent lock; launch accounting is recorded atomically with the count so an older recovered occurrence cannot be counted twice. Legacy child columns remain NULL without guessed ordering or accounting backfill. All three fields are internal and omitted from repository responses. Both once-task recovery paths lock the parent, defer while any occurrence row is active (sequenced or not), and otherwise project only from the highest sequence (`can_project`), the same rule as the launch, completion, and queue-failure writes. Chains after `0019_thread_incarnations`.
 - `migrations/versions/0023_run_change_seq.py` — adds `runs.change_seq`, its global singleton allocation clock, and owner-aware cursor indexes. Legacy rows remain at zero and page by run id; lifecycle, cancellation, and model-name mutations allocate monotonically increasing positions in their own transaction. Atomic replacement uses one position for every affected row. Progress snapshots and lease heartbeats do not advance the clock. `0023_user_preferences` follows this revision. Its migration test verifies membership in the single-head chain and the expected predecessor rather than pinning the latest head, so later migrations can extend the chain.
 - `migrations/versions/0024_project_documents.py` — creates the `project_documents` shelf table (id/project_id/user_id/name/stored_relpath/sha256/size_bytes, nullable promotion provenance and trash fields, timestamps) with indexes on project_id, user_id, sha256 and trashed_at; no DB-level foreign key on project_id by design (project delete trashes the shelf inside its own transaction). New table, so the bootstrap forward-compat floor is unchanged; chains after `0023_user_preferences` (renumbered from 0023 after the rebase)
-- `migrations/versions/0025_repair_run_change_seq.py` — heals databases that skipped `0023_run_change_seq` because it was inserted ahead of the already-shipped `0023_user_preferences` (#5516): re-applies the guarded `run_change_clock` table, `runs.change_seq` column, and cursor indexes on upgrade; no-ops on healthy shapes. Its downgrade is a deliberate no-op — the schema and allocated clock positions are owned by ancestor 0023, and dropping them on a repair rollback would recreate the skipped-revision hole and discard cursor data; downgrading to 0023 itself removes the schema through that revision's own downgrade. Chains after `0024_project_documents`
+- `migrations/versions/0025_repair_run_change_seq.py` — heals databases that skipped `0023_run_change_seq` because it was inserted ahead of the already-shipped `0023_user_preferences` (#5516): re-applies the guarded `run_change_clock` table, `runs.change_seq` column, and cursor indexes on upgrade; no-ops on healthy shapes; chains after `0024_project_documents`. Its downgrade is intentionally a no-op: the repaired objects belong to ancestor `0023_run_change_seq`, remain required at 0024, and must retain their existing change positions. Only the original 0023 downgrade removes them. `tests/test_run_change_repair_history.py` reconstructs both pre-insertion published descendants and verifies historical upgrade, unchanged healthy positions, and usable run-store writes after downgrade and re-upgrade
 - `persistence/bootstrap.py` — `bootstrap_schema(engine, backend=...)`, the three-branch provisioning decision, locked revision validation, and the narrow 0019 forward-compatibility exception
 - `extensions/loader.py::load_extensions` — registers each spec's `table_prefix` with `register_extension_table_prefix()`
 - Tests: `tests/test_persistence_bootstrap.py` (branches), `tests/test_persistence_bootstrap_concurrency.py` (concurrency), `tests/test_persistence_bootstrap_regression.py` (issue #3682), `tests/test_persistence_migrations_env.py` (filter, including extension-owned tables), `tests/test_extension_loader.py::TestTablePrefixRegistration` (spec-to-filter wiring), `tests/blocking_io/test_persistence_bootstrap.py` (asyncio.to_thread anchor), `tests/test_migration_0004_run_ownership_dedupe.py` + `tests/test_migration_0007_scheduled_run_active_dedupe.py` (dedupe-before-unique-index pre-steps), `tests/test_migration_0025_repair_run_change_seq.py` (issue #5516 skipped-revision heal)

backend/tests/test_run_change_repair_history.py

@@ -0,0 +1,114 @@
+"""Exercise the published pre-#5405 migration graph, not a new-schema stamp."""
+
+from __future__ import annotations
+
+import asyncio
+import shutil
+from pathlib import Path
+
+import pytest
+import sqlalchemy as sa
+from alembic import command
+from alembic.script import ScriptDirectory
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from deerflow.persistence.bootstrap import _MIGRATIONS_DIR, _get_alembic_config, bootstrap_schema
+from deerflow.persistence.run import RunRepository
+
+pytestmark = pytest.mark.asyncio
+PREVIOUS = "0024_project_documents"
+
+
+def _historical_config(engine, tmp_path):
+    """Reconstruct the two published descendants before their ancestor changed."""
+    target = tmp_path / "historical_migrations"
+    shutil.copytree(_MIGRATIONS_DIR, target)
+    script = ScriptDirectory(str(_MIGRATIONS_DIR))
+    historical_files = {Path(revision.path).name for revision in script.walk_revisions(base="base", head=PREVIOUS) if revision.revision != "0023_run_change_seq"}
+    for path in (target / "versions").glob("*.py"):
+        if path.name not in historical_files:
+            path.unlink()
+    preferences = target / "versions" / "0023_user_preferences.py"
+    text = preferences.read_text(encoding="utf-8")
+    expected_parent = 'down_revision = "0023_run_change_seq"'
+    assert expected_parent in text, f"{preferences.name}: expected {expected_parent!r}; update the historical migration rewrite"
+    historical_text = text.replace(expected_parent, 'down_revision = "0022_scheduled_occurrence_seq"')
+    assert historical_text != text, f"{preferences.name}: historical migration rewrite did not change the parent revision"
+    preferences.write_text(historical_text, encoding="utf-8")
+    cfg = _get_alembic_config(engine)
+    cfg.set_main_option("script_location", str(target))
+    return cfg
+
+
+async def _assert_schema(engine):
+    async with engine.connect() as connection:
+
+        def inspect(conn):
+            inspector = sa.inspect(conn)
+            assert inspector.has_table("run_change_clock")
+            assert "change_seq" in {column["name"] for column in inspector.get_columns("runs")}
+            indexes = {index["name"]: index["column_names"] for index in inspector.get_indexes("runs")}
+            assert indexes["ix_runs_change_seq"] == ["change_seq", "run_id"]
+            assert indexes["ix_runs_user_change_seq"] == ["user_id", "change_seq", "run_id"]
+
+        await connection.run_sync(inspect)
+
+
+@pytest.mark.parametrize("revision", ["0023_user_preferences", PREVIOUS])
+async def test_startup_repairs_skipped_ancestor_and_allows_thread_delete(tmp_path, revision):
+    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'historical.db'}")
+    try:
+        await asyncio.to_thread(command.upgrade, _historical_config(engine, tmp_path), revision)
+        async with engine.begin() as connection:
+            assert not await connection.run_sync(lambda conn: sa.inspect(conn).has_table("run_change_clock"))
+            await connection.execute(
+                sa.text(
+                    "INSERT INTO runs (run_id, thread_id, user_id, status, operation_kind, metadata_json, kwargs_json, "
+                    "multitask_strategy, message_count, total_input_tokens, total_output_tokens, total_tokens, llm_call_count, "
+                    "lead_agent_tokens, subagent_tokens, middleware_tokens, created_at, updated_at) "
+                    "VALUES ('legacy', 'legacy-thread', 'user-1', 'success', 'run', '{}', '{}', 'reject', "
+                    "0, 0, 0, 0, 0, 0, 0, 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
+                )
+            )
+        await bootstrap_schema(engine, backend="sqlite")
+        await _assert_schema(engine)
+        repo = RunRepository(async_sessionmaker(engine, expire_on_commit=False))
+        legacy = await repo.list_changed(after_change_seq=-1, after_run_id="", user_id="user-1", limit=10)
+        assert [(row["run_id"], row["change_seq"]) for row in legacy] == [("legacy", 0)]
+        operation, _ = await repo.create_thread_operation_atomic("delete-op", thread_id="legacy-thread", user_id="user-1", owner_worker_id="worker", lease_expires_at=None, operation_kind="thread_delete")
+        assert operation["change_seq"] > 0
+        assert await repo.update_status("delete-op", "success")
+        async with engine.connect() as connection:
+            assert await connection.scalar(sa.text("SELECT change_seq FROM runs WHERE run_id = 'delete-op'")) > operation["change_seq"]
+    finally:
+        await engine.dispose()
+
+
+async def test_repair_preserves_clock_and_downgrade_keeps_ancestor_schema(tmp_path):
+    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'healthy.db'}")
+    cfg = _get_alembic_config(engine)
+    try:
+        await asyncio.to_thread(command.upgrade, cfg, PREVIOUS)
+        repo = RunRepository(async_sessionmaker(engine, expire_on_commit=False))
+        await repo.put("existing", thread_id="existing-thread", user_id="user-1", status="pending")
+        async with engine.begin() as connection:
+            await connection.execute(sa.text("UPDATE runs SET change_seq = 37"))
+            await connection.execute(sa.text("UPDATE run_change_clock SET value = 100 WHERE id = 1"))
+        await bootstrap_schema(engine, backend="sqlite")
+        await bootstrap_schema(engine, backend="sqlite")
+        async with engine.connect() as connection:
+            assert await connection.scalar(sa.text("SELECT value FROM run_change_clock WHERE id = 1")) == 100
+            assert await connection.scalar(sa.text("SELECT change_seq FROM runs WHERE run_id = 'existing'")) == 37
+        await asyncio.to_thread(command.downgrade, cfg, PREVIOUS)
+        await _assert_schema(engine)
+        # The previous application's run store must remain usable before any
+        # re-upgrade; a schema-only assertion would miss sequence resets.
+        assert await repo.start_run("existing")
+        async with engine.connect() as connection:
+            assert await connection.scalar(sa.text("SELECT change_seq FROM runs WHERE run_id = 'existing'")) == 101
+        await asyncio.to_thread(command.upgrade, cfg, "head")
+        await repo.update_status("existing", "error")
+        changed = await repo.list_changed(after_change_seq=100, after_run_id="", user_id="user-1", limit=10)
+        assert [(row["run_id"], row["change_seq"]) for row in changed] == [("existing", 102)]
+    finally:
+        await engine.dispose()