fix(runner): single durable activation ownership record owned by the orchestrator

Makefile

@@ -14,10 +14,11 @@ DATAFLOW_PG_DATABASE ?= dataflow_test
 DATAFLOW_PG_USERNAME ?= dataflow
 DATAFLOW_PG_PASSWORD ?= dataflow
 WIPPY ?= wippy
+UPGRADE_FROM ?= v0.7.19
 TESTS ?=
 PUBLISH_DRY_RUN_TOKEN ?= wpy_ci_dry_run_0123456789abcdef0123456789abcdef
 
-.PHONY: test test-sqlite test-postgres test-restart-sqlite test-restart-postgres test-static lint install verify-lock verify-package clean
+.PHONY: test test-sqlite test-postgres test-restart-sqlite test-restart-postgres test-upgrade-sqlite test-upgrade-postgres test-static lint install verify-lock verify-package clean
 
 test: test-sqlite
 
@@ -44,6 +45,21 @@ test-restart-postgres: test-static
 	DATAFLOW_PG_PASSWORD=$(DATAFLOW_PG_PASSWORD) \
 	./scripts/restart-proof.sh
 
+# Upgrade proofs start the first runtime on an older release ($(UPGRADE_FROM))
+# and need its modules installed, so they are not part of the default test.
+test-upgrade-sqlite: test-static
+	DATAFLOW_RESTART_FROM=$(UPGRADE_FROM) ./scripts/restart-proof.sh
+
+test-upgrade-postgres: test-static
+	DATAFLOW_RESTART_FROM=$(UPGRADE_FROM) \
+	DATAFLOW_RESTART_DIALECT=postgres \
+	DATAFLOW_PG_HOST=$(DATAFLOW_PG_HOST) \
+	DATAFLOW_PG_PORT=$(DATAFLOW_PG_PORT) \
+	DATAFLOW_PG_DATABASE=$(DATAFLOW_PG_DATABASE)_upgrade \
+	DATAFLOW_PG_USERNAME=$(DATAFLOW_PG_USERNAME) \
+	DATAFLOW_PG_PASSWORD=$(DATAFLOW_PG_PASSWORD) \
+	./scripts/restart-proof.sh
+
 test-static:
 	@command -v rg >/dev/null 2>&1 || { echo "test-static requires ripgrep (rg)"; exit 1; }
 	@if rg -n "keeper\\.views\\.dataflow|dataflow-link|Open full view" src/session/views/state.jet; then \

README.md

@@ -15,6 +15,18 @@
 
 </div>
 
+## Caller security
+
+A workflow runs under the actor and scope of the caller that created it; the
+orchestrator needs no grant beyond what that scope already allows for running the
+workflow. To classify ownership it reads the module's runtime epoch
+(`userspace.dataflow.env:runtime_epoch`, action `env.get`). The orchestrator entry
+carries the `userspace.dataflow.security:epoch_reader` group for this, and wippy
+merges that group into the caller's scope, so callers need no grant for it. A
+caller policy that explicitly **denies** `env.get` on that resource (or on `*`)
+overrides the merged allow and stops every orchestrator of that caller; exclude
+`userspace.dataflow.env:runtime_epoch` from such a deny.
+
 ## Durable external waits
 
 Nodes that start external work and then wait for a signal use the declarative park contract:

scripts/restart-proof.sh

@@ -13,6 +13,10 @@ pg_port="${DATAFLOW_PG_PORT:-5432}"
 pg_database="${DATAFLOW_PG_DATABASE:-dataflow_restart_test}"
 pg_username="${DATAFLOW_PG_USERNAME:-dataflow}"
 pg_password="${DATAFLOW_PG_PASSWORD:-dataflow}"
+# With DATAFLOW_RESTART_FROM set to a git ref, the first runtime runs that
+# older release and the second runtime upgrades its database in place.
+from_ref="${DATAFLOW_RESTART_FROM:-}"
+phase1_dir="$test_dir"
 
 case "$dialect" in
     sqlite|postgres) ;;
@@ -68,7 +72,7 @@ start_runtime() {
             --set "vars.postgres_password=$pg_password" >"$runtime_log" 2>&1 &
     else
         wippy run -s --profile sqlite --profile "$restart_profile" \
-            --set vars.sqlite_file=./.wippy/restart-proof.db >"$runtime_log" 2>&1 &
+            --set "vars.sqlite_file=$db_file" >"$runtime_log" 2>&1 &
     fi
     runtime_pid=$!
 }
@@ -82,7 +86,16 @@ if [ "$dialect" = "postgres" ]; then
         -h "$pg_host" -p "$pg_port" -U "$pg_username" "$pg_database"
 fi
 
-cd "$test_dir"
+if [ -n "$from_ref" ]; then
+    old_root="$test_dir/.wippy/restart-from"
+    rm -rf "$old_root"
+    mkdir -p "$old_root"
+    git -C "$repo_dir" archive "$from_ref" | tar -x -C "$old_root"
+    (cd "$old_root/test" && wippy install >/dev/null)
+    phase1_dir="$old_root/test"
+fi
+
+cd "$phase1_dir"
 start_runtime restart_create "$phase1_log"
 
 phase1=""
@@ -168,6 +181,10 @@ while [ "$attempts" -lt 100 ]; do
 done
 [ "$rolling_refilled" = "1" ] || { echo "rolling window did not refill before the slow iteration completed" >&2; exit 1; }
 
+# The first runtime's probe can race its migrations and leave an unstarted
+# workflow behind; only workflows created after the restart are duplicates.
+workflows_before_restart=$(query "SELECT COUNT(*) FROM dataflows;")
+
 stop_runtime "$runtime_pid"
 runtime_pid=""
 
@@ -181,6 +198,18 @@ preserved=$(query "
 ")
 [ "$preserved" = "1" ] || { echo "graceful runtime shutdown destroyed active intent" >&2; exit 1; }
 
+orphan_id=""
+if [ -n "$from_ref" ] && [ "$dialect" = "sqlite" ]; then
+    # Releases before the activation ownership record deleted workflows without
+    # their activation row on SQLite, which does not enforce the cascade.
+    orphan_id="00000000-0000-7000-8000-000000000001"
+    query "INSERT INTO dataflow_activations(dataflow_id, generation, desired_active, requested_at, updated_at)
+        VALUES ('$orphan_id', 1, 1, '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z');
+        INSERT INTO dataflow_wakes(dataflow_id, wake_key, wake_at)
+        VALUES ('$orphan_id', 'yield:orphan', '2026-01-01T00:00:00Z');"
+fi
+
+cd "$test_dir"
 start_runtime restart_observe "$phase2_log"
 
 second_epoch=""
@@ -268,6 +297,20 @@ yield_result_rows=$(query "
 }
 
 workflow_count=$(query "SELECT COUNT(*) FROM dataflows;")
-[ "$workflow_count" = "1" ] || { echo "restart created duplicate workflows: $workflow_count" >&2; exit 1; }
+[ "$workflow_count" = "$workflows_before_restart" ] || {
+    echo "restart created duplicate workflows: $workflows_before_restart -> $workflow_count" >&2
+    exit 1
+}
+
+ownership=$(query "SELECT owner_phase || '|' || CASE WHEN owner_token IS NULL THEN 'none' ELSE 'token' END
+    || '|' || CASE WHEN owner_pid IS NULL THEN 'none' ELSE 'pid' END
+    FROM dataflow_activations WHERE dataflow_id = '$dataflow_id';")
+[ "$ownership" = "released|token|pid" ] || { echo "recovered workflow has unexpected ownership: $ownership" >&2; exit 1; }
+
+if [ -n "$orphan_id" ]; then
+    orphans=$(query "SELECT (SELECT COUNT(*) FROM dataflow_activations WHERE dataflow_id = '$orphan_id')
+        + (SELECT COUNT(*) FROM dataflow_wakes WHERE dataflow_id = '$orphan_id');")
+    [ "$orphans" = "0" ] || { echo "upgrade left activation or wake rows of a deleted workflow" >&2; exit 1; }
+fi
 
-echo "$dialect restart proof passed: $dataflow_id $first_epoch -> $second_epoch"
+echo "$dialect restart proof passed${from_ref:+ from $from_ref}: $dataflow_id $first_epoch -> $second_epoch"

src/_index.yaml

@@ -268,6 +268,10 @@ entries:
         path: ".meta.target_db"
       - entry: userspace.dataflow.migrations:09_add_iteration_uniqueness_guard
         path: ".meta.target_db"
+      - entry: userspace.dataflow.migrations:10_add_activation_ownership
+        path: ".meta.target_db"
+      - entry: userspace.dataflow.migrations:11_remove_orphaned_activation_rows
+        path: ".meta.target_db"
       - entry: userspace.dataflow.runner:overseer.service
         path: ".lifecycle.depends_on +="
       - entry: userspace.dataflow.env:retention_db

src/consts.lua

@@ -4,6 +4,13 @@ local consts = {}
 consts.HOST_ID = "app:processes"
 consts.APP_DB = "app:db"
 consts.ORCHESTRATOR = "userspace.dataflow.runner:orchestrator"
+consts.RUNTIME_EPOCH_ENV = "userspace.dataflow.env:runtime_epoch"
+
+-- Phases of the orchestrator ownership record on an activation.
+consts.OWNER_PHASE = {
+    RUNNING = "running",
+    RELEASED = "released",
+}
 
 -- Topic constants for actor state transitions
 consts.TOPIC = {

src/migrations/10_add_activation_ownership.lua

@@ -0,0 +1,57 @@
+local function execute_or_error(db, query)
+    local success, err = db:execute(query)
+    if err then error(err) end
+    return success
+end
+
+-- The ownership record of an activation. owner_token identifies one
+-- orchestrator incarnation, owner_pid its process, owner_epoch (added by
+-- migration 08) the runtime it runs in, and owner_phase is running or
+-- released; a row without a phase has never been owned. An orchestrator
+-- admits itself as the running owner; a completion releases it.
+local COLUMNS = { "owner_token", "owner_pid", "owner_phase" }
+
+local function sqlite_columns(db)
+    local columns, columns_err = db:query("PRAGMA table_info(dataflow_activations)")
+    if columns_err then error(columns_err) end
+    local present = {}
+    for _, column in ipairs(columns or {}) do present[column.name] = true end
+    return present
+end
+
+return require("migration").define(function()
+    migration("Record the orchestrator that owns an activation", function()
+        database("postgres", function()
+            up(function(db)
+                for _, column in ipairs(COLUMNS) do
+                    execute_or_error(db, "ALTER TABLE dataflow_activations ADD COLUMN IF NOT EXISTS " ..
+                        column .. " TEXT")
+                end
+            end)
+            down(function(db)
+                for _, column in ipairs(COLUMNS) do
+                    execute_or_error(db, "ALTER TABLE dataflow_activations DROP COLUMN IF EXISTS " .. column)
+                end
+            end)
+        end)
+
+        database("sqlite", function()
+            up(function(db)
+                local present = sqlite_columns(db)
+                for _, column in ipairs(COLUMNS) do
+                    if not present[column] then
+                        execute_or_error(db, "ALTER TABLE dataflow_activations ADD COLUMN " .. column .. " TEXT")
+                    end
+                end
+            end)
+            down(function(db)
+                local present = sqlite_columns(db)
+                for _, column in ipairs(COLUMNS) do
+                    if present[column] then
+                        execute_or_error(db, "ALTER TABLE dataflow_activations DROP COLUMN " .. column)
+                    end
+                end
+            end)
+        end)
+    end)
+end)

src/migrations/11_remove_orphaned_activation_rows.lua

@@ -0,0 +1,32 @@
+local function execute_or_error(db, query)
+    local success, err = db:execute(query)
+    if err then error(err) end
+    return success
+end
+
+-- SQLite does not enforce the cascading foreign keys, and releases before the
+-- activation ownership record deleted workflows without their activation row.
+-- Such rows, and wakes left the same way, belong to no workflow.
+local ORPHANS = {
+    [[DELETE FROM dataflow_wakes WHERE NOT EXISTS (
+        SELECT 1 FROM dataflows WHERE dataflows.dataflow_id = dataflow_wakes.dataflow_id)]],
+    [[DELETE FROM dataflow_activations WHERE NOT EXISTS (
+        SELECT 1 FROM dataflows WHERE dataflows.dataflow_id = dataflow_activations.dataflow_id)]],
+}
+
+local function remove_orphans(db)
+    for _, statement in ipairs(ORPHANS) do execute_or_error(db, statement) end
+end
+
+return require("migration").define(function()
+    migration("Remove activation and wake rows of deleted workflows", function()
+        database("postgres", function()
+            up(remove_orphans)
+            down(function() end)
+        end)
+        database("sqlite", function()
+            up(remove_orphans)
+            down(function() end)
+        end)
+    end)
+end)

src/migrations/_index.yaml

@@ -198,3 +198,31 @@ entries:
     imports:
       migration: wippy.migration:migration
     method: migrate
+
+  - name: 10_add_activation_ownership
+    kind: function.lua
+    meta:
+      type: migration
+      tags: [dataflows, activation, durability]
+      description: Record the orchestrator incarnation that owns an activation and whether it is running or released
+      depends_on: [ns:wippy.migration]
+      target_db: app:db
+      timestamp: "2026-09-22T12:00:00Z"
+    source: file://10_add_activation_ownership.lua
+    imports:
+      migration: wippy.migration:migration
+    method: migrate
+
+  - name: 11_remove_orphaned_activation_rows
+    kind: function.lua
+    meta:
+      type: migration
+      tags: [dataflows, activation, durability]
+      description: Remove activation and wake rows whose workflow was deleted where foreign keys did not cascade
+      depends_on: [ns:wippy.migration]
+      target_db: app:db
+      timestamp: "2026-09-23T12:00:00Z"
+    source: file://11_remove_orphaned_activation_rows.lua
+    imports:
+      migration: wippy.migration:migration
+    method: migrate

src/persist/_index.yaml

@@ -8,7 +8,7 @@ entries:
       private: true
       comment: Generation-fenced desired activation state for durable orchestrator lives.
     source: file://activation_repo.lua
-    modules: [sql, json]
+    modules: [sql, json, time]
     imports:
       dataflow_consts: userspace.dataflow:consts
 
@@ -206,30 +206,6 @@ entries:
       test: wippy.test:test
     method: run_tests
 
-  - name: wake_repo
-    kind: library.lua
-    meta:
-      private: true
-      comment: Indexed persistence for independently addressable signal and yield wake transitions.
-    source: file://wake_repo.lua
-    modules: [sql]
-    imports:
-      dataflow_consts: userspace.dataflow:consts
-
-  - name: wake_repo_test
-    kind: function.lua
-    meta:
-      type: test
-      suite: Workflow
-      group: Workflow
-      comment: Proves per-trigger registration, exact consumption, and dataflow-wide terminal cleanup.
-    source: file://wake_repo_test.lua
-    modules: [sql, uuid, time]
-    imports:
-      test: wippy.test:test
-      wake_repo: userspace.dataflow.persist:wake_repo
-    method: run_tests
-
   # userspace.dataflow.persist:node_reader
   - name: node_reader
     kind: library.lua