From 156974471da0a6e7d4ca1240dc309fda3e9721ab Mon Sep 17 00:00:00 2001 From: Kyber5323 Date: Fri, 16 Oct 2026 11:25:12 +0800 Subject: [PATCH 1/9] feat(core): bind WorkflowRun identity and workspace lifecycle to the control plane specs#99 D3/D8 step 4 (run identity): the WorkflowRun becomes an execution aggregate the control plane accepts, while session execution stays unwired - the scheduler still fails runs on unwired agent nodes until the executor-slot resolution bridge lands, per the slots decision's landing order (cloud behavior is unchanged). - migration 0035: workflow_runs freezes created_by/project_id; workspaces bind workflow_run_id exclusively with issue_run_id; execution_work carries the same optional ownership under an exactly-one constraint with one partial unregistered index per owner; runtime_controls.maintenance_run_id drops its IssueRun FK, since the live run of either kind is validated where it is granted on. - createWorkflowRun freezes created_by from the authenticated Identity. - createWorkflowRunWorkspace/deleteWorkflowRunWorkspace: the WorkflowRun counterpart of the IssueRun workspace lifecycle, keyed on the frozen identity (project and actor required; idempotent; busy semantics shared). - the six public-visibility filters and the document key strip exclude workflow-run workspaces exactly as they exclude issue-run ones. - enqueue, dispatch and runtime permit accept either run kind; the three settlement hooks (workspace settled/deleted, thread events) accept a WorkflowRun and write nothing - the evidence-driven scheduler reads terminal facts directly. Co-Authored-By: Claude Code --- integration/revision_upgrade_test.go | 5 +- internal/core/agent_run_release.go | 8 + internal/core/agent_run_settle.go | 9 + internal/core/agent_run_thread.go | 11 +- internal/core/execution_work.go | 27 +- .../migrations/0035_workflow_run_identity.sql | 43 +++ internal/core/node_execution.go | 6 +- internal/core/plugins.go | 4 +- internal/core/public.go | 2 +- internal/core/run_workspace.go | 61 +++++ internal/core/runtime_control.go | 11 +- internal/core/runtime_control_delivery.go | 5 + internal/core/runtime_permissions.go | 2 +- internal/core/store.go | 5 +- internal/core/workflow_run.go | 8 +- .../core/workflow_run_identity_db_test.go | 255 ++++++++++++++++++ 16 files changed, 447 insertions(+), 15 deletions(-) create mode 100644 internal/core/migrations/0035_workflow_run_identity.sql create mode 100644 internal/core/workflow_run_identity_db_test.go diff --git a/integration/revision_upgrade_test.go b/integration/revision_upgrade_test.go index 1d69401c9..e666eea28 100644 --- a/integration/revision_upgrade_test.go +++ b/integration/revision_upgrade_test.go @@ -93,7 +93,10 @@ func TestRevisionUpgradePreservesPublishedWorkflowSchema(t *testing.T) { must(t, err) } must(t, tx.Commit()) - const documents = `SELECT jsonb_build_object('workflow',to_jsonb(w),'snapshot',to_jsonb(s),'run',to_jsonb(r))::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` + // The run document strips the identity columns a later migration (0035) added: 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'])::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 a8ad20ddf..6da3ea814 100644 --- a/internal/core/agent_run_release.go +++ b/internal/core/agent_run_release.go @@ -284,6 +284,14 @@ 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. + if t.one("SELECT id FROM workflow_runs WHERE id=$1 AND deleted_at IS NULL", runID) != nil { + return nil + } return fmt.Errorf("runWorkspaceDeleted: issue_run %s not found", runID) } if o.S("executorType") != "agent" { diff --git a/internal/core/agent_run_settle.go b/internal/core/agent_run_settle.go index d3492e46b..2ff57e752 100644 --- a/internal/core/agent_run_settle.go +++ b/internal/core/agent_run_settle.go @@ -35,6 +35,15 @@ func (s *Store) settleRunWorkspace(t *transaction, runID, status 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 create_workspace terminal firing + // under a WorkflowRun reaches this hook with the WorkflowRun's id; the evidence-driven + // scheduler reads terminal operation facts directly, so settlement writes nothing — but + // it must ACCEPT the fact, or the terminal transaction aborts and a workflow run's + // Workspace can never settle. An id that is neither a live issue run nor a live + // workflow run is still an invariant violation, not a replay. + if t.one("SELECT id FROM workflow_runs WHERE id=$1 AND deleted_at IS NULL", runID) != nil { + return nil + } // A create_workspace terminal firing for a run that does not exist is an invariant // violation, not a replay: returning an error aborts the whole transaction. return fmt.Errorf("runWorkspaceSettled: issue_run %s not found", runID) diff --git a/internal/core/agent_run_thread.go b/internal/core/agent_run_thread.go index 2316ea7ec..f3c16fa04 100644 --- a/internal/core/agent_run_thread.go +++ b/internal/core/agent_run_thread.go @@ -90,8 +90,15 @@ func (s *Store) settleThreadEvents(t *transaction, runID, executionID string, ev o := t.one("SELECT * FROM issue_runs WHERE id=$1 AND deleted_at IS NULL", runID) if o == nil { - // No authoritative run to accept the events: an invariant violation, not a replay, so the - // batch must not be acked (unknown run). + // D8: a session may execute under a WorkflowRun. Phase-1 (D7) defers workflow thread + // history — thread_entries is keyed on issue_runs — so the receipts fence (A-side + // node_event_receipts, already recorded in the same transaction) is the durable record + // and nothing is appended here. Accepting the batch is what lets the terminal result + // commit; an id that is neither a live issue run nor a live workflow run is still an + // invariant violation, not a replay, so the batch must not be acked (unknown run). + if t.one("SELECT id FROM workflow_runs WHERE id=$1 AND deleted_at IS NULL", runID) != nil { + return nil + } return fmt.Errorf("settleThreadEvents: issue_run %s not found", runID) } if o.S("executorType") != "agent" { diff --git a/internal/core/execution_work.go b/internal/core/execution_work.go index e5fa44f2b..4420317ac 100644 --- a/internal/core/execution_work.go +++ b/internal/core/execution_work.go @@ -16,7 +16,9 @@ func (s *Store) EnqueueExecutionWork(ctx context.Context, runID, kind string, in func enqueueExecutionWork(t *transaction, runID, kind string, input, target Object, availableAt time.Time) Object { require(validID(runID), 400, "invalid_dispatch") run := t.one("SELECT * FROM issue_runs WHERE id=$1 AND deleted_at IS NULL", runID) - require(run != nil, 404, "not_found") + if run == nil { + return enqueueWorkflowExecutionWork(t, runID, kind, input, target, availableAt) + } require(t.one("SELECT run_id FROM agent_run_residuals WHERE run_id=$1", runID) == nil, 409, "dispatch_conflict") if kind == "deliver_revision" { if t.one("SELECT run_id FROM revision_delivery_skips WHERE run_id=$1", runID) != nil { @@ -54,6 +56,29 @@ func enqueueExecutionWork(t *transaction, runID, kind string, input, target Obje return t.one("SELECT * FROM execution_work WHERE id=$1", id) } +// enqueueWorkflowExecutionWork releases one workflow run's session work for Controllers to claim. +// D8: the item registers against workflow_runs — exactly one unregistered item per run (the +// workflow partial index), kind fixed to agent_session, and no delivery variant (D7 defers +// 成果交付). Repeating the same unregistered input returns the original item; a different input conflicts. +func enqueueWorkflowExecutionWork(t *transaction, runID, kind string, input, target Object, availableAt time.Time) Object { + require(t.one("SELECT id FROM workflow_runs WHERE id=$1 AND deleted_at IS NULL", runID) != nil, 404, "not_found") + require(kind == "agent_session", 400, "invalid_dispatch") + require(input.S("kind") == kind, 400, "invalid_dispatch") + require(target.S("workspaceId") != "" && target.S("sandboxInstanceId") != "" && target.S("nodeId") != "", 400, "invalid_dispatch") + if existing := t.one("SELECT * FROM execution_work WHERE workflow_run_id=$1 AND execution_id IS NULL", runID); existing != nil { + require(existing.S("kind") == kind && jsonText(existing.O("input")) == jsonText(input) && jsonText(existing.O("target")) == jsonText(target), 409, "dispatch_conflict") + return existing + } + id := newID() + if availableAt.IsZero() { + t.exec("INSERT INTO execution_work(id,workflow_run_id,kind,input,target) VALUES($1,$2,$3,$4,$5)", id, runID, kind, jsonText(input), jsonText(target)) + } else { + t.exec("INSERT INTO execution_work(id,workflow_run_id,kind,input,target,available_at) VALUES($1,$2,$3,$4,$5,$6)", id, runID, kind, jsonText(input), jsonText(target), availableAt) + } + t.workSignals = append(t.workSignals, runID) + return t.one("SELECT * FROM execution_work WHERE id=$1", id) +} + // deliveryWorkInput fixes server-owned locations before an execution identity exists. // Copying the caller's map keeps retries from accidentally retaining an earlier attempt's keys. func deliveryWorkInput(t *transaction, run Object, workID string, input Object) Object { diff --git a/internal/core/migrations/0035_workflow_run_identity.sql b/internal/core/migrations/0035_workflow_run_identity.sql new file mode 100644 index 000000000..56ab30374 --- /dev/null +++ b/internal/core/migrations/0035_workflow_run_identity.sql @@ -0,0 +1,43 @@ +-- WorkflowRun identity and bindings (specs decisions/cloud/workflow +-- 0-cloud-side-scheduler-owns-workflow-execution.md D3/D8, landing order step 4 "运行身份"; +-- semantics inherited from decisions/cloud/workflow/20261015-executor-slots-replace-agent-model-ids.md +-- and decisions/cloud/model-access/20261015-executor-slot-mappings-and-per-slot-run-bindings.md). +-- +-- The WorkflowRun is the execution aggregate for workflow invocations, so it needs the frozen identity +-- the run Workspace lifecycle reads (who invoked it, which project its Workspace provisions in), and +-- the control plane's work registry must accept a WorkflowRun as an alternative work owner without a +-- new Controller contract (D8: 判别列 + 恰好一个约束). Session execution itself stays unwired until +-- the executor-slot resolution bridge lands per the slots decision's 落地顺序 — until then Cloud does +-- not execute workflows, and behavior is unchanged. + +-- 1. Frozen run identity. Both nullable: rows written before this migration carry no recoverable +-- actor, and project_id is filled by the invocation path that knows a project (the @-path; the +-- editor's run button does not). New runs freeze both at creation. +ALTER TABLE workflow_runs ADD COLUMN created_by uuid REFERENCES users(id); +ALTER TABLE workflow_runs ADD COLUMN project_id uuid REFERENCES projects(id); + +-- 2. A run Workspace can belong to a WorkflowRun. The binding is exclusive with the IssueRun one; +-- workspaces without either stay the public kind. +ALTER TABLE workspaces ADD COLUMN workflow_run_id uuid UNIQUE REFERENCES workflow_runs(id); +ALTER TABLE workspaces ADD CONSTRAINT workspace_run_binding_exclusive + CHECK (issue_run_id IS NULL OR workflow_run_id IS NULL); + +-- 3. Execution work carries the same optional ownership: a work item names exactly one run, of either +-- kind. The single-column uniqueness splits into one partial index per owner, so a run of each kind +-- keeps at most one unregistered item and the two ownerships never collide. +ALTER TABLE execution_work ALTER COLUMN run_id DROP NOT NULL; +ALTER TABLE execution_work ADD COLUMN workflow_run_id uuid REFERENCES workflow_runs(id); +ALTER TABLE execution_work ADD CONSTRAINT execution_work_owner_exclusive + CHECK ((run_id IS NULL) <> (workflow_run_id IS NULL)); +DROP INDEX one_unregistered_execution_work; +CREATE UNIQUE INDEX one_unregistered_execution_work ON execution_work(run_id) + WHERE execution_id IS NULL AND workflow_run_id IS NULL; +CREATE UNIQUE INDEX one_unregistered_workflow_execution_work ON execution_work(workflow_run_id) + WHERE execution_id IS NULL AND run_id IS NULL; + +-- 4. Runtime control's run reservation names the run of either kind that owns the Workspace's next +-- control epoch. The IssueRun FK is dropped, not reworked: a polymorphic run reference cannot be a +-- foreign key, so the liveness check moves to the code that grants on it (runtime_permit re-reads +-- the live run before a Node binds, reconciliation clears the reservation), and 0026's own CHECKs — +-- exactly one of operation/run id while maintenance is reserved — are unchanged. +ALTER TABLE runtime_controls DROP CONSTRAINT runtime_controls_maintenance_run_id_fkey; diff --git a/internal/core/node_execution.go b/internal/core/node_execution.go index ae13c8a1b..47cac23cb 100644 --- a/internal/core/node_execution.go +++ b/internal/core/node_execution.go @@ -43,7 +43,9 @@ func dispatchPluginExecution(t *transaction, r *ControlRequest, operation, execu func dispatchRunExecution(t *transaction, r *ControlRequest, runID, execution, node string, input Object) Object { require(t.one("SELECT run_id FROM agent_run_residuals WHERE run_id=$1", runID) == nil, 409, "dispatch_conflict") - work := t.one("SELECT * FROM execution_work WHERE run_id=$1 AND execution_id IS NULL AND kind=$2", runID, input.S("kind")) + // The work item names its owner — IssueRun or WorkflowRun (D8); operation_id is that run's id + // either way, so the ownership check re-reads whichever column the row carries. + work := t.one("SELECT * FROM execution_work WHERE (run_id=$1 OR workflow_run_id=$1) AND execution_id IS NULL AND kind=$2", runID, input.S("kind")) require(work != nil, 409, "dispatch_conflict") require(jsonText(work.O("input")) == jsonText(input), 409, "dispatch_conflict") target := work.O("target") @@ -51,7 +53,7 @@ func dispatchRunExecution(t *transaction, r *ControlRequest, runID, execution, n wid := target.S("workspaceId") requireNoForceStop(t, wid) w := t.one("SELECT * FROM workspaces WHERE id=$1", wid) - require(w.S("issueRunId") == runID && w.S("observedState") == "ready" && w.B("admissionOpen"), 409, "dispatch_conflict") + require((w.S("issueRunId") == runID || w.S("workflowRunId") == runID) && w.S("observedState") == "ready" && w.B("admissionOpen"), 409, "dispatch_conflict") c := runtimeControl(t, wid) require(c.S("state") == "maintenance" && c.S("maintenanceRunId") == runID, 409, "runtime_control_required") current := currentNode(t, wid) diff --git a/internal/core/plugins.go b/internal/core/plugins.go index ea41cc9d1..1ee93e221 100644 --- a/internal/core/plugins.go +++ b/internal/core/plugins.go @@ -141,7 +141,7 @@ func pluginIdentity(id string) (namespace, identifier string, ok bool) { func livePluginWorkspaces(t *transaction, spaceID string) []Object { // Run Workspaces install plugins through their own create_workspace step. They are not fan-out // targets: a user install must not queue an operation against a Workspace the user cannot see. - return t.list(`SELECT w.* FROM workspaces w JOIN projects p ON p.id=w.project_id WHERE p.space_id=$1 AND w.deleted_at IS NULL AND w.issue_run_id IS NULL ORDER BY w.id`, spaceID) + return t.list(`SELECT w.* FROM workspaces w JOIN projects p ON p.id=w.project_id WHERE p.space_id=$1 AND w.deleted_at IS NULL AND w.issue_run_id IS NULL AND w.workflow_run_id IS NULL ORDER BY w.id`, spaceID) } // installSpacePlugin records an administrator's pinned selection. Execution waits durably; @@ -271,7 +271,7 @@ func pluginInstanceWriteback(t *transaction, op Object, pluginID, state, version } // Run Workspace instances are execution facts for that run. They do not change the Space aggregate, // so one disposable install failure cannot retire an Agent the Space still has selected. - states := t.list("SELECT observed_state FROM workspace_plugin_instances wi WHERE source_namespace=$1 AND identifier=$2 AND workspace_id IN (SELECT w.id FROM workspaces w JOIN projects p ON p.id=w.project_id WHERE p.space_id=$3 AND w.deleted_at IS NULL AND w.issue_run_id IS NULL)", namespace, identifier, row.S("spaceId")) + states := t.list("SELECT observed_state FROM workspace_plugin_instances wi WHERE source_namespace=$1 AND identifier=$2 AND workspace_id IN (SELECT w.id FROM workspaces w JOIN projects p ON p.id=w.project_id WHERE p.space_id=$3 AND w.deleted_at IS NULL AND w.issue_run_id IS NULL AND w.workflow_run_id IS NULL)", namespace, identifier, row.S("spaceId")) aggregate := "installed" if desired := t.one("SELECT desired_state,tenant_id FROM space_plugins WHERE space_id=$1 AND source_namespace=$2 AND identifier=$3", row.S("spaceId"), namespace, identifier); desired != nil { desiredState := desired.S("desiredState") diff --git a/internal/core/public.go b/internal/core/public.go index 688abad2e..01a4c51d8 100644 --- a/internal/core/public.go +++ b/internal/core/public.go @@ -492,7 +492,7 @@ func readPublic(t *transaction, r *PublicRequest, uid string) Object { case strings.HasSuffix(r.Path, "/issues"): return issueList(t, r) case strings.HasSuffix(r.Path, "/resource-status"): - return page(t, "SELECT w.id,w.project_id,w.owner_user_id,w.kind,w.desired_state,w.observed_state,w.runtime_generation,w.version FROM workspaces w WHERE w.tenant_id=$1 AND w.deleted_at IS NULL AND w.issue_run_id IS NULL", []any{r.TenantID}, "w.id", r) + return page(t, "SELECT w.id,w.project_id,w.owner_user_id,w.kind,w.desired_state,w.observed_state,w.runtime_generation,w.version FROM workspaces w WHERE w.tenant_id=$1 AND w.deleted_at IS NULL AND w.issue_run_id IS NULL AND w.workflow_run_id IS NULL", []any{r.TenantID}, "w.id", r) case strings.Contains(r.Path, "/workflows"): // A snapshot-scoped form read is checked before the bare snapshot read, because both carry a // snapshot id and only the path distinguishes projecting a version from describing it. diff --git a/internal/core/run_workspace.go b/internal/core/run_workspace.go index a28f33d72..e3e7ae874 100644 --- a/internal/core/run_workspace.go +++ b/internal/core/run_workspace.go @@ -46,6 +46,67 @@ func runTriggerActor(t *transaction, run Object) string { return actors[0].S("actorId") } +// createWorkflowRunWorkspace is the WorkflowRun counterpart of createRunWorkspace: it idempotently +// creates the isolated Workspace and its create_workspace operation for one workflow run. The run's +// frozen identity is the whole basis (specs#99 D3): project_id names the Project the Workspace +// provisions in, and created_by is the actor the Workspace and its operation record. A run without a +// frozen project has no place to provision in — the caller fails it honestly rather than guessing a +// Project; an unbound run is the editor path today, and the invocation path (#71) freezes one. +func createWorkflowRunWorkspace(t *transaction, runID string) Object { + run := t.one("SELECT * FROM workflow_runs WHERE id=$1 AND deleted_at IS NULL", runID) + require(run != nil, 404, "not_found") + require(run.S("projectId") != "", 409, "workflow_run_project_required") + require(run.S("createdBy") != "", 409, "workflow_run_actor_required") + if existing := t.one("SELECT * FROM workspaces WHERE workflow_run_id=$1", runID); existing != nil { + op := t.one("SELECT * FROM operations WHERE workspace_id=$1 AND kind='create_workspace' ORDER BY created_at,id LIMIT 1", existing.S("id")) + return Object{"busy": false, "workspace": existing, "operation": op} + } + p := t.one("SELECT * FROM projects WHERE id=$1 AND tenant_id=$2 AND deleted_at IS NULL", run.S("projectId"), run.S("tenantId")) + require(p != nil, 404, "not_found") + if p.S("lifecycle") != "active" || t.one("SELECT id FROM operations WHERE project_id=$1 AND state IN ('queued','running','retry_wait','blocked')", p.S("id")) != nil { + return Object{"busy": true} + } + wid := newID() + ref := p.S("defaultBranch") + require(ref != "" && ref != "HEAD", 400, "default_branch_required") + insertWorkspace(t, run.S("tenantId"), p.S("ownerUserId"), run.S("createdBy"), p.S("id"), wid, "isolated", ref, "Workflow run") + t.exec("UPDATE workspaces SET workflow_run_id=$2 WHERE id=$1", wid, runID) + op := newOperation(t, &PublicRequest{TenantID: run.S("tenantId"), Key: "workflow-run-" + runID + "-create"}, run.S("createdBy"), p.S("id"), wid, "create_workspace", "sandbox", requestHash("workflow-run-workspace", runID, Object{"kind": "create"}), Object{}) + return Object{"busy": false, "workspace": t.one("SELECT * FROM workspaces WHERE id=$1", wid), "operation": op} +} + +// deleteWorkflowRunWorkspace is the WorkflowRun counterpart of deleteRunWorkspace: it idempotently +// queues deletion of the run Workspace bound to a workflow run, sharing the busy/quiesce semantics and +// the pre-declaration snapshot. The snapshot omits workflow_run_id for the same reason its IssueRun +// counterpart omits issue_run_id: the operation claim response deliberately carries no run binding, +// and the snapshot is what a refused quiesce restores from. +func deleteWorkflowRunWorkspace(t *transaction, runID string) Object { + w := t.one("SELECT * FROM workspaces WHERE workflow_run_id=$1", runID) + require(w != nil, 404, "not_found") + if w["deletedAt"] != nil { + return Object{"busy": false, "workspace": w} + } + if existing := t.one("SELECT * FROM operations WHERE workspace_id=$1 AND kind='delete_workspace' AND state IN ('queued','running','retry_wait','blocked','succeeded') ORDER BY created_at DESC,id DESC LIMIT 1", w.S("id")); existing != nil { + return Object{"busy": false, "workspace": w, "operation": existing} + } + if t.one("SELECT id FROM operations WHERE project_id=$1 AND state IN ('queued','running','retry_wait','blocked')", w.S("projectId")) != nil { + return Object{"busy": true} + } + require(t.one("SELECT run_id FROM agent_run_residuals WHERE workspace_id=$1 AND cleared_at IS NULL", w.S("id")) == nil, 409, "resource_in_use") + closeAdmission(t, w, "deleted") + actor := w.S("creatorUserId") + snapshot := Object{} + for k, v := range w { + if k != "workflowRunId" { + snapshot[k] = v + } + } + req := Object{"previous": Object{w.S("id"): snapshot}} + op := newOperation(t, &PublicRequest{TenantID: w.S("tenantId"), Key: "workflow-run-" + runID + "-delete"}, actor, w.S("projectId"), w.S("id"), "delete_workspace", "quiesce", requestHash("workflow-run-workspace-delete", runID, Object{}), req) + reserveRuntimeMaintenance(t, w.S("id"), op.S("id")) + return Object{"busy": false, "workspace": t.one("SELECT * FROM workspaces WHERE id=$1", w.S("id")), "operation": op} +} + func deleteRunWorkspace(t *transaction, runID string) Object { w := t.one("SELECT * FROM workspaces WHERE issue_run_id=$1", runID) require(w != nil, 404, "not_found") diff --git a/internal/core/runtime_control.go b/internal/core/runtime_control.go index 42d7b3693..213070231 100644 --- a/internal/core/runtime_control.go +++ b/internal/core/runtime_control.go @@ -126,10 +126,15 @@ func reserveRuntimeMaintenance(t *transaction, wid, oid string) { // finishRuntimeMaintenance requests closure after the fixed intent finishes; it cannot grant // another session on a still-running Node before that Node confirms its durable responsibilities. func finishRuntimeMaintenance(t *transaction, oid string) { - if run := t.one("SELECT w.id,w.issue_run_id FROM workspaces w JOIN operations o ON o.workspace_id=w.id WHERE o.id=$1 AND o.kind='create_workspace' AND o.state='succeeded' AND w.issue_run_id IS NOT NULL", oid); run != nil { + if run := t.one("SELECT w.id,w.issue_run_id,w.workflow_run_id FROM workspaces w JOIN operations o ON o.workspace_id=w.id WHERE o.id=$1 AND o.kind='create_workspace' AND o.state='succeeded' AND (w.issue_run_id IS NOT NULL OR w.workflow_run_id IS NOT NULL)", oid); run != nil { // Initialization handed back terminal evidence for clone/plugins. The next control epoch - // belongs to this run; Node must acknowledge it before any session or delivery is permitted. - t.exec("UPDATE runtime_controls SET state='maintenance',maintenance_operation_id=NULL,maintenance_run_id=$2,control_epoch=control_epoch+1,binding_confirmed=false,input_closed=false,version=version+1,updated_at=clock_timestamp() WHERE workspace_id=$1", run.S("id"), run.S("issueRunId")) + // belongs to this run — IssueRun or WorkflowRun, whichever owns the Workspace (D8); Node + // must acknowledge it before any session or delivery is permitted. + owner := run.S("issueRunId") + if owner == "" { + owner = run.S("workflowRunId") + } + t.exec("UPDATE runtime_controls SET state='maintenance',maintenance_operation_id=NULL,maintenance_run_id=$2,control_epoch=control_epoch+1,binding_confirmed=false,input_closed=false,version=version+1,updated_at=clock_timestamp() WHERE workspace_id=$1", run.S("id"), owner) auditRuntimeControl(t, runtimeControl(t, run.S("id")), "run_reserved", "") return } diff --git a/internal/core/runtime_control_delivery.go b/internal/core/runtime_control_delivery.go index 76d38c45a..d9e0ff6c9 100644 --- a/internal/core/runtime_control_delivery.go +++ b/internal/core/runtime_control_delivery.go @@ -47,7 +47,12 @@ func runtimeControlCommand(t *transaction, r *ControlRequest) Object { require(e != nil && (e.S("kind") == "agent_session" || e.S("kind") == "deliver_revision") && e.S("operationId") == c.S("maintenanceRunId") && e.S("nodeId") == n.S("nodeId") && e["result"] == nil && e["terminatedByForceStopId"] == nil, 409, "stale_execution") work := t.one("SELECT target FROM execution_work WHERE id=$1", e.S("workId")) require(work.O("target").S("sandboxInstanceId") == n.S("sandboxInstanceId"), 409, "stale_node") + // The maintenance holder is a live run of either kind (D8): the IssueRun join first, the + // WorkflowRun row when the Workspace belongs to a workflow invocation. run := t.one("SELECT r.id FROM issue_runs r JOIN issues i ON i.id=r.issue_id WHERE r.id=$1 AND r.deleted_at IS NULL AND i.deleted_at IS NULL", c.S("maintenanceRunId")) + if run == nil { + run = t.one("SELECT id FROM workflow_runs WHERE id=$1 AND deleted_at IS NULL", c.S("maintenanceRunId")) + } w := t.one("SELECT * FROM workspaces WHERE id=$1", c.S("workspaceId")) require(run != nil && runtimeUsable(t, w, w.S("creatorUserId")), 403, "runtime_use_forbidden") requireRepositoryAccess(t, w.S("projectId")) diff --git a/internal/core/runtime_permissions.go b/internal/core/runtime_permissions.go index fb4cd4e82..cb1b10efa 100644 --- a/internal/core/runtime_permissions.go +++ b/internal/core/runtime_permissions.go @@ -20,7 +20,7 @@ func runtimeOverview(w Object, usable bool) Object { // runtimeListing keeps shared safe summaries while exposing clone/task content only to its users. func runtimeListing(t *transaction, r *PublicRequest, uid, pid string) Object { - out := page(t, "SELECT w.*,wt.branch_name,task.title FROM workspaces w LEFT JOIN workspace_worktrees wt ON wt.workspace_id=w.id LEFT JOIN tasks task ON task.workspace_id=w.id WHERE w.project_id=$1 AND w.tenant_id=$2 AND w.deleted_at IS NULL AND w.issue_run_id IS NULL", []any{pid, r.TenantID}, "w.id", r) + out := page(t, "SELECT w.*,wt.branch_name,task.title FROM workspaces w LEFT JOIN workspace_worktrees wt ON wt.workspace_id=w.id LEFT JOIN tasks task ON task.workspace_id=w.id WHERE w.project_id=$1 AND w.tenant_id=$2 AND w.deleted_at IS NULL AND w.issue_run_id IS NULL AND w.workflow_run_id IS NULL", []any{pid, r.TenantID}, "w.id", r) items, _ := out["items"].([]Object) for i, w := range items { usable := runtimeUsable(t, w, uid) diff --git a/internal/core/store.go b/internal/core/store.go index f6dc87d3a..1452d77eb 100644 --- a/internal/core/store.go +++ b/internal/core/store.go @@ -278,6 +278,9 @@ func (t *transaction) list(q string, args ...any) []Object { if o["issueRunId"] == nil { delete(o, "issueRunId") } + if o["workflowRunId"] == nil { + delete(o, "workflowRunId") + } out = append(out, o) } if e = rows.Err(); e != nil { @@ -602,7 +605,7 @@ func project(t *transaction, tid, uid, pid string) Object { // workspace loads authorized runtime content. Safe summaries use runtimeOverview instead. func workspace(t *transaction, tid, uid, wid string, admin bool) Object { require(validID(wid), 404, "not_found") - w := t.one("SELECT * FROM workspaces WHERE id=$1 AND tenant_id=$2 AND deleted_at IS NULL AND issue_run_id IS NULL", wid, tid) + w := t.one("SELECT * FROM workspaces WHERE id=$1 AND tenant_id=$2 AND deleted_at IS NULL AND issue_run_id IS NULL AND workflow_run_id IS NULL", wid, tid) require(w != nil, 404, "not_found") require(runtimeUsable(t, w, uid), 403, "runtime_use_forbidden") if admin { diff --git a/internal/core/workflow_run.go b/internal/core/workflow_run.go index bb9937d41..dc6acddb2 100644 --- a/internal/core/workflow_run.go +++ b/internal/core/workflow_run.go @@ -91,7 +91,13 @@ func createWorkflowRun(t *transaction, r *PublicRequest, uid string) Object { input := r.Body.O("input") requireWorkflowRunInput(s.O("graph"), input) id := newID() - t.exec("INSERT INTO workflow_runs(id,tenant_id,workflow_id,snapshot_id,name,status,input) VALUES($1,$2,$3,$4,$5,$6,$7)", id, r.TenantID, r.WorkflowID, s.S("id"), name, "pending", jsonText(input)) + // The invocation identity freezes here (D3): created_by is the aggregate's actor for the whole + // run — the run Workspace lifecycle reads it instead of re-resolving who pressed Run. The + // authenticated Identity, not a /users path parameter, is who that is. project_id stays unset on + // this path; the invocation path that knows a project (@, #71) freezes it too. + require(r.Identity != nil, 401, "user_credential_required") + createdBy := identityWithAlias(t, r.Identity).S("id") + t.exec("INSERT INTO workflow_runs(id,tenant_id,workflow_id,snapshot_id,name,status,input,created_by) VALUES($1,$2,$3,$4,$5,$6,$7,$8)", id, r.TenantID, r.WorkflowID, s.S("id"), name, "pending", jsonText(input), createdBy) if t.simulator != nil { nodeStates, rounds, status := t.simulator.SimulateWorkflowRun(s.O("graph"), input) t.exec("UPDATE workflow_runs SET node_states=$1,rounds=$2,status=$3,started_at=now(),finished_at=now(),version=version+1,updated_at=now() WHERE id=$4", jsonText(nodeStates), jsonText(rounds), status, id) diff --git a/internal/core/workflow_run_identity_db_test.go b/internal/core/workflow_run_identity_db_test.go new file mode 100644 index 000000000..ab86c3b87 --- /dev/null +++ b/internal/core/workflow_run_identity_db_test.go @@ -0,0 +1,255 @@ +package core + +// DB tests for the WorkflowRun execution identity (migration 0035, specs#99 D3/D8): a run's frozen +// identity drives the workspace lifecycle, the workspace binds to workflow_runs, and the control +// plane's seams — enqueue, dispatch, runtime permit and the settlement hooks — accept a WorkflowRun +// as the run of either kind. Session execution stays unwired (the scheduler still fails runs on +// unwired agent nodes); these tests exercise the seams the execution wiring lands on. + +import ( + "context" + "database/sql" + "strings" + "testing" + "time" + + "github.com/google/uuid" +) + +// seedWorkflowIdentityScene loads one tenant with an active project, a workflow, a frozen snapshot and +// a pending run carrying the frozen identity columns (created_by, project_id). withProject=false +// stages the editor-path run as it exists today — identity unbound — for the refusal test. +func seedWorkflowIdentityScene(t *testing.T, store *Store, withProject bool) (runID, project, actor string) { + t.Helper() + uid, tid := uuid.NewString(), uuid.NewString() + wid, sid, runID := uuid.NewString(), uuid.NewString(), uuid.NewString() + project, space := uuid.NewString(), uuid.NewString() + var projectID any + if withProject { + projectID = project + } + tx, err := store.Pool.Begin() + if err != nil { + t.Fatalf("seed begin: %v", err) + } + exec := func(name, q string, args ...any) { + if _, err := tx.Exec(q, args...); err != nil { + t.Fatalf("seed %s: %v", name, err) + } + } + exec("user", `INSERT INTO users(id,display_name,status) VALUES($1,'wfi','active')`, uid) + exec("tenant", `INSERT INTO tenants(id,name,status) VALUES($1,'wfi','active')`, tid) + exec("membership", `INSERT INTO tenant_memberships(tenant_id,user_id,role,status) VALUES($1,$2,'admin','active')`, tid, uid) + exec("space", `INSERT INTO collab_workspaces(id,tenant_id,name,slug,created_by) VALUES($1,$2,'wfi','wfi-'||left($4::text,8),$3)`, space, tid, uid, project) + exec("project", `INSERT INTO projects(id,tenant_id,owner_user_id,space_id,name,repository_url,default_branch,lifecycle) VALUES($1,$2,$3,$4,'wfi','https://example.invalid/wfi.git','main','active')`, project, tid, uid, space) + // A live project must carry exactly one main workspace (deferred project_main trigger). + exec("main-workspace", `INSERT INTO workspaces(id,tenant_id,owner_user_id,creator_user_id,creator_evidence,project_id,kind,desired_state,observed_state,runtime_generation,requested_ref) VALUES($1,$2,$3,$3,'verified_request',$4,'main','running','ready',1,'HEAD')`, uuid.NewString(), tid, uid, project) + exec("workflow", `INSERT INTO workflows(id,tenant_id,name,graph) VALUES($1,$2,'wfi','{"nodes":[],"edges":[]}'::jsonb)`, wid, tid) + exec("snapshot", `INSERT INTO workflow_snapshots(id,tenant_id,workflow_id,version,name,graph) VALUES($1,$2,$3,1,'v1','{"nodes":[],"edges":[]}'::jsonb)`, sid, tid, wid) + exec("run", `INSERT INTO workflow_runs(id,tenant_id,workflow_id,snapshot_id,name,status,input,created_by,project_id) VALUES($1,$2,$3,$4,'disp','pending','{}'::jsonb,$5,$6)`, runID, tid, wid, sid, uid, projectID) + if err := tx.Commit(); err != nil { + t.Fatalf("seed commit: %v", err) + } + return runID, project, uid +} + +// The lifecycle a wired run will drive: the frozen identity provisions the Workspace (created_by is +// the creator), the create_workspace terminal hands runtime control to the run, and the declared +// delete settles — every step accepting the WorkflowRun as the Workspace's owner. +func TestWorkflowRunWorkspaceLifecycle(t *testing.T) { + store := dispatcherDB(t) + runID, _, actor := seedWorkflowIdentityScene(t, store, true) + + var wsID, createOp string + if _, err := store.transact(context.Background(), func(tx *transaction) Object { + created := createWorkflowRunWorkspace(tx, runID) + if created.B("busy") { + t.Fatal("the project must be idle") + } + wsID, createOp = created.O("workspace").S("id"), created.O("operation").S("id") + return Object{} + }); err != nil { + t.Fatalf("create workflow run workspace: %v", err) + } + var creator, boundRun, kind string + if err := store.Pool.QueryRow(`SELECT creator_user_id,COALESCE(workflow_run_id::text,''),kind FROM workspaces WHERE id=$1`, wsID). + Scan(&creator, &boundRun, &kind); err != nil { + t.Fatalf("read workspace: %v", err) + } + if creator != actor || boundRun != runID || kind != "isolated" { + t.Fatalf("workspace creator=%q run=%q kind=%q; want the frozen actor, the owning run, isolated", creator, boundRun, kind) + } + + // The create_workspace terminal step the control plane itself drives: the operation succeeds, + // runtime control is handed back reserved for this run, and the settlement hook accepts a + // workflow-owned workspace — writing no issue-run facts. + if _, err := store.transact(context.Background(), func(tx *transaction) Object { + settleCreateWorkspaceOperation(t, tx, createOp) + runWorkspaceSettled(tx, runID, "ready") + return Object{} + }); err != nil { + t.Fatalf("settle create_workspace terminal: %v", err) + } + var state, maintenanceRun string + if err := store.Pool.QueryRow(`SELECT state,COALESCE(maintenance_run_id::text,'') FROM runtime_controls WHERE workspace_id=$1`, wsID). + Scan(&state, &maintenanceRun); err != nil { + t.Fatalf("read runtime control: %v", err) + } + if state != "maintenance" || maintenanceRun != runID { + t.Fatalf("runtime control state=%q run=%q; want maintenance reserved for the workflow run", state, maintenanceRun) + } + + // 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 { + declared := deleteWorkflowRunWorkspace(tx, runID) + if declared.B("busy") { + t.Fatal("declare delete: the project must be idle") + } + deleteOp := declared.O("operation").S("id") + tx.exec(`UPDATE operations SET state='succeeded', step='done', version=version+1, updated_at=now() WHERE id=$1`, deleteOp) + tx.exec(`UPDATE workspaces SET deleted_at=now(), observed_state='deleted', version=version+1 WHERE id=$1`, wsID) + runWorkspaceDeleted(tx, runID) + finishRuntimeMaintenance(tx, deleteOp) + return Object{} + }); err != nil { + t.Fatalf("settle delete_workspace terminal: %v", err) + } + var deleted sql.NullString + if err := store.Pool.QueryRow(`SELECT observed_state FROM workspaces WHERE id=$1`, wsID).Scan(&deleted); err != nil { + t.Fatalf("read deleted workspace: %v", err) + } + if deleted.String != "deleted" { + t.Fatalf("workspace observed_state=%q; want deleted", deleted.String) + } + if err := store.Pool.QueryRow(`SELECT state FROM runtime_controls WHERE workspace_id=$1`, wsID).Scan(&state); err != nil { + t.Fatalf("read runtime control after delete: %v", err) + } + if state != "idle" { + t.Fatalf("runtime control state=%q after the delete settled; want idle", state) + } +} + +// A run whose identity never learned a project — the editor path today — has no place to provision a +// Workspace in, and says so instead of guessing one. +func TestWorkflowRunWorkspaceRequiresProject(t *testing.T) { + store := dispatcherDB(t) + runID, _, _ := seedWorkflowIdentityScene(t, store, false) + _, err := store.transact(context.Background(), func(tx *transaction) Object { + createWorkflowRunWorkspace(tx, runID) + return Object{} + }) + if ErrorCode(err).Code != "workflow_run_project_required" { + t.Fatalf("error = %v; want workflow_run_project_required", err) + } +} + +// Execution work names exactly one run of either kind: the discriminator constraint rejects a row +// carrying both owners and a row carrying neither, while the IssueRun-only shape keeps inserting. +func TestExecutionWorkOwnerExclusivity(t *testing.T) { + store := dispatcherDB(t) + seed := seedDispatchScene(t, store.Pool, false) + wfRun, _, _ := seedWorkflowIdentityScene(t, store, true) + doc := `'{"kind":"agent_session"}'::jsonb` + if _, err := store.Pool.Exec(`INSERT INTO execution_work(id,run_id,workflow_run_id,kind,input,target) VALUES($1,$2,$3,'agent_session',`+doc+`,`+doc+`)`, uuid.NewString(), seed.run, wfRun); err == nil || !strings.Contains(err.Error(), "execution_work_owner_exclusive") { + t.Fatalf("both owners: err = %v; want execution_work_owner_exclusive", err) + } + if _, err := store.Pool.Exec(`INSERT INTO execution_work(id,kind,input,target) VALUES($1,'agent_session',`+doc+`,`+doc+`)`, uuid.NewString()); err == nil || !strings.Contains(err.Error(), "execution_work_owner_exclusive") { + t.Fatalf("neither owner: err = %v; want execution_work_owner_exclusive", err) + } + if _, err := store.Pool.Exec(`INSERT INTO execution_work(id,run_id,kind,input,target) VALUES($1,$2,'agent_session',`+doc+`,`+doc+`)`, uuid.NewString(), seed.run); err != nil { + t.Fatalf("issue-run-only work must still insert: %v", err) + } +} + +// The seam a wired run drives end to end: work registers against the workflow run, repeats of the +// same input are idempotent, the Node's dispatch is accepted under workflow ownership, and the run +// reserved on runtime control is permitted to bind the Node. +func TestWorkflowExecutionWorkDispatchSeam(t *testing.T) { + store := dispatcherDB(t) + runID, _, actor := seedWorkflowIdentityScene(t, store, true) + + var wsID string + if _, err := store.transact(context.Background(), func(tx *transaction) Object { + created := createWorkflowRunWorkspace(tx, runID) + if created.B("busy") { + t.Fatal("the project must be idle") + } + wsID = created.O("workspace").S("id") + settleCreateWorkspaceOperation(t, tx, created.O("operation").S("id")) + runWorkspaceSettled(tx, runID, "ready") + tx.exec(`UPDATE workspaces SET observed_state='ready', admission_open=true, version=version+1 WHERE id=$1`, wsID) + return Object{} + }); err != nil { + t.Fatalf("stage ready workspace: %v", err) + } + sandbox, nodeService := bindConnectedNode(t, store, wsID) + var nodeRow string + if err := store.Pool.QueryRow(`SELECT id FROM node_instances WHERE workspace_id=$1 AND ended_at IS NULL`, wsID).Scan(&nodeRow); err != nil { + t.Fatalf("read node row: %v", err) + } + + input := Object{"kind": "agent_session"} + target := Object{"workspaceId": wsID, "sandboxInstanceId": sandbox, "nodeId": nodeService} + work, err := store.EnqueueExecutionWork(context.Background(), runID, "agent_session", input, target, time.Time{}) + if err != nil { + t.Fatalf("enqueue workflow work: %v", err) + } + if work.S("workflowRunId") != runID || work.S("runId") != "" { + t.Fatalf("work %v must register against the workflow run alone", work) + } + again, err := store.EnqueueExecutionWork(context.Background(), runID, "agent_session", input, target, time.Time{}) + if err != nil || again.S("id") != work.S("id") { + t.Fatalf("same unregistered input returned %v (err %v); want the original item", again, err) + } + _, err = store.EnqueueExecutionWork(context.Background(), runID, "agent_session", input, Object{"workspaceId": wsID, "sandboxInstanceId": sandbox, "nodeId": "node-elsewhere"}, time.Time{}) + if ErrorCode(err).Code != "dispatch_conflict" { + t.Fatalf("conflicting target: err = %v; want dispatch_conflict", err) + } + + execution := "exec-" + runID[:8] + if _, err := store.transact(context.Background(), func(tx *transaction) Object { + nodeDispatch(tx, &ControlRequest{Body: Object{ + "operationId": runID, "executionId": execution, "nodeId": nodeService, + "input": Object{"kind": "agent_session"}, "epoch": int64(1), + }}) + return Object{} + }); err != nil { + t.Fatalf("dispatch under workflow ownership: %v", err) + } + var boundExecution string + if err := store.Pool.QueryRow(`SELECT COALESCE(execution_id::text,'') FROM execution_work WHERE id=$1`, work.S("id")).Scan(&boundExecution); err != nil { + t.Fatalf("read work: %v", err) + } + if boundExecution != execution { + t.Fatalf("work registered execution %q; want %q", boundExecution, execution) + } + + // The Node acks its binding and takes the permit: the maintenance holder is the workflow run, and + // the binding's actor is the run's frozen creator. + var epoch, controlVersion int64 + if err := store.Pool.QueryRow(`SELECT control_epoch,version FROM runtime_controls WHERE workspace_id=$1`, wsID).Scan(&epoch, &controlVersion); err != nil { + t.Fatalf("read control epoch: %v", err) + } + if _, err := store.transact(context.Background(), func(tx *transaction) Object { + runtimeControlCommand(tx, &ControlRequest{Action: "runtime_ack", Body: Object{ + "workspaceId": wsID, "controlEpoch": epoch, "controlVersion": controlVersion, "nodeInstanceId": nodeRow, + "inputClosed": false, "unfinishedExecutionIds": []any{}, + }}) + return Object{} + }); err != nil { + t.Fatalf("runtime ack: %v", err) + } + var permit Object + if _, err := store.transact(context.Background(), func(tx *transaction) Object { + permit = runtimeControlCommand(tx, &ControlRequest{Action: "runtime_permit", Body: Object{ + "workspaceId": wsID, "controlEpoch": epoch, "nodeInstanceId": nodeRow, "executionId": execution, + }}) + return Object{} + }); err != nil { + t.Fatalf("runtime permit: %v", err) + } + if permit.S("actorUserId") != actor || permit.S("operationId") != runID { + t.Fatalf("permit binding actor=%q operation=%q; want the frozen creator and the workflow run", permit.S("actorUserId"), permit.S("operationId")) + } +} From 88b56f66b7e5442910432c0bf77020b06e2866c2 Mon Sep 17 00:00:00 2001 From: Kyber5323 Date: Sat, 17 Oct 2026 10:38:06 +0800 Subject: [PATCH 2/9] fix(core): freeze the run Project a direct run states MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit specs#99 D3, run identity: workflow_runs.project_id had no production writer at all. Migration 0035 added the column and createWorkflowRunWorkspace already refuses a run without one, but the only creation path that exists — the editor's Run button — never wrote one, so every direct run was identity-incomplete and could only ever fail at its first agent node with workflow_run_project_required. Nothing ever named the Project a run provisions in. - createWorkflowRun accepts an optional projectId from the request body, freezes it in the same INSERT as created_by, and authorizes it through the same `project` helper every project-scoped route uses: an unknown, malformed, other tenant's or archived Project is one 404, and the frozen value is the row's own id rather than the caller's string. No hardcoded or default Project — a workflow document belongs to a tenant, not to a Project, so there is nothing to default to. - A Project already `deleting` is refused (409 resource_unavailable, the rule the rest of the Project API applies). Freezing one would commit a run whose Workspace can never be created — createWorkflowRunWorkspace reads a deleting Project as busy and a deleted one as gone — leaving the run `running` for good, one failed advance per tick and no terminal state. A `provisioning` Project is still accepted: that state is transient, and the scheduler already waits for it the way it waits for any other admission. - Omitting projectId stays legal: a graph with no agent node needs no Workspace, and the editor's Run button has no Project to offer. Such a run stays identity-incomplete and fails honestly at its first agent node rather than guessing. - The route's field list, api/openapi.json and the generated client carry the new optional field. Tests (integration/workflow_run_project_test.go) drive the real HTTP path rather than seeding rows. The freeze test proves the creation transaction wrote the value — read off the row, not off the response, since the run resource carries no project field — and that deleting the Project afterwards cannot change what an already-created run was created against. The refusal test proves no run row survives a rejected Project, and the `deleting` case proves the hang guard. Both were confirmed failing before this change: the old router rejects the field outright (400 unknown_field field:projectId), and the `deleting` case answered 400 instead of 409. Co-Authored-By: Claude Code --- api/openapi.json | 5 + frontend/src/api/generated.schemas.ts | 2 + integration/workflow_run_project_test.go | 163 +++++++++++++++++++++++ internal/api/router/router.go | 2 +- internal/contract/openapi.go | 6 + internal/core/workflow_run.go | 41 +++++- 6 files changed, 215 insertions(+), 4 deletions(-) create mode 100644 integration/workflow_run_project_test.go diff --git a/api/openapi.json b/api/openapi.json index 362505ded..0e5b77b3e 100644 --- a/api/openapi.json +++ b/api/openapi.json @@ -22305,6 +22305,11 @@ "name": { "type": "string" }, + "projectId": { + "description": "The Project the run's Workspace provisions in. Frozen at creation for the run's whole lifetime. Authorized against the caller's own collaboration space: an unknown, archived or foreign Project is 404. Omit it for a graph with no agent node; a run without one fails honestly at its first agent node.", + "format": "uuid", + "type": "string" + }, "snapshotId": { "type": "string" } diff --git a/frontend/src/api/generated.schemas.ts b/frontend/src/api/generated.schemas.ts index d98bc99cb..d02b4fd06 100644 --- a/frontend/src/api/generated.schemas.ts +++ b/frontend/src/api/generated.schemas.ts @@ -3247,6 +3247,8 @@ export type PostApiV1TenantsTidWorkflowsWfidRunsBodyInput = { [key: string]: unk export type PostApiV1TenantsTidWorkflowsWfidRunsBody = { input?: PostApiV1TenantsTidWorkflowsWfidRunsBodyInput; name?: string; + /** The Project the run's Workspace provisions in. Frozen at creation for the run's whole lifetime. Authorized against the caller's own collaboration space: an unknown, archived or foreign Project is 404. Omit it for a graph with no agent node; a run without one fails honestly at its first agent node. */ + projectId?: string; snapshotId?: string; }; diff --git a/integration/workflow_run_project_test.go b/integration/workflow_run_project_test.go new file mode 100644 index 000000000..98d7faf10 --- /dev/null +++ b/integration/workflow_run_project_test.go @@ -0,0 +1,163 @@ +package integration + +import ( + "database/sql" + "testing" + + "github.com/google/uuid" + + "github.com/wanglongan587/cloud/internal/core" +) + +// frozenRunProject reads the Project a run froze, straight off the row. The run resource carries no +// project field, so the only way to tell "the API accepted the value" from "the creation transaction +// wrote it" is to look at what the transaction left behind. +func frozenRunProject(t *testing.T, f *fixture, runID string) string { + t.Helper() + var frozen sql.NullString + if err := f.store.Pool.QueryRow(`SELECT project_id FROM workflow_runs WHERE id=$1`, runID).Scan(&frozen); err != nil { + t.Fatalf("read run project: %v", err) + } + return frozen.String +} + +func runCount(t *testing.T, f *fixture, wfid string) int { + t.Helper() + return len(issueItems(f.call("GET", f.path("/workflows/"+wfid+"/runs"), nil, "", 200))) +} + +// archiveProject takes a fully provisioned Project into `deleting` and stops there — the caller +// drains when it wants the delete to settle. The runtime control the clone left on the Project's main +// workspace has to be handed back first, or the Project API refuses the delete as held. +func archiveProject(t *testing.T, f *fixture, pid, key string) { + t.Helper() + active := f.call("GET", f.path("/projects/"+pid), nil, "", 200) + if active.S("lifecycle") != "active" { + t.Fatalf("project lifecycle = %q; want active before the delete", active.S("lifecycle")) + } + f.call("DELETE", f.path("/projects/"+pid), core.Object{"version": active.N("version")}, key, 202) +} + +// foreignProject provisions a second tenant and returns a live Project id inside it. The row is +// seeded directly because the fixture actor is not a member of that tenant and so cannot create one +// there through the API — which is exactly the property under test: the id names a Project that +// really exists and is unreachable from the calling tenant, so the refusal cannot be explained by +// the Project merely being absent. The Project and its main workspace go in together because the +// deferred project_main trigger requires a live Project to carry exactly one. +func foreignProject(t *testing.T, f *fixture) string { + t.Helper() + other, err := f.store.Bootstrap(t.Context(), "Other", "corp", "other", "Other") + must(t, err) + pid := uuid.NewString() + tx, err := f.store.Pool.Begin() + if err != nil { + t.Fatalf("foreign project begin: %v", err) + } + defer tx.Rollback() + if _, err := tx.Exec(`INSERT INTO projects(id,tenant_id,owner_user_id,space_id,name,repository_url,default_branch,lifecycle) VALUES($1,$2,$3,$4,'Foreign','https://example.invalid/foreign.git','main','active')`, + pid, other.S("tenantId"), other.S("userId"), other.S("spaceId")); err != nil { + t.Fatalf("seed foreign project: %v", err) + } + if _, err := tx.Exec(`INSERT INTO workspaces(id,tenant_id,owner_user_id,creator_user_id,creator_evidence,project_id,kind,desired_state,observed_state,runtime_generation,requested_ref) VALUES($1,$2,$3,$3,'verified_request',$4,'main','running','ready',1,'HEAD')`, + uuid.NewString(), other.S("tenantId"), other.S("userId"), pid); err != nil { + t.Fatalf("seed foreign project workspace: %v", err) + } + if err := tx.Commit(); err != nil { + t.Fatalf("seed foreign project commit: %v", err) + } + return pid +} + +// The direct run's Project is stated by the caller and frozen in the creation transaction (D3). +// Before this, no production path wrote workflow_runs.project_id at all, so every editor-created run +// was identity-incomplete and could only ever fail at its first agent node +// (workflow_run_project_required); the run's Workspace had nowhere to provision and the project +// access a run implies was never checked against anyone. +func TestWorkflowRunFreezesCallerProject(t *testing.T) { + f := setup(t) + seedPersonalModel(t, f) + pid := f.call("POST", f.path("/projects"), core.Object{ + "name": "Run project", "repositoryUrl": "https://example.invalid/repo.git", "defaultBranch": "main", + }, "run-project", 202).O("resource").S("id") + wfid := f.call("POST", f.path("/workflows"), core.Object{ + "name": "Project run", "graph": runGraph(), + }, "run-project-wf", 200).O("resource").S("id") + f.call("POST", f.path("/workflows/"+wfid+"/publish"), core.Object{}, "run-project-pub", 200) + + run := f.call("POST", f.path("/workflows/"+wfid+"/runs"), core.Object{"projectId": pid}, "run-project-1", 200).O("resource") + if got := frozenRunProject(t, f, run.S("id")); got != pid { + t.Fatalf("run froze project %q; want %q", got, pid) + } + // The same run replayed under its idempotency key keeps the one frozen value. + replay := f.call("POST", f.path("/workflows/"+wfid+"/runs"), core.Object{"projectId": pid}, "run-project-1", 200).O("resource") + if replay.S("id") != run.S("id") || frozenRunProject(t, f, replay.S("id")) != pid { + t.Fatalf("replayed create diverged: %v", replay) + } + + // The identity is frozen, not re-derived: deleting the Project afterwards cannot change what an + // already-created run was created against. + f.drain() + f.acknowledgeSimulatorBindings() + archiveProject(t, f, pid, "run-project-del") + f.drain() + if got := frozenRunProject(t, f, run.S("id")); got != pid { + t.Fatalf("project after delete = %q; want the frozen %q", got, pid) + } +} + +// A Project the caller cannot reach is refused, and refused inside the creation transaction: the run +// does not exist afterwards. The check is the same one every project-scoped route makes, so an +// unknown, a malformed and another tenant's Project are one indistinguishable 404 — a run is never +// created against a Project its creator could not open, and never against a default nobody named. +func TestWorkflowRunRefusesUnreachableProject(t *testing.T) { + f := setup(t) + seedPersonalModel(t, f) + wfid := f.call("POST", f.path("/workflows"), core.Object{ + "name": "Guarded run", "graph": runGraph(), + }, "guard-run-wf", 200).O("resource").S("id") + f.call("POST", f.path("/workflows/"+wfid+"/publish"), core.Object{}, "guard-run-pub", 200) + baseline := runCount(t, f, wfid) + + // A Project the caller owns but is deleting. Not drained, so it sits in `deleting` — the state a + // run must never be frozen against, because its Workspace could never be created and the run would + // stay `running` for good instead of reaching a terminal state. + deleting := f.call("POST", f.path("/projects"), core.Object{ + "name": "Deleting", "repositoryUrl": "https://example.invalid/repo.git", "defaultBranch": "main", + }, "guard-run-deleting-project", 202).O("resource").S("id") + f.drain() + f.acknowledgeSimulatorBindings() + archiveProject(t, f, deleting, "guard-run-deleting-del") + + for _, c := range []struct { + name, projectID string + status int + code string + }{ + {"unknown", uuid.NewString(), 404, "not_found"}, + {"malformed", "not-a-uuid", 404, "not_found"}, + {"other tenant", foreignProject(t, f), 404, "not_found"}, + {"deleting", deleting, 409, "resource_unavailable"}, + } { + refused := f.call("POST", f.path("/workflows/"+wfid+"/runs"), core.Object{"projectId": c.projectID}, "guard-run-"+c.name, c.status) + if refused.S("code") != c.code { + t.Fatalf("%s project: code = %q; want %q", c.name, refused.S("code"), c.code) + } + if got := runCount(t, f, wfid); got != baseline { + t.Fatalf("%s project created a run anyway: %d rows, want %d", c.name, got, baseline) + } + } + + // Once the delete settles the same Project is gone, and reads as the same 404 an unknown id does. + f.drain() + if refused := f.call("POST", f.path("/workflows/"+wfid+"/runs"), core.Object{"projectId": deleting}, "guard-run-deleted", 404); refused.S("code") != "not_found" { + t.Fatalf("deleted project: code = %q; want not_found", refused.S("code")) + } + + // Omitting the Project stays accepted: a graph with no agent node needs no Workspace, and the + // editor's Run button has no Project to offer. Such a run is identity-incomplete and fails + // honestly at its first agent node rather than guessing a Project. + unbound := f.call("POST", f.path("/workflows/"+wfid+"/runs"), core.Object{}, "guard-run-unbound", 200).O("resource") + if got := frozenRunProject(t, f, unbound.S("id")); got != "" { + t.Fatalf("unbound run froze project %q; want none", got) + } +} diff --git a/internal/api/router/router.go b/internal/api/router/router.go index 0fe946d8b..e2be6b25c 100644 --- a/internal/api/router/router.go +++ b/internal/api/router/router.go @@ -116,7 +116,7 @@ func Routes() []Route { {"GET", "/api/v1/tenants/:tid/workflows/:wfid/snapshots/:snapshotId/form", "", nil}, {"PUT", "/api/v1/tenants/:tid/workflows/:wfid/snapshots/:snapshotId/restore", "", []string{"version"}}, {"GET", "/api/v1/tenants/:tid/workflows/:wfid/runs", "", nil}, - {"POST", "/api/v1/tenants/:tid/workflows/:wfid/runs", "", []string{"name", "snapshotId", "input"}}, + {"POST", "/api/v1/tenants/:tid/workflows/:wfid/runs", "", []string{"name", "snapshotId", "input", "projectId"}}, {"GET", "/api/v1/tenants/:tid/workflows/:wfid/runs/:rid", "", nil}, {"GET", "/api/v1/tenants/:tid/issue-views", "", nil}, {"POST", "/api/v1/tenants/:tid/issue-views", "", []string{"name", "filter"}}, diff --git a/internal/contract/openapi.go b/internal/contract/openapi.go index f53a1be98..6b117d590 100644 --- a/internal/contract/openapi.go +++ b/internal/contract/openapi.go @@ -885,6 +885,10 @@ func optionalField(name string, r router.Route) bool { case "snapshotId": // A run may omit the snapshot to pin the latest published one. return strings.HasSuffix(r.Path, "/runs") + case "projectId": + // A run states the Project its Workspace provisions in; a graph with no agent node + // needs no Workspace, so a run may omit it and fail honestly at its first agent node. + return strings.HasSuffix(r.Path, "/runs") case "pluginVersion": // Omitted install pins the catalog's current version. return true @@ -958,6 +962,8 @@ func inputSchema(name string, r router.Route) obj { return enumeration("unstarted", "started", "done", "closed") case "labelId", "userId", "assigneeId", "executorId", "refId", "parentId", "projectRef", "targetId": return uuid() + case "projectId": + return obj{"type": "string", "format": "uuid", "description": "The Project the run's Workspace provisions in. Frozen at creation for the run's whole lifetime. Authorized against the caller's own collaboration space: an unknown, archived or foreign Project is 404. Omit it for a graph with no agent node; a run without one fails honestly at its first agent node."} case "ids": return array(uuid()) case "filter", "properties", "input", "values", "graph": diff --git a/internal/core/workflow_run.go b/internal/core/workflow_run.go index dc6acddb2..3796ce2e4 100644 --- a/internal/core/workflow_run.go +++ b/internal/core/workflow_run.go @@ -81,6 +81,9 @@ func composeWorkflowRun(t *transaction, run Object) Object { // entry accepts exactly what a manual workflow IssueRun and a confirmed @ interaction accept. The // initiator matters here: the executor slots are resolved against *their* personal configuration, // which is why the uid is part of the request rather than inferred later. +// +// The run's Project comes from the caller (D3): see resolveWorkflowRunProject for why it is stated +// rather than inferred, and why omitting it is still accepted. func createWorkflowRun(t *transaction, r *PublicRequest, uid string) Object { w := workflow(t, r.TenantID, r.WorkflowID) name := w.S("name") @@ -93,14 +96,46 @@ func createWorkflowRun(t *transaction, r *PublicRequest, uid string) Object { id := newID() // The invocation identity freezes here (D3): created_by is the aggregate's actor for the whole // run — the run Workspace lifecycle reads it instead of re-resolving who pressed Run. The - // authenticated Identity, not a /users path parameter, is who that is. project_id stays unset on - // this path; the invocation path that knows a project (@, #71) freezes it too. + // authenticated Identity, not a /users path parameter, is who that is. The project freezes in the + // same statement, so the two halves of the identity can never disagree about who ran what where. require(r.Identity != nil, 401, "user_credential_required") createdBy := identityWithAlias(t, r.Identity).S("id") - t.exec("INSERT INTO workflow_runs(id,tenant_id,workflow_id,snapshot_id,name,status,input,created_by) VALUES($1,$2,$3,$4,$5,$6,$7,$8)", id, r.TenantID, r.WorkflowID, s.S("id"), name, "pending", jsonText(input), createdBy) + projectID := resolveWorkflowRunProject(t, r, uid) + t.exec("INSERT INTO workflow_runs(id,tenant_id,workflow_id,snapshot_id,name,status,input,created_by,project_id) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9)", id, r.TenantID, r.WorkflowID, s.S("id"), name, "pending", jsonText(input), createdBy, projectID) if t.simulator != nil { nodeStates, rounds, status := t.simulator.SimulateWorkflowRun(s.O("graph"), input) t.exec("UPDATE workflow_runs SET node_states=$1,rounds=$2,status=$3,started_at=now(),finished_at=now(),version=version+1,updated_at=now() WHERE id=$4", jsonText(nodeStates), jsonText(rounds), status, id) } return composeWorkflowRun(t, t.one("SELECT * FROM workflow_runs WHERE id=$1 AND tenant_id=$2", id, r.TenantID)) } + +// resolveWorkflowRunProject authorizes and returns the optional Project a direct run freezes (D3). +// The Project is the one thing that says where the run's Workspace may provision — +// createWorkflowRunWorkspace refuses a run without one — and it is stated by the caller rather than +// inferred, because nothing else knows it: a workflow document belongs to a tenant, not to a Project, +// so there is no project to default to and defaulting to one would place a run's Workspace, its clone +// and its whole checkout in a Project nobody named. +// +// The value is authorized by the same helper every project-scoped route uses, so an unknown, foreign, +// archived or foreign-space project is the same 404 and a run can never be created against a Project +// its creator could not open. The frozen value is the row's own id, never the string the caller sent. +// +// A Project already being deleted is refused outright (the `resource_unavailable` rule the rest of the +// Project API applies). Freezing one would be worse than refusing it: the run would commit, and its +// Workspace — which can never be created, because createWorkflowRunWorkspace reads a deleting Project +// as busy and a deleted one as gone — would leave the run `running` for good, one failed advance per +// tick and no terminal state. A Project still provisioning is accepted: that state is transient, and +// the scheduler waits for it the same way it waits for any other admission. +// +// Omitting it stays legal. A graph whose nodes are all start/condition/output needs no Workspace, and +// the editor's Run button has no Project to offer; such a run stays identity-incomplete and fails +// honestly at its first agent node (workflow_run_project_required) rather than guessing. +func resolveWorkflowRunProject(t *transaction, r *PublicRequest, uid string) any { + pid := r.Body.S("projectId") + if pid == "" { + return nil + } + p := project(t, r.TenantID, uid, pid) + require(p.S("lifecycle") != "deleting", 409, "resource_unavailable") + return p.S("id") +} From 8d5b875df1269aef0c2f29f352f80cfecd9af647 Mon Sep 17 00:00:00 2001 From: t Date: Sat, 17 Oct 2026 16:24:49 +0800 Subject: [PATCH 3/9] fix(core): refuse a direct run that names no Project MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Owner ruling on #141 (2026-10-17): `project_id` is half of a run's frozen identity (specs#99 D3), so a direct run that names no Project is refused at creation rather than persisted. The refusal reuses the code createWorkflowRunWorkspace already raises for the same condition (`workflow_run_project_required`, 409) instead of inventing a second one. - `resolveWorkflowRunProject` requires a non-empty `projectId`. The gate lives on the resolver rather than on its one caller so that every entry which creates a run — the editor today, and the @ confirm / manual run once they are linked to the same creation point — makes one shared decision inside the creation transaction instead of each writing its own "no Project" rule. This is the single creation point the owner asked for; today the INSERT exists only here. - This reverses the previous "omitting it stays legal" contract. That reading treated a Project-less run as merely incomplete and deferred the failure to the run's first agent node. It is not recoverable there: the run can never provision the Workspace those nodes need, so the honest place to report the gap is before the row exists. - A graph that genuinely needs no Workspace still names one. The rule is unconditional, which is the part a "only graphs that provision" reading would get wrong, so a start -> input graph is the case the new test pins. - `createWorkflowRunWorkspace`'s guard is kept: it now guards rows written before this gate, or by a future entry that bypasses the shared creation point. Tests (integration/workflow_run_project_test.go) drive the real HTTP path: - `TestWorkflowRunWithoutProjectIsRefused` — a start -> input graph is refused with `workflow_run_project_required`, leaves no run row, and runs once the caller names a Project, so the refusal is about the missing identity and not about the graph shape. - `TestWorkflowRunRefusesUnreachableProject` — its trailing "omitting stays accepted" case is now the missing-Project refusal, asserted against the same run-count baseline the unreachable-Project table uses. - The tests that start runs legitimately (`TestWorkflowRunLifecycle`, `TestWorkflowRunTenantIsolation`, the admission and snapshot-scope cases) now name a Project through the new `runProject` helper. A missing Project is checked after admission and input validation, so the cases asserting a snapshot, input or slot refusal are unchanged. Co-Authored-By: Claude Code --- integration/workflow_admission_test.go | 10 +++-- integration/workflow_run_project_test.go | 57 +++++++++++++++++++++--- integration/workflow_run_test.go | 12 ++--- internal/core/run_workspace.go | 4 +- internal/core/workflow_run.go | 23 +++++++--- 5 files changed, 83 insertions(+), 23 deletions(-) diff --git a/integration/workflow_admission_test.go b/integration/workflow_admission_test.go index 8ed03b6d8..b2a24db7c 100644 --- a/integration/workflow_admission_test.go +++ b/integration/workflow_admission_test.go @@ -227,7 +227,7 @@ func TestWorkflowRunSnapshotIsScopedToItsWorkflow(t *testing.T) { publishWorkflow(t, f, second.S("id"), "adm-scope-pub-b") // The workflow's own snapshot runs. - run := f.call("POST", f.path("/workflows/"+first.S("id")+"/runs"), core.Object{"snapshotId": mine.S("id")}, "adm-scope-1", 200).O("resource") + run := f.call("POST", f.path("/workflows/"+first.S("id")+"/runs"), core.Object{"snapshotId": mine.S("id"), "projectId": runProject(t, f, "adm-scope-project")}, "adm-scope-1", 200).O("resource") if run.S("snapshotId") != mine.S("id") { t.Fatalf("the run did not execute the version it named: %v", run) } @@ -246,6 +246,7 @@ func TestWorkflowRunSnapshotIsScopedToItsWorkflow(t *testing.T) { // and the newest published version is what an unqualified run gets. func TestWorkflowRunValidatesAgainstTheVersionItRuns(t *testing.T) { f := setup(t) + pid := runProject(t, f, "adm-hist-project") graph := func(names ...string) core.Object { variables := []any{} for _, name := range names { @@ -268,7 +269,7 @@ func TestWorkflowRunValidatesAgainstTheVersionItRuns(t *testing.T) { // The older version does not demand a variable that did not exist when it was frozen... f.call("POST", f.path("/workflows/"+wf.S("id")+"/runs"), core.Object{ - "snapshotId": older.S("id"), "input": core.Object{"topic": "auth"}, + "snapshotId": older.S("id"), "input": core.Object{"topic": "auth"}, "projectId": pid, }, "adm-hist-1", 200) // ...and still enforces its own required variable. f.call("POST", f.path("/workflows/"+wf.S("id")+"/runs"), core.Object{ @@ -282,7 +283,7 @@ func TestWorkflowRunValidatesAgainstTheVersionItRuns(t *testing.T) { t.Fatalf("the newest version was not the one applied: %v", missing) } newest := f.call("POST", f.path("/workflows/"+wf.S("id")+"/runs"), core.Object{ - "snapshotId": newer.S("id"), "input": core.Object{"topic": "auth", "added-later": "x"}, + "snapshotId": newer.S("id"), "input": core.Object{"topic": "auth", "added-later": "x"}, "projectId": pid, }, "adm-hist-4", 200).O("resource") // The version a run executed is recorded on the run row, not left to be re-derived later. @@ -388,6 +389,7 @@ func TestManualWorkflowRunCannotNameAForeignWorkflow(t *testing.T) { // as itself, and no run is created while it is unresolved. func TestWorkflowRunDefaultSlotGate(t *testing.T) { f := setup(t) + pid := runProject(t, f, "adm-gate-project") wf := f.call("POST", f.path("/workflows"), core.Object{ "name": "Gated flow", "graph": agentRunGraph("", false, nil), }, "adm-gate-wf", 200).O("resource") @@ -422,7 +424,7 @@ func TestWorkflowRunDefaultSlotGate(t *testing.T) { _, err := f.store.PutModelCredential(context.Background(), &f.user, connection.S("id"), connection.N("version"), "adm-gate-credential", uuid.NewString(), make([]byte, 32), "fixture-key") must(t, err) - run := f.call("POST", runPath, core.Object{}, "adm-gate-3", 200).O("resource") + run := f.call("POST", runPath, core.Object{"projectId": pid}, "adm-gate-3", 200).O("resource") if run.S("status") != "succeeded" { t.Fatalf("the admitted run did not execute: %v", run) } diff --git a/integration/workflow_run_project_test.go b/integration/workflow_run_project_test.go index 98d7faf10..d1ebbd3ca 100644 --- a/integration/workflow_run_project_test.go +++ b/integration/workflow_run_project_test.go @@ -26,6 +26,18 @@ func runCount(t *testing.T, f *fixture, wfid string) int { return len(issueItems(f.call("GET", f.path("/workflows/"+wfid+"/runs"), nil, "", 200))) } +// runProject creates a Project the fixture actor may start runs in and returns its id. Every direct +// run names one, so a test that creates a run needs a reachable Project even when its graph never +// provisions a Workspace. The Project is deliberately left `provisioning`: a Project still converging +// is an accepted place to freeze a run against, and draining it would make each of these tests wait +// on an operation it does not care about. +func runProject(t *testing.T, f *fixture, key string) string { + t.Helper() + return f.call("POST", f.path("/projects"), core.Object{ + "name": "Run project", "repositoryUrl": "https://example.invalid/repo.git", "defaultBranch": "main", + }, key, 202).O("resource").S("id") +} + // archiveProject takes a fully provisioned Project into `deleting` and stops there — the caller // drains when it wants the delete to settle. The runtime control the clone left on the Project's main // workspace has to be handed back first, or the Project API refuses the delete as held. @@ -153,11 +165,44 @@ func TestWorkflowRunRefusesUnreachableProject(t *testing.T) { t.Fatalf("deleted project: code = %q; want not_found", refused.S("code")) } - // Omitting the Project stays accepted: a graph with no agent node needs no Workspace, and the - // editor's Run button has no Project to offer. Such a run is identity-incomplete and fails - // honestly at its first agent node rather than guessing a Project. - unbound := f.call("POST", f.path("/workflows/"+wfid+"/runs"), core.Object{}, "guard-run-unbound", 200).O("resource") - if got := frozenRunProject(t, f, unbound.S("id")); got != "" { - t.Fatalf("unbound run froze project %q; want none", got) + // Omitting the Project is refused at creation, and refused as the same condition + // createWorkflowRunWorkspace raises it as: a run with no frozen Project is an aggregate that can + // never provision the Workspace its agent nodes need, so it must not be persisted at all. A + // missing Project is a distinct case from an unreachable one — nothing was named, so nothing was + // authorized — which is why it is asserted here rather than as another row of the table above. + missing := f.call("POST", f.path("/workflows/"+wfid+"/runs"), core.Object{}, "guard-run-unbound", 409) + if missing.S("code") != "workflow_run_project_required" { + t.Fatalf("missing project: code = %q; want workflow_run_project_required", missing.S("code")) } + if got := runCount(t, f, wfid); got != baseline { + t.Fatalf("a run naming no project was created anyway: %d rows, want %d", got, baseline) + } +} + +// Naming a Project is mandatory for every direct run, including one whose graph never provisions a +// Workspace. project_id is half of the run's frozen identity (D3), so a run created without one would +// be born unable to finish rather than merely incomplete, and the place to say so is the creation +// transaction rather than an execution that can never proceed. A start -> input graph is refused +// exactly like an agent graph: that is the part a "only graphs that need a Workspace" reading of the +// rule would get wrong, so it is the case this test pins. +func TestWorkflowRunWithoutProjectIsRefused(t *testing.T) { + f := setup(t) + wf := f.call("POST", f.path("/workflows"), core.Object{ + "name": "No project", "graph": startInputGraph(nil), + }, "no-project-wf", 200).O("resource") + wfid := wf.S("id") + publishWorkflow(t, f, wfid, "no-project-pub") + baseline := runCount(t, f, wfid) + + refused := f.call("POST", f.path("/workflows/"+wfid+"/runs"), core.Object{}, "no-project-run", 409) + if refused.S("code") != "workflow_run_project_required" { + t.Fatalf("code = %q; want workflow_run_project_required", refused.S("code")) + } + if got := runCount(t, f, wfid); got != baseline { + t.Fatalf("a refused start left %d runs, want %d", got, baseline) + } + + // The same graph runs once the caller names a Project, so the refusal is about the missing + // identity and not about the graph shape. + f.call("POST", f.path("/workflows/"+wfid+"/runs"), core.Object{"projectId": runProject(t, f, "no-project-project")}, "no-project-run-2", 200) } diff --git a/integration/workflow_run_test.go b/integration/workflow_run_test.go index 007d2a3a9..3fd64dac1 100644 --- a/integration/workflow_run_test.go +++ b/integration/workflow_run_test.go @@ -66,6 +66,8 @@ func TestWorkflowRunLifecycle(t *testing.T) { // The document's agent node runs on the initiator's personal default model, so the actor needs one // before any start can be admitted. seedPersonalModel(t, f) + // Every direct run names the Project it freezes, whether or not its graph provisions a Workspace. + pid := runProject(t, f, "run-project") wf := f.call("POST", f.path("/workflows"), core.Object{ "name": "Run target", "graph": runGraph(), }, "run-wf", 200).O("resource") @@ -83,7 +85,7 @@ func TestWorkflowRunLifecycle(t *testing.T) { snap := f.call("POST", f.path("/workflows/"+wfid+"/publish"), core.Object{}, "run-pub", 200).O("resource") snapID := snap.S("id") run := f.call("POST", f.path("/workflows/"+wfid+"/runs"), core.Object{ - "name": "Trial run", "input": core.Object{"branch": "main"}, + "name": "Trial run", "input": core.Object{"branch": "main"}, "projectId": pid, }, "run-1", 200).O("resource") if run.S("workflowId") != wfid || run.S("name") != "Trial run" || run.S("status") != "succeeded" { t.Fatalf("create run wrong: %v", run) @@ -111,12 +113,12 @@ func TestWorkflowRunLifecycle(t *testing.T) { // Same idempotency key replays the same run; a fresh key with no name defaults to the workflow name. replay := f.call("POST", f.path("/workflows/"+wfid+"/runs"), core.Object{ - "name": "Trial run", "input": core.Object{"branch": "main"}, + "name": "Trial run", "input": core.Object{"branch": "main"}, "projectId": pid, }, "run-1", 200).O("resource") if replay.S("id") != run.S("id") { t.Fatalf("idempotent create made another run: %v", replay) } - f.call("POST", f.path("/workflows/"+wfid+"/runs"), core.Object{}, "run-2", 200).O("resource") + f.call("POST", f.path("/workflows/"+wfid+"/runs"), core.Object{"projectId": pid}, "run-2", 200).O("resource") // Read one run back and list them (newest of the run-2 page is visible by reading ids). one := f.call("GET", f.path("/workflows/"+wfid+"/runs/"+run.S("id")), nil, "", 200) @@ -135,7 +137,7 @@ func TestWorkflowRunLifecycle(t *testing.T) { // A run pinned to a specific, still-owned snapshot works; a foreign snapshot id is read 404. explicit := f.call("POST", f.path("/workflows/"+wfid+"/runs"), core.Object{ - "snapshotId": snapID, + "snapshotId": snapID, "projectId": pid, }, "run-3", 200).O("resource") if explicit.S("snapshotId") != snapID { t.Fatalf("explicit snapshot create wrong: %v", explicit) @@ -164,7 +166,7 @@ func TestWorkflowRunTenantIsolation(t *testing.T) { }, "iso-run-wf", 200).O("resource") wfid := wf.S("id") snap := f.call("POST", f.path("/workflows/"+wfid+"/publish"), core.Object{}, "iso-run-pub", 200).O("resource") - run := f.call("POST", f.path("/workflows/"+wfid+"/runs"), core.Object{}, "iso-run-1", 200).O("resource") + run := f.call("POST", f.path("/workflows/"+wfid+"/runs"), core.Object{"projectId": runProject(t, f, "iso-run-project")}, "iso-run-1", 200).O("resource") other, e := f.store.Bootstrap(context.Background(), "Other", "corp", "other", "Other") must(t, e) diff --git a/internal/core/run_workspace.go b/internal/core/run_workspace.go index e3e7ae874..638f915e1 100644 --- a/internal/core/run_workspace.go +++ b/internal/core/run_workspace.go @@ -51,7 +51,9 @@ func runTriggerActor(t *transaction, run Object) string { // frozen identity is the whole basis (specs#99 D3): project_id names the Project the Workspace // provisions in, and created_by is the actor the Workspace and its operation record. A run without a // frozen project has no place to provision in — the caller fails it honestly rather than guessing a -// Project; an unbound run is the editor path today, and the invocation path (#71) freezes one. +// Project. Creation now refuses to persist such a run at all (resolveWorkflowRunProject), so the two +// checks agree by construction: this one still guards runs written before that gate, or by a future +// entry that inserts a run row without going through the shared creation point. func createWorkflowRunWorkspace(t *transaction, runID string) Object { run := t.one("SELECT * FROM workflow_runs WHERE id=$1 AND deleted_at IS NULL", runID) require(run != nil, 404, "not_found") diff --git a/internal/core/workflow_run.go b/internal/core/workflow_run.go index 3796ce2e4..8b116ab66 100644 --- a/internal/core/workflow_run.go +++ b/internal/core/workflow_run.go @@ -83,7 +83,7 @@ func composeWorkflowRun(t *transaction, run Object) Object { // which is why the uid is part of the request rather than inferred later. // // The run's Project comes from the caller (D3): see resolveWorkflowRunProject for why it is stated -// rather than inferred, and why omitting it is still accepted. +// rather than inferred, and why a request that names none is refused rather than accepted. func createWorkflowRun(t *transaction, r *PublicRequest, uid string) Object { w := workflow(t, r.TenantID, r.WorkflowID) name := w.S("name") @@ -127,14 +127,23 @@ func createWorkflowRun(t *transaction, r *PublicRequest, uid string) Object { // tick and no terminal state. A Project still provisioning is accepted: that state is transient, and // the scheduler waits for it the same way it waits for any other admission. // -// Omitting it stays legal. A graph whose nodes are all start/condition/output needs no Workspace, and -// the editor's Run button has no Project to offer; such a run stays identity-incomplete and fails -// honestly at its first agent node (workflow_run_project_required) rather than guessing. +// Naming one is mandatory, not optional. A run with no frozen Project is not a run that has not +// needed one yet — it is an identity-incomplete aggregate that can never provision the Workspace its +// agent nodes require, so accepting it would persist a row that is born unable to finish and would +// only ever report the gap later, from inside an execution that cannot proceed. This is the same +// reason created_by is required one line up: project_id is the "where" half of the frozen identity, +// and freezing half an identity is not freezing one. A graph that genuinely needs no Workspace still +// names the Project it would provision in, which costs its caller one choice and keeps every run's +// identity whole. +// +// The gate lives on this resolver rather than on the one caller so that every entry which creates a +// run — the editor today, the @ confirm and the manual run once they are linked to the same creation +// point — makes one shared decision in the creation transaction instead of each writing its own +// "no Project" rule. Refusal reuses the code createWorkflowRunWorkspace already raises for the same +// condition, so a missing Project reads the same whether it is caught here or there. func resolveWorkflowRunProject(t *transaction, r *PublicRequest, uid string) any { pid := r.Body.S("projectId") - if pid == "" { - return nil - } + require(pid != "", 409, "workflow_run_project_required") p := project(t, r.TenantID, uid, pid) require(p.S("lifecycle") != "deleting", 409, "resource_unavailable") return p.S("id") From b249aa06637566d0ece1ce6df528e142dd5b40fa Mon Sep 17 00:00:00 2001 From: t Date: Sat, 17 Oct 2026 16:55:16 +0800 Subject: [PATCH 4/9] fix(core): hand the Controller the run a claimed work item belongs to MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `ClaimWork` projected `OperationId` from `run_id` alone. A workflow work item carries `workflow_run_id` and leaves `run_id` NULL by the mutual-exclusion CHECK 0035 added, so the Controller was handed the empty string; the dispatch that echoes it back then matched no row in `dispatchRunExecution` and was refused as `dispatch_conflict` forever. The whole session dispatch chain was dead for workflow runs, and the existing seam test could not see it because it calls `nodeDispatch` directly instead of claiming first. The claim now projects `coalesce(run_id, workflow_run_id)` under its own name. The alias is deliberate: rows are serialized with `row_to_json`, which keeps both copies of a repeated column name, and which one the decoder keeps is not something this query should depend on. New DB test drives the whole chain through the real control actions — `clone_claim` -> `clone_dispatch` -> `node_executions` — and asserts the claim projects the workflow run id and that the dispatch it authorizes binds both the execution and the work item. Against the old projection it fails with the empty string, which is the bug. Co-Authored-By: Claude Code --- internal/controlgrpc/executions.go | 5 +- internal/core/execution_work.go | 16 ++++- .../core/workflow_run_identity_db_test.go | 65 +++++++++++++++++++ 3 files changed, 84 insertions(+), 2 deletions(-) diff --git a/internal/controlgrpc/executions.go b/internal/controlgrpc/executions.go index 3442cb58f..b32710345 100644 --- a/internal/controlgrpc/executions.go +++ b/internal/controlgrpc/executions.go @@ -28,7 +28,10 @@ func (s *executionService) ClaimWork(ctx context.Context, req *controlpb.ClaimWo if work := out.O("work"); work.S("id") != "" { target := work.O("target") return &controlpb.ClaimWorkResponse{Item: &controlpb.WorkItem{ - OperationId: work.S("runId"), + // claimedRunId is the claim's own projection of "the run this work belongs to" — run_id + // for agent work, workflow_run_id for workflow work (claimWorkItem explains why one + // column cannot answer for both). + OperationId: work.S("claimedRunId"), Input: input(work.O("input")), Target: &controlpb.WorkTarget{WorkspaceId: target.S("workspaceId"), SandboxInstanceId: target.S("sandboxInstanceId"), NodeId: target.S("nodeId")}, }}, nil diff --git a/internal/core/execution_work.go b/internal/core/execution_work.go index 4420317ac..783a880e2 100644 --- a/internal/core/execution_work.go +++ b/internal/core/execution_work.go @@ -105,7 +105,21 @@ func claimWorkItem(t *transaction) Object { } else if t.one("SELECT id FROM clone_requests WHERE state='queued'") != nil { reject(410, "runtime_scope_required") } - work := t.one("SELECT * FROM execution_work ew WHERE execution_id IS NULL AND available_at<=clock_timestamp() AND NOT EXISTS(SELECT 1 FROM agent_run_residuals ar WHERE ar.run_id=ew.run_id) ORDER BY created_at,id LIMIT 1") + // A work item names the run it belongs to in one of two columns: agent work carries run_id (an + // issue_runs id), workflow work carries workflow_run_id (a workflow_runs id) and leaves run_id + // NULL by the mutual-exclusion CHECK 0035 added. The Controller is handed the run id as the + // operation id and echoes it back on dispatch, which resolves either column — so the claim has to + // project one value. Reading the agent-only column instead hands back the empty string a workflow + // row produces, and every dispatch of that item is then refused as a conflict it cannot resolve. + // + // The alias is deliberate rather than coalescing over run_id: the row is serialized with + // row_to_json, which keeps both copies of a repeated column name, and which one the decoder ends + // up with is not something this query should depend on. + work := t.one(`SELECT ew.*, coalesce(ew.run_id, ew.workflow_run_id) AS claimed_run_id + FROM execution_work ew + WHERE execution_id IS NULL AND available_at<=clock_timestamp() + AND NOT EXISTS(SELECT 1 FROM agent_run_residuals ar WHERE ar.run_id=ew.run_id) + ORDER BY created_at,id LIMIT 1`) if work != nil && (request == nil || work.S("createdAt") < request.S("createdAt")) { return Object{"work": work} } diff --git a/internal/core/workflow_run_identity_db_test.go b/internal/core/workflow_run_identity_db_test.go index ab86c3b87..66952d2c5 100644 --- a/internal/core/workflow_run_identity_db_test.go +++ b/internal/core/workflow_run_identity_db_test.go @@ -253,3 +253,68 @@ func TestWorkflowExecutionWorkDispatchSeam(t *testing.T) { t.Fatalf("permit binding actor=%q operation=%q; want the frozen creator and the workflow run", permit.S("actorUserId"), permit.S("operationId")) } } + +// The claim hands the Controller the run its work belongs to, and the dispatch that echoes that value +// back is accepted. This is the whole chain — clone_claim -> clone_dispatch -> node_executions — +// driven through the real control actions rather than by calling nodeDispatch directly. +// +// The seam test above dispatches directly and so cannot see this: a workflow work item carries +// workflow_run_id and leaves run_id NULL by 0035's mutual-exclusion CHECK, so a claim that projects +// the agent-only column hands the Controller the empty string, and every dispatch of that item then +// matches no row and is refused as a conflict it can never resolve. +func TestWorkflowExecutionWorkClaimCarriesItsRunID(t *testing.T) { + store := dispatcherDB(t) + runID, _, _ := seedWorkflowIdentityScene(t, store, true) + seedControllerLease(t, store) + + var wsID string + if _, err := store.transact(context.Background(), func(tx *transaction) Object { + created := createWorkflowRunWorkspace(tx, runID) + if created.B("busy") { + t.Fatal("the project must be idle") + } + wsID = created.O("workspace").S("id") + settleCreateWorkspaceOperation(t, tx, created.O("operation").S("id")) + runWorkspaceSettled(tx, runID, "ready") + tx.exec(`UPDATE workspaces SET observed_state='ready', admission_open=true, version=version+1 WHERE id=$1`, wsID) + return Object{} + }); err != nil { + t.Fatalf("stage ready workspace: %v", err) + } + sandbox, nodeService := bindConnectedNode(t, store, wsID) + + input := Object{"kind": "agent_session"} + target := Object{"workspaceId": wsID, "sandboxInstanceId": sandbox, "nodeId": nodeService} + work, err := store.EnqueueExecutionWork(context.Background(), runID, "agent_session", input, target, time.Time{}) + if err != nil { + t.Fatalf("enqueue workflow work: %v", err) + } + + claims := controllerClaims() + pick, err := store.Control(context.Background(), &ControlRequest{Action: "clone_claim", Body: Object{"epoch": 1}, Service: claims}) + if err != nil { + t.Fatalf("claim: %v", err) + } + claimed := pick.O("work") + if claimed.S("id") != work.S("id") { + t.Fatalf("claim returned work %v; want the declared workflow item", claimed) + } + if claimed.S("claimedRunId") != runID { + t.Fatalf("claim projected run %q; want the workflow run %q — an empty value is refused at every dispatch", claimed.S("claimedRunId"), runID) + } + + execution := "exec-claim-" + runID[:8] + if _, err := store.Control(context.Background(), &ControlRequest{ + Action: "clone_dispatch", SubmissionID: "claim-dispatch", Service: claims, + Body: Object{"epoch": int64(1), "operationId": claimed.S("claimedRunId"), "executionId": execution, "nodeId": nodeService, "input": input}, + }); err != nil { + t.Fatalf("dispatch the claimed run id: %v", err) + } + var operation, workID string + if err := store.Pool.QueryRow(`SELECT coalesce(operation_id::text,''),coalesce(work_id::text,'') FROM node_executions WHERE execution_id=$1`, execution).Scan(&operation, &workID); err != nil { + t.Fatalf("read node execution: %v", err) + } + if operation != runID || workID != work.S("id") { + t.Fatalf("node execution operation=%q work=%q; want the workflow run and the claimed item", operation, workID) + } +} From 65ed9d36ffb998e18a07a8fadec743201150e96e Mon Sep 17 00:00:00 2001 From: t Date: Sat, 17 Oct 2026 17:18:24 +0800 Subject: [PATCH 5/9] test(core): register migration 0035 and cover it fresh, upgraded and rerun MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review round on #68. The migration shipped without the two things every other migration in this directory has: an entry in the READMEs and a fresh/upgrade/ rerun test. `TestMigration0035WorkflowRunIdentityAppliesFreshAndUpgrades` asserts the four columns, both exactly-one-owner CHECKs, both partial unique indexes, and — the one this migration does by deletion rather than addition — that runtime_controls' IssueRun foreign key is gone while the column it constrained stays. A polymorphic run reference cannot be a foreign key, so the liveness check moved into the code that grants on it, and that is worth a test rather than a comment. The migration is pinned to LF at the same time. It is introduced by this PR and no released database has applied it, so this is the last moment the pin is free: after merge the bytes are in a ledger this change could not reach. 0034, which is on main, keeps its platform-dependent newline for exactly that reason, and 0035_model_attempt_statistics.sql is left to the PR that owns it. Co-Authored-By: Claude Code --- .gitattributes | 7 +++ integration/migration_upgrade_path_test.go | 60 +++++++++++++++++++++- internal/core/migrations/README.en.md | 15 ++++++ internal/core/migrations/README.md | 12 +++++ 4 files changed, 93 insertions(+), 1 deletion(-) diff --git a/.gitattributes b/.gitattributes index 571382da5..bd6ef6b67 100644 --- a/.gitattributes +++ b/.gitattributes @@ -11,3 +11,10 @@ internal/core/migrations/0028_workflow_snapshots.sql text eol=lf internal/core/migrations/0029_workflow_runs.sql text eol=lf internal/core/migrations/0033_personal_model_connections.sql text eol=lf internal/core/migrations/0033_revision_resume.sql text eol=lf +# 0034 is the one migration since the allowlist lapsed that is already on main, so its checksum is +# recorded in databases this branch cannot reach; pinning it now would fail every one of them with +# migration_checksum_mismatch. It stays as it is. (0035_model_attempt_statistics.sql is in the same +# position for a different reason — it belongs to another PR, which has to make this call for it.) +# Everything introduced since main is pinned here instead, while no released database has recorded +# it yet; after merge that would no longer be true. +internal/core/migrations/0035_workflow_run_identity.sql text eol=lf diff --git a/integration/migration_upgrade_path_test.go b/integration/migration_upgrade_path_test.go index 2a3d54778..9ffa9d39c 100644 --- a/integration/migration_upgrade_path_test.go +++ b/integration/migration_upgrade_path_test.go @@ -1018,7 +1018,7 @@ func TestMigration0033RevisionResumeAppliesFreshAndUpgrades(t *testing.T) { constraints := func(pool *sql.DB) string { t.Helper() var names string - must(t, pool.QueryRow(`SELECT string_agg(conname, ',' ORDER BY conname) FROM pg_constraint + must(t, pool.QueryRow(`SELECT coalesce(string_agg(conname, ',' ORDER BY conname),'') FROM pg_constraint WHERE conrelid='revisions'::regclass AND contype='c' AND conname IN ('revisions_check','revisions_bundle_shape','revisions_resume_refused')`).Scan(&names)) return names } @@ -1047,3 +1047,61 @@ func TestMigration0033RevisionResumeAppliesFreshAndUpgrades(t *testing.T) { t.Fatalf("after 0033: columns %d constraints %q", got, names) } } + +// 0035 turns the WorkflowRun into an execution aggregate the control plane accepts as a work owner: +// the frozen run identity, the workspace binding, the exactly-one-owner discriminator on work and +// the polymorphic runtime reservation. Fresh databases get all of it, an upgrade from 0034 gets it +// without a backfill, and a rerun changes nothing. +func TestMigration0035WorkflowRunIdentityAppliesFreshAndUpgrades(t *testing.T) { + columns := func(pool *sql.DB) int { + t.Helper() + var n int + must(t, pool.QueryRow(`SELECT count(*) FROM information_schema.columns WHERE table_schema=current_schema() AND + ((table_name='workflow_runs' AND column_name IN ('created_by','project_id')) + OR (table_name='workspaces' AND column_name='workflow_run_id') + OR (table_name='execution_work' AND column_name='workflow_run_id'))`).Scan(&n)) + return n + } + constraints := func(pool *sql.DB) string { + t.Helper() + var names string + must(t, pool.QueryRow(`SELECT coalesce(string_agg(conname, ',' ORDER BY conname),'') FROM pg_constraint + WHERE conrelid IN ('workspaces'::regclass,'execution_work'::regclass) AND contype='c' + AND conname IN ('workspace_run_binding_exclusive','execution_work_owner_exclusive')`).Scan(&names)) + return names + } + indexes := func(pool *sql.DB) string { + t.Helper() + var names string + must(t, pool.QueryRow(`SELECT coalesce(string_agg(indexname, ',' ORDER BY indexname),'') FROM pg_indexes + WHERE schemaname=current_schema() AND tablename='execution_work' + AND indexname IN ('one_unregistered_execution_work','one_unregistered_workflow_execution_work')`).Scan(&names)) + return names + } + // A polymorphic run reference cannot be a foreign key, so 0035 drops the IssueRun FK instead of + // reworking it. The column stays; only the constraint goes. Resolved through the search path + // rather than by name alone, which would count another schema's constraint of the same name. + reservationFK := func(pool *sql.DB) int { + t.Helper() + var n int + must(t, pool.QueryRow(`SELECT count(*) FROM pg_constraint + WHERE conrelid='runtime_controls'::regclass AND conname='runtime_controls_maintenance_run_id_fkey'`).Scan(&n)) + return n + } + + freshPool, _ := testSchema(t, "test_wfi35f_") + migratedTwice(t, newStoreOnSchema(t, freshPool)) + if got, names, idx, fk := columns(freshPool), constraints(freshPool), indexes(freshPool), reservationFK(freshPool); got != 4 || names != "execution_work_owner_exclusive,workspace_run_binding_exclusive" || idx != "one_unregistered_execution_work,one_unregistered_workflow_execution_work" || fk != 0 { + t.Fatalf("fresh: columns %d constraints %q indexes %q reservation fk %d", got, names, idx, fk) + } + + pool, _ := testSchema(t, "test_wfi35_") + applyMigrationsUpTo(t, pool, migrationsBefore(t, "0035_workflow_run_identity.sql")) + if got, names, fk := columns(pool), constraints(pool), reservationFK(pool); got != 0 || names != "" || fk != 1 { + t.Fatalf("before 0035: columns %d constraints %q reservation fk %d", got, names, fk) + } + migratedTwice(t, newStoreOnSchema(t, pool)) + if got, names, idx, fk := columns(pool), constraints(pool), indexes(pool), reservationFK(pool); got != 4 || names != "execution_work_owner_exclusive,workspace_run_binding_exclusive" || idx != "one_unregistered_execution_work,one_unregistered_workflow_execution_work" || fk != 0 { + t.Fatalf("after 0035: columns %d constraints %q indexes %q reservation fk %d", got, names, idx, fk) + } +} diff --git a/internal/core/migrations/README.en.md b/internal/core/migrations/README.en.md index 42c3fe688..ce78619bc 100644 --- a/internal/core/migrations/README.en.md +++ b/internal/core/migrations/README.en.md @@ -114,3 +114,18 @@ only validate schema. `TestMigration0035ModelAttemptsFreshAndUpgradeWithoutBackf ## 0038: Bounded abandonment of a blocked run Workspace creation Adds issue_runs.workspace_blocked_at, workspace_deadline_at, workspace_blocked_source (blocked_write / first_observed) with a provisioning-phase partial index, and widens agent_run_residuals' reason constraint to agent_start_failed / workspace_unavailable. The first transaction persisting `blocked` anchors the absolute deadline; legacy blocked creations have no immutable first-blocked evidence, are not backfilled, and receive one default window from the first policy scan as first_observed. Defer, manual retry, re-claiming, restarts and configuration changes never renew; only the creation's real terminal ends the window. Purely forward and retry-safe, without rewriting historical outcomes. Fresh/upgrade/repeat coverage: TestMigration0038WorkspaceBlockedDeadlineAppliesFreshAndUpgrades. + +## 0035: WorkflowRun identity + +`0035_workflow_run_identity.sql` hands run identity and work-item ownership to the control plane. +`workflow_runs` gains `created_by` and `project_id` (both nullable: rows written before this migration +carry no recoverable actor, and `project_id` is filled by the invocation path that knows a Project), +which new runs freeze at creation. `workspaces` gains a mutually exclusive `workflow_run_id`, so a run +Workspace can belong to a WorkflowRun. `execution_work` takes the same exactly-one-owner discriminator +shape, its single-column uniqueness splitting into one partial index per owner, so a run of each kind +keeps at most one unregistered item and the two ownerships never collide. Runtime control's reservation +becomes a polymorphic run reference: the `maintenance_run_id` foreign key is dropped rather than +reworked, because a polymorphic reference cannot be a foreign key, and the liveness check moves to the +code that grants on it. Additive: columns, constraints and indexes, no backfill. + +`TestMigration0035WorkflowRunIdentityAppliesFreshAndUpgrades` covers fresh, 0034 upgrade and rerun. diff --git a/internal/core/migrations/README.md b/internal/core/migrations/README.md index d23dfd558..863861109 100644 --- a/internal/core/migrations/README.md +++ b/internal/core/migrations/README.md @@ -113,3 +113,15 @@ Cloud 领域方法。冻结与撤销规则见 [个人模型连接](../../../docs ## 0038：运行 Workspace 创建 blocked 的有界放弃 新增 issue_runs.workspace_blocked_at、workspace_deadline_at、workspace_blocked_source（blocked_write / first_observed）与 provisioning 期部分索引，并把 agent_run_residuals 的 reason 约束扩为 agent_start_failed / workspace_unavailable。首次持久写入 blocked 的同事务锚定绝对截止；存量 blocked 无不可变首次证据，不回填，由首次策略扫描以 first_observed 给一次默认窗口。defer、人工 retry、重新领取、重启与配置调整不续期；只有原创建真实终态结束窗口。纯前向、可重复执行，不改历史执行结果。新库/升级/重复执行见 TestMigration0038WorkspaceBlockedDeadlineAppliesFreshAndUpgrades。 + +## 0035：WorkflowRun 运行身份 + +`0035_workflow_run_identity.sql` 把运行身份与工作项所有权交给控制面。`workflow_runs` 新增 +`created_by` 与 `project_id` 两列（均可空：本迁移之前的行没有可恢复的发起人，而 `project_id` 由知道 +Project 的调用路径写入），新运行在创建时冻结二者；`workspaces` 新增互斥的 `workflow_run_id`，使运行 +Workspace 可以属于 WorkflowRun；`execution_work` 同样变为「恰有一个所有者」的判别形状，原单列唯一索引 +拆为每种所有者各一条部分索引，因此两类所有者各自最多保留一个未注册工作项且互不冲突； +`runtime_controls` 的预留改为多态运行引用，`maintenance_run_id` 的外键被删除而非改写——多态引用无法做 +外键，存活校验移到授予它的代码里。只新增列、约束与索引，不回填。 + +`TestMigration0035WorkflowRunIdentityAppliesFreshAndUpgrades` 验证空库、从 0034 升级及重复迁移。 From 0fbdafad79d889d2f6d8a09ead5c5d6e0485754f Mon Sep 17 00:00:00 2001 From: t Date: Sat, 17 Oct 2026 19:16:41 +0800 Subject: [PATCH 6/9] fix(core): read the Project's repository through its Binding Rebasing onto main surfaced three conflicts the textual merge could not see: 0037 (cloud#75, already on main) moved a Project's repository out of the projects table into repository_bindings, and this branch still read it where it used to live. - createWorkflowRunWorkspace selected the projects row whole and read defaultBranch from it. That column is gone, so every run Workspace was refused with default_branch_required. It now reads through projectView, exactly as its IssueRun counterpart in the same file does. - TestWorkflowRunRefusesUnreachableProject and the dispatcherDB seed inserted projects(repository_url, default_branch). Both now insert the Project and its Binding together, the idiom every other direct seed in this repository uses. Behavior is unchanged: the same branch name, read from the row that now holds it. Co-Authored-By: Claude Code --- integration/workflow_run_project_test.go | 4 +++- internal/core/run_workspace.go | 5 ++++- internal/core/workflow_run_identity_db_test.go | 2 +- 3 files changed, 8 insertions(+), 3 deletions(-) diff --git a/integration/workflow_run_project_test.go b/integration/workflow_run_project_test.go index d1ebbd3ca..077af0134 100644 --- a/integration/workflow_run_project_test.go +++ b/integration/workflow_run_project_test.go @@ -66,7 +66,9 @@ func foreignProject(t *testing.T, f *fixture) string { t.Fatalf("foreign project begin: %v", err) } defer tx.Rollback() - if _, err := tx.Exec(`INSERT INTO projects(id,tenant_id,owner_user_id,space_id,name,repository_url,default_branch,lifecycle) VALUES($1,$2,$3,$4,'Foreign','https://example.invalid/foreign.git','main','active')`, + // The repository lives in the Project's Binding, not in Project columns (0037): the two go in + // together, the way every other direct seed in this repository now does it. + if _, err := tx.Exec(`WITH p AS (INSERT INTO projects(id,tenant_id,owner_user_id,space_id,name,lifecycle) VALUES($1,$2,$3,$4,'Foreign','active') RETURNING id,tenant_id) INSERT INTO repository_bindings(id,tenant_id,project_id,repository_url,branch) SELECT gen_random_uuid(),tenant_id,id,'https://example.invalid/foreign.git','main' FROM p`, pid, other.S("tenantId"), other.S("userId"), other.S("spaceId")); err != nil { t.Fatalf("seed foreign project: %v", err) } diff --git a/internal/core/run_workspace.go b/internal/core/run_workspace.go index 638f915e1..40e193489 100644 --- a/internal/core/run_workspace.go +++ b/internal/core/run_workspace.go @@ -63,7 +63,10 @@ func createWorkflowRunWorkspace(t *transaction, runID string) Object { op := t.one("SELECT * FROM operations WHERE workspace_id=$1 AND kind='create_workspace' ORDER BY created_at,id LIMIT 1", existing.S("id")) return Object{"busy": false, "workspace": existing, "operation": op} } - p := t.one("SELECT * FROM projects WHERE id=$1 AND tenant_id=$2 AND deleted_at IS NULL", run.S("projectId"), run.S("tenantId")) + // projectView, not the projects table: 0037 moved the repository to the Project's Binding, so + // `default_branch` is read back under the Project's published name from there. It is the same + // read its IssueRun counterpart above makes, and the reason both now require a Binding. + p := t.one("SELECT * FROM "+projectView+" p WHERE id=$1 AND tenant_id=$2 AND deleted_at IS NULL", run.S("projectId"), run.S("tenantId")) require(p != nil, 404, "not_found") if p.S("lifecycle") != "active" || t.one("SELECT id FROM operations WHERE project_id=$1 AND state IN ('queued','running','retry_wait','blocked')", p.S("id")) != nil { return Object{"busy": true} diff --git a/internal/core/workflow_run_identity_db_test.go b/internal/core/workflow_run_identity_db_test.go index 66952d2c5..e010bd0a8 100644 --- a/internal/core/workflow_run_identity_db_test.go +++ b/internal/core/workflow_run_identity_db_test.go @@ -41,7 +41,7 @@ func seedWorkflowIdentityScene(t *testing.T, store *Store, withProject bool) (ru exec("tenant", `INSERT INTO tenants(id,name,status) VALUES($1,'wfi','active')`, tid) exec("membership", `INSERT INTO tenant_memberships(tenant_id,user_id,role,status) VALUES($1,$2,'admin','active')`, tid, uid) exec("space", `INSERT INTO collab_workspaces(id,tenant_id,name,slug,created_by) VALUES($1,$2,'wfi','wfi-'||left($4::text,8),$3)`, space, tid, uid, project) - exec("project", `INSERT INTO projects(id,tenant_id,owner_user_id,space_id,name,repository_url,default_branch,lifecycle) VALUES($1,$2,$3,$4,'wfi','https://example.invalid/wfi.git','main','active')`, project, tid, uid, space) + exec("project", `WITH p AS (INSERT INTO projects(id,tenant_id,owner_user_id,space_id,name,lifecycle) VALUES($1,$2,$3,$4,'wfi','active') RETURNING id,tenant_id) INSERT INTO repository_bindings(id,tenant_id,project_id,repository_url,branch) SELECT gen_random_uuid(),tenant_id,id,'https://example.invalid/wfi.git','main' FROM p`, project, tid, uid, space) // A live project must carry exactly one main workspace (deferred project_main trigger). exec("main-workspace", `INSERT INTO workspaces(id,tenant_id,owner_user_id,creator_user_id,creator_evidence,project_id,kind,desired_state,observed_state,runtime_generation,requested_ref) VALUES($1,$2,$3,$3,'verified_request',$4,'main','running','ready',1,'HEAD')`, uuid.NewString(), tid, uid, project) exec("workflow", `INSERT INTO workflows(id,tenant_id,name,graph) VALUES($1,$2,'wfi','{"nodes":[],"edges":[]}'::jsonb)`, wid, tid) From 83a121c71850e6bc49a6b92ed3a1d55a1cbed431 Mon Sep 17 00:00:00 2001 From: t Date: Sat, 17 Oct 2026 20:43:58 +0800 Subject: [PATCH 7/9] fix(contract): require the Project a direct run freezes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The runtime refuses a run that names no Project with 409 `workflow_run_project_required`, but the contract still described the opposite: `projectId` sat in `optionalField`'s `/runs` branch, so the published schema left it out of `required`, and its description promised that a run without one "fails honestly at its first agent node". Neither holds — a run that names none is never created, so it never reaches a node. `resolveWorkflowRunProject`'s own doc comment contradicted itself the same way, calling the Project optional two paragraphs above "Naming one is mandatory, not optional." `projectId` is now required on the run-creation route and nowhere else, the description states the refusal, and the doc comment says what the body does. The same change is carried through `api/openapi.json` and `frontend/src/api`, so the generated client stops offering a request the service always rejects. That makes the editor's create mutation unable to build its body at all: it has no Project to state, and the run form that picks one is not built yet. The mutation refuses such a run before sending it rather than inventing a placeholder Project to fill the field — the service answers the same code, and no run is frozen against a Project nobody chose. `useCreateWorkflowRun` is where that gate lives, so the form that lands later states a Project and the gate opens; the toolbar tests pin the refusal on both sides of the app's real routes. Co-Authored-By: Claude Code --- api/openapi.json | 5 +- frontend/src/api/generated.schemas.ts | 4 +- frontend/src/features/workflows/api.ts | 19 +++++ .../workflows/editor/workflow-editor.test.tsx | 7 +- .../editor/workflow-run-tools.test.tsx | 73 ++++++++++++++++--- internal/contract/openapi.go | 6 +- internal/contract/openapi_test.go | 15 ++++ internal/core/workflow_run.go | 2 +- 8 files changed, 109 insertions(+), 22 deletions(-) diff --git a/api/openapi.json b/api/openapi.json index 0e5b77b3e..4bf4a78ff 100644 --- a/api/openapi.json +++ b/api/openapi.json @@ -22306,7 +22306,7 @@ "type": "string" }, "projectId": { - "description": "The Project the run's Workspace provisions in. Frozen at creation for the run's whole lifetime. Authorized against the caller's own collaboration space: an unknown, archived or foreign Project is 404. Omit it for a graph with no agent node; a run without one fails honestly at its first agent node.", + "description": "The Project the run's Workspace provisions in. Required: a run that names none is refused with 409 workflow_run_project_required, never persisted without one. Frozen at creation for the run's whole lifetime. Authorized against the caller's own collaboration space: an unknown, archived or foreign Project is 404.", "format": "uuid", "type": "string" }, @@ -22314,6 +22314,9 @@ "type": "string" } }, + "required": [ + "projectId" + ], "type": "object" } } diff --git a/frontend/src/api/generated.schemas.ts b/frontend/src/api/generated.schemas.ts index d02b4fd06..289a80b51 100644 --- a/frontend/src/api/generated.schemas.ts +++ b/frontend/src/api/generated.schemas.ts @@ -3247,8 +3247,8 @@ export type PostApiV1TenantsTidWorkflowsWfidRunsBodyInput = { [key: string]: unk export type PostApiV1TenantsTidWorkflowsWfidRunsBody = { input?: PostApiV1TenantsTidWorkflowsWfidRunsBodyInput; name?: string; - /** The Project the run's Workspace provisions in. Frozen at creation for the run's whole lifetime. Authorized against the caller's own collaboration space: an unknown, archived or foreign Project is 404. Omit it for a graph with no agent node; a run without one fails honestly at its first agent node. */ - projectId?: string; + /** The Project the run's Workspace provisions in. Required: a run that names none is refused with 409 workflow_run_project_required, never persisted without one. Frozen at creation for the run's whole lifetime. Authorized against the caller's own collaboration space: an unknown, archived or foreign Project is 404. */ + projectId: string; snapshotId?: string; }; diff --git a/frontend/src/features/workflows/api.ts b/frontend/src/features/workflows/api.ts index 1ac6ffbc4..485d114c7 100644 --- a/frontend/src/features/workflows/api.ts +++ b/frontend/src/features/workflows/api.ts @@ -364,6 +364,13 @@ export function useWorkflowRun( /** Input for running a workflow once against a published snapshot. */ export interface CreateWorkflowRunInput { workflowId: string + /** + * The Project the run's Workspace provisions in. The contract requires it — a run that names + * none is refused with 409 `workflow_run_project_required` — so the mutation refuses a caller + * that has none instead of sending a body the contract does not describe. Optional here only + * because the run form that picks a Project is not built yet. + */ + projectId?: string /** Optional display name; defaults to the workflow's name. */ name?: string /** The exact snapshot to execute, or omit to pin the latest published one. */ @@ -382,6 +389,10 @@ export interface CreateWorkflowRunInput { * idempotency key, so a retried run replays the stored response instead of * minting a second execution. * + * The run's Project is required by the contract — a run that names none is refused with 409 + * `workflow_run_project_required` — and this editor states none yet: the run form that picks a + * Project is not built, so a run started here is refused until that lands. + * * @param tenantId - Tenant the workflow belongs to. * @returns The create mutation; `data` is the stored run. */ @@ -391,10 +402,18 @@ export function useCreateWorkflowRun(tenantId: string | undefined) { return useMutation, CreateWorkflowRunInput>({ mutationFn: (input: CreateWorkflowRunInput) => { if (!tenantId) throw new Error('cloud tenant not resolved') + // Refused here rather than sent: a run with no frozen Project is an identity-incomplete + // aggregate that can never provision the Workspace its agent nodes need, so the service + // answers this same code (409) and there is no value the client could put in the field + // instead — the caller either names a Project or names none. + if (input.projectId === undefined || input.projectId === '') { + throw new Error('workflow_run_project_required') + } return postApiV1TenantsTidWorkflowsWfidRuns( tenantId, input.workflowId, { + projectId: input.projectId, ...(input.name === undefined || input.name === '' ? {} : { name: input.name }), ...(input.snapshotId === undefined || input.snapshotId === '' ? {} diff --git a/frontend/src/features/workflows/editor/workflow-editor.test.tsx b/frontend/src/features/workflows/editor/workflow-editor.test.tsx index 5a7464b8b..af9194271 100644 --- a/frontend/src/features/workflows/editor/workflow-editor.test.tsx +++ b/frontend/src/features/workflows/editor/workflow-editor.test.tsx @@ -1021,7 +1021,7 @@ describe('WorkflowEditor run tools', () => { expect(screen.getByText('还没有运行记录。')).toBeInTheDocument() }) - it('creates a run from the toolbar once a snapshot is published', async () => { + it('starts no run from the toolbar, which cannot state the Project a run freezes', async () => { const published: WorkflowSnapshot = { id: 'snap-1', tenantId: TEST_TENANT_ID, @@ -1047,6 +1047,9 @@ describe('WorkflowEditor run tools', () => { ) fireEvent.click(runButton) - await waitFor(() => expect(bodies).toEqual([{}])) + // The run form that picks the Project is not built yet, so the mutation refuses the run before + // the request is sent and this capture stays empty. + await waitFor(() => expect(runButton).toBeEnabled()) + expect(bodies).toEqual([]) }) }) diff --git a/frontend/src/features/workflows/editor/workflow-run-tools.test.tsx b/frontend/src/features/workflows/editor/workflow-run-tools.test.tsx index d2e4185f5..3fe8ec6ae 100644 --- a/frontend/src/features/workflows/editor/workflow-run-tools.test.tsx +++ b/frontend/src/features/workflows/editor/workflow-run-tools.test.tsx @@ -1,7 +1,9 @@ -import { fireEvent, screen, waitFor } from '@testing-library/react' +import { QueryClient, QueryClientProvider } from '@tanstack/react-query' +import { fireEvent, renderHook, screen, waitFor } from '@testing-library/react' import { http, HttpResponse } from 'msw' import { beforeEach, describe, expect, it } from 'vitest' import type { WorkflowRun, WorkflowSnapshot } from '@/api/generated.schemas' +import { useCreateWorkflowRun } from '@/features/workflows/api' import { WorkflowRunTools } from '@/features/workflows/editor/workflow-run-tools' import { routes } from '@/routes' import { @@ -15,6 +17,8 @@ import { renderRoutes } from '@/test/render' import { stubViewportGeometry } from '@/test/viewport' const WORKFLOW_ID = '44444444-4444-4444-4444-444444444444' +/** The Project a run freezes; the editor has none to state yet, so only the hook tests name one. */ +const PROJECT_ID = '55555555-5555-5555-5555-555555555555' /** * The readable workspace slug the browser route carries. It is deliberately not * `TEST_TENANT_ID`: the tenant UUID scopes API calls, and using it as the slug is @@ -117,6 +121,49 @@ beforeEach(() => { stubViewportGeometry() }) +/** A client whose mutations never retry, so one refusal settles once. */ +function wrapper() { + const client = new QueryClient({ defaultOptions: { mutations: { retry: false } } }) + return function Wrapper({ children }: { children: React.ReactNode }) { + return {children} + } +} + +/** + * The create mutation's own gate, pinned at the hook because the editor cannot reach it: a run must + * name the Project its Workspace provisions in, and the editor has none to state until the run form + * that picks one lands. A caller that states a Project still gets a run, and the Project it stated + * is what the request carries. + */ +describe('useCreateWorkflowRun', () => { + it('refuses a run that names no Project without sending it', async () => { + const bodies: unknown[] = [] + installRunHandlers({ onCreate: (body) => bodies.push(body) }) + const { result } = renderHook(() => useCreateWorkflowRun(TEST_TENANT_ID), { + wrapper: wrapper(), + }) + + result.current.mutate({ workflowId: WORKFLOW_ID }) + + await waitFor(() => expect(result.current.isError).toBe(true)) + expect(result.current.error?.message).toBe('workflow_run_project_required') + expect(bodies).toEqual([]) + }) + + it('sends the stated Project with the run', async () => { + const bodies: unknown[] = [] + installRunHandlers({ onCreate: (body) => bodies.push(body) }) + const { result } = renderHook(() => useCreateWorkflowRun(TEST_TENANT_ID), { + wrapper: wrapper(), + }) + + result.current.mutate({ workflowId: WORKFLOW_ID, projectId: PROJECT_ID }) + + await waitFor(() => expect(result.current.isSuccess).toBe(true)) + expect(bodies).toEqual([{ projectId: PROJECT_ID }]) + }) +}) + describe('WorkflowRunTools', () => { it('disables the run button while no snapshot exists to execute', async () => { renderRunTools({ snapshots: [] }) @@ -128,7 +175,7 @@ describe('WorkflowRunTools', () => { expect(runButton).toHaveAttribute('title', '请先发布一个版本，才能运行这个工作流') }) - it('arms the run button once a snapshot exists, creating the run and navigating to it', async () => { + it('arms the run button once a snapshot exists, but starts no run it cannot state a Project for', async () => { const bodies: unknown[] = [] const router = renderRunTools({ runs: [], onCreate: (body) => bodies.push(body) }) const runButton = await screen.findByRole('button', { name: '运行' }) @@ -136,9 +183,12 @@ describe('WorkflowRunTools', () => { fireEvent.click(runButton) - // The browser path carries the workspace slug, never the tenant UUID. - await waitFor(() => expect(router.state.location.pathname).toBe(`${EDITOR_PATH}/runs/run-new`)) - expect(bodies).toEqual([{}]) + // A run must name the Project its Workspace provisions in, and the editor has none to state + // yet: the refusal happens before the request, so nothing is created and nothing navigates. + // The button goes pending and comes back armed, which is where the refusal settles. + await waitFor(() => expect(runButton).toBeEnabled()) + expect(bodies).toEqual([]) + expect(router.state.location.pathname).toBe(EDITOR_PATH) }) it('lists run history and opens a run from a row', async () => { @@ -182,18 +232,19 @@ describe('workflow run navigation through the app routes', () => { return { ...view, bodies } } - it('creates one run and stays on the same workspace route instead of the UUID one', async () => { + it('starts no run, and stays on the same workspace route instead of the UUID one', async () => { const { router, bodies } = renderAppAt(EDITOR_PATH) const runButton = await screen.findByRole('button', { name: '运行' }) await waitFor(() => expect(runButton).toBeEnabled()) fireEvent.click(runButton) - // The destination is the real run detail page: a UUID workspace path would have - // been redirected to Issues by the Layout before this page ever mounted. - expect(await screen.findByRole('heading', { name: '运行详情' })).toBeInTheDocument() - expect(router.state.location.pathname).toBe(`${EDITOR_PATH}/runs/run-new`) - expect(bodies).toHaveLength(1) + // A UUID workspace path would have been redirected to Issues by the Layout before + // this page ever mounted, so the route is what proves the click stayed put. + await waitFor(() => expect(runButton).toBeEnabled()) + expect(bodies).toHaveLength(0) + expect(router.state.location.pathname).toBe(EDITOR_PATH) + expect(screen.queryByRole('heading', { name: '运行详情' })).not.toBeInTheDocument() }) it('opens an existing run from the history with the same slug and creates nothing', async () => { diff --git a/internal/contract/openapi.go b/internal/contract/openapi.go index 6b117d590..f47554b42 100644 --- a/internal/contract/openapi.go +++ b/internal/contract/openapi.go @@ -885,10 +885,6 @@ func optionalField(name string, r router.Route) bool { case "snapshotId": // A run may omit the snapshot to pin the latest published one. return strings.HasSuffix(r.Path, "/runs") - case "projectId": - // A run states the Project its Workspace provisions in; a graph with no agent node - // needs no Workspace, so a run may omit it and fail honestly at its first agent node. - return strings.HasSuffix(r.Path, "/runs") case "pluginVersion": // Omitted install pins the catalog's current version. return true @@ -963,7 +959,7 @@ func inputSchema(name string, r router.Route) obj { case "labelId", "userId", "assigneeId", "executorId", "refId", "parentId", "projectRef", "targetId": return uuid() case "projectId": - return obj{"type": "string", "format": "uuid", "description": "The Project the run's Workspace provisions in. Frozen at creation for the run's whole lifetime. Authorized against the caller's own collaboration space: an unknown, archived or foreign Project is 404. Omit it for a graph with no agent node; a run without one fails honestly at its first agent node."} + return obj{"type": "string", "format": "uuid", "description": "The Project the run's Workspace provisions in. Required: a run that names none is refused with 409 workflow_run_project_required, never persisted without one. Frozen at creation for the run's whole lifetime. Authorized against the caller's own collaboration space: an unknown, archived or foreign Project is 404."} case "ids": return array(uuid()) case "filter", "properties", "input", "values", "graph": diff --git a/internal/contract/openapi_test.go b/internal/contract/openapi_test.go index 3c23159b7..9539e59c3 100644 --- a/internal/contract/openapi_test.go +++ b/internal/contract/openapi_test.go @@ -114,3 +114,18 @@ func TestPluginRoutesInDocument(t *testing.T) { t.Fatalf("DELETE plugins required fields = %v", deleteRequired) } } + +// A direct run states the Project it freezes, and the document has to say so: the runtime refuses a +// run that names none with 409 workflow_run_project_required, so a schema that let a client omit +// projectId would describe a request the service always rejects. The other three fields stay +// optional — a run defaults its name to the workflow's, pins the latest snapshot when none is given, +// and may carry no input — which is why the assertion is the exact set rather than membership. +func TestWorkflowRunRouteRequiresProject(t *testing.T) { + paths := Document()["paths"].(map[string]any) + post := paths["/api/v1/tenants/{tid}/workflows/{wfid}/runs"].(map[string]any)["post"].(map[string]any) + body := post["requestBody"].(map[string]any)["content"].(map[string]any)["application/json"].(map[string]any)["schema"].(map[string]any) + required := body["required"].([]string) + if len(required) != 1 || required[0] != "projectId" { + t.Fatalf("POST workflow runs required fields = %v; want [projectId]", required) + } +} diff --git a/internal/core/workflow_run.go b/internal/core/workflow_run.go index 8b116ab66..4dbc26bd5 100644 --- a/internal/core/workflow_run.go +++ b/internal/core/workflow_run.go @@ -109,7 +109,7 @@ func createWorkflowRun(t *transaction, r *PublicRequest, uid string) Object { return composeWorkflowRun(t, t.one("SELECT * FROM workflow_runs WHERE id=$1 AND tenant_id=$2", id, r.TenantID)) } -// resolveWorkflowRunProject authorizes and returns the optional Project a direct run freezes (D3). +// resolveWorkflowRunProject authorizes and returns the Project a direct run freezes (D3). // The Project is the one thing that says where the run's Workspace may provision — // createWorkflowRunWorkspace refuses a run without one — and it is stated by the caller rather than // inferred, because nothing else knows it: a workflow document belongs to a tenant, not to a Project, From 0198ee0bf9996dc78b844a7fdf1041ed9fd127f4 Mon Sep 17 00:00:00 2001 From: Kyber5323 Date: Sat, 17 Oct 2026 21:44:54 +0800 Subject: [PATCH 8/9] chore(core): renumber the run-identity migration to 0039 Main now carries 0038_agent_workspace_blocked_deadline.sql (cloud#62), so the run-identity migration moves one further along: 0038_workflow_run_identity.sql becomes 0039, with the test name, upgrade baseline (from 0038), migrations README entries (zh and en) and the eol pin following it. Main keeps its own 0038 section and file untouched. Co-Authored-By: Claude Code --- .gitattributes | 2 +- integration/migration_upgrade_path_test.go | 18 +++++++++--------- integration/revision_upgrade_test.go | 2 +- internal/core/execution_work.go | 2 +- ...tity.sql => 0039_workflow_run_identity.sql} | 0 internal/core/migrations/README.en.md | 6 +++--- internal/core/migrations/README.md | 6 +++--- internal/core/workflow_run_identity_db_test.go | 4 ++-- 8 files changed, 20 insertions(+), 20 deletions(-) rename internal/core/migrations/{0035_workflow_run_identity.sql => 0039_workflow_run_identity.sql} (100%) diff --git a/.gitattributes b/.gitattributes index bd6ef6b67..9ef377e92 100644 --- a/.gitattributes +++ b/.gitattributes @@ -17,4 +17,4 @@ internal/core/migrations/0033_revision_resume.sql text eol=lf # position for a different reason — it belongs to another PR, which has to make this call for it.) # Everything introduced since main is pinned here instead, while no released database has recorded # it yet; after merge that would no longer be true. -internal/core/migrations/0035_workflow_run_identity.sql text eol=lf +internal/core/migrations/0039_workflow_run_identity.sql text eol=lf diff --git a/integration/migration_upgrade_path_test.go b/integration/migration_upgrade_path_test.go index 9ffa9d39c..81c1893fc 100644 --- a/integration/migration_upgrade_path_test.go +++ b/integration/migration_upgrade_path_test.go @@ -1048,11 +1048,11 @@ func TestMigration0033RevisionResumeAppliesFreshAndUpgrades(t *testing.T) { } } -// 0035 turns the WorkflowRun into an execution aggregate the control plane accepts as a work owner: +// 0039 turns the WorkflowRun into an execution aggregate the control plane accepts as a work owner: // the frozen run identity, the workspace binding, the exactly-one-owner discriminator on work and -// the polymorphic runtime reservation. Fresh databases get all of it, an upgrade from 0034 gets it +// the polymorphic runtime reservation. Fresh databases get all of it, an upgrade from 0037 gets it // without a backfill, and a rerun changes nothing. -func TestMigration0035WorkflowRunIdentityAppliesFreshAndUpgrades(t *testing.T) { +func TestMigration0039WorkflowRunIdentityAppliesFreshAndUpgrades(t *testing.T) { columns := func(pool *sql.DB) int { t.Helper() var n int @@ -1078,7 +1078,7 @@ func TestMigration0035WorkflowRunIdentityAppliesFreshAndUpgrades(t *testing.T) { AND indexname IN ('one_unregistered_execution_work','one_unregistered_workflow_execution_work')`).Scan(&names)) return names } - // A polymorphic run reference cannot be a foreign key, so 0035 drops the IssueRun FK instead of + // A polymorphic run reference cannot be a foreign key, so 0039 drops the IssueRun FK instead of // reworking it. The column stays; only the constraint goes. Resolved through the search path // rather than by name alone, which would count another schema's constraint of the same name. reservationFK := func(pool *sql.DB) int { @@ -1089,19 +1089,19 @@ func TestMigration0035WorkflowRunIdentityAppliesFreshAndUpgrades(t *testing.T) { return n } - freshPool, _ := testSchema(t, "test_wfi35f_") + freshPool, _ := testSchema(t, "test_wfi39f_") migratedTwice(t, newStoreOnSchema(t, freshPool)) if got, names, idx, fk := columns(freshPool), constraints(freshPool), indexes(freshPool), reservationFK(freshPool); got != 4 || names != "execution_work_owner_exclusive,workspace_run_binding_exclusive" || idx != "one_unregistered_execution_work,one_unregistered_workflow_execution_work" || fk != 0 { t.Fatalf("fresh: columns %d constraints %q indexes %q reservation fk %d", got, names, idx, fk) } - pool, _ := testSchema(t, "test_wfi35_") - applyMigrationsUpTo(t, pool, migrationsBefore(t, "0035_workflow_run_identity.sql")) + pool, _ := testSchema(t, "test_wfi39_") + applyMigrationsUpTo(t, pool, migrationsBefore(t, "0039_workflow_run_identity.sql")) if got, names, fk := columns(pool), constraints(pool), reservationFK(pool); got != 0 || names != "" || fk != 1 { - t.Fatalf("before 0035: columns %d constraints %q reservation fk %d", got, names, fk) + t.Fatalf("before 0039: columns %d constraints %q reservation fk %d", got, names, fk) } migratedTwice(t, newStoreOnSchema(t, pool)) if got, names, idx, fk := columns(pool), constraints(pool), indexes(pool), reservationFK(pool); got != 4 || names != "execution_work_owner_exclusive,workspace_run_binding_exclusive" || idx != "one_unregistered_execution_work,one_unregistered_workflow_execution_work" || fk != 0 { - t.Fatalf("after 0035: columns %d constraints %q indexes %q reservation fk %d", got, names, idx, fk) + t.Fatalf("after 0039: columns %d constraints %q indexes %q reservation fk %d", got, names, idx, fk) } } diff --git a/integration/revision_upgrade_test.go b/integration/revision_upgrade_test.go index e666eea28..860523b1f 100644 --- a/integration/revision_upgrade_test.go +++ b/integration/revision_upgrade_test.go @@ -93,7 +93,7 @@ func TestRevisionUpgradePreservesPublishedWorkflowSchema(t *testing.T) { must(t, err) } must(t, tx.Commit()) - // The run document strips the identity columns a later migration (0035) added: they land as + // The run document strips the identity columns a later migration (0039) added: 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'])::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` diff --git a/internal/core/execution_work.go b/internal/core/execution_work.go index 783a880e2..ad8d46a21 100644 --- a/internal/core/execution_work.go +++ b/internal/core/execution_work.go @@ -107,7 +107,7 @@ func claimWorkItem(t *transaction) Object { } // A work item names the run it belongs to in one of two columns: agent work carries run_id (an // issue_runs id), workflow work carries workflow_run_id (a workflow_runs id) and leaves run_id - // NULL by the mutual-exclusion CHECK 0035 added. The Controller is handed the run id as the + // NULL by the mutual-exclusion CHECK 0039 added. The Controller is handed the run id as the // operation id and echoes it back on dispatch, which resolves either column — so the claim has to // project one value. Reading the agent-only column instead hands back the empty string a workflow // row produces, and every dispatch of that item is then refused as a conflict it cannot resolve. diff --git a/internal/core/migrations/0035_workflow_run_identity.sql b/internal/core/migrations/0039_workflow_run_identity.sql similarity index 100% rename from internal/core/migrations/0035_workflow_run_identity.sql rename to internal/core/migrations/0039_workflow_run_identity.sql diff --git a/internal/core/migrations/README.en.md b/internal/core/migrations/README.en.md index ce78619bc..3a48209e0 100644 --- a/internal/core/migrations/README.en.md +++ b/internal/core/migrations/README.en.md @@ -115,9 +115,9 @@ only validate schema. `TestMigration0035ModelAttemptsFreshAndUpgradeWithoutBackf Adds issue_runs.workspace_blocked_at, workspace_deadline_at, workspace_blocked_source (blocked_write / first_observed) with a provisioning-phase partial index, and widens agent_run_residuals' reason constraint to agent_start_failed / workspace_unavailable. The first transaction persisting `blocked` anchors the absolute deadline; legacy blocked creations have no immutable first-blocked evidence, are not backfilled, and receive one default window from the first policy scan as first_observed. Defer, manual retry, re-claiming, restarts and configuration changes never renew; only the creation's real terminal ends the window. Purely forward and retry-safe, without rewriting historical outcomes. Fresh/upgrade/repeat coverage: TestMigration0038WorkspaceBlockedDeadlineAppliesFreshAndUpgrades. -## 0035: WorkflowRun identity +## 0039: WorkflowRun identity -`0035_workflow_run_identity.sql` hands run identity and work-item ownership to the control plane. +`0039_workflow_run_identity.sql` hands run identity and work-item ownership to the control plane. `workflow_runs` gains `created_by` and `project_id` (both nullable: rows written before this migration carry no recoverable actor, and `project_id` is filled by the invocation path that knows a Project), which new runs freeze at creation. `workspaces` gains a mutually exclusive `workflow_run_id`, so a run @@ -128,4 +128,4 @@ becomes a polymorphic run reference: the `maintenance_run_id` foreign key is dro reworked, because a polymorphic reference cannot be a foreign key, and the liveness check moves to the code that grants on it. Additive: columns, constraints and indexes, no backfill. -`TestMigration0035WorkflowRunIdentityAppliesFreshAndUpgrades` covers fresh, 0034 upgrade and rerun. +`TestMigration0039WorkflowRunIdentityAppliesFreshAndUpgrades` covers fresh, 0038 upgrade and rerun. diff --git a/internal/core/migrations/README.md b/internal/core/migrations/README.md index 863861109..c30eefdb0 100644 --- a/internal/core/migrations/README.md +++ b/internal/core/migrations/README.md @@ -114,9 +114,9 @@ Cloud 领域方法。冻结与撤销规则见 [个人模型连接](../../../docs 新增 issue_runs.workspace_blocked_at、workspace_deadline_at、workspace_blocked_source（blocked_write / first_observed）与 provisioning 期部分索引，并把 agent_run_residuals 的 reason 约束扩为 agent_start_failed / workspace_unavailable。首次持久写入 blocked 的同事务锚定绝对截止；存量 blocked 无不可变首次证据，不回填，由首次策略扫描以 first_observed 给一次默认窗口。defer、人工 retry、重新领取、重启与配置调整不续期；只有原创建真实终态结束窗口。纯前向、可重复执行，不改历史执行结果。新库/升级/重复执行见 TestMigration0038WorkspaceBlockedDeadlineAppliesFreshAndUpgrades。 -## 0035：WorkflowRun 运行身份 +## 0039：WorkflowRun 运行身份 -`0035_workflow_run_identity.sql` 把运行身份与工作项所有权交给控制面。`workflow_runs` 新增 +`0039_workflow_run_identity.sql` 把运行身份与工作项所有权交给控制面。`workflow_runs` 新增 `created_by` 与 `project_id` 两列（均可空：本迁移之前的行没有可恢复的发起人，而 `project_id` 由知道 Project 的调用路径写入），新运行在创建时冻结二者；`workspaces` 新增互斥的 `workflow_run_id`，使运行 Workspace 可以属于 WorkflowRun；`execution_work` 同样变为「恰有一个所有者」的判别形状，原单列唯一索引 @@ -124,4 +124,4 @@ Workspace 可以属于 WorkflowRun；`execution_work` 同样变为「恰有一 `runtime_controls` 的预留改为多态运行引用，`maintenance_run_id` 的外键被删除而非改写——多态引用无法做 外键，存活校验移到授予它的代码里。只新增列、约束与索引，不回填。 -`TestMigration0035WorkflowRunIdentityAppliesFreshAndUpgrades` 验证空库、从 0034 升级及重复迁移。 +`TestMigration0039WorkflowRunIdentityAppliesFreshAndUpgrades` 验证空库、从 0038 升级及重复迁移。 diff --git a/internal/core/workflow_run_identity_db_test.go b/internal/core/workflow_run_identity_db_test.go index e010bd0a8..8a9423259 100644 --- a/internal/core/workflow_run_identity_db_test.go +++ b/internal/core/workflow_run_identity_db_test.go @@ -1,6 +1,6 @@ package core -// DB tests for the WorkflowRun execution identity (migration 0035, specs#99 D3/D8): a run's frozen +// DB tests for the WorkflowRun execution identity (migration 0039, specs#99 D3/D8): a run's frozen // identity drives the workspace lifecycle, the workspace binds to workflow_runs, and the control // plane's seams — enqueue, dispatch, runtime permit and the settlement hooks — accept a WorkflowRun // as the run of either kind. Session execution stays unwired (the scheduler still fails runs on @@ -259,7 +259,7 @@ func TestWorkflowExecutionWorkDispatchSeam(t *testing.T) { // driven through the real control actions rather than by calling nodeDispatch directly. // // The seam test above dispatches directly and so cannot see this: a workflow work item carries -// workflow_run_id and leaves run_id NULL by 0035's mutual-exclusion CHECK, so a claim that projects +// workflow_run_id and leaves run_id NULL by 0039's mutual-exclusion CHECK, so a claim that projects // the agent-only column hands the Controller the empty string, and every dispatch of that item then // matches no row and is refused as a conflict it can never resolve. func TestWorkflowExecutionWorkClaimCarriesItsRunID(t *testing.T) { From 73a81893f7eb1d77a7aa71a4492b32c9d4c983ca Mon Sep 17 00:00:00 2001 From: Kyber5323 Date: Mon, 19 Oct 2026 09:47:59 +0800 Subject: [PATCH 9/9] docs(workflows): record the frozen run identity and run Workspace ownership main rebuilt docs/ into bilingual per-domain documents (cloud#85), so the behavior this PR adds is documented there: the workflow run section gains the mandatory projectId and the created_by/project_id frozen at creation, and the run Workspace ownership (workspaces.workflow_run_id, mutually exclusive with issue_run_id) is stated where the Workspace aggregates and the runtime control reservation are described. Both languages change together, per the Documentation rules. Co-Authored-By: Claude Code --- docs/agent-runs/issue-runs.en.md | 3 ++- docs/agent-runs/issue-runs.md | 3 ++- docs/runtime/runtime-control.en.md | 5 +++-- docs/runtime/runtime-control.md | 4 ++-- docs/runtime/workspaces.en.md | 2 ++ docs/runtime/workspaces.md | 2 ++ docs/workflows/workflows.en.md | 9 +++++++++ docs/workflows/workflows.md | 6 ++++++ 8 files changed, 28 insertions(+), 6 deletions(-) diff --git a/docs/agent-runs/issue-runs.en.md b/docs/agent-runs/issue-runs.en.md index 8c8681137..5cac12f19 100644 --- a/docs/agent-runs/issue-runs.en.md +++ b/docs/agent-runs/issue-runs.en.md @@ -54,7 +54,8 @@ Cloud in the transaction that takes over the previous phase's evidence. [the delete-side limit](#limits)). A run Workspace belongs to exactly this run (`workspaces.issue_run_id` and `issue_runs.workspace_id` -reference each other uniquely). Its creator and operation actor are the user who triggered the run, it +reference each other uniquely; a WorkflowRun's run Workspace likewise belongs only to it, see +[Workflows](../workflows/workflows.en.md#runs)). Its creator and operation actor are the user who triggered the run, it is cloned from the Project's default branch, it never appears in any public Workspace list or endpoint, and it accepts no other run or interactive session. diff --git a/docs/agent-runs/issue-runs.md b/docs/agent-runs/issue-runs.md index 55cd175df..a26c7327b 100644 --- a/docs/agent-runs/issue-runs.md +++ b/docs/agent-runs/issue-runs.md @@ -40,7 +40,8 @@ Cloud 在接管上一阶段证据的事务里写入。 - 创建操作停在 `blocked` 超过上限时，运行按 `failed`、`done` 收尾并保留残留责任，原创建操作失败、释放 Project 的操作槽（见 [创建 blocked 的有界放弃](startup-settlement.md#创建-blocked-的有界放弃)）。 - **`done` 不代表运行 Workspace 已经删除**，只有删除操作成功才是（见 [删除侧的上限](#上限)）。 -运行 Workspace 只属于这一次运行（`workspaces.issue_run_id` 与 `issue_runs.workspace_id` 互相唯一引用），以触发运行的用户为创建者 +运行 Workspace 只属于这一次运行（`workspaces.issue_run_id` 与 `issue_runs.workspace_id` 互相唯一引用；一次 WorkflowRun 的运行 Workspace 同样只属于它， +见 [工作流](../workflows/workflows.md#运行)），以触发运行的用户为创建者 和操作发起者，从 Project 默认分支克隆，不出现在任何公开的 Workspace 列表或接口里，也不接受其他运行或交互式会话。 ## 后台循环 diff --git a/docs/runtime/runtime-control.en.md b/docs/runtime/runtime-control.en.md index e08fcb43a..a4ce62752 100644 --- a/docs/runtime/runtime-control.en.md +++ b/docs/runtime/runtime-control.en.md @@ -75,8 +75,9 @@ idle ───────────────▶ acquiring ───── **Lifecycle operations share the same write boundary.** As soon as a lifecycle operation is registered it puts control into `maintenance` (recording the operation ID and incrementing the control epoch), so page sessions cannot operate meanwhile. When the operation ends, control becomes `draining` -and settles by the rules above. After an Agent run's run Workspace is created, control becomes a -`maintenance` owned by that run, which bounds its sessions and deliveries from then on. +and settles by the rules above. After a run Workspace's creation operation succeeds, control becomes a +`maintenance` owned by the run that created it — the reservation records the run id, and both IssueRuns +and WorkflowRuns count — which bounds its sessions and deliveries from then on. An administrator's ordinary operations cannot take over control someone else holds; force-stop is the way to interrupt immediately. diff --git a/docs/runtime/runtime-control.md b/docs/runtime/runtime-control.md index eca9ffbfc..57203a0cf 100644 --- a/docs/runtime/runtime-control.md +++ b/docs/runtime/runtime-control.md @@ -60,8 +60,8 @@ idle ───────────────▶ acquiring ───── - 幂等重放返回历史响应，但不会让已撤回的资格复活。每次状态变化都写入 `runtime_control_events`，保留持有者和原会话。 **生命周期操作与资格共用同一个写边界。** 生命周期操作一旦登记，就把资格置为 `maintenance`（记录操作 ID，控制代次加一）， -页面会话在此期间不能操作；操作结束后资格进入 `draining`，按上面的规则结清。Agent 运行的运行 Workspace 创建成功后，资格转为 -属于该运行的 `maintenance`，之后的会话和交付都以它为边界。 +页面会话在此期间不能操作；操作结束后资格进入 `draining`，按上面的规则结清。运行 Workspace 的创建操作成功后，资格转为属于创建它的那次运行的 +`maintenance`（预留记录运行 ID，IssueRun 与 WorkflowRun 都认），之后的会话和交付都以它为边界。 管理员的普通操作不能抢占别人持有的资格；需要立即打断时使用强停。 diff --git a/docs/runtime/workspaces.en.md b/docs/runtime/workspaces.en.md index 1a034b844..0449d608f 100644 --- a/docs/runtime/workspaces.en.md +++ b/docs/runtime/workspaces.en.md @@ -76,6 +76,8 @@ by the Controller on its behalf. **Run Workspaces.** Agent IssueRuns also use isolated Workspaces (`issue_run_id` set). Cloud creates and deletes them under the identity of the user who triggered the run, through the same operations, but they never appear in any public list or endpoint (see [Agent IssueRuns](../agent-runs/issue-runs.en.md)). +A run Workspace belongs to exactly one run: `issue_run_id` and `workflow_run_id` are mutually exclusive +columns, and the latter belongs to one [WorkflowRun](../workflows/workflows.en.md#runs). ## Durable operations diff --git a/docs/runtime/workspaces.md b/docs/runtime/workspaces.md index de6c67782..9e4b43984 100644 --- a/docs/runtime/workspaces.md +++ b/docs/runtime/workspaces.md @@ -64,6 +64,8 @@ Cloud 不直接调用 Substrate 或 Node。它只保存意图和证据，并按 **运行 Workspace。** Agent IssueRun 也使用 isolated Workspace（`issue_run_id` 非空），由 Cloud 以触发运行的用户 身份建删，走同一套操作，但不出现在任何公开列表或接口里（见 [Agent IssueRun](../agent-runs/issue-runs.md)）。 +运行 Workspace 恰好属于一次运行：`issue_run_id` 与 `workflow_run_id` 两列互斥，后者属于一次 +[WorkflowRun](../workflows/workflows.md#运行)。 ## 持久操作 diff --git a/docs/workflows/workflows.en.md b/docs/workflows/workflows.en.md index 27b7b0f52..79746d019 100644 --- a/docs/workflows/workflows.en.md +++ b/docs/workflows/workflows.en.md @@ -76,6 +76,15 @@ Workflows have three start entries, all sharing one admission gate: If any check fails, no run, claimed interaction or activity is left behind. The executed version is frozen onto the run, so the run still knows which version it executed after the workflow is edited or published again. +**A run's identity is frozen at creation.** A direct run must name its `projectId`: a missing one is `409 +workflow_run_project_required`, an unusable, archived or foreign-space Project is refused as `404`, and one being +deleted is `409 resource_unavailable`. The run writes the initiating user (`created_by`) and that Project +(`project_id`) into its row and never changes them — a workflow document belongs to a tenant, not to a Project, so +there is nothing to default to, and the run's Workspace, its clone and its whole checkout land in the Project that +was named. A run Workspace belongs to the run that created it: `workspaces.workflow_run_id` is mutually exclusive +with an IssueRun's `issue_run_id`, and the `maintenance` reserved for it in runtime control is recorded against the +same run identity (see [Runtime control](../runtime/runtime-control.en.md)). + ## Scheduling WorkflowRuns are carried forward by a scheduling loop in the Cloud server (`AdvanceWorkflowRunsOnce`, every 10 diff --git a/docs/workflows/workflows.md b/docs/workflows/workflows.md index ff2796894..2f647692d 100644 --- a/docs/workflows/workflows.md +++ b/docs/workflows/workflows.md @@ -57,6 +57,12 @@ Agent 节点不在图里写死 Agent CLI 或模型 ID，而是引用一个**执 任何一项不通过都不会留下运行、已认领的交互或活动。被执行的版本冻结在运行上，运行之后工作流再被编辑或发布，它仍然知道自己执行的是哪个版本。 +**运行的身份在创建时冻结。** 直接创建运行时 `projectId` 必填：缺失返回 `409 workflow_run_project_required`，不可用、已归档或属于其他空间的 Project +按 `404` 拒绝，正在删除的返回 `409 resource_unavailable`。运行把发起人（`created_by`）和这个 Project（`project_id`）写进运行行，之后不再变化——工作流文档 +属于租户而不是某个 Project，没有可以默认的对象，而运行的 Workspace、它的 clone 和整份检出都落在被指定的 Project 里。运行 Workspace 归属于创建它的那次运行： +`workspaces.workflow_run_id` 与 IssueRun 的 `issue_run_id` 互斥，运行时控制里为它预留的 `maintenance` 同样按运行身份记录（见 +[运行时控制](../runtime/runtime-control.md)）。 + ## 调度 WorkflowRun 由 Cloud server 里的调度循环推进（`AdvanceWorkflowRunsOnce`，每 10 秒一轮）：