test(builder): verify dispatch migration in forward history

.github/workflows/builder-safe-execution-tests.yml

@@ -63,7 +63,7 @@ on:
       test_scope:
         description: Select a focused lane; all always runs on build branch pushes
         type: choice
-        options: [all, native-agent, native-catalog, native-continuation, native-console, native-dispatch]
+        options: [all, native-agent, native-catalog, native-continuation, native-console, native-dispatch, native-postgres]
         default: all
 permissions:
   contents: read
@@ -72,7 +72,7 @@ concurrency:
   cancel-in-progress: true
 jobs:
   native-postgres:
-    if: github.event_name != 'workflow_dispatch' || (!inputs.contract_codegen && (inputs.test_scope == 'all'))
+    if: github.event_name != 'workflow_dispatch' || (!inputs.contract_codegen && (inputs.test_scope == 'all' || inputs.test_scope == 'native-postgres'))
     runs-on: ubuntu-24.04
     timeout-minutes: 60
     env:

api/tests/test_containers_integration_tests/core/workflow/test_builder_execution_policy_persistence.py

@@ -198,7 +198,8 @@ def test_real_postgres_forward_merge_preserves_both_branches(flask_app_with_cont
     from extensions.ext_database import db
 
     merge_revision = "f9a3c62de418"
-    current_head = "f47a1c90d2e8"
+    current_head = "e59b7d32a6c1"
+    owner_head = "f47a1c90d2e8"
     continuation_head = "d28f0e3c912a"
     native_head = "c67e12a904bf"
     previous_merge = "a61e9b2c7d40"
@@ -208,7 +209,10 @@ def test_real_postgres_forward_merge_preserves_both_branches(flask_app_with_cont
     config.set_main_option("script_location", str(Path(__file__).parents[4] / "migrations"))
     scripts = ScriptDirectory.from_config(config)
     assert scripts.get_heads() == [current_head]
-    owner_revision = scripts.get_revision(current_head)
+    dispatch_revision = scripts.get_revision(current_head)
+    assert dispatch_revision is not None
+    assert dispatch_revision.down_revision == owner_head
+    owner_revision = scripts.get_revision(owner_head)
     assert owner_revision is not None
     assert owner_revision.down_revision == continuation_head
     continuation_revision = scripts.get_revision(continuation_head)
@@ -343,7 +347,11 @@ def test_real_postgres_forward_merge_preserves_both_branches(flask_app_with_cont
         }
 
         def witness(
-            *, include_archive: bool = False, include_native: bool = False, include_continuation: bool = False
+            *,
+            include_archive: bool = False,
+            include_native: bool = False,
+            include_continuation: bool = False,
+            include_dispatch_claims: bool = False,
         ) -> dict[str, object]:
             inspector = inspect(connection)
             expected_tables = {*tables, "alembic_version"}
@@ -353,6 +361,8 @@ def test_real_postgres_forward_merge_preserves_both_branches(flask_app_with_cont
                 expected_tables.update(native_tables)
             if include_continuation:
                 expected_tables.update(continuation_tables)
+            if include_dispatch_claims:
+                expected_tables.add("dify_builder_native_effect_dispatch_claims")
             assert set(inspector.get_table_names(schema=schema)) == expected_tables
             return {
                 table: {
@@ -498,7 +508,11 @@ def test_real_postgres_forward_merge_preserves_both_branches(flask_app_with_cont
             for table in sorted(continuation_tables)
         }
         assert migrate("head") == (current_head,)
-        assert witness(include_archive=True, include_native=True, include_continuation=True) == before_continuation
+        assert (
+            witness(include_archive=True, include_native=True, include_continuation=True, include_dispatch_claims=True)
+            == before_continuation
+        )
+        assert connection.scalar(text("SELECT count(*) FROM dify_builder_native_effect_dispatch_claims")) == 0
         assert native_ledger_witness() == ledger_before
         assert archive_witness() == archive_before
         for table, previous in continuation_before.items():