diff --git a/.gitattributes b/.gitattributes index 571382da5..9ef377e92 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/0039_workflow_run_identity.sql text eol=lf diff --git a/api/openapi.json b/api/openapi.json index 362505ded..4bf4a78ff 100644 --- a/api/openapi.json +++ b/api/openapi.json @@ -22305,10 +22305,18 @@ "name": { "type": "string" }, + "projectId": { + "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" + }, "snapshotId": { "type": "string" } }, + "required": [ + "projectId" + ], "type": "object" } } 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 秒一轮）： diff --git a/frontend/src/api/generated.schemas.ts b/frontend/src/api/generated.schemas.ts index d98bc99cb..289a80b51 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. 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/integration/migration_upgrade_path_test.go b/integration/migration_upgrade_path_test.go index 2a3d54778..81c1893fc 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) } } + +// 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 0037 gets it +// without a backfill, and a rerun changes nothing. +func TestMigration0039WorkflowRunIdentityAppliesFreshAndUpgrades(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 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 { + 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_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_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 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 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 1d69401c9..860523b1f 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 (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` 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/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 new file mode 100644 index 000000000..077af0134 --- /dev/null +++ b/integration/workflow_run_project_test.go @@ -0,0 +1,210 @@ +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))) +} + +// 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. +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() + // 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) + } + 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 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/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..f47554b42 100644 --- a/internal/contract/openapi.go +++ b/internal/contract/openapi.go @@ -958,6 +958,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. 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/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/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..ad8d46a21 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 { @@ -80,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 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. + // + // 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/migrations/0039_workflow_run_identity.sql b/internal/core/migrations/0039_workflow_run_identity.sql new file mode 100644 index 000000000..56ab30374 --- /dev/null +++ b/internal/core/migrations/0039_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/migrations/README.en.md b/internal/core/migrations/README.en.md index 42c3fe688..3a48209e0 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. + +## 0039: WorkflowRun identity + +`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 +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. + +`TestMigration0039WorkflowRunIdentityAppliesFreshAndUpgrades` covers fresh, 0038 upgrade and rerun. diff --git a/internal/core/migrations/README.md b/internal/core/migrations/README.md index d23dfd558..c30eefdb0 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。 + +## 0039：WorkflowRun 运行身份 + +`0039_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` 的外键被删除而非改写——多态引用无法做 +外键，存活校验移到授予它的代码里。只新增列、约束与索引，不回填。 + +`TestMigration0039WorkflowRunIdentityAppliesFreshAndUpgrades` 验证空库、从 0038 升级及重复迁移。 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..40e193489 100644 --- a/internal/core/run_workspace.go +++ b/internal/core/run_workspace.go @@ -46,6 +46,72 @@ 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. 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") + 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} + } + // 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} + } + 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..4dbc26bd5 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 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") @@ -91,10 +94,57 @@ 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. 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") + 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 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. +// +// 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") + 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") +} 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..8a9423259 --- /dev/null +++ b/internal/core/workflow_run_identity_db_test.go @@ -0,0 +1,320 @@ +package core + +// 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 +// 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", `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) + 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")) + } +} + +// 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 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) { + 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) + } +}