feat(core): record and bound the WorkflowRun Workspace release

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.

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-只拥有消费侧的契约)）。

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"])
+}

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))

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)
 	}