From 609d5b329f63ae912700dfd13e12b88f6037c3b2 Mon Sep 17 00:00:00 2001
From: Kyber5323 <Kyber5323@users.noreply.github.com>
Date: Tue, 20 Oct 2026 22:14:26 +0800
Subject: [PATCH] feat(core): record and bound the WorkflowRun Workspace
 release

A terminal WorkflowRun said how the workflow ended but nothing about the
isolated Workspace it had provisioned: Cloud declared the delete and
re-declared it, yet no run-level fact recorded whether the Workspace was
released, and a delete stuck behind an unreachable Node was retried forever.

Migration 0055 adds a forward-only release state
(pending/releasing/released/abandoned/not_required), the give-up reason, and
the release instant, with derived-state CHECK constraints and a backfill that
reads only evidence already in the database: a terminal run with no Workspace
is not_required, one whose Workspace is soft-deleted is released with that
row's deletion instant, and one with a live Workspace is releasing.

Terminal settlement now writes the intent in the transaction that makes the
run terminal; the delete-completion hook, reachable for workflow runs now that
control.go reads both bindings, writes released in the transaction that writes
the operation succeeded and soft-deletes the Workspace; and a reconciliation
pass heals Workspaces deleted by another path and re-declares refused deletes.
The give-up reuses IssueRun D8's single condition through the shared
runWorkspaceReleaseGivenUp: a live sandbox whose Node has been silent past
DeliveryUnreachableAfter abandons the release, fails the unfinished operation
as node_unavailable, keeps the Workspace row and never writes a release
instant. Only released claims the Workspace is gone.

Refs ora-space/circus#377

Co-Authored-By: Claude Code <noreply@anthropic.com>
---
 docs/workflows/workflows.en.md                |  12 +
 docs/workflows/workflows.md                   |   6 +
 integration/migration_upgrade_path_test.go    | 206 ++++++++++
 integration/revision_upgrade_test.go          |   9 +-
 internal/core/agent_run_release.go            |  43 +--
 internal/core/control.go                      |  24 +-
 .../0055_workflow_run_workspace_release.sql   |  55 +++
 internal/core/migrations/README.en.md         |   4 +-
 internal/core/migrations/README.md            |   4 +-
 .../core/workflow_run_identity_db_test.go     |   7 +
 internal/core/workflow_run_release.go         | 336 +++++++++++++++++
 internal/core/workflow_run_release_db_test.go | 353 ++++++++++++++++++
 internal/core/workflow_run_session.go         |  13 +-
 internal/core/workflow_scheduler.go           |  49 +--
 internal/core/workflow_scheduler_db_test.go   |  12 +-
 15 files changed, 1038 insertions(+), 95 deletions(-)
 create mode 100644 internal/core/migrations/0055_workflow_run_workspace_release.sql
 create mode 100644 internal/core/workflow_run_release.go
 create mode 100644 internal/core/workflow_run_release_db_test.go

diff --git a/docs/workflows/workflows.en.md b/docs/workflows/workflows.en.md
index 0c2c17566..c74f1172b 100644
--- a/docs/workflows/workflows.en.md
+++ b/docs/workflows/workflows.en.md
@@ -150,6 +150,18 @@ seconds):
   `cancelled`. A Node that can no longer report is a cancellation too, not `node_unavailable`. Every terminal path
   declares the run Workspace's deletion, and a result arriving after the run is terminal is kept on the execution row
   without rewriting the run.
+- **A Workspace release is an independent fact after the terminal state, and it is bounded.** In the same transaction
+  that settles the run `succeeded`/`failed`/`cancelled`, Cloud records the release state on the run row and declares the
+  run Workspace's deletion: a run that never had a Workspace records `not_required`, every other run records
+  `releasing`. The transaction that writes the delete operation `succeeded` and soft-deletes the Workspace records
+  `released`, with that instant as the release fact; a Workspace deleted by another path — a project deletion, for
+  example — is healed to `released` from the Workspace's own `deleted_at` instead of staying `releasing`. When the
+  Workspace still holds a live sandbox whose Node has sent no heartbeat for longer than
+  `issue_runs.delivery_unreachable_after` (30 minutes by default; an unconfigured window never gives up), the scheduler
+  gives up the release: the unfinished delete operation fails as `node_unavailable`, the run records `abandoned` with
+  its reason, the Workspace row stays readable, and no release instant is written. **Only `released` means Cloud
+  observed the Workspace deleted**; `abandoned` is never reported as a successful release and is never re-declared or
+  retried by a later pass. Replayed delete completions and repeated reconciliation passes are no-ops.
 
 Development fixtures (`collaboration.development_fixtures`) install a deterministic run simulator that fills in the
 execution trace at run creation and says in its output that nothing really ran; production never installs it.
diff --git a/docs/workflows/workflows.md b/docs/workflows/workflows.md
index a8108484a..dd537fb0c 100644
--- a/docs/workflows/workflows.md
+++ b/docs/workflows/workflows.md
@@ -100,6 +100,12 @@ WorkflowRun 由 Cloud server 里的调度循环推进（`AdvanceWorkflowRunsOnce
   调度器复用该节点已有的结束请求（没有就放出 `end_session{cancelled}`），等待上限仍是 `issue_runs.agent_end_confirm_timeout`，超过后节点和运行都以
   `cancelled` 结束；Node 已不可达时取消优先于 `node_unavailable`，同样以 `cancelled` 收尾。每种终态都声明删除运行的 Workspace；请求记录之后才到达的结果
   保留在执行行上，不改写已经终态的运行。
+- **Workspace 的释放是终态之后的独立事实，并且有界。** 运行进入终态（`succeeded`/`failed`/`cancelled`）的同一个事务里，Cloud 在运行行上记录释放状态并声明
+  删除这次运行的 Workspace：没有 Workspace 的运行记为 `not_required`，其余记为 `releasing`。删除操作成功、Workspace 被软删除的同一个事务把它记为 `released`，
+  释放时刻就是那一刻；Workspace 被项目删除等其他路径直接删掉的，调度器按 Workspace 自己的 `deleted_at` 补记为 `released`，不会一直停在 `releasing`。
+  Workspace 仍有存活沙箱、而它的 Node 超过 `issue_runs.delivery_unreachable_after`（默认 30 分钟；未配置时不放弃）没有任何心跳时，调度器放弃这次释放：
+  未完成的删除操作以 `node_unavailable` 失败，运行记为 `abandoned` 并写明原因，Workspace 行保留可查，也不写释放时刻。**只有 `released` 表示 Cloud
+  观察到 Workspace 已被删除**；`abandoned` 永远不会被报告成释放成功，也不会被后续轮次重新声明或重试。重复的删除完成回执和重复的清扫都是无操作。
 
 开发夹具（`collaboration.development_fixtures`）会装入一个确定性的运行模拟器，在创建运行时就填好执行轨迹，并在输出里说明没有真实执行；生产环境
 不装它。工作流 IssueRun 在生产中目前没有执行端，保持 `queued`（见 [Issue 协作](../collaboration/interactions.md#端口issue-只拥有消费侧的契约)）。
diff --git a/integration/migration_upgrade_path_test.go b/integration/migration_upgrade_path_test.go
index 2470fd908..afe528f16 100644
--- a/integration/migration_upgrade_path_test.go
+++ b/integration/migration_upgrade_path_test.go
@@ -1492,3 +1492,209 @@ func TestMigration0052LegacyWorkflowIssueRunSettlementAppliesFreshAndUpgrades(t
 	migratedTwice(t, newStoreOnSchema(t, pool))
 	assertUpgraded(t, pool, ids, "replay")
 }
+
+// 0055 adds the WorkflowRun Workspace release fact (circus#75 S5; specs decisions/cloud/workflow
+// 0-cloud-side-scheduler-owns-workflow-execution D3/D7). The upgrade path is what matters: a
+// deployment that stopped at 0052 can hold terminal runs that never provisioned a Workspace, terminal
+// runs whose Workspace is still live, Workspaces already deleted by a project deletion, and runs
+// still executing. The backfill records only the fact each row supports — `released` always takes an
+// observed deletion instant, never a claimed one — and the derived-state CHECK constraints refuse the
+// shapes that would claim a release without evidence. A fresh database has nothing to backfill,
+// reaches the same shape, and a second Migrate changes nothing.
+func TestMigration0055WorkflowRunWorkspaceReleaseAppliesFreshAndUpgrades(t *testing.T) {
+	columns := func(pool *sql.DB) map[string]string {
+		t.Helper()
+		rows, e := pool.Query(`SELECT column_name, is_nullable, coalesce(column_default,'')
+			FROM information_schema.columns WHERE table_schema=current_schema() AND table_name='workflow_runs'
+			  AND column_name IN ('workspace_release_state','workspace_release_error','workspace_released_at')`)
+		must(t, e)
+		defer func() { _ = rows.Close() }()
+		out := map[string]string{}
+		for rows.Next() {
+			var name, nullable, dflt string
+			must(t, rows.Scan(&name, &nullable, &dflt))
+			out[name] = nullable + "|" + dflt
+		}
+		must(t, rows.Err())
+		return out
+	}
+	constraints := func(pool *sql.DB) map[string]string {
+		t.Helper()
+		rows, e := pool.Query(`SELECT conname, pg_get_constraintdef(oid) FROM pg_constraint
+			WHERE conrelid='workflow_runs'::regclass AND contype='c'
+			  AND conname IN ('workflow_run_release_released_at','workflow_run_release_abandoned_error')`)
+		must(t, e)
+		defer func() { _ = rows.Close() }()
+		out := map[string]string{}
+		for rows.Next() {
+			var name, def string
+			must(t, rows.Scan(&name, &def))
+			out[name] = def
+		}
+		must(t, rows.Err())
+		return out
+	}
+	assertShape := func(t *testing.T, pool *sql.DB, stage string) {
+		t.Helper()
+		cols := columns(pool)
+		if got := cols["workspace_release_state"]; got != "NO|'pending'::text" {
+			t.Fatalf("%s: workspace_release_state = %q; want NOT NULL default pending", stage, got)
+		}
+		if got := cols["workspace_release_error"]; got != "NO|''::text" {
+			t.Fatalf("%s: workspace_release_error = %q; want NOT NULL default ''", stage, got)
+		}
+		if got := cols["workspace_released_at"]; got != "YES|" {
+			t.Fatalf("%s: workspace_released_at = %q; want nullable with no default", stage, got)
+		}
+		cons := constraints(pool)
+		if len(cons) != 2 || !strings.Contains(cons["workflow_run_release_released_at"], "workspace_released_at IS NOT NULL") {
+			t.Fatalf("%s: release constraints = %v; want the released-at guard", stage, cons)
+		}
+		if !strings.Contains(cons["workflow_run_release_abandoned_error"], "btrim(workspace_release_error)") {
+			t.Fatalf("%s: abandoned-error constraint = %q; want a non-blank reason guard", stage, cons["workflow_run_release_abandoned_error"])
+		}
+	}
+
+	type releaseShape struct {
+		state, reason string
+		releasedAt    sql.NullTime
+		version       int64
+	}
+	read := func(t *testing.T, pool *sql.DB, id string) releaseShape {
+		t.Helper()
+		var s releaseShape
+		must(t, pool.QueryRow(`SELECT workspace_release_state, workspace_release_error, workspace_released_at, version
+			FROM workflow_runs WHERE id=$1`, id).Scan(&s.state, &s.reason, &s.releasedAt, &s.version))
+		return s
+	}
+	seedRuns := func(t *testing.T, pool *sql.DB) (ids, workspaces map[string]string) {
+		t.Helper()
+		uid, tid, space, project := uuid.NewString(), uuid.NewString(), uuid.NewString(), uuid.NewString()
+		wf, snap := uuid.NewString(), uuid.NewString()
+		ids = map[string]string{
+			"noWorkspace": uuid.NewString(), "live": uuid.NewString(),
+			"deleted": uuid.NewString(), "running": uuid.NewString(),
+		}
+		workspaces = map[string]string{"live": uuid.NewString(), "deleted": uuid.NewString(), "running": uuid.NewString()}
+		tx, e := pool.Begin()
+		must(t, e)
+		defer func() { _ = tx.Rollback() }()
+		for _, s := range []struct {
+			query string
+			args  []any
+		}{
+			{`INSERT INTO users(id,display_name,status) VALUES($1,'Release user','active')`, []any{uid}},
+			{`INSERT INTO tenants(id,name,status) VALUES($1,'Release tenant','active')`, []any{tid}},
+			{`INSERT INTO tenant_memberships(tenant_id,user_id,role,status) VALUES($1,$2,'admin','active')`, []any{tid, uid}},
+			{`INSERT INTO collab_workspaces(id,tenant_id,name,slug,created_by) VALUES($1,$2,'Release space',$3,$4)`, []any{space, tid, "release-" + space[:8], uid}},
+			{`INSERT INTO projects(id,tenant_id,owner_user_id,space_id,name,lifecycle)
+				VALUES($1,$2,$3,$4,'Release project','active')`, []any{project, tid, uid, space}},
+			// A live project requires its main Workspace; without it the deferred aggregate check refuses
+			// the commit before the run Workspaces this test is about are ever examined.
+			{`INSERT INTO workspaces(id,tenant_id,owner_user_id,creator_evidence,project_id,kind,desired_state,observed_state,runtime_generation,requested_ref)
+				VALUES($1,$2,$3,'unknown',$4,'main','running','ready',1,'HEAD')`, []any{uuid.NewString(), tid, uid, project}},
+			{`INSERT INTO workflows(id,tenant_id,name,graph) VALUES($1,$2,'Release workflow','{"nodes":[],"edges":[]}'::jsonb)`, []any{wf, tid}},
+			{`INSERT INTO workflow_snapshots(id,tenant_id,workflow_id,version,name,graph)
+				VALUES($1,$2,$3,1,'v1','{"nodes":[],"edges":[]}'::jsonb)`, []any{snap, tid, wf}},
+		} {
+			_, e := tx.Exec(s.query, s.args...)
+			must(t, e)
+		}
+		insertRun := func(id, status string) {
+			_, e := tx.Exec(`INSERT INTO workflow_runs(id,tenant_id,workflow_id,snapshot_id,name,status)
+				VALUES($1,$2,$3,$4,'Release run',$5)`, id, tid, wf, snap, status)
+			must(t, e)
+		}
+		insertWorkspace := func(id, runID string, deleted bool) {
+			_, e := tx.Exec(`INSERT INTO workspaces(id,tenant_id,owner_user_id,creator_evidence,project_id,kind,
+				desired_state,observed_state,runtime_generation,requested_ref,workflow_run_id,deleted_at)
+				VALUES($1,$2,$3,'unknown',$4,'isolated',CASE WHEN $6 THEN 'deleted' ELSE 'running' END,
+				CASE WHEN $6 THEN 'deleted' ELSE 'ready' END,1,'HEAD',$5,
+				CASE WHEN $6 THEN clock_timestamp() END)`, id, tid, uid, project, runID, deleted)
+			must(t, e)
+			// A live isolated Workspace carries exactly one task identity; a deleted one carries none.
+			if !deleted {
+				_, e = tx.Exec(`INSERT INTO tasks(id,workspace_id,title) VALUES($1,$2,'Release task')`, uuid.NewString(), id)
+				must(t, e)
+			}
+		}
+		insertRun(ids["noWorkspace"], "succeeded")
+		insertRun(ids["live"], "failed")
+		insertWorkspace(workspaces["live"], ids["live"], false)
+		insertRun(ids["deleted"], "cancelled")
+		insertWorkspace(workspaces["deleted"], ids["deleted"], true)
+		insertRun(ids["running"], "running")
+		insertWorkspace(workspaces["running"], ids["running"], false)
+		must(t, tx.Commit())
+		return ids, workspaces
+	}
+	assertBackfilled := func(t *testing.T, pool *sql.DB, ids, workspaces map[string]string, stage string) {
+		t.Helper()
+		for name, want := range map[string]struct {
+			state   string
+			instant bool
+		}{
+			"noWorkspace": {"not_required", false},
+			"live":        {"releasing", false},
+			"deleted":     {"released", true},
+			"running":     {"pending", false},
+		} {
+			got := read(t, pool, ids[name])
+			if got.state != want.state || got.reason != "" || got.releasedAt.Valid != want.instant {
+				t.Fatalf("%s: %s run release = %+v; want %s (instant=%v) with no reason", stage, name, got, want.state, want.instant)
+			}
+			if got.version != 1 {
+				t.Fatalf("%s: %s run version = %d; the backfill must not invent a transition", stage, name, got.version)
+			}
+		}
+		// The one run that may claim a release carries the Workspace's own deletion instant, not the
+		// migration's clock.
+		var same bool
+		must(t, pool.QueryRow(`SELECT r.workspace_released_at = w.deleted_at FROM workflow_runs r
+			JOIN workspaces w ON w.workflow_run_id=r.id WHERE r.id=$1`, ids["deleted"]).Scan(&same))
+		if !same {
+			t.Fatalf("%s: released instant does not equal the Workspace deletion instant", stage)
+		}
+		// And the live/deleted Workspaces themselves are untouched: a migration never deletes data.
+		var live, gone int
+		must(t, pool.QueryRow(`SELECT count(*) FILTER (WHERE deleted_at IS NULL), count(*) FILTER (WHERE deleted_at IS NOT NULL)
+			FROM workspaces WHERE id IN ($1,$2,$3)`,
+			workspaces["live"], workspaces["deleted"], workspaces["running"]).Scan(&live, &gone))
+		if live != 2 || gone != 1 {
+			t.Fatalf("%s: Workspace rows live=%d deleted=%d; want 2/1 untouched", stage, live, gone)
+		}
+	}
+
+	freshPool, _ := testSchema(t, "test_wfr53f_")
+	migratedTwice(t, newStoreOnSchema(t, freshPool))
+	assertShape(t, freshPool, "fresh")
+	var rows int
+	must(t, freshPool.QueryRow(`SELECT count(*) FROM workflow_runs`).Scan(&rows))
+	if rows != 0 {
+		t.Fatalf("fresh database has %d workflow runs; want 0", rows)
+	}
+
+	pool, _ := testSchema(t, "test_wfr55_")
+	applyMigrationsUpTo(t, pool, migrationsBefore(t, "0055_workflow_run_workspace_release.sql"))
+	ids, workspaces := seedRuns(t, pool)
+	migratedTwice(t, newStoreOnSchema(t, pool))
+	assertShape(t, pool, "upgrade")
+	assertBackfilled(t, pool, ids, workspaces, "upgrade")
+	migratedTwice(t, newStoreOnSchema(t, pool))
+	assertBackfilled(t, pool, ids, workspaces, "replay")
+
+	// The constraints are the guard rail the code relies on: no state may claim a release without an
+	// observed instant, and no abandonment may be silent about why.
+	refuse := func(query string, args ...any) {
+		t.Helper()
+		tx, e := pool.Begin()
+		must(t, e)
+		defer func() { _ = tx.Rollback() }()
+		if _, e := tx.Exec(query, args...); e == nil {
+			t.Fatalf("statement %q was accepted; want the derived-state CHECK to refuse it", query)
+		}
+	}
+	refuse(`UPDATE workflow_runs SET workspace_release_state='released', workspace_released_at=NULL WHERE id=$1`, ids["running"])
+	refuse(`UPDATE workflow_runs SET workspace_release_state='abandoned', workspace_release_error='   ' WHERE id=$1`, ids["running"])
+	refuse(`UPDATE workflow_runs SET workspace_release_state='settled' WHERE id=$1`, ids["running"])
+}
diff --git a/integration/revision_upgrade_test.go b/integration/revision_upgrade_test.go
index fb8b5ab9d..b500ffda1 100644
--- a/integration/revision_upgrade_test.go
+++ b/integration/revision_upgrade_test.go
@@ -94,10 +94,11 @@ func TestRevisionUpgradePreservesPublishedWorkflowSchema(t *testing.T) {
 	}
 	must(t, tx.Commit())
 	// The run document strips the columns later migrations added to the published row: 0039's
-	// identity columns, 0042's frozen plugin version and 0051's cancellation request. They land as
-	// NULL on published rows, which is an intended shape change, not a data change — the comparison
-	// keeps guarding the row data the published schema carried.
-	const documents = `SELECT jsonb_build_object('workflow',to_jsonb(w),'snapshot',to_jsonb(s),'run',to_jsonb(r) - ARRAY['created_by','project_id','plugin_version','cancel_requested_at'])::text FROM workflows w JOIN workflow_snapshots s ON s.workflow_id=w.id JOIN workflow_runs r ON r.snapshot_id=s.id WHERE r.id=$1`
+	// identity columns, 0042's frozen plugin version, 0051's cancellation request and 0055's
+	// Workspace release fact. They land as NULL or a constant default on published rows, which is an
+	// intended shape change, not a data change — the comparison keeps guarding the row data the
+	// published schema carried.
+	const documents = `SELECT jsonb_build_object('workflow',to_jsonb(w),'snapshot',to_jsonb(s),'run',to_jsonb(r) - ARRAY['created_by','project_id','plugin_version','cancel_requested_at','workspace_release_state','workspace_release_error','workspace_released_at'])::text FROM workflows w JOIN workflow_snapshots s ON s.workflow_id=w.id JOIN workflow_runs r ON r.snapshot_id=s.id WHERE r.id=$1`
 	const ledger = `SELECT jsonb_object_agg(version,checksum)::text FROM schema_migrations WHERE version <> '0027_verified_revisions.sql' AND version < '0030'`
 	var beforeDocuments, beforeLedger string
 	must(t, pool.QueryRow(documents, run).Scan(&beforeDocuments))
diff --git a/internal/core/agent_run_release.go b/internal/core/agent_run_release.go
index 3a93e5428..e76e5f790 100644
--- a/internal/core/agent_run_release.go
+++ b/internal/core/agent_run_release.go
@@ -194,32 +194,14 @@ func runWorkspaceNodeUnknown(t *transaction, runID, wid string, window time.Dura
 		wid, runID, window.Seconds()) != nil
 }
 
-// releaseGivenUp reports whether D8's single give-up condition holds for a run still in `releasing`:
-// the run Workspace still holds a sandbox and that sandbox's Node state has been unknown for longer
-// than `DeliveryUnreachableAfter` (D5's window, reused verbatim — D8 introduces no second limit and no
-// "delete failed N times" count).
-//
-// The live-sandbox requirement is what makes the shared heartbeat judgement the right one here. A
-// Workspace usually has a sandbox for the whole `releasing` phase, because the delete cannot confirm
-// quiescence without a Node; a Workspace whose sandbox is already terminated has finished the step
-// that needed it and is being taken through `cleanup`, which runs against the Substrate with no Node
-// involved. That case is not an unknown Node — there is no Node left to be unknown — and abandoning it
-// would report a release as failed while a live Controller was still completing it. So a silent Node
-// only counts as D8's condition while the sandbox it belongs to is still there.
-//
-// The window is judged in the caller's transaction, on the database clock, exactly as D5 judges its
-// own, and an unconfigured window fails closed. An explicit quiesce refusal does not reach here at
-// all: refusing leaves the Node connected and freshly seen, so the heartbeat is present and the
-// condition is false — which is D8's "a failed delete is not a reason to give up", not an extra rule.
+// releaseGivenUp reports whether D8's single give-up condition holds for a run still in `releasing`.
+// The judgement itself — the run Workspace still holds a sandbox and that sandbox's Node state has
+// been unknown for longer than `DeliveryUnreachableAfter` (D5's window, reused verbatim; D8 introduces
+// no second limit and no "delete failed N times" count) — lives in runWorkspaceReleaseGivenUp, which
+// the workflow scheduler's release pass asks for the same fact. Neither run kind can drift into a
+// second definition of "the release should be abandoned".
 func (s *Store) releaseGivenUp(t *transaction, o Object) bool {
-	wid := o.S("workspaceId")
-	if !validID(wid) {
-		return false
-	}
-	if t.one(`SELECT 1 FROM sandbox_instances WHERE workspace_id=$1 AND terminated_at IS NULL`, wid) == nil {
-		return false
-	}
-	return runWorkspaceNodeUnknown(t, o.S("id"), wid, s.DeliveryUnreachableAfter)
+	return runWorkspaceReleaseGivenUp(t, o.S("id"), o.S("workspaceId"), s.DeliveryUnreachableAfter)
 }
 
 // giveUpRunWorkspaceRelease is D8's terminal transition, run on the caller-owned *transaction: an
@@ -295,13 +277,12 @@ func (s *Store) settleRunWorkspaceDeleted(t *transaction, runID string) error {
 	}
 	o := t.one("SELECT * FROM issue_runs WHERE id=$1 AND deleted_at IS NULL", runID)
 	if o == nil {
-		// D8: a run Workspace may belong to a WorkflowRun. A delete_workspace terminal firing
-		// under a WorkflowRun reaches this hook with the WorkflowRun's id; the scheduler owns the
-		// run-level facts, so settlement writes nothing — but it must ACCEPT the fact, or the
-		// terminal transaction aborts and a workflow run's delete can never settle. An id that is
-		// neither a live issue run nor a live workflow run is still an invariant violation.
+		// A run Workspace may belong to a WorkflowRun: the same delete_workspace terminal fires this
+		// hook with the WorkflowRun's id, and the release fact it proves is the WorkflowRun's
+		// (`settleWorkflowRunWorkspaceDeleted`, 0055). An id that is neither a live issue run nor a
+		// live workflow run is still an invariant violation.
 		if t.one("SELECT id FROM workflow_runs WHERE id=$1 AND deleted_at IS NULL", runID) != nil {
-			return nil
+			return settleWorkflowRunWorkspaceDeleted(t, runID)
 		}
 		return fmt.Errorf("runWorkspaceDeleted: issue_run %s not found", runID)
 	}
diff --git a/internal/core/control.go b/internal/core/control.go
index 9e69d2ab7..93ab2d101 100644
--- a/internal/core/control.go
+++ b/internal/core/control.go
@@ -468,12 +468,24 @@ func advance(t *transaction, r *ControlRequest, o Object) Object {
 	if next == "done" {
 		t.exec("UPDATE operations SET step='done',state='succeeded',error_code=NULL,retry_at=NULL,result=jsonb_build_object('resourceId',COALESCE(workspace_id,project_id)),version=version+1,updated_at=now() WHERE id=$1", o.S("id"))
 		finishRuntimeMaintenance(t, o.S("id"))
-		if w := t.one("SELECT issue_run_id FROM workspaces WHERE id=$1", nullable(wid)); w.S("issueRunId") != "" {
-			switch o.S("kind") {
-			case "create_workspace":
-				runWorkspaceSettled(t, w.S("issueRunId"), "ready")
-			case "delete_workspace":
-				runWorkspaceDeleted(t, w.S("issueRunId"))
+		if w := t.one("SELECT issue_run_id, workflow_run_id FROM workspaces WHERE id=$1", nullable(wid)); w != nil {
+			// A Workspace belongs to at most one run of either kind (0039's XOR), so at most one binding
+			// fires. `create_workspace` settlement is IssueRun business: a workflow run reads its
+			// Workspace's readiness from the row itself. `delete_workspace` is the one terminal every run
+			// Workspace shares, and both kinds record their own release fact from it.
+			runID := w.S("issueRunId")
+			if runID == "" {
+				runID = w.S("workflowRunId")
+			}
+			if runID != "" {
+				switch o.S("kind") {
+				case "create_workspace":
+					if w.S("issueRunId") != "" {
+						runWorkspaceSettled(t, runID, "ready")
+					}
+				case "delete_workspace":
+					runWorkspaceDeleted(t, runID)
+				}
 			}
 		}
 	} else {
diff --git a/internal/core/migrations/0055_workflow_run_workspace_release.sql b/internal/core/migrations/0055_workflow_run_workspace_release.sql
new file mode 100644
index 000000000..0718bff82
--- /dev/null
+++ b/internal/core/migrations/0055_workflow_run_workspace_release.sql
@@ -0,0 +1,55 @@
+-- WorkflowRun Workspace release (circus#75 S5; specs decisions/cloud/workflow
+-- 0-cloud-side-scheduler-owns-workflow-execution.md D3/D7).
+--
+-- A WorkflowRun's isolated Workspace is a real resource with a real owner: Cloud provisions it, and
+-- once the run is terminal it must either be deleted under the operation contract or be recorded as
+-- abandoned with a reason. Before this migration that fact was implicit — the scheduler declared the
+-- delete and nothing recorded how it ended — so a terminal run could not say whether its Workspace
+-- was gone, and a delete operation stuck behind an unreachable Node was re-declared forever.
+--
+-- The state machine is forward-only and has one writer per transition:
+--
+--   pending    -> releasing      terminal cleanup declared the Workspace delete
+--   pending    -> not_required   the run never provisioned a Workspace
+--   releasing  -> released       the delete_workspace operation succeeded, in the same transaction
+--   releasing  -> abandoned      the Workspace's Node stayed unreachable past the give-up window
+--
+-- `released` is the only state that claims the Workspace is gone, and the abort path never writes it:
+-- `abandoned` carries the reason, keeps the Workspace row (the record of a residual resource) and
+-- leaves `workspace_released_at` NULL. `not_required` is the honest answer for a run whose graph
+-- never needed a Workspace. A run never moves out of a terminal release state: a late delete success
+-- after a give-up replays an operation the give-up already failed, and must not rewrite the fact.
+--
+-- Additive; edits no applied migration.
+
+ALTER TABLE workflow_runs
+  ADD COLUMN workspace_release_state text NOT NULL DEFAULT 'pending'
+    CHECK (workspace_release_state IN ('pending','releasing','released','abandoned','not_required')),
+  ADD COLUMN workspace_release_error text NOT NULL DEFAULT '',
+  ADD COLUMN workspace_released_at timestamptz,
+  ADD CONSTRAINT workflow_run_release_released_at
+    CHECK (workspace_release_state <> 'released' OR workspace_released_at IS NOT NULL),
+  ADD CONSTRAINT workflow_run_release_abandoned_error
+    CHECK (workspace_release_state <> 'abandoned' OR btrim(workspace_release_error) <> '');
+
+-- Backfill existing rows from the evidence already authoritative in the database. A terminal run
+-- with no Workspace row never provisioned one; one whose Workspace row is soft-deleted had it
+-- released by the very delete operation that removed it in the same commit; one with a live
+-- Workspace either waits for its delete or has one in flight, and the scheduler's reconciliation
+-- pass finishes that job. Non-terminal runs stay `pending`: their release has not started.
+--
+-- `workspace_released_at` for a backfilled release is the Workspace's own deletion instant, which is
+-- the commit in which the delete operation settled — no row is stamped with the migration's clock.
+UPDATE workflow_runs r
+SET workspace_release_state = CASE
+      WHEN r.status NOT IN ('succeeded','failed','cancelled') THEN 'pending'
+      WHEN w.id IS NULL THEN 'not_required'
+      WHEN w.deleted_at IS NOT NULL THEN 'released'
+      ELSE 'releasing'
+    END,
+    workspace_released_at = CASE
+      WHEN r.status IN ('succeeded','failed','cancelled') AND w.deleted_at IS NOT NULL THEN w.deleted_at
+    END
+FROM workflow_runs current
+LEFT JOIN workspaces w ON w.workflow_run_id = current.id
+WHERE r.id = current.id;
diff --git a/internal/core/migrations/README.en.md b/internal/core/migrations/README.en.md
index 51d4f3fd7..1b2c04590 100644
--- a/internal/core/migrations/README.en.md
+++ b/internal/core/migrations/README.en.md
@@ -6,7 +6,7 @@ SQL is embedded in the binary through `embed.FS` in `core/store.go`. `Store.Migr
 
 ## File index
 
-There are currently 54 SQL files.
+There are currently 57 SQL files.
 
 | SQL file | Schema change |
 |---|---|
@@ -66,6 +66,7 @@ There are currently 54 SQL files.
 | [0052_legacy_workflow_issue_run_settlement.sql](0052_legacy_workflow_issue_run_settlement.sql) | One-time settlement of legacy queued workflow IssueRuns |
 | [0053_retire_tenant_level_clones.sql](0053_retire_tenant_level_clones.sql) | Retire tenant-level clones |
 | [0054_git_delivery_refs.sql](0054_git_delivery_refs.sql) | Revisions delivered over Git |
+| [0055_workflow_run_workspace_release.sql](0055_workflow_run_workspace_release.sql) | Workspace release facts for WorkflowRuns |
 
 0050 copies no personal connection or key and backfills no historical run. Clearing configuration retains its revision; run options are stored separately from shared run input.
 
@@ -83,6 +84,7 @@ There are currently 54 SQL files.
 - 0052 settles the live `executor_type='workflow'` IssueRuns still `queued`/`dispatched` at upgrade time: they have no executor and no frozen workflow identity, so the ADR's D3 marks them failed with `failure_reason='workflow_execution_unavailable'` instead of migrating them. Other runs and soft-deleted rows are untouched. Rows of that shape stop being produced only once the workflow launch gate binds a manual run to a real WorkflowRun; until that gate (circus#71) lands, `createRun` keeps producing new ones, and this migration runs once at upgrade time and does not settle them. Coverage is `TestMigration0052LegacyWorkflowIssueRunSettlementAppliesFreshAndUpgrades`.
 - 0053 is the switch to Workspaces cloned from Git Server snapshots and migrates no data: clone executions that belong to no Workspace and their receipts are deleted, `clone_executions.clone_request_id` and the `one_request_execution` index are dropped, `clone_executions.workspace_id` becomes NOT NULL, and `clone_requests` is dropped; Workspace clone executions and their receipts stay, and it is retry-safe. See [Workspaces](../../../docs/runtime/workspaces.en.md); upgrading a database that still holds tenant-level clones is covered by `TestMigration0053RetiresTenantLevelClonesKeepingWorkspaceClones`.
 - 0054 is the switch to Revisions delivered over Git and migrates no data: `revisions` gains `delivery_ref`, and `revisions_bundle_shape` is replaced by `revisions_shape` (with a NULL `delivery_ref` a row keeps a legacy bundle shape; with one it must be a well-formed `refs/ora/deliveries/<UUID>` and carry no bundle column); the partial unique index `revisions_delivery_ref` lets one delivery ref belong to one Revision only. Existing bundle rows are never rewritten, and it is retry-safe. See [Revisions](../../../docs/agent-runs/revisions.en.md); the constraints are covered by `TestRevisionShapeKeepsGitAndBundleRowsApart`.
+- 0055 adds the forward-only `workspace_release_state` (`pending`/`releasing`/`released`/`abandoned`/`not_required`), `workspace_release_error` and `workspace_released_at` to `workflow_runs`, and backfills by upgrade-time state: non-terminal rows stay `pending`; terminal rows with no Workspace become `not_required`, those whose Workspace is already soft-deleted become `released` (using that row's `deleted_at`), and the rest become `releasing`. Two CHECK constraints require a release instant for `released` and a reason for `abandoned`. Only `released` means Cloud observed the Workspace deleted; `abandoned` keeps the Workspace row and writes no release instant. Semantics are in [workflows](../../../docs/workflows/workflows.en.md#scheduling); coverage is `TestMigration0055WorkflowRunWorkspaceReleaseAppliesFreshAndUpgrades` and `internal/core/workflow_run_release_db_test.go`.
 - New migrations must verify fresh databases, upgrades from the preceding schema, repeated application and key constraints. See [migration_upgrade_path_test.go](../../../integration/migration_upgrade_path_test.go) and domain upgrade tests in integration.
 
 See [Persistence](../../../docs/core/persistence.en.md) for ownership/transactions, and [Workflows](../../../docs/workflows/workflows.en.md) / [External Git line](../../../docs/repositories/external-git.en.md) for runtime semantics.
diff --git a/internal/core/migrations/README.md b/internal/core/migrations/README.md
index bc1efb627..40f613780 100644
--- a/internal/core/migrations/README.md
+++ b/internal/core/migrations/README.md
@@ -6,7 +6,7 @@ SQL 通过 `core/store.go` 的 `embed.FS` 编入二进制。`Store.Migrate` 在
 
 ## 文件索引
 
-当前共 54 个 SQL 文件。
+当前共 57 个 SQL 文件。
 
 | SQL 文件 | schema 变化 |
 |---|---|
@@ -66,6 +66,7 @@ SQL 通过 `core/store.go` 的 `embed.FS` 编入二进制。`Store.Migrate` 在
 | [0052_legacy_workflow_issue_run_settlement.sql](0052_legacy_workflow_issue_run_settlement.sql) | 存量 queued 工作流 IssueRun 的一次性结算 |
 | [0053_retire_tenant_level_clones.sql](0053_retire_tenant_level_clones.sql) | 退役租户级 clone |
 | [0054_git_delivery_refs.sql](0054_git_delivery_refs.sql) | 经 Git 交付的 Revision |
+| [0055_workflow_run_workspace_release.sql](0055_workflow_run_workspace_release.sql) | WorkflowRun 的 Workspace 释放事实 |
 
 0050 不复制个人连接或 Key，不回填历史运行；清除配置保留 revision，运行选项独立于共享运行输入保存。
 
@@ -83,6 +84,7 @@ SQL 通过 `core/store.go` 的 `embed.FS` 编入二进制。`Store.Migrate` 在
 - 0052 结算升级时仍为 `queued`/`dispatched` 的、存活的 `executor_type='workflow'` IssueRun：它们没有执行器、也没有冻结的工作流身份，按 ADR D3 标记失败并写 `failure_reason='workflow_execution_unavailable'`，不做迁移；其他状态的行与软删除行保持原样。工作流启动门禁把 manual 运行绑定到真实 WorkflowRun 之后才会停止产生这种行；门禁（circus#71）落地前 `createRun` 仍会造出新行，而本迁移只在升级时执行一次，不会回头结算它们。覆盖见 `TestMigration0052LegacyWorkflowIssueRunSettlementAppliesFreshAndUpgrades`。
 - 0053 是 Workspace 改为从 Git Server 快照 clone 的切换，不迁移数据：删除不属于 Workspace 的 clone execution 及其收据，去掉 `clone_executions.clone_request_id` 与索引 `one_request_execution`，`clone_executions.workspace_id` 改为 NOT NULL，删除 `clone_requests`；Workspace 的 clone execution 与收据保留，可重复执行。语义见 [Workspace](../../../docs/runtime/workspaces.md)，仍有租户级 clone 的库的升级覆盖见 `TestMigration0053RetiresTenantLevelClonesKeepingWorkspaceClones`。
 - 0054 是 Revision 改为经 Git 交付的切换，不迁移数据：`revisions` 增加 `delivery_ref`，`revisions_bundle_shape` 换成 `revisions_shape`（`delivery_ref` 为 NULL 时仍是原有的 bundle 形状，非 NULL 时必须是格式正确的 `refs/ora/deliveries/<UUID>` 且不带任何 bundle 列），部分唯一索引 `revisions_delivery_ref` 让一个交付 ref 只属于一个 Revision；既有 bundle 行不改写，可重复执行。语义见 [Revision](../../../docs/agent-runs/revisions.md)，约束覆盖见 `TestRevisionShapeKeepsGitAndBundleRowsApart`。
+- 0055 给 `workflow_runs` 增加单向的 `workspace_release_state`（`pending`/`releasing`/`released`/`abandoned`/`not_required`）、`workspace_release_error` 和 `workspace_released_at`，并按升级时的状态回填：非终态行保持 `pending`；终态行没有 Workspace 的记 `not_required`，Workspace 已软删除的记 `released`（释放时刻取该行的 `deleted_at`），其余记 `releasing`。两条 CHECK 保证 `released` 必有释放时刻、`abandoned` 必有原因。只有 `released` 表示 Cloud 观察到 Workspace 已被删除；`abandoned` 保留 Workspace 行、不写释放时刻。语义见[工作流](../../../docs/workflows/workflows.md#调度)，覆盖见 `TestMigration0055WorkflowRunWorkspaceReleaseAppliesFreshAndUpgrades` 与 `internal/core/workflow_run_release_db_test.go`。
 - 新增迁移须验证空库、前一 schema 升级、重复应用与关键约束。覆盖见 [migration_upgrade_path_test.go](../../../integration/migration_upgrade_path_test.go) 和 integration 的领域升级测试。
 
 数据归属与事务规则见 [持久化](../../../docs/core/persistence.md)，运行语义见 [工作流](../../../docs/workflows/workflows.md) 和 [外部 Git 线](../../../docs/repositories/external-git.md)。
diff --git a/internal/core/workflow_run_identity_db_test.go b/internal/core/workflow_run_identity_db_test.go
index 8a9423259..c600157ab 100644
--- a/internal/core/workflow_run_identity_db_test.go
+++ b/internal/core/workflow_run_identity_db_test.go
@@ -99,6 +99,13 @@ func TestWorkflowRunWorkspaceLifecycle(t *testing.T) {
 		t.Fatalf("runtime control state=%q run=%q; want maintenance reserved for the workflow run", state, maintenanceRun)
 	}
 
+	// The deletion below belongs to a terminal run: since 0055 the delete hook records the WorkflowRun's
+	// release fact and refuses a completion whose run never declared a terminal state, so the run is
+	// settled first, exactly as the scheduler settles it before declaring the delete.
+	if _, err := store.Pool.Exec(`UPDATE workflow_runs SET status='succeeded', version=version+1 WHERE id=$1`, runID); err != nil {
+		t.Fatalf("settle the run terminal: %v", err)
+	}
+
 	// Deletion: declared like the scheduler declares it, settled through the terminal step, and
 	// accepted by the delete hook — the terminal transaction must not abort on a workflow run id.
 	if _, err := store.transact(context.Background(), func(tx *transaction) Object {
diff --git a/internal/core/workflow_run_release.go b/internal/core/workflow_run_release.go
new file mode 100644
index 000000000..235929516
--- /dev/null
+++ b/internal/core/workflow_run_release.go
@@ -0,0 +1,336 @@
+package core
+
+import (
+	"context"
+	"fmt"
+	"time"
+)
+
+// The WorkflowRun Workspace release (circus#75 S5; specs decisions/cloud/workflow
+// 0-cloud-side-scheduler-owns-workflow-execution.md D3/D7). A run's terminal status says how the
+// workflow ended; it says nothing about the isolated Workspace Cloud provisioned for it. These are
+// two independent facts, and keeping them apart is the whole point of this file: a run that finished
+// is not a Workspace that was released.
+//
+// The durable state machine (0055) is forward-only:
+//
+//	pending    -> releasing      terminal cleanup declared the Workspace delete
+//	pending    -> not_required   the run never provisioned a Workspace
+//	releasing  -> released       the delete_workspace operation succeeded, in that same transaction
+//	releasing  -> abandoned      the Workspace's Node stayed unreachable past the give-up window
+//
+// Only `released` claims the Workspace is gone. `abandoned` keeps the Workspace row, carries the
+// reason and leaves `workspace_released_at` NULL — the same "never report a release Cloud did not
+// observe" rule IssueRun D8 applies, for the run kind that has no phase column to carry it. No
+// transition leaves a terminal release state, so a replay of a settled delete is a no-op rather than
+// a second fact.
+//
+// Every writer runs inside the transaction that owns the evidence it reacts to:
+//
+//   - terminal settlement declares the delete and writes `releasing`, so a terminal run is never
+//     without a durable release intent;
+//   - the delete-completion hook writes `released` in the transaction that writes the operation
+//     `succeeded` and soft-deletes the Workspace, so the two cannot disagree;
+//   - the bounded give-up pass writes `abandoned` in the transaction that fails the unfinished
+//     operation, so a stuck delete cannot keep a run in `releasing` forever.
+
+const (
+	workflowReleasePending     = "pending"
+	workflowReleaseReleasing   = "releasing"
+	workflowReleaseReleased    = "released"
+	workflowReleaseAbandoned   = "abandoned"
+	workflowReleaseNotRequired = "not_required"
+)
+
+// workflowReleaseAbandonedReason is the run-level reason a give-up records: Cloud could not reach the
+// Workspace's Node, so the release did not happen. It is the code IssueRun D8 already uses for the
+// same fact, kept identical so a client reading either run kind reads the same word.
+const workflowReleaseAbandonedReason = "workspace_unavailable"
+
+// declareWorkflowRunWorkspaceRelease records the release intent of a terminal run and declares the
+// Workspace delete. It is the only writer of `not_required` and the only writer of the first
+// `releasing`; every later call for the same run is a no-op on the state and at most re-declares an
+// absent delete operation.
+//
+// A run with no Workspace row never needed one (its graph had no agent node, or it was cancelled
+// before provisioning), and `not_required` says exactly that instead of leaving the release
+// ambiguous. A run whose Workspace row is already soft-deleted had it deleted by another path (a
+// project deletion deletes every Workspace inline and never fires the per-run delete hook), so its
+// own deletion instant is the release fact — the same value the completion hook would have written.
+//
+// Declaring the delete can legitimately do nothing: a busy Project refuses the declaration, and the
+// reconciliation pass re-declares it later. The state stays `releasing` either way, because the
+// intent is already durable.
+func declareWorkflowRunWorkspaceRelease(t *transaction, runID string) {
+	run := t.one("SELECT status, workspace_release_state FROM workflow_runs WHERE id=$1 AND deleted_at IS NULL", runID)
+	if run == nil {
+		return
+	}
+	switch run.S("status") {
+	case "succeeded", "failed", "cancelled":
+	default:
+		// A release starts only when the run is terminal. A running run whose Workspace was deleted
+		// is an operator intervention, not this scheduler's fact to invent.
+		return
+	}
+	switch run.S("workspaceReleaseState") {
+	case workflowReleaseReleased, workflowReleaseAbandoned, workflowReleaseNotRequired:
+		return
+	}
+	w := t.one("SELECT id, deleted_at FROM workspaces WHERE workflow_run_id=$1", runID)
+	if w == nil {
+		t.exec(`UPDATE workflow_runs
+			SET workspace_release_state=$2, workspace_release_error='', version=version+1, updated_at=now()
+			WHERE id=$1 AND workspace_release_state='pending'`, runID, workflowReleaseNotRequired)
+		return
+	}
+	if w["deletedAt"] != nil {
+		t.exec(`UPDATE workflow_runs
+			SET workspace_release_state=$2, workspace_release_error='', workspace_released_at=COALESCE($3::timestamptz, now()),
+			    version=version+1, updated_at=now()
+			WHERE id=$1 AND workspace_release_state IN ('pending','releasing')`,
+			runID, workflowReleaseReleased, w["deletedAt"])
+		return
+	}
+	t.exec(`UPDATE workflow_runs
+		SET workspace_release_state=$2, workspace_release_error='', version=version+1, updated_at=now()
+		WHERE id=$1 AND workspace_release_state='pending'`, runID, workflowReleaseReleasing)
+	// deleteWorkflowRunWorkspace is idempotent: an existing live delete operation is returned, a busy
+	// Project reports busy, and an absent one is declared. Its result decides nothing about the release
+	// state — the intent is already committed with this transaction.
+	_ = deleteWorkflowRunWorkspace(t, runID)
+}
+
+// settleWorkflowRunWorkspaceDeleted is the WorkflowRun half of the delete-completion hook
+// (settleRunWorkspaceDeleted): it runs in the transaction that writes the run Workspace's
+// delete_workspace operation `succeeded`, and records the one fact that operation proves.
+//
+// `released` is written from `pending`, `releasing` or `not_required`. The last one would contradict
+// the operation's own existence (a Workspace row bound to this run means something needed releasing),
+// but the delete's success is the stronger evidence and the row is already being deleted in this
+// transaction, so it is corrected rather than refused.
+//
+// A run already `released` is a replay and writes nothing. A run `abandoned` writes nothing either:
+// the give-up failed the very operation this hook would settle, so reaching here means a late or
+// duplicated completion for an operation Cloud already gave up on, and D8's rule is that a give-up is
+// never turned back into a reported release. Any other state is a contradiction and aborts the
+// caller's transaction rather than silently accepting a delete the run's lifecycle never declared.
+func settleWorkflowRunWorkspaceDeleted(t *transaction, runID string) error {
+	run := t.one("SELECT status, workspace_release_state FROM workflow_runs WHERE id=$1 AND deleted_at IS NULL", runID)
+	if run == nil {
+		return fmt.Errorf("runWorkspaceDeleted: workflow run %s not found", runID)
+	}
+	switch run.S("status") {
+	case "succeeded", "failed", "cancelled":
+	default:
+		return fmt.Errorf("runWorkspaceDeleted: workflow run %s is status=%q, not terminal", runID, run.S("status"))
+	}
+	switch run.S("workspaceReleaseState") {
+	case workflowReleaseReleased, workflowReleaseAbandoned:
+		return nil
+	case workflowReleasePending, workflowReleaseReleasing, workflowReleaseNotRequired:
+		if t.execRows(`UPDATE workflow_runs
+			SET workspace_release_state=$2, workspace_release_error='', workspace_released_at=now(),
+			    version=version+1, updated_at=now()
+			WHERE id=$1 AND workspace_release_state IN ('pending','releasing','not_required')`,
+			runID, workflowReleaseReleased) != 1 {
+			return fmt.Errorf("runWorkspaceDeleted: workflow run %s was not moved to released", runID)
+		}
+		return nil
+	default:
+		return fmt.Errorf("runWorkspaceDeleted: workflow run %s carries release state %q, outside the closed set", runID, run.S("workspaceReleaseState"))
+	}
+}
+
+// abandonWorkflowRunWorkspaceRelease is D8's terminal transition for a WorkflowRun: an unreachable
+// run Workspace stops being retried, and the run records that Cloud gave up without ever claiming the
+// Workspace was released.
+//
+// The writes are the same three facts as the IssueRun give-up, and nothing else:
+//
+//   - the unfinished delete_workspace operation becomes terminally failed. Zero rows is a legal
+//     outcome: the run can be `releasing` with no operation (a busy Project that has not accepted the
+//     declaration yet), and the condition is about the Workspace, not about an operation existing.
+//   - `releasing -> abandoned` (or `pending -> abandoned`) as one CAS on the state the caller read.
+//     Zero affected rows is an invariant contradiction and rolls the operation failure back with it.
+//   - `workspace_release_error = workspace_unavailable`, and `workspace_released_at` stays NULL.
+//
+// Deliberately not here: anything about `workspaces`. The Workspace is still there, and clearing the
+// binding, soft-deleting the row or writing a success-shaped operation result is the lie D8 forbids.
+// The row is the record of a residual resource and must stay readable.
+func abandonWorkflowRunWorkspaceRelease(t *transaction, runID, wid string) error {
+	// `node_unavailable` is the published operation failure code for an unreachable Node, the same code
+	// the IssueRun give-up and the session settlement use. The operation keeps its step and request:
+	// they are the record of how far the delete got.
+	t.exec(`UPDATE operations SET state='failed', error_code='node_unavailable', version=version+1, updated_at=now()
+		WHERE workspace_id=$1 AND kind='delete_workspace' AND state IN ('queued','running','retry_wait','blocked')`, wid)
+	if t.execRows(`UPDATE workflow_runs
+		SET workspace_release_state=$2, workspace_release_error=$3, version=version+1, updated_at=now()
+		WHERE id=$1 AND deleted_at IS NULL AND status IN ('succeeded','failed','cancelled')
+		  AND workspace_release_state IN ('pending','releasing')`,
+		runID, workflowReleaseAbandoned, workflowReleaseAbandonedReason) != 1 {
+		run := t.one("SELECT status, workspace_release_state FROM workflow_runs WHERE id=$1", runID)
+		return fmt.Errorf("abandon workflow run release: run %s was not moved to abandoned (status=%q release=%q)",
+			runID, run.S("status"), run.S("workspaceReleaseState"))
+	}
+	return nil
+}
+
+// giveUpWorkflowRunWorkspaceReleases is one bounded pass of D8's give-up policy for WorkflowRuns: it
+// finds terminal runs still holding a live Workspace whose release has not settled, and abandons the
+// ones whose Node state has been unknown past the configured window. Each decision runs in its own
+// short transaction; a per-run failure rolls that transaction back and the next tick retries it, so
+// one transiently stuck run never cancels the pass.
+//
+// The condition is runWorkspaceReleaseGivenUp — the same live-sandbox-plus-heartbeat judgement the
+// IssueRun release paths use, reusing the shared runWorkspaceNodeUnknown definition of "the Node is
+// gone". It is never taken from the delete operation's state: an operation that failed repeatedly, or
+// one that never existed, is not evidence about reachability, and a refusal by a reachable Node is
+// not a reason to give up. `pending` is included because a terminal run can wait a long time for a
+// busy Project to accept its delete declaration; if the Node is already gone, waiting for the Project
+// changes nothing.
+//
+// An unconfigured window (zero or negative) is a no-op, not an immediate give-up: runWorkspaceReleaseGivenUp
+// fails closed, and a deployment that never set the limit loses nothing it was promised. This is the
+// same fail-closed rule and the same `DeliveryUnreachableAfter` window as IssueRun D5/D8 — there is no
+// second limit to tune.
+func (s *Store) giveUpWorkflowRunWorkspaceReleases(ctx context.Context) error {
+	if s.DeliveryUnreachableAfter <= 0 {
+		return nil
+	}
+	type releaseCandidate struct{ runID, workspaceID string }
+	rows, err := s.Pool.QueryContext(ctx, `
+		SELECT r.id, w.id FROM workflow_runs r
+		JOIN workspaces w ON w.workflow_run_id=r.id
+		WHERE r.status IN ('succeeded','failed','cancelled') AND r.deleted_at IS NULL
+		  AND r.workspace_release_state IN ('pending','releasing')
+		  AND w.deleted_at IS NULL
+		ORDER BY r.updated_at, r.id
+		LIMIT $1`, workflowAdvanceBatchSize)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	var candidates []releaseCandidate
+	for rows.Next() {
+		var c releaseCandidate
+		if err := rows.Scan(&c.runID, &c.workspaceID); err != nil {
+			return err
+		}
+		candidates = append(candidates, c)
+	}
+	if err := rows.Err(); err != nil {
+		return err
+	}
+	for _, c := range candidates {
+		if _, err := s.transact(ctx, func(t *transaction) Object {
+			// Re-read the predicate inside the transaction that writes: a run that settled its release
+			// between the scan and this pass is a deterministic no-op, never a wrongful abandonment.
+			if t.one(`
+				SELECT r.id FROM workflow_runs r
+				JOIN workspaces w ON w.workflow_run_id=r.id
+				WHERE r.id=$1 AND w.id=$2 AND r.status IN ('succeeded','failed','cancelled')
+				  AND r.deleted_at IS NULL AND r.workspace_release_state IN ('pending','releasing')
+				  AND w.deleted_at IS NULL`, c.runID, c.workspaceID) == nil {
+				return Object{}
+			}
+			if !runWorkspaceReleaseGivenUp(t, c.runID, c.workspaceID, s.DeliveryUnreachableAfter) {
+				return Object{}
+			}
+			if err := abandonWorkflowRunWorkspaceRelease(t, c.runID, c.workspaceID); err != nil {
+				panic(databaseFailure{err})
+			}
+			return Object{}
+		}); err != nil {
+			_ = err
+		}
+	}
+	return nil
+}
+
+// reconcileWorkflowRunWorkspaceReleases finishes the release fact of every terminal run the scheduler
+// itself no longer scans, in two steps.
+//
+// The first heals runs whose Workspace was already deleted by a path that never fires the per-run
+// delete hook — a project deletion deletes every Workspace of the project inline. The Workspace row's
+// own deletion instant is the release fact, so the state is corrected to `released` rather than left
+// `releasing` beside a Workspace that no longer exists.
+//
+// The second re-declares the delete of every terminal run still holding a live Workspace and no
+// unfinished delete operation. Terminal settlement declares it in the settling transaction; a busy
+// Project refuses that declaration, so this scan is the retry. Idempotent:
+// deleteWorkflowRunWorkspace treats an already-declared or already-deleted Workspace as done.
+//
+// Runs whose release already reached a terminal state are out of scope in both steps. In particular
+// an `abandoned` run is never re-entered: D8's residual Workspace is intentionally left with no
+// operation and no retry.
+func (s *Store) reconcileWorkflowRunWorkspaceReleases(ctx context.Context) error {
+	if _, err := s.Pool.ExecContext(ctx, `
+		UPDATE workflow_runs r
+		SET workspace_release_state='released', workspace_release_error='',
+		    workspace_released_at=w.deleted_at, version=r.version+1, updated_at=now()
+		FROM workspaces w
+		WHERE w.workflow_run_id=r.id AND w.deleted_at IS NOT NULL
+		  AND r.status IN ('succeeded','failed','cancelled') AND r.deleted_at IS NULL
+		  AND r.workspace_release_state IN ('pending','releasing')`); err != nil {
+		return err
+	}
+	rows, err := s.Pool.QueryContext(ctx, `
+		SELECT r.id FROM workflow_runs r
+		JOIN workspaces w ON w.workflow_run_id=r.id
+		WHERE r.status IN ('failed','cancelled','succeeded') AND r.deleted_at IS NULL
+		  AND r.workspace_release_state IN ('pending','releasing')
+		  AND w.deleted_at IS NULL
+		  AND NOT EXISTS (SELECT 1 FROM operations o WHERE o.workspace_id=w.id AND o.kind='delete_workspace'
+		   AND o.state IN ('queued','running','retry_wait','blocked','succeeded'))
+		ORDER BY r.created_at, r.id LIMIT $1`, workflowAdvanceBatchSize)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	var ids []string
+	for rows.Next() {
+		var id string
+		if err := rows.Scan(&id); err != nil {
+			return err
+		}
+		ids = append(ids, id)
+	}
+	if err := rows.Err(); err != nil {
+		return err
+	}
+	for _, id := range ids {
+		if _, err := s.transact(ctx, func(t *transaction) Object {
+			finishWorkflowRunCleanup(t, id)
+			return Object{}
+		}); err != nil {
+			_ = err
+		}
+	}
+	return nil
+}
+
+// runWorkspaceReleaseGivenUp is D8's single give-up condition, shared by both run kinds: the run
+// Workspace still holds a sandbox, and that sandbox's Node state has been unknown for longer than the
+// deployment's unreachability window.
+//
+// The live-sandbox requirement is what makes the shared heartbeat judgement the right one here. A
+// Workspace usually has a sandbox for the whole release, because the delete cannot confirm quiescence
+// without a Node; once that sandbox is terminated the remaining cleanup runs against the Substrate
+// with no Node involved, and abandoning it would report a release as failed while a live Controller
+// was still completing it. So a silent Node only counts while the sandbox it belongs to is still
+// there.
+//
+// The window is judged on the database clock through runWorkspaceNodeUnknown, and an unconfigured
+// window fails closed. An explicit quiesce refusal does not reach here at all: refusing leaves the
+// Node connected and freshly seen, so the heartbeat is present and the condition is false — "a failed
+// delete is not a reason to give up", not an extra rule.
+func runWorkspaceReleaseGivenUp(t *transaction, runID, wid string, window time.Duration) bool {
+	if !validID(wid) {
+		return false
+	}
+	if t.one(`SELECT 1 FROM sandbox_instances WHERE workspace_id=$1 AND terminated_at IS NULL`, wid) == nil {
+		return false
+	}
+	return runWorkspaceNodeUnknown(t, runID, wid, window)
+}
diff --git a/internal/core/workflow_run_release_db_test.go b/internal/core/workflow_run_release_db_test.go
new file mode 100644
index 000000000..77a01e39d
--- /dev/null
+++ b/internal/core/workflow_run_release_db_test.go
@@ -0,0 +1,353 @@
+package core
+
+// White-box acceptance for the WorkflowRun Workspace release (circus#75 S5; specs decisions/cloud/
+// workflow 0-cloud-side-scheduler-owns-workflow-execution D3/D7). Everything here is real PostgreSQL
+// in an isolated schema, and the transitions are driven the way production drives them: the terminal
+// cleanup in the transaction that settles the run, the delete-completion hook in the transaction
+// that writes the operation `succeeded`, and the scheduler's own bounded pass for the give-up and
+// reconciliation half.
+//
+// The obligation under test is a pair of independent facts: a terminal run may not leave its
+// Workspace's fate unrecorded, and Cloud may never report a Workspace `released` that it did not see
+// deleted. The end-to-end chain (a real Controller taking a delete_workspace operation through every
+// step) is the acceptance stack's job; these pin the transaction-scoped transitions and the give-up
+// bound without standing up a Node.
+
+import (
+	"context"
+	"database/sql"
+	"testing"
+	"time"
+)
+
+// workflowReleaseShape is the durable release fact plus the version it was written under.
+type workflowReleaseShape struct {
+	state      string
+	reason     string
+	releasedAt sql.NullTime
+	version    int64
+}
+
+func readWorkflowRelease(t *testing.T, store *Store, runID string) workflowReleaseShape {
+	t.Helper()
+	var s workflowReleaseShape
+	if err := store.Pool.QueryRow(`SELECT workspace_release_state, workspace_release_error, workspace_released_at, version
+		FROM workflow_runs WHERE id=$1`, runID).Scan(&s.state, &s.reason, &s.releasedAt, &s.version); err != nil {
+		t.Fatalf("read workflow release: %v", err)
+	}
+	return s
+}
+
+// finishWorkflowRunNow calls the scheduler's real terminal transition on a run whose workspace scene
+// is staged. It runs in one transaction, exactly as a scheduler pass does.
+func finishWorkflowRunNow(t *testing.T, store *Store, runID, status string) {
+	t.Helper()
+	if _, err := store.transact(context.Background(), func(tx *transaction) Object {
+		run := tx.one("SELECT version FROM workflow_runs WHERE id=$1", runID)
+		finishWorkflowRun(tx, runID, run.N("version"), status, "", nil)
+		return Object{}
+	}); err != nil {
+		t.Fatalf("finish workflow run: %v", err)
+	}
+}
+
+// stageProvisionedWorkflowRun seeds a run whose Workspace was provisioned and settled ready with a
+// connected Node, then settles the run terminal through the scheduler's own cleanup. It is the state
+// every release transition starts from: the delete has been declared and the Node that would execute
+// it is live.
+func stageProvisionedWorkflowRun(t *testing.T, store *Store) (runID, wsID string) {
+	t.Helper()
+	runID, _, _, _, _ = seedWorkflowSessionScene(t, store)
+	if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil {
+		t.Fatalf("advance to provision the workspace: %v", err)
+	}
+	wsID = settleSessionWorkspace(t, store, runID)
+	bindConnectedNode(t, store, wsID)
+	finishWorkflowRunNow(t, store, runID, "succeeded")
+	return runID, wsID
+}
+
+func countWorkflowDeleteOperations(t *testing.T, store *Store, runID string) int {
+	t.Helper()
+	var n int
+	if err := store.Pool.QueryRow(`SELECT count(*) FROM operations o JOIN workspaces w ON w.id=o.workspace_id
+		WHERE w.workflow_run_id=$1 AND o.kind='delete_workspace'`, runID).Scan(&n); err != nil {
+		t.Fatalf("count delete operations: %v", err)
+	}
+	return n
+}
+
+// A terminal run with a live Workspace commits its release intent and the delete declaration in the
+// settling transaction. Repeating the cleanup writes nothing new: the intent is already durable and
+// the existing operation is returned rather than duplicated.
+func TestWorkflowRunCleanupRecordsTheReleaseIntent(t *testing.T) {
+	store := dispatcherDB(t)
+	runID, wsID := stageProvisionedWorkflowRun(t, store)
+
+	release := readWorkflowRelease(t, store, runID)
+	if release.state != workflowReleaseReleasing || release.reason != "" || release.releasedAt.Valid {
+		t.Fatalf("release after terminal cleanup = %+v; want releasing with no error and no release instant", release)
+	}
+	var deletedAt sql.NullTime
+	if err := store.Pool.QueryRow(`SELECT deleted_at FROM workspaces WHERE id=$1`, wsID).Scan(&deletedAt); err != nil {
+		t.Fatal(err)
+	}
+	if deletedAt.Valid {
+		t.Fatal("the terminal cleanup only declares the delete; it must not delete the Workspace itself")
+	}
+	if n := countWorkflowDeleteOperations(t, store, runID); n != 1 {
+		t.Fatalf("%d delete operations declared; want exactly 1", n)
+	}
+
+	// The reconciliation pass calls the same cleanup again on every tick until the delete settles.
+	// It must not bump the version or declare a second operation.
+	before := readWorkflowRelease(t, store, runID)
+	if _, err := store.transact(context.Background(), func(tx *transaction) Object {
+		finishWorkflowRunCleanup(tx, runID)
+		return Object{}
+	}); err != nil {
+		t.Fatalf("replayed cleanup: %v", err)
+	}
+	after := readWorkflowRelease(t, store, runID)
+	if after.state != before.state || after.version != before.version {
+		t.Fatalf("a replayed cleanup changed %+v to %+v", before, after)
+	}
+	if n := countWorkflowDeleteOperations(t, store, runID); n != 1 {
+		t.Fatalf("a replayed cleanup declared %d delete operations; want 1", n)
+	}
+}
+
+// The delete-completion hook is the only writer of `released`. Driven through the control plane's
+// own `advance` at the cleanup step — the transaction that writes the operation `succeeded` and
+// soft-deletes the Workspace — it records the release instant on the WorkflowRun, proving both the
+// workflow-run binding in control.go and the B-side transition.
+func TestWorkflowRunDeleteCompletionRecordsReleased(t *testing.T) {
+	store := dispatcherDB(t)
+	runID, wsID := stageProvisionedWorkflowRun(t, store)
+	var opID, projectID string
+	if err := store.Pool.QueryRow(`SELECT o.id, o.project_id FROM operations o WHERE o.workspace_id=$1 AND o.kind='delete_workspace'`, wsID).
+		Scan(&opID, &projectID); err != nil {
+		t.Fatalf("read delete operation: %v", err)
+	}
+	// The Controller has taken the operation and finished the data-delete effect; the next advance is
+	// the cleanup step that deletes the Workspace row and settles the operation.
+	if _, err := store.Pool.Exec(`UPDATE operations SET state='running', step='cleanup', controller_epoch=1, version=version+1 WHERE id=$1`, opID); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := store.Pool.Exec(`INSERT INTO external_effects(id,operation_id,project_id,workspace_id,kind,state,request,reconciled_epoch)
+		VALUES($1,$2,$3,$4,'workspace_data_delete','succeeded','{}'::jsonb,1)`, newID(), opID, projectID, wsID); err != nil {
+		t.Fatalf("seed completed effect: %v", err)
+	}
+	before := readWorkflowRelease(t, store, runID)
+	if _, err := store.transact(context.Background(), func(tx *transaction) Object {
+		o := tx.one("SELECT * FROM operations WHERE id=$1", opID)
+		advance(tx, &ControlRequest{Body: Object{"epoch": int64(1), "version": o.N("version")}}, o)
+		return Object{}
+	}); err != nil {
+		t.Fatalf("advance delete to done: %v", err)
+	}
+
+	release := readWorkflowRelease(t, store, runID)
+	if release.state != workflowReleaseReleased || release.reason != "" || !release.releasedAt.Valid {
+		t.Fatalf("release after the delete completed = %+v; want released with an instant and no error", release)
+	}
+	var observed string
+	var deletedAt sql.NullTime
+	if err := store.Pool.QueryRow(`SELECT observed_state, deleted_at FROM workspaces WHERE id=$1`, wsID).Scan(&observed, &deletedAt); err != nil {
+		t.Fatal(err)
+	}
+	if observed != "deleted" || !deletedAt.Valid {
+		t.Fatalf("workspace observed=%q deleted_at=%v; want the delete committed in the same transaction", observed, deletedAt)
+	}
+
+	// A replayed hook invocation finds `released` and writes nothing: the fact is already committed.
+	if _, err := store.transact(context.Background(), func(tx *transaction) Object {
+		if err := store.settleRunWorkspaceDeleted(tx, runID); err != nil {
+			panic(databaseFailure{err})
+		}
+		return Object{}
+	}); err != nil {
+		t.Fatalf("replayed delete settle: %v", err)
+	}
+	after := readWorkflowRelease(t, store, runID)
+	if after.state != workflowReleaseReleased || after.version != release.version {
+		t.Fatalf("a replayed delete settle changed %+v to %+v", release, after)
+	}
+	if release.version != before.version+1 {
+		t.Fatalf("the release took %d version bumps; want exactly 1", release.version-before.version)
+	}
+}
+
+// A delete completion for a run that is still executing cannot be justified: no terminal transition
+// declared that delete, so the hook refuses and the caller's transaction rolls back. Accepting it
+// would record a Workspace release for a run whose nodes are still executing against that Workspace.
+func TestWorkflowRunDeleteCompletionRefusesANonTerminalRun(t *testing.T) {
+	store := dispatcherDB(t)
+	runID, _, _, _, _ := seedWorkflowSessionScene(t, store)
+	if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil {
+		t.Fatalf("advance: %v", err)
+	}
+	before := readWorkflowRelease(t, store, runID)
+	err := func() error {
+		_, err := store.transact(context.Background(), func(tx *transaction) Object {
+			if e := store.settleRunWorkspaceDeleted(tx, runID); e != nil {
+				panic(databaseFailure{e})
+			}
+			return Object{}
+		})
+		return err
+	}()
+	if err == nil {
+		t.Fatal("a delete completion for a running run must be refused")
+	}
+	if after := readWorkflowRelease(t, store, runID); after != before {
+		t.Fatalf("a refused delete completion wrote %+v; want %+v", after, before)
+	}
+}
+
+// A run whose graph never needed a Workspace records `not_required` at settlement. That is a
+// different fact from `released`, and the two must never be conflated: nothing was ever released.
+func TestWorkflowRunWithoutWorkspaceRecordsNotRequired(t *testing.T) {
+	store := dispatcherDB(t)
+	runID := seedWorkflowRunScene(t, store, wfGraph(
+		[]Object{wfNode("s", "start", ""), wfNode("o", "output", "")},
+		[][3]string{{"s", "o", ""}},
+	))
+	if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil {
+		t.Fatalf("advance: %v", err)
+	}
+	release := readWorkflowRelease(t, store, runID)
+	if release.state != workflowReleaseNotRequired || release.reason != "" || release.releasedAt.Valid {
+		t.Fatalf("release = %+v; want not_required with no release instant", release)
+	}
+}
+
+// D8's bound, for a WorkflowRun: a terminal run whose Workspace still holds a sandbox whose Node has
+// been silent past the configured window is abandoned once. The unfinished delete operation is failed
+// with the published unreachable-Node code, the Workspace row and its binding stay readable, and the
+// run claims no release instant. A later pass does not re-declare the delete or re-enter the release.
+func TestWorkflowRunGiveUpAbandonsAnUnreachableWorkspace(t *testing.T) {
+	store := dispatcherDB(t)
+	store.DeliveryUnreachableAfter = 30 * time.Minute
+	runID, wsID := stageProvisionedWorkflowRun(t, store)
+	if _, err := store.Pool.Exec(`UPDATE node_instances SET last_seen_at=clock_timestamp()-interval '2 hours' WHERE workspace_id=$1`, wsID); err != nil {
+		t.Fatalf("age the Node heartbeat: %v", err)
+	}
+	if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil {
+		t.Fatalf("advance with the Node gone: %v", err)
+	}
+
+	release := readWorkflowRelease(t, store, runID)
+	if release.state != workflowReleaseAbandoned || release.reason != workflowReleaseAbandonedReason || release.releasedAt.Valid {
+		t.Fatalf("release = %+v; want abandoned/%s with no release instant", release, workflowReleaseAbandonedReason)
+	}
+	var opState, opError string
+	if err := store.Pool.QueryRow(`SELECT state, COALESCE(error_code,'') FROM operations
+		WHERE workspace_id=$1 AND kind='delete_workspace' ORDER BY created_at DESC, id DESC LIMIT 1`, wsID).
+		Scan(&opState, &opError); err != nil {
+		t.Fatalf("read delete operation: %v", err)
+	}
+	if opState != "failed" || opError != "node_unavailable" {
+		t.Fatalf("delete operation = %s/%q; want failed/node_unavailable", opState, opError)
+	}
+	var deletedAt sql.NullTime
+	var binding sql.NullString
+	if err := store.Pool.QueryRow(`SELECT deleted_at, workflow_run_id FROM workspaces WHERE id=$1`, wsID).Scan(&deletedAt, &binding); err != nil {
+		t.Fatal(err)
+	}
+	if deletedAt.Valid || binding.String != runID {
+		t.Fatalf("the abandoned Workspace must stay live and bound (deleted_at=%v run=%q)", deletedAt, binding.String)
+	}
+
+	// A second pass neither re-declares another delete nor moves the release: the abandoned run is out
+	// of both scans.
+	if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil {
+		t.Fatalf("second advance: %v", err)
+	}
+	if after := readWorkflowRelease(t, store, runID); after.state != workflowReleaseAbandoned || after.reason != release.reason {
+		t.Fatalf("a second pass changed %+v to %+v", release, after)
+	}
+	if n := countWorkflowDeleteOperations(t, store, runID); n != 1 {
+		t.Fatalf("a second pass declared %d delete operations; want only the failed one", n)
+	}
+}
+
+// The give-up window is a bound on silence, not a reaction to any silence: a Node heard from inside
+// the window holds the release, and an unconfigured window fails closed — the deployment that never
+// set the limit loses only the give-up, never a Workspace it was still waiting for.
+func TestWorkflowRunReleaseHoldsWithinTheWindowAndWithoutOne(t *testing.T) {
+	for _, tc := range []struct {
+		name      string
+		window    time.Duration
+		heartbeat time.Duration
+	}{
+		{name: "heartbeat inside the window", window: 30 * time.Minute, heartbeat: 5 * time.Minute},
+		{name: "unconfigured window", window: 0, heartbeat: 2 * time.Hour},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			store := dispatcherDB(t)
+			store.DeliveryUnreachableAfter = tc.window
+			runID, wsID := stageProvisionedWorkflowRun(t, store)
+			if _, err := store.Pool.Exec(`UPDATE node_instances SET last_seen_at=clock_timestamp()-make_interval(secs => $2) WHERE workspace_id=$1`, wsID, tc.heartbeat.Seconds()); err != nil {
+				t.Fatalf("age the heartbeat: %v", err)
+			}
+			if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil {
+				t.Fatalf("advance: %v", err)
+			}
+			release := readWorkflowRelease(t, store, runID)
+			if release.state != workflowReleaseReleasing || release.reason != "" {
+				t.Fatalf("release = %+v; want still releasing", release)
+			}
+			if n := countWorkflowDeleteOperations(t, store, runID); n != 1 {
+				t.Fatalf("%d delete operations; want the declared one still outstanding", n)
+			}
+		})
+	}
+}
+
+// A Workspace whose sandbox is already terminated is not an abandoned release: no Node is left to be
+// unreachable, and the remaining cleanup runs against the Substrate without one. Abandoning it would
+// report a release as failed while a live Controller was still completing it.
+func TestWorkflowRunReleaseDoesNotAbandonATerminatedSandbox(t *testing.T) {
+	store := dispatcherDB(t)
+	store.DeliveryUnreachableAfter = 30 * time.Minute
+	runID, wsID := stageProvisionedWorkflowRun(t, store)
+	if _, err := store.Pool.Exec(`UPDATE sandbox_instances SET terminated_at=clock_timestamp(), observed_state='terminated' WHERE workspace_id=$1`, wsID); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := store.Pool.Exec(`UPDATE node_instances SET ended_at=clock_timestamp(), connection_state='ended', version=version+1 WHERE workspace_id=$1`, wsID); err != nil {
+		t.Fatal(err)
+	}
+	if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil {
+		t.Fatalf("advance: %v", err)
+	}
+	if release := readWorkflowRelease(t, store, runID); release.state != workflowReleaseReleasing {
+		t.Fatalf("release = %+v; want still releasing while the delete can proceed without a Node", release)
+	}
+}
+
+// A Workspace deleted by another path — a project deletion deletes every Workspace inline and never
+// fires the per-run delete hook — is healed to `released` by the reconciliation pass, with the
+// Workspace's own deletion instant as the release fact. The state is not left `releasing` beside a
+// Workspace that no longer exists, and it is not claimed without evidence.
+func TestWorkflowRunReleaseHealsAWorkspaceDeletedByAnotherPath(t *testing.T) {
+	store := dispatcherDB(t)
+	runID, wsID := stageProvisionedWorkflowRun(t, store)
+	if _, err := store.Pool.Exec(`UPDATE workspaces SET observed_state='deleted', deleted_at=clock_timestamp(), version=version+1 WHERE id=$1`, wsID); err != nil {
+		t.Fatal(err)
+	}
+	var deletedAt time.Time
+	if err := store.Pool.QueryRow(`SELECT deleted_at FROM workspaces WHERE id=$1`, wsID).Scan(&deletedAt); err != nil {
+		t.Fatal(err)
+	}
+	if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil {
+		t.Fatalf("advance: %v", err)
+	}
+	release := readWorkflowRelease(t, store, runID)
+	if release.state != workflowReleaseReleased || !release.releasedAt.Valid {
+		t.Fatalf("release = %+v; want released with the Workspace's deletion instant", release)
+	}
+	if !release.releasedAt.Time.Equal(deletedAt) {
+		t.Fatalf("release instant = %v; want the Workspace's own deletion instant %v", release.releasedAt.Time, deletedAt)
+	}
+}
diff --git a/internal/core/workflow_run_session.go b/internal/core/workflow_run_session.go
index 6f286d124..20327b0a6 100644
--- a/internal/core/workflow_run_session.go
+++ b/internal/core/workflow_run_session.go
@@ -266,15 +266,14 @@ func finishWorkflowRun(t *transaction, runID string, version int64, status, runE
 	finishWorkflowRunCleanup(t, runID)
 }
 
-// finishWorkflowRunCleanup deletes the run's unregistered execution work and re-declares the
-// run Workspace's deletion for a terminal run. Safe to call on any terminal run in any pass.
+// finishWorkflowRunCleanup deletes the run's unregistered execution work and records the run
+// Workspace's release: `not_required` when the run never provisioned one, otherwise the `releasing`
+// intent together with the delete declaration (declareWorkflowRunWorkspaceRelease). The release fact
+// and the declaration commit in the transaction that settles the run, so a terminal run is never
+// without either. Safe to call on any terminal run in any pass.
 func finishWorkflowRunCleanup(t *transaction, runID string) {
 	t.exec("DELETE FROM execution_work WHERE workflow_run_id=$1 AND execution_id IS NULL", runID)
-	if w := t.one("SELECT * FROM workspaces WHERE workflow_run_id=$1", runID); w != nil {
-		// Re-declaring an already-declared delete is a no-op; a busy Project leaves the
-		// declaration to a later pass — a terminal run's Workspace is never left live forever.
-		deleteWorkflowRunWorkspace(t, runID)
-	}
+	declareWorkflowRunWorkspaceRelease(t, runID)
 }
 
 // workflowSessionTarget resolves the Workspace's current connected Node the way currentNode does,
diff --git a/internal/core/workflow_scheduler.go b/internal/core/workflow_scheduler.go
index 87252b8ec..c6aefa972 100644
--- a/internal/core/workflow_scheduler.go
+++ b/internal/core/workflow_scheduler.go
@@ -86,6 +86,11 @@ func (s *Store) scanAdvancableWorkflowRuns(ctx context.Context, limit int) ([]st
 // that same window through the same shared check (runWorkspaceNodeUnknown) the IssueRun release
 // paths use. An unconfigured window (zero or negative) fails closed there — a session is held rather
 // than failed — so a deployment that never set the limit loses nothing it was promised.
+//
+// The pass also closes each terminal run's release: the bounded give-up whose Node is gone, and the
+// reconciliation that heals or re-declares the Workspace delete. Together with the delete-completion
+// hook they leave every terminal run with a durable release fact and never report `released` for a
+// Workspace still holding resources.
 func (s *Store) AdvanceWorkflowRunsOnce(ctx context.Context) error {
 	if ctx.Err() != nil {
 		return ctx.Err()
@@ -102,47 +107,13 @@ func (s *Store) AdvanceWorkflowRunsOnce(ctx context.Context) error {
 			_ = err
 		}
 	}
-	return s.redeclareWorkflowRunWorkspaceDeletes(ctx)
-}
-
-// redeclareWorkflowRunWorkspaceDeletes re-declares the Workspace deletion of every terminal run
-// still holding a live Workspace. finishWorkflowRunCleanup declares delete in the pass that
-// settles the run; a busy Project (another operation in flight) returns busy, so the declaration
-// waits for a later pass — this second scan is the retry. Idempotent: deleteWorkflowRunWorkspace
-// treats an already-declared or already-deleted Workspace as done.
-func (s *Store) redeclareWorkflowRunWorkspaceDeletes(ctx context.Context) error {
-	rows, err := s.Pool.QueryContext(ctx, `
-		SELECT r.id FROM workflow_runs r
-		JOIN workspaces w ON w.workflow_run_id=r.id
-		WHERE r.status IN ('failed','cancelled','succeeded') AND r.deleted_at IS NULL
-		  AND w.deleted_at IS NULL
-		  AND NOT EXISTS (SELECT 1 FROM operations o WHERE o.workspace_id=w.id AND o.kind='delete_workspace'
-		   AND o.state IN ('queued','running','retry_wait','blocked','succeeded'))
-		ORDER BY r.created_at, r.id LIMIT $1`, workflowAdvanceBatchSize)
-	if err != nil {
-		return err
-	}
-	defer rows.Close()
-	var ids []string
-	for rows.Next() {
-		var id string
-		if err := rows.Scan(&id); err != nil {
-			return err
-		}
-		ids = append(ids, id)
-	}
-	if err := rows.Err(); err != nil {
+	// The release half of the terminal lifecycle: runs the scheduler no longer scans still owe a
+	// Workspace release fact. The give-up pass runs first so a run whose Node is gone is abandoned in
+	// this tick instead of having one more delete declared for it.
+	if err := s.giveUpWorkflowRunWorkspaceReleases(ctx); err != nil {
 		return err
 	}
-	for _, id := range ids {
-		if _, err := s.transact(ctx, func(t *transaction) Object {
-			finishWorkflowRunCleanup(t, id)
-			return Object{}
-		}); err != nil {
-			_ = err
-		}
-	}
-	return nil
+	return s.reconcileWorkflowRunWorkspaceReleases(ctx)
 }
 
 // workflowNodeView is one node of the frozen snapshot graph the scheduler executes.
diff --git a/internal/core/workflow_scheduler_db_test.go b/internal/core/workflow_scheduler_db_test.go
index 4da3bfb32..f908937ad 100644
--- a/internal/core/workflow_scheduler_db_test.go
+++ b/internal/core/workflow_scheduler_db_test.go
@@ -92,8 +92,8 @@ func TestWorkflowAdvanceSucceedsLinearRun(t *testing.T) {
 	if status != "succeeded" || runErr != "" {
 		t.Fatalf("run settled %q error %q; want succeeded", status, runErr)
 	}
-	if version != 3 { // claim +1, settle +1
-		t.Fatalf("version = %d; want 3", version)
+	if version != 4 { // claim +1, settle +1, release fact +1 (not_required: the graph has no agent node)
+		t.Fatalf("version = %d; want 4", version)
 	}
 	for id, want := range map[string]string{"s": "succeeded", "o": "succeeded"} {
 		state, ok := asWorkflowObject(states[id])
@@ -361,8 +361,8 @@ func TestWorkflowAdvanceFailsACyclicSnapshot(t *testing.T) {
 	if len(states) != 0 {
 		t.Fatalf("a refused cyclic run committed node states: %v", states)
 	}
-	if version != 3 { // claim +1, settle +1, and no node ever committed
-		t.Fatalf("version = %d; want 3", version)
+	if version != 4 { // claim +1, settle +1, release fact +1, and no node ever committed
+		t.Fatalf("version = %d; want 4", version)
 	}
 	// The refusal is terminal: the retry loop never rescans the run, so it cannot drift to succeeded.
 	if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil {
@@ -415,8 +415,8 @@ func TestWorkflowAdvanceIsIdempotent(t *testing.T) {
 	if status != "succeeded" || runErr != "" {
 		t.Fatalf("run settled %q error %q; want succeeded", status, runErr)
 	}
-	if version != 3 {
-		t.Fatalf("version = %d after three passes; want 3 (claim + settle only)", version)
+	if version != 4 {
+		t.Fatalf("version = %d after three passes; want 4 (claim + settle + the release fact, with later passes adding nothing)", version)
 	}
 }