diff --git a/frontend/src/features/issues/components/workflow-interaction-composer.tsx b/frontend/src/features/issues/components/workflow-interaction-composer.tsx
index 7d6f7e1a2..1d33d0c4f 100644
--- a/frontend/src/features/issues/components/workflow-interaction-composer.tsx
+++ b/frontend/src/features/issues/components/workflow-interaction-composer.tsx
@@ -1,5 +1,6 @@
import { ChevronDown, ChevronUp, Sparkles, Workflow, X } from 'lucide-react'
-import { useEffect, useRef, useState } from 'react'
+import { useMutation } from '@tanstack/react-query'
+import { useRef, useState } from 'react'
import { Badge } from '@/components/ui/badge'
import { Button } from '@/components/ui/button'
import { Skeleton } from '@/components/ui/skeleton'
@@ -8,6 +9,7 @@ import {
useAssistWorkflow,
useConfirmWorkflow,
useFormDescriptor,
+ useWorkflowSnapshotForm,
} from '@/features/issues/api'
import { AssistSuggestions } from '@/features/issues/components/assist-suggestions'
import { ConfirmReview } from '@/features/issues/components/confirm-review'
@@ -22,16 +24,21 @@ import type {
} from '@/features/issues/types'
/** Seeds the editable values from the descriptor's declared defaults. */
-function seedValues(descriptor: FormDescriptor): FormValues {
+function seedValues(descriptor: FormDescriptor, snapshotId?: string): FormValues {
const values: FormValues = {}
for (const field of descriptor.fields) {
if (field.defaultValue !== undefined && field.defaultValue !== null) {
values[field.key] = field.defaultValue
}
}
+ if (snapshotId && descriptor.fields.some(isSnapshotSelector)) values['snapshotId'] = snapshotId
return values
}
+function isSnapshotSelector(field: FormField): boolean {
+ return field.key === 'snapshotId' && field.type === 'select'
+}
+
/** Required fields the user has not filled yet — the only client-side gate on Review (§38.15). */
function missingRequired(descriptor: FormDescriptor, values: FormValues): FormField[] {
return descriptor.fields.filter((field) => {
@@ -96,18 +103,11 @@ function mergeContextRefs(applied: ContextRefRef[], picked: ContextRefRef[]): Co
}
/** Editable form state for one Workflow interaction. Pure UI state — nothing here is persisted. */
-function useWorkflowForm(descriptor: FormDescriptor | undefined) {
- const [values, setValues] = useState
({})
+function useWorkflowForm(descriptor: FormDescriptor, snapshotId?: string) {
+ const [values, setValues] = useState(() => seedValues(descriptor, snapshotId))
const [contextRefs, setContextRefs] = useState([])
const [suggestion, setSuggestion] = useState(null)
const [appliedKeys, setAppliedKeys] = useState([])
- const seeded = useRef(null)
-
- useEffect(() => {
- if (!descriptor || seeded.current === descriptor.formRef) return
- seeded.current = descriptor.formRef
- setValues(seedValues(descriptor))
- }, [descriptor])
function setField(key: string, next: unknown) {
setValues((current) => ({ ...current, [key]: next }))
@@ -174,11 +174,16 @@ function WorkflowFormBody(props: {
missing: FormField[]
onAssist: () => void
onReview: () => void
+ onFieldChange: (key: string, next: unknown) => void
}) {
const { descriptor, form } = props
return (
-
+
{form.suggestion ? (
void
onRemove: () => void
}) {
@@ -255,6 +262,7 @@ function ComposerHeader({
type="button"
className="ml-auto rounded p-0.5 hover:bg-muted"
onClick={onRemove}
+ disabled={pending}
aria-label="取消"
>
@@ -278,6 +286,7 @@ function ComposerBody(props: {
onReview: () => void
onBack: () => void
onConfirm: () => void
+ onFieldChange: (key: string, next: unknown) => void
}) {
if (props.reviewing) {
return (
@@ -307,6 +316,7 @@ function ComposerBody(props: {
missing={props.missing}
onAssist={props.onAssist}
onReview={props.onReview}
+ onFieldChange={props.onFieldChange}
/>
{props.assistFailed ? (
AI 建议当前不可用。
@@ -315,6 +325,53 @@ function ComposerBody(props: {
)
}
+/** Keeps one interaction across retries and locks before React can render a second click. */
+function useWorkflowSubmission({
+ slug,
+ issueId,
+ target,
+ onConfirmed,
+}: {
+ slug: string
+ issueId: string
+ target: CollaborationTargetSummary
+ onConfirmed: () => void
+}) {
+ const interactionId = useRef(null)
+ const locked = useRef(false)
+ const confirm = useConfirmWorkflow(slug, issueId)
+ const submission = useMutation({
+ mutationFn: async (input: { values: FormValues; contextRefs?: ContextRefRef[] }) => {
+ if (!interactionId.current) {
+ const interaction = await openWorkflowInteraction(
+ slug,
+ issueId,
+ target.id,
+ `@${target.displayName}`,
+ )
+ interactionId.current = interaction.id
+ }
+ return confirm.mutateAsync({ interactionId: interactionId.current, ...input })
+ },
+ onSuccess: onConfirmed,
+ onError: () => {
+ locked.current = false
+ },
+ })
+
+ function submit(input: { values: FormValues; contextRefs?: ContextRefRef[] }) {
+ if (locked.current) return
+ locked.current = true
+ submission.mutate(input)
+ }
+
+ return {
+ submit,
+ isPending: submission.isPending || submission.isSuccess,
+ isError: submission.isError,
+ }
+}
+
/**
* The Form Mode configuration panel for a workflow the user has selected but not yet submitted. It is
* a **draft**: the comment, the interaction and the run are all created only when 确认执行 is pressed,
@@ -334,23 +391,72 @@ export function WorkflowInteractionComposer({
onRemove: () => void
}) {
const formRef = target.interactionDescriptor.formRef ?? undefined
- const descriptorQuery = useFormDescriptor(slug, formRef, issueId)
- const assist = useAssistWorkflow(slug, issueId)
- const confirm = useConfirmWorkflow(slug, issueId)
- const [reviewing, setReviewing] = useState(false)
- const [collapsed, setCollapsed] = useState(false)
+ const [snapshotId, setSnapshotId] = useState()
+ const live = useFormDescriptor(slug, formRef, issueId)
+ const snapshot = useWorkflowSnapshotForm(slug, target.id, snapshotId, issueId)
+ const descriptorQuery = snapshotId ? snapshot : live
+ const confirm = useWorkflowSubmission({ slug, issueId, target, onConfirmed })
const descriptor = descriptorQuery.data
- const form = useWorkflowForm(descriptor)
if (descriptorQuery.isPending) return
if (descriptorQuery.isError || !descriptor) {
return (
-
- 当前部署无法加载该 Workflow 的表单（form descriptor 不可用）。
-
+
+
+ 当前部署无法加载该 Workflow 的表单（form descriptor 不可用）。
+
+
+
{descriptor.description}
) : null}
- {!collapsed && (
+ {props.loading && }
+ {!collapsed && !props.loading && (
setReviewing(true)}
onBack={() => setReviewing(false)}
- onConfirm={() => {
- void submit()
- }}
+ onConfirm={() => confirm.submit(confirmInput)}
+ onFieldChange={setField}
/>
)}
diff --git a/frontend/src/features/issues/components/workflow-interaction.test.tsx b/frontend/src/features/issues/components/workflow-interaction.test.tsx
index 187322ccb..3dc9270fa 100644
--- a/frontend/src/features/issues/components/workflow-interaction.test.tsx
+++ b/frontend/src/features/issues/components/workflow-interaction.test.tsx
@@ -1,4 +1,4 @@
-import { screen, waitFor, within } from '@testing-library/react'
+import { fireEvent, screen, waitFor, within } from '@testing-library/react'
import userEvent from '@testing-library/user-event'
import { http, HttpResponse } from 'msw'
import { describe, expect, it, vi } from 'vitest'
@@ -238,6 +238,30 @@ async function reviewAndConfirm(user: ReturnType) {
await user.click(await screen.findByRole('button', { name: '确认执行' }))
}
+function serveOldSnapshot() {
+ const read = vi.fn<(issueId: string | null) => void>()
+ server.use(
+ http.get('/api/v1/tenants/t1/workflows/wf1/snapshots/snap-2/form', ({ request }) => {
+ read(new URL(request.url).searchParams.get('issueId'))
+ return HttpResponse.json({
+ ...launchDescriptor,
+ snapshotId: 'snap-2',
+ fields: [
+ ...launchDescriptor.fields.filter((field) => field.key !== 'prompt'),
+ {
+ key: 'oldInput',
+ label: 'Old input',
+ type: 'text',
+ required: true,
+ defaultValue: 'old default',
+ },
+ ],
+ })
+ }),
+ )
+ return read
+}
+
describe('WorkflowInteractionComposer', () => {
it('renders exactly the fields the descriptor declares', async () => {
serveComposer()
@@ -391,6 +415,92 @@ describe('WorkflowInteractionComposer', () => {
)
})
+ it('locks both confirm clicks while the first interaction is still being created', async () => {
+ const { comment, confirm } = serveComposer({ descriptor: platformDescriptor })
+ let release = vi.fn<() => void>()
+ const gate = new Promise((resolve) => {
+ release = vi.fn<() => void>(resolve)
+ })
+ server.use(
+ http.post('/api/v1/tenants/t1/issues/i1/comments', async ({ request }) => {
+ comment(await request.json())
+ await gate
+ return HttpResponse.json({ resource: { id: 'c1' } })
+ }),
+ )
+ const user = userEvent.setup()
+ const { onConfirmed } = renderComposer()
+ await user.click(await screen.findByRole('button', { name: '审阅' }))
+ const button = screen.getByRole('button', { name: '确认执行' })
+ fireEvent.click(button)
+ fireEvent.click(button)
+
+ try {
+ await waitFor(() => expect(comment).toHaveBeenCalledTimes(1))
+ expect(screen.getByRole('button', { name: '确认中…' })).toBeDisabled()
+ expect(screen.getByRole('button', { name: '返回修改' })).toBeDisabled()
+ expect(screen.getByRole('button', { name: '取消' })).toBeDisabled()
+ expect(confirm).not.toHaveBeenCalled()
+ } finally {
+ release()
+ }
+ await waitFor(() => expect(onConfirmed).toHaveBeenCalledTimes(1))
+ expect(comment).toHaveBeenCalledTimes(1)
+ expect(confirm).toHaveBeenCalledTimes(1)
+ })
+
+ it('reuses its interaction after a rejected confirm, including edits before retry', async () => {
+ const { comment, confirm } = serveComposer({ descriptor: platformDescriptor })
+ let attempts = 0
+ server.use(
+ http.post('/api/v1/tenants/t1/issues/i1/interactions/ix1/confirm', async ({ request }) => {
+ confirm(await request.json())
+ attempts += 1
+ return attempts === 1
+ ? HttpResponse.json({ code: 'invalid_input' }, { status: 400 })
+ : HttpResponse.json({ resource: run() })
+ }),
+ )
+ const user = userEvent.setup()
+ const { onConfirmed } = renderComposer()
+ await screen.findByLabelText(/仓库地址/)
+ await reviewAndConfirm(user)
+ expect(await screen.findByText('确认失败，请重试。')).toBeInTheDocument()
+ expect(onConfirmed).not.toHaveBeenCalled()
+
+ await user.click(screen.getByRole('button', { name: '返回修改' }))
+ await user.type(screen.getByLabelText('Notes'), 'retry notes')
+ await reviewAndConfirm(user)
+
+ await waitFor(() => expect(onConfirmed).toHaveBeenCalledTimes(1))
+ expect(comment).toHaveBeenCalledTimes(1)
+ expect(confirm).toHaveBeenCalledTimes(2)
+ expect(confirm).toHaveBeenLastCalledWith({
+ values: {
+ repository: 'https://github.com/ora/cloud',
+ prompt: 'Review the diff',
+ notes: 'retry notes',
+ },
+ })
+ })
+
+ it('shows interaction creation failures and unlocks the form for a retry', async () => {
+ serveComposer({ descriptor: platformDescriptor })
+ server.use(
+ http.post('/api/v1/tenants/t1/issues/i1/comments', () =>
+ HttpResponse.json({ code: 'unavailable' }, { status: 503 }),
+ ),
+ )
+ const user = userEvent.setup()
+ renderComposer()
+ await screen.findByLabelText(/仓库地址/)
+ await reviewAndConfirm(user)
+
+ expect(await screen.findByText('确认失败，请重试。')).toBeInTheDocument()
+ expect(screen.getByRole('button', { name: '确认执行' })).toBeEnabled()
+ expect(screen.getByRole('button', { name: '返回修改' })).toBeEnabled()
+ })
+
it('carries the ticked context references, once each even when Assist suggested one too', async () => {
const { confirm } = serveComposer({
descriptor: launchDescriptor,
@@ -426,4 +536,109 @@ describe('WorkflowInteractionComposer', () => {
}),
)
})
+
+ it('renders the selected snapshot defaults and retains its echo instead of the newest default', async () => {
+ const { confirm } = serveComposer({ descriptor: launchDescriptor })
+ const read = serveOldSnapshot()
+ const user = userEvent.setup()
+ renderComposer()
+ await user.type(await screen.findByLabelText(/提示词/), ' obsolete draft')
+ await user.click(selectTrigger('运行版本'))
+ await user.click(await screen.findByRole('option', { name: 'v2 · 发布 2' }))
+
+ expect(await screen.findByLabelText(/Old input/)).toHaveValue('old default')
+ expect(screen.queryByLabelText(/提示词/)).not.toBeInTheDocument()
+ expect(selectTrigger('运行版本')).toHaveTextContent('v2 · 发布 2')
+ expect(read).toHaveBeenCalledWith('i1')
+ await reviewAndConfirm(user)
+ await waitFor(() =>
+ expect(confirm).toHaveBeenCalledWith({
+ values: {
+ repository: 'https://github.com/ora/cloud',
+ branch: 'main',
+ snapshotId: 'snap-2',
+ oldInput: 'old default',
+ },
+ }),
+ )
+ })
+
+ it('blocks stale review while a selected snapshot loads and after a read failure', async () => {
+ serveComposer({ descriptor: launchDescriptor })
+ let release = vi.fn<() => void>()
+ const gate = new Promise((resolve) => {
+ release = vi.fn<() => void>(resolve)
+ })
+ server.use(
+ http.get('/api/v1/tenants/t1/workflows/wf1/snapshots/snap-2/form', async () => {
+ await gate
+ return HttpResponse.json({ code: 'not_found' }, { status: 404 })
+ }),
+ )
+ const user = userEvent.setup()
+ renderComposer()
+ await user.click(await screen.findByLabelText(/运行版本/))
+ await user.click(await screen.findByRole('option', { name: 'v2 · 发布 2' }))
+ try {
+ expect(screen.queryByLabelText(/提示词/)).not.toBeInTheDocument()
+ expect(screen.queryByRole('button', { name: '审阅' })).not.toBeInTheDocument()
+ } finally {
+ release()
+ }
+ expect(await screen.findByText(/当前部署无法加载/)).toBeInTheDocument()
+ expect(screen.queryByRole('button', { name: '审阅' })).not.toBeInTheDocument()
+ const read = serveOldSnapshot()
+ await user.click(screen.getByRole('button', { name: '重试加载表单' }))
+ expect(await screen.findByLabelText(/Old input/)).toHaveValue('old default')
+ expect(read).toHaveBeenCalledTimes(1)
+ })
+
+ it('keeps an authors text variable named snapshotId as ordinary input', async () => {
+ const { confirm } = serveComposer({
+ descriptor: {
+ formRef: 'security-review',
+ fields: [
+ { key: 'snapshotId', label: 'Author snapshot name', type: 'text', required: true },
+ ],
+ },
+ })
+ const user = userEvent.setup()
+ renderComposer()
+ await user.type(await screen.findByLabelText(/Author snapshot name/), 'my snapshot')
+ await reviewAndConfirm(user)
+ await waitFor(() =>
+ expect(confirm).toHaveBeenCalledWith({ values: { snapshotId: 'my snapshot' } }),
+ )
+ })
+
+ it('keeps the same interaction when a rejected confirm is retried after switching snapshot', async () => {
+ const { comment, confirm } = serveComposer({ descriptor: launchDescriptor })
+ serveOldSnapshot()
+ server.use(
+ http.post('/api/v1/tenants/t1/issues/i1/interactions/ix1/confirm', () =>
+ HttpResponse.json({ code: 'workflow_version_fields_changed' }, { status: 409 }),
+ ),
+ )
+ const user = userEvent.setup()
+ const { onConfirmed } = renderComposer()
+ await screen.findByLabelText(/提示词/)
+ await reviewAndConfirm(user)
+ await screen.findByText('确认失败，请重试。')
+ await user.click(screen.getByRole('button', { name: '返回修改' }))
+ await user.click(selectTrigger('运行版本'))
+ await user.click(await screen.findByRole('option', { name: 'v2 · 发布 2' }))
+ await screen.findByLabelText(/Old input/)
+ server.use(
+ http.post('/api/v1/tenants/t1/issues/i1/interactions/ix1/confirm', async ({ request }) => {
+ confirm(await request.json())
+ return HttpResponse.json({ resource: run() })
+ }),
+ )
+ await reviewAndConfirm(user)
+ await waitFor(() => expect(onConfirmed).toHaveBeenCalledTimes(1))
+ expect(comment).toHaveBeenCalledTimes(1)
+ expect(confirm).toHaveBeenCalledWith(
+ expect.objectContaining({ values: expect.objectContaining({ snapshotId: 'snap-2' }) }),
+ )
+ })
})
diff --git a/frontend/src/features/issues/types.ts b/frontend/src/features/issues/types.ts
index 0cbb76592..da4ea1020 100644
--- a/frontend/src/features/issues/types.ts
+++ b/frontend/src/features/issues/types.ts
@@ -88,6 +88,8 @@ export interface IssueRun {
issueId: string
executorType: 'agent' | 'team' | 'workflow'
executorId: string
+ /** Workflow runs: the linked WorkflowRun id, whose status is authoritative for execution. */
+ workflowInvocationRef?: string | null
input: Record
status: string
/** Agent runs: the settled delivery, `{revisionId, deliveryState}`, once the run released its Workspace. */
From 09b75af662336b142cc6596efb7ee749f5668e61 Mon Sep 17 00:00:00 2001
From: creegon2 <315603738+creegon2@users.noreply.github.com>
Date: Fri, 16 Oct 2026 17:43:39 -0700
Subject: [PATCH 05/12] fix(workflow): address invocation migration and form
review
Move invocation linkage to migration 0037, reserving 0035 and 0036 for the earlier identity and execution work. Explain why the composite foreign key fails closed on unexpected dangling references.
Identify the platform snapshot selector using all published snapshot ids from the existing paginated endpoint, preserving an author's same-named select as ordinary input when its options are custom. Explain how to resolve workflow_version_fields_changed in the composer without changing admission policy.
Validation: Go build, the core package, all migration and invocation integration cases with a 30-minute timeout, and touched-file frontend format, lint, types, component/API and module gates. No scheduler or identity changes.
Co-Authored-By: Claude Opus 5.5
---
frontend/src/features/issues/README.en.md | 2 +
frontend/src/features/issues/README.md | 2 +
frontend/src/features/issues/api.test.tsx | 70 +++++++++++++++++++
frontend/src/features/issues/api.ts | 28 +++++++-
.../features/issues/components/README.en.md | 6 +-
.../src/features/issues/components/README.md | 6 +-
.../workflow-interaction-composer.tsx | 60 +++++++++++-----
.../components/workflow-interaction.test.tsx | 62 +++++++++++++++-
.../workflow_invocation_migration_test.go | 2 +-
...=> 0037_workflow_invocation_reference.sql} | 1 +
internal/core/migrations/README.en.md | 8 +--
internal/core/migrations/README.md | 6 +-
12 files changed, 223 insertions(+), 30 deletions(-)
rename internal/core/migrations/{0036_workflow_invocation_reference.sql => 0037_workflow_invocation_reference.sql} (91%)
diff --git a/frontend/src/features/issues/README.en.md b/frontend/src/features/issues/README.en.md
index efae32116..697bf2d4f 100644
--- a/frontend/src/features/issues/README.en.md
+++ b/frontend/src/features/issues/README.en.md
@@ -38,3 +38,5 @@ Detail tests use SessionProvider and CurrentSpaceProvider to cover versioned ass
`IssueRun.workflowInvocationRef` retains the server's WorkflowRun reference for activity status and navigation; the API test checks that list reads preserve it. Workflow confirmation exclusion, retry and detail-link interaction tests live in `components/`.
`useWorkflowSnapshotForm` reuses the generated snapshot-form GET with tenant/workflow/snapshot/issue-scoped query keys. The API test varies every scope so versions and Issues cannot share the wrong field defaults.
+
+`useWorkflowPublishedSnapshotIds` caches all published ids by tenant/workflow, following the generated list endpoint's cursors with a page limit of 100 and the caller's abort signal; it stores no graph copies. The activity form uses those ids to identify published-version options. API tests cover the second-page cursor, complete id list and tenant/workflow isolation.
diff --git a/frontend/src/features/issues/README.md b/frontend/src/features/issues/README.md
index c2af1e21e..e58a226f3 100644
--- a/frontend/src/features/issues/README.md
+++ b/frontend/src/features/issues/README.md
@@ -34,3 +34,5 @@ MSW 模拟真实 Cloud 接口；`onUnhandledRequest: 'error'` 下每个测试都
`IssueRun.workflowInvocationRef` 保留服务端返回的 WorkflowRun 引用，供活动栏读取真实运行状态并导航；API 测试检查列表没有丢失该关联。Workflow 确认去重、失败重试和详情链接的交互测试位于 `components/`。
`useWorkflowSnapshotForm` 复用生成的 snapshot form GET，查询键包含 tenant/workflow/snapshot/issue；API 测试逐项切换这些范围，避免不同版本或 Issue 共用表单默认值。
+
+`useWorkflowPublishedSnapshotIds` 按 tenant/workflow 缓存全部已发布版本 id，使用生成的列表接口逐页读取（每页 100，传递取消信号），不缓存图副本。活动表单用它判断下拉选项是否都属于已发布版本；API 测试验证第二页游标、完整 id 列表及租户/Workflow 隔离。
diff --git a/frontend/src/features/issues/api.test.tsx b/frontend/src/features/issues/api.test.tsx
index 690b813db..46584c844 100644
--- a/frontend/src/features/issues/api.test.tsx
+++ b/frontend/src/features/issues/api.test.tsx
@@ -11,6 +11,7 @@ import {
useMoveIssue,
useRuns,
useUpdateIssue,
+ useWorkflowPublishedSnapshotIds,
useWorkflowSnapshotForm,
} from './api'
@@ -266,3 +267,72 @@ describe('useWorkflowSnapshotForm', () => {
expect(queryClient.getQueryCache().getAll()).toHaveLength(2)
})
})
+
+describe('useWorkflowPublishedSnapshotIds', () => {
+ it.each([
+ { scope: 'tenant', tenant: 't2', workflow: 'wf1' },
+ { scope: 'workflow', tenant: 't1', workflow: 'wf2' },
+ ])('follows all pages and isolates ids when $scope changes', async ({ tenant, workflow }) => {
+ const requests: { path: string; limit: string | null; cursor: string | null }[] = []
+ server.use(
+ http.get('/api/v1/tenants/:tid/workflows/:wfid/snapshots', ({ request, params }) => {
+ const url = new URL(request.url)
+ const cursor = url.searchParams.get('cursor')
+ requests.push({ path: url.pathname, limit: url.searchParams.get('limit'), cursor })
+ const prefix = `${String(params['tid'])}/${String(params['wfid'])}`
+ return HttpResponse.json({
+ items: [{ id: `${prefix}/${cursor ? 'second' : 'first'}` }],
+ nextCursor: cursor ? '' : 'page-two',
+ })
+ }),
+ )
+ const cache = new QueryClient()
+ const { result, rerender } = renderHook(
+ (props: { tenant: string; workflow: string }) =>
+ useWorkflowPublishedSnapshotIds(props.tenant, props.workflow),
+ { wrapper: wrapper(cache), initialProps: { tenant: 't1', workflow: 'wf1' } },
+ )
+ await waitFor(() => expect(result.current.data).toEqual(['t1/wf1/first', 't1/wf1/second']))
+ rerender({ tenant, workflow })
+ await waitFor(() =>
+ expect(result.current.data).toEqual([
+ `${tenant}/${workflow}/first`,
+ `${tenant}/${workflow}/second`,
+ ]),
+ )
+ expect(requests).toEqual([
+ { path: '/api/v1/tenants/t1/workflows/wf1/snapshots', limit: '100', cursor: null },
+ { path: '/api/v1/tenants/t1/workflows/wf1/snapshots', limit: '100', cursor: 'page-two' },
+ {
+ path: `/api/v1/tenants/${tenant}/workflows/${workflow}/snapshots`,
+ limit: '100',
+ cursor: null,
+ },
+ {
+ path: `/api/v1/tenants/${tenant}/workflows/${workflow}/snapshots`,
+ limit: '100',
+ cursor: 'page-two',
+ },
+ ])
+ expect(
+ cache
+ .getQueryCache()
+ .getAll()
+ .map((query) => query.queryKey),
+ ).toEqual([
+ ['workflow-published-snapshot-ids', 't1', 'wf1'],
+ ['workflow-published-snapshot-ids', tenant, workflow],
+ ])
+ })
+
+ it.each([
+ { tenant: '', workflow: 'wf1' },
+ { tenant: 't1', workflow: undefined },
+ ])('does not fetch until tenant and workflow ids are present: %j', ({ tenant, workflow }) => {
+ const { result } = renderHook(() => useWorkflowPublishedSnapshotIds(tenant, workflow), {
+ wrapper: wrapper(new QueryClient()),
+ })
+ expect(result.current.fetchStatus).toBe('idle')
+ expect(result.current.data).toBeUndefined()
+ })
+})
diff --git a/frontend/src/features/issues/api.ts b/frontend/src/features/issues/api.ts
index f24255e51..c1d77b05a 100644
--- a/frontend/src/features/issues/api.ts
+++ b/frontend/src/features/issues/api.ts
@@ -1,5 +1,8 @@
import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'
-import { getApiV1TenantsTidWorkflowsWfidSnapshotsSnapshotIdForm } from '@/api/workflows/workflows'
+import {
+ getApiV1TenantsTidWorkflowsWfidSnapshots,
+ getApiV1TenantsTidWorkflowsWfidSnapshotsSnapshotIdForm,
+} from '@/api/workflows/workflows'
import { AXIOS_INSTANCE } from '@/lib/api-client'
import type {
AssistSuggestion,
@@ -313,6 +316,29 @@ export function useFormDescriptor(tid: string, formRef: string | undefined, issu
})
}
+/** Lists every published snapshot id so version-selector checks do not depend on UUID page order. */
+export function useWorkflowPublishedSnapshotIds(tid: string, workflowId: string | undefined) {
+ return useQuery({
+ queryKey: ['workflow-published-snapshot-ids', tid, workflowId ?? ''],
+ queryFn: ({ signal }) => {
+ const ids: string[] = []
+ async function collectPage(cursor = ''): Promise {
+ const page = await getApiV1TenantsTidWorkflowsWfidSnapshots(
+ tid,
+ workflowId ?? '',
+ { limit: 100, ...(cursor ? { cursor } : {}) },
+ undefined,
+ signal,
+ )
+ ids.push(...page.items.map((item) => item.id))
+ return page.nextCursor ? collectPage(page.nextCursor) : ids
+ }
+ return collectPage()
+ },
+ enabled: !!tid && !!workflowId,
+ })
+}
+
/** Reads a chosen frozen version's form, isolated by tenant, workflow, snapshot and issue context. */
export function useWorkflowSnapshotForm(
tid: string,
diff --git a/frontend/src/features/issues/components/README.en.md b/frontend/src/features/issues/components/README.en.md
index d4bfcc13c..94d6783a1 100644
--- a/frontend/src/features/issues/components/README.en.md
+++ b/frontend/src/features/issues/components/README.en.md
@@ -28,7 +28,9 @@ Creation drafts remount by tenant/space; activity drafts by tenant/space/issue,
Workflow confirmation locks duplicate submissions, editing and removal from interaction creation onward; a rejected confirm reuses that interaction on retry, including after a version switch. Confirmed rows use `features/workflows/api` to read the WorkflowRun referenced by `IssueRun.workflowInvocationRef`; detail links combine `executorId`, that reference and the route's workspace slug.
-The initial form reads the live descriptor by `formRef + issueId`. Changing the `snapshotId` select loads the existing snapshot-form route with a tenant/workflow/snapshot/issue-scoped key, clears the old field draft and initializes the selected version's defaults. The selector retains the echoed snapshotId rather than its newest-version default. Loading or failed reads cannot review stale fields; failed reads can retry. An author's text variable with the same name remains ordinary input. Reading a form is not run admission; Confirm's `workflow_version_fields_changed` gate is unchanged.
+The initial form reads the live descriptor by `formRef + issueId`. A `snapshotId` select is recognized as the platform version selector only when every option belongs to this workflow's published snapshots; the form waits for that list and allows retry after a failed read. An author's same-named text field or select with custom options remains ordinary input.
+
+Changing the platform version selector loads the existing snapshot-form route with a tenant/workflow/snapshot/issue-scoped key, clears the old field draft and initializes the selected version's defaults. The selector retains the echoed snapshotId rather than its newest-version default. Loading or failed reads cannot review stale fields; failed reads can retry. Reading a form is not run admission; Confirm's `workflow_version_fields_changed` gate is unchanged. That refusal explains the selected version differs from the current draft and directs the user to publish the draft or pick another version.
Historical runs without a WorkflowRun reference retain their IssueRun status, make no WorkflowRun request and show no detail link. Linked runs always use the WorkflowRun's authoritative status.
@@ -36,4 +38,4 @@ Historical runs without a WorkflowRun reference retain their IssueRun status, ma
Component tests use MSW; properties and Task Mode→Thread are covered by `issue-detail-page.test.tsx`. Loading, empty, refused and retry states are distinct; creation, ordinary-comment and target failures retain drafts, while scope changes clear them.
-Workflow tests cover double-click exclusion, rejected confirms reusing an interaction across version switches, snapshot fields/defaults/echoes, pending and failed reads, and WorkflowRun status and detail links.
+Workflow tests cover double-click exclusion, rejected confirms reusing an interaction across version switches, snapshot fields/defaults/echoes, an author's same-named select making no snapshot-form request, actionable field-divergence feedback, pending and failed reads, and WorkflowRun status and detail links.
diff --git a/frontend/src/features/issues/components/README.md b/frontend/src/features/issues/components/README.md
index 339e41b12..356a67eb7 100644
--- a/frontend/src/features/issues/components/README.md
+++ b/frontend/src/features/issues/components/README.md
@@ -28,7 +28,9 @@
Workflow 确认从创建 interaction 开始锁住重复提交、返回修改与取消；确认失败后仍用同一 interaction 重试，包括切换版本后的重试。确认行通过 `features/workflows/api` 读取 `IssueRun.workflowInvocationRef` 对应的 WorkflowRun 状态，详情链接使用 `executorId`、该引用和当前路由的空间 slug。
-初始表单按 `formRef + issueId` 读取活稿。用户切换 `snapshotId` 下拉选择器时，按 tenant/workflow/snapshot/issue 读取既有 snapshot form 接口，清空旧字段草稿并初始化新版本默认值；选择器保留响应回显的 snapshotId，不采用它仍指向最新版本的默认值。加载或失败期间不能审阅旧字段，失败可重试；作者自定义的同名文本字段仍是普通输入。读取表单不代表服务端准入通过，Confirm 的 `workflow_version_fields_changed` 门禁保持不变。
+初始表单按 `formRef + issueId` 读取活稿。只有 `snapshotId` select 的全部选项都属于该 Workflow 的已发布快照时，才识别为平台版本选择器；列表读取完成前不开放表单，失败可重试。作者自定义的同名文本字段或含自定义选项的 select 仍是普通输入。
+
+用户切换平台版本选择器时，按 tenant/workflow/snapshot/issue 读取既有 snapshot form 接口，清空旧字段草稿并初始化新版本默认值；选择器保留响应回显的 snapshotId，不采用它仍指向最新版本的默认值。加载或失败期间不能审阅旧字段，失败可重试。读取表单不代表服务端准入通过，Confirm 的 `workflow_version_fields_changed` 门禁保持不变；收到该拒绝时，界面提示所选版本字段与当前草稿不一致，并引导先发布草稿或选择其他版本。
历史运行没有 WorkflowRun 引用时，确认行仍显示 IssueRun 状态，不请求 WorkflowRun，也不显示运行详情链接；有关联的运行始终以 WorkflowRun 状态为准。
@@ -36,4 +38,4 @@ Workflow 确认从创建 interaction 开始锁住重复提交、返回修改与
组件测试用 MSW；属性栏和 Task Mode→Thread 经 `issue-detail-page.test.tsx` 覆盖。测试区分加载、空列表、拒绝、重试；创建、普通评论和目标提交失败保留草稿，作用域切换则清空草稿。
-Workflow 测试覆盖双击互斥、确认失败后复用 interaction（含切换版本）、版本字段/default/回显切换、读取等待与失败，以及 WorkflowRun 状态和详情链接。
+Workflow 测试覆盖双击互斥、确认失败后复用 interaction（含切换版本）、版本字段/default/回显切换、作者同名 select 不发快照表单请求、字段差异拒绝的操作提示、读取等待与失败，以及 WorkflowRun 状态和详情链接。
diff --git a/frontend/src/features/issues/components/workflow-interaction-composer.tsx b/frontend/src/features/issues/components/workflow-interaction-composer.tsx
index 1d33d0c4f..4d1d0ea19 100644
--- a/frontend/src/features/issues/components/workflow-interaction-composer.tsx
+++ b/frontend/src/features/issues/components/workflow-interaction-composer.tsx
@@ -9,6 +9,7 @@ import {
useAssistWorkflow,
useConfirmWorkflow,
useFormDescriptor,
+ useWorkflowPublishedSnapshotIds,
useWorkflowSnapshotForm,
} from '@/features/issues/api'
import { AssistSuggestions } from '@/features/issues/components/assist-suggestions'
@@ -22,6 +23,7 @@ import type {
FormField,
FormValues,
} from '@/features/issues/types'
+import { faultCode } from '@/lib/api-client'
/** Seeds the editable values from the descriptor's declared defaults. */
function seedValues(descriptor: FormDescriptor, snapshotId?: string): FormValues {
@@ -31,12 +33,25 @@ function seedValues(descriptor: FormDescriptor, snapshotId?: string): FormValues
values[field.key] = field.defaultValue
}
}
- if (snapshotId && descriptor.fields.some(isSnapshotSelector)) values['snapshotId'] = snapshotId
+ if (snapshotId) values['snapshotId'] = snapshotId
return values
}
-function isSnapshotSelector(field: FormField): boolean {
- return field.key === 'snapshotId' && field.type === 'select'
+// An author's snapshotId field suppresses platform injection, so its name and type are insufficient.
+function isSnapshotSelector(field: FormField, publishedIds: string[]): boolean {
+ return (
+ field.key === 'snapshotId' &&
+ field.type === 'select' &&
+ !!field.options?.length &&
+ field.options.every((option) => publishedIds.includes(option.value))
+ )
+}
+
+function confirmFailure(error: unknown): string {
+ if (faultCode(error) === 'workflow_version_fields_changed') {
+ return '所选版本的字段与当前草稿不一致，请先发布当前草稿或选择其他版本。'
+ }
+ return '确认失败，请重试。'
}
/** Required fields the user has not filled yet — the only client-side gate on Review (§38.15). */
@@ -281,7 +296,7 @@ function ComposerBody(props: {
assistPending: boolean
confirmPending: boolean
assistFailed: boolean
- confirmFailed: boolean
+ confirmError: string | undefined
onAssist: () => void
onReview: () => void
onBack: () => void
@@ -301,8 +316,10 @@ function ComposerBody(props: {
onBack={props.onBack}
onConfirm={props.onConfirm}
/>
- {props.confirmFailed ? (
- 确认失败，请重试。
+ {props.confirmError ? (
+
+ {props.confirmError}
+
) : null}
)
@@ -368,7 +385,7 @@ function useWorkflowSubmission({
return {
submit,
isPending: submission.isPending || submission.isSuccess,
- isError: submission.isError,
+ errorMessage: submission.isError ? confirmFailure(submission.error) : undefined,
}
}
@@ -397,15 +414,25 @@ export function WorkflowInteractionComposer({
const descriptorQuery = snapshotId ? snapshot : live
const confirm = useWorkflowSubmission({ slug, issueId, target, onConfirmed })
const descriptor = descriptorQuery.data
+ const candidate = descriptor?.fields.find(
+ (field) => field.key === 'snapshotId' && field.type === 'select',
+ )
+ const published = useWorkflowPublishedSnapshotIds(slug, candidate ? target.id : undefined)
- if (descriptorQuery.isPending) return
- if (descriptorQuery.isError || !descriptor) {
+ if (descriptorQuery.isPending || (candidate && published.isPending)) {
+ return
+ }
+ if (descriptorQuery.isError || !descriptor || (candidate && published.isError)) {
return (
当前部署无法加载该 Workflow 的表单（form descriptor 不可用）。
-
@@ -419,6 +446,9 @@ export function WorkflowInteractionComposer({
issueId={issueId}
target={target}
descriptor={descriptor}
+ versionField={descriptor.fields.find((field) =>
+ isSnapshotSelector(field, published.data ?? []),
+ )}
snapshotId={snapshot.data?.snapshotId}
loading={descriptorQuery.isFetching}
confirm={confirm}
@@ -434,6 +464,7 @@ function WorkflowInteractionForm(props: {
issueId: string
target: CollaborationTargetSummary
descriptor: FormDescriptor
+ versionField: FormField | undefined
snapshotId: string | undefined
loading: boolean
confirm: ReturnType
@@ -444,13 +475,10 @@ function WorkflowInteractionForm(props: {
const assist = useAssistWorkflow(slug, issueId)
const [reviewing, setReviewing] = useState(false)
const [collapsed, setCollapsed] = useState(false)
- const form = useWorkflowForm(descriptor, props.snapshotId)
+ const form = useWorkflowForm(descriptor, props.versionField ? props.snapshotId : undefined)
function setField(key: string, next: unknown) {
- if (
- typeof next === 'string' &&
- descriptor.fields.some((field) => field.key === key && isSnapshotSelector(field))
- ) {
+ if (typeof next === 'string' && key === props.versionField?.key) {
props.onSnapshotChange(next)
} else {
form.setField(key, next)
@@ -488,7 +516,7 @@ function WorkflowInteractionForm(props: {
assistPending={assist.isPending}
confirmPending={confirm.isPending}
assistFailed={assist.isError}
- confirmFailed={confirm.isError}
+ confirmError={confirm.errorMessage}
onAssist={() =>
assist.mutate(
{ targetId: target.id, values: form.values },
diff --git a/frontend/src/features/issues/components/workflow-interaction.test.tsx b/frontend/src/features/issues/components/workflow-interaction.test.tsx
index 3dc9270fa..c247ce7bf 100644
--- a/frontend/src/features/issues/components/workflow-interaction.test.tsx
+++ b/frontend/src/features/issues/components/workflow-interaction.test.tsx
@@ -180,6 +180,9 @@ function serveComposer(overrides: { descriptor?: unknown; suggestion?: AssistSug
http.get('/api/v1/tenants/t1/collaboration/forms/security-review', () =>
HttpResponse.json(overrides.descriptor ?? descriptor),
),
+ http.get('/api/v1/tenants/t1/workflows/wf1/snapshots', () =>
+ HttpResponse.json({ items: [{ id: 'snap-3' }, { id: 'snap-2' }], nextCursor: '' }),
+ ),
http.post('/api/v1/tenants/t1/issues/i1/collaboration/assist', async ({ request }) => {
assist(await request.json())
return HttpResponse.json(overrides.suggestion ?? suggestion)
@@ -611,6 +614,63 @@ describe('WorkflowInteractionComposer', () => {
)
})
+ it('keeps an authors select named snapshotId as ordinary input when any option is not published', async () => {
+ const { confirm } = serveComposer({
+ descriptor: {
+ formRef: 'security-review',
+ fields: [
+ {
+ key: 'snapshotId',
+ label: 'Author snapshot choice',
+ type: 'select',
+ required: true,
+ defaultValue: 'snap-3',
+ options: [
+ { value: 'snap-3', label: 'Author default' },
+ { value: 'custom-choice', label: 'Author custom choice' },
+ ],
+ },
+ ],
+ },
+ })
+ const formRead = vi.fn<() => void>()
+ server.use(
+ http.get('/api/v1/tenants/t1/workflows/wf1/snapshots/:id/form', () => {
+ formRead()
+ return HttpResponse.json({ code: 'not_found' }, { status: 404 })
+ }),
+ )
+ const user = userEvent.setup()
+ renderComposer()
+ await user.click(await screen.findByLabelText(/Author snapshot choice/))
+ await user.click(await screen.findByRole('option', { name: 'Author custom choice' }))
+ expect(selectTrigger('Author snapshot choice')).toHaveTextContent('Author custom choice')
+ await reviewAndConfirm(user)
+ await waitFor(() =>
+ expect(confirm).toHaveBeenCalledWith({ values: { snapshotId: 'custom-choice' } }),
+ )
+ expect(formRead).not.toHaveBeenCalled()
+ })
+
+ it('explains how to resolve workflow version field divergence after confirm', async () => {
+ serveComposer({ descriptor: launchDescriptor })
+ server.use(
+ http.post('/api/v1/tenants/t1/issues/i1/interactions/ix1/confirm', () =>
+ HttpResponse.json({ code: 'workflow_version_fields_changed' }, { status: 409 }),
+ ),
+ )
+ const user = userEvent.setup()
+ const { onConfirmed } = renderComposer()
+ await screen.findByLabelText(/提示词/)
+ await reviewAndConfirm(user)
+ expect(await screen.findByRole('alert')).toHaveTextContent(
+ '所选版本的字段与当前草稿不一致，请先发布当前草稿或选择其他版本。',
+ )
+ expect(screen.queryByText('workflow_version_fields_changed')).not.toBeInTheDocument()
+ expect(screen.getByRole('button', { name: '返回修改' })).toBeEnabled()
+ expect(onConfirmed).not.toHaveBeenCalled()
+ })
+
it('keeps the same interaction when a rejected confirm is retried after switching snapshot', async () => {
const { comment, confirm } = serveComposer({ descriptor: launchDescriptor })
serveOldSnapshot()
@@ -623,7 +683,7 @@ describe('WorkflowInteractionComposer', () => {
const { onConfirmed } = renderComposer()
await screen.findByLabelText(/提示词/)
await reviewAndConfirm(user)
- await screen.findByText('确认失败，请重试。')
+ await screen.findByText('所选版本的字段与当前草稿不一致，请先发布当前草稿或选择其他版本。')
await user.click(screen.getByRole('button', { name: '返回修改' }))
await user.click(selectTrigger('运行版本'))
await user.click(await screen.findByRole('option', { name: 'v2 · 发布 2' }))
diff --git a/integration/workflow_invocation_migration_test.go b/integration/workflow_invocation_migration_test.go
index aba699c5a..1d7afcc1d 100644
--- a/integration/workflow_invocation_migration_test.go
+++ b/integration/workflow_invocation_migration_test.go
@@ -7,7 +7,7 @@ import (
"github.com/google/uuid"
)
-const workflowInvocationMigration = "0036_workflow_invocation_reference.sql"
+const workflowInvocationMigration = "0037_workflow_invocation_reference.sql"
// seedInvocationWorkflowRun creates real rows for the composite reference under test. The graph is
// deliberately minimal: these tests exercise database integrity rather than graph admission.
diff --git a/internal/core/migrations/0036_workflow_invocation_reference.sql b/internal/core/migrations/0037_workflow_invocation_reference.sql
similarity index 91%
rename from internal/core/migrations/0036_workflow_invocation_reference.sql
rename to internal/core/migrations/0037_workflow_invocation_reference.sql
index 9b8c5d8bd..7229efe3f 100644
--- a/internal/core/migrations/0036_workflow_invocation_reference.sql
+++ b/internal/core/migrations/0037_workflow_invocation_reference.sql
@@ -12,6 +12,7 @@ ALTER TABLE workflow_runs
ALTER TABLE issue_runs
ADD CONSTRAINT issue_runs_workflow_invocation_fk
+ -- workflow_invocation_ref had no writer before this migration, so dangling references are not expected and the FK fails closed.
FOREIGN KEY (workflow_invocation_ref, tenant_id) REFERENCES workflow_runs(id, tenant_id),
ADD CONSTRAINT issue_runs_workflow_invocation_uniq UNIQUE (workflow_invocation_ref),
ADD CONSTRAINT issue_runs_workflow_invocation_shape
diff --git a/internal/core/migrations/README.en.md b/internal/core/migrations/README.en.md
index d8abb3488..d2716ba1f 100644
--- a/internal/core/migrations/README.en.md
+++ b/internal/core/migrations/README.en.md
@@ -102,15 +102,15 @@ Both `0033` files retain their complete filenames and original SQL. The complete
Adds startup_ready_at, startup_deadline_at, startup_resume, startup_ending_deadline_at and agent_run_residuals without rewriting historical outcomes. The serialized policy scan freezes legacy active windows once. Residual ownership commits with failed/done; late true terminal facts add evidence without granting cleanup authority. Fresh/upgrade/repeat coverage: TestMigration0034StartupWindowsPreserveExistingRunOnUpgrade.
-## 0036: WorkflowRun invocation references
+## 0037: WorkflowRun invocation references
-`0036_workflow_invocation_reference.sql` first settles every non-terminal unlinked historical
+`0037_workflow_invocation_reference.sql` first settles every non-terminal unlinked historical
workflow IssueRun (including running and deferred) as failed, with
`workflow_execution_unavailable_before_invocation_linkage`. It then adds a tenant-scoped composite
foreign key and a unique invocation reference. The CHECK requires a reference for non-terminal
workflow executors and forbids it for other executors. Historical terminal workflow rows may retain
NULL references and remain editable; their input and outcomes are preserved without inventing runs.
The CHECK is unconditionally validated, so fresh and upgraded databases enforce the same schema.
-0035 is reserved for the earlier WorkflowRun identity migration in #68; invocation linkage follows as
-0036. Fresh, seeded upgrade, repeated migration,
+0035 and 0036 are reserved for the earlier cloud#68 and cloud#72 migrations, respectively; invocation linkage follows as
+0037. Fresh, seeded upgrade, repeated migration,
constraint validation and historical update coverage is in `integration/workflow_invocation_migration_test.go`.
diff --git a/internal/core/migrations/README.md b/internal/core/migrations/README.md
index 80a39ff47..4d3f5ed10 100644
--- a/internal/core/migrations/README.md
+++ b/internal/core/migrations/README.md
@@ -101,13 +101,13 @@ Cloud 领域方法。冻结与撤销规则见 [模型连接契约](../../../docs
新增 startup_ready_at、startup_deadline_at、startup_resume、startup_ending_deadline_at 及 agent_run_residuals。迁移不改历史结果；活动存量由串行策略扫描一次固定窗口。残留账本与 failed/done 同事务提交，迟到真实终态仅补证据，不代替清理授权。新库/升级/重复执行见 TestMigration0034StartupWindowsPreserveExistingRunOnUpgrade。
-## 0036：WorkflowRun 调用引用
+## 0037：WorkflowRun 调用引用
-`0036_workflow_invocation_reference.sql` 先将所有未关联的历史非终态 workflow IssueRun
+`0037_workflow_invocation_reference.sql` 先将所有未关联的历史非终态 workflow IssueRun
（含 running、deferred）结算为 failed，原因是
`workflow_execution_unavailable_before_invocation_linkage`，再添加租户一致的复合外键及唯一引用。
CHECK 要求非终态 workflow 行具有引用，其他 executor 不得携带引用。历史终态 workflow 行可保留
NULL 引用并继续更新；迁移保留其输入和结果，不虚构执行。CHECK 无条件 VALIDATE，空库和升级库
-采用同一份已验证约束。0035 留给依赖链中更早的 #68 WorkflowRun 身份迁移，本调用关联迁移使用 0036。
+采用同一份已验证约束。0035、0036 分别留给依赖链中更早的 cloud#68、cloud#72 迁移，本调用关联迁移使用 0037。
空库、种子升级、重复迁移、约束验证状态及
历史行更新测试见 `integration/workflow_invocation_migration_test.go`。
From 78458700ccb5fe613920d5c80a6381af66a9732c Mon Sep 17 00:00:00 2001
From: creegon2 <315603738+creegon2@users.noreply.github.com>
Date: Fri, 16 Oct 2026 17:49:09 -0700
Subject: [PATCH 06/12] test(frontend): share the field-divergence confirm
setup
The two field-divergence tests repeated the same 409 handler, render and
confirm steps, which the jscpd duplication gate rejects.
Docs-Unchanged: test-only refactor; no behavior or module contract changed
Co-Authored-By: Claude Opus 5.5
---
.../components/workflow-interaction.test.tsx | 34 +++++++++----------
1 file changed, 16 insertions(+), 18 deletions(-)
diff --git a/frontend/src/features/issues/components/workflow-interaction.test.tsx b/frontend/src/features/issues/components/workflow-interaction.test.tsx
index c247ce7bf..b75fddfd6 100644
--- a/frontend/src/features/issues/components/workflow-interaction.test.tsx
+++ b/frontend/src/features/issues/components/workflow-interaction.test.tsx
@@ -241,6 +241,20 @@ async function reviewAndConfirm(user: ReturnType) {
await user.click(await screen.findByRole('button', { name: '确认执行' }))
}
+/** Renders the composer and confirms once against a server that answers the field-divergence 409. */
+async function confirmIntoFieldsChanged() {
+ server.use(
+ http.post('/api/v1/tenants/t1/issues/i1/interactions/ix1/confirm', () =>
+ HttpResponse.json({ code: 'workflow_version_fields_changed' }, { status: 409 }),
+ ),
+ )
+ const user = userEvent.setup()
+ const { onConfirmed } = renderComposer()
+ await screen.findByLabelText(/提示词/)
+ await reviewAndConfirm(user)
+ return { user, onConfirmed }
+}
+
function serveOldSnapshot() {
const read = vi.fn<(issueId: string | null) => void>()
server.use(
@@ -654,15 +668,7 @@ describe('WorkflowInteractionComposer', () => {
it('explains how to resolve workflow version field divergence after confirm', async () => {
serveComposer({ descriptor: launchDescriptor })
- server.use(
- http.post('/api/v1/tenants/t1/issues/i1/interactions/ix1/confirm', () =>
- HttpResponse.json({ code: 'workflow_version_fields_changed' }, { status: 409 }),
- ),
- )
- const user = userEvent.setup()
- const { onConfirmed } = renderComposer()
- await screen.findByLabelText(/提示词/)
- await reviewAndConfirm(user)
+ const { onConfirmed } = await confirmIntoFieldsChanged()
expect(await screen.findByRole('alert')).toHaveTextContent(
'所选版本的字段与当前草稿不一致，请先发布当前草稿或选择其他版本。',
)
@@ -674,15 +680,7 @@ describe('WorkflowInteractionComposer', () => {
it('keeps the same interaction when a rejected confirm is retried after switching snapshot', async () => {
const { comment, confirm } = serveComposer({ descriptor: launchDescriptor })
serveOldSnapshot()
- server.use(
- http.post('/api/v1/tenants/t1/issues/i1/interactions/ix1/confirm', () =>
- HttpResponse.json({ code: 'workflow_version_fields_changed' }, { status: 409 }),
- ),
- )
- const user = userEvent.setup()
- const { onConfirmed } = renderComposer()
- await screen.findByLabelText(/提示词/)
- await reviewAndConfirm(user)
+ const { user, onConfirmed } = await confirmIntoFieldsChanged()
await screen.findByText('所选版本的字段与当前草稿不一致，请先发布当前草稿或选择其他版本。')
await user.click(screen.getByRole('button', { name: '返回修改' }))
await user.click(selectTrigger('运行版本'))
From 7088e26dbc17d126c69c8bc07dc8d3850b1e05f6 Mon Sep 17 00:00:00 2001
From: creegon2 <315603738+creegon2@users.noreply.github.com>
Date: Sat, 17 Oct 2026 01:19:29 -0700
Subject: [PATCH 07/12] test(workflow): cover imported snapshots in issue
invocations
Reuse the real .orax, HTTPS marketplace and PostgreSQL harness to prove imported workflows enter the Issue target directory and their selected snapshot forms remain stable across administrators and import-trigger deactivation. Compare pending run bindings across those identities.
Verify Confirm links its IssueRun to a WorkflowRun bound to the imported snapshot, direct starts select the same snapshot, and editing the draft cannot replace the frozen graph. After provider uninstall, preserve the old snapshot, publish a new version and create a run from it.
Evidence stops at WorkflowRun creation and binding; real node execution remains with #70. No import implementation or specs changes.
Validation: go build ./..., the core package, three new import invocation cases and 23 existing invocation/confirm/assets regressions with 30-minute timeouts. The merge has no overlap with this branch's frontend edits, so the conditional frontend gate does not apply.
Co-Authored-By: Claude Opus 5.5
---
integration/workflow_asset_invocation_test.go | 215 ++++++++++++++++++
integration/workflow_asset_retained_test.go | 58 +++++
2 files changed, 273 insertions(+)
create mode 100644 integration/workflow_asset_invocation_test.go
create mode 100644 integration/workflow_asset_retained_test.go
diff --git a/integration/workflow_asset_invocation_test.go b/integration/workflow_asset_invocation_test.go
new file mode 100644
index 000000000..c0a1886e7
--- /dev/null
+++ b/integration/workflow_asset_invocation_test.go
@@ -0,0 +1,215 @@
+package integration
+
+import (
+ "encoding/json"
+ "reflect"
+ "testing"
+
+ "github.com/wanglongan587/cloud/internal/core"
+ "github.com/wanglongan587/cloud/internal/workflowasset"
+)
+
+type importedWorkflowEvidence struct {
+ f *fixture
+ spaceID string
+ workflow string
+ snapshot core.Object
+}
+
+// importWorkflowForInvocation reuses the real .orax/HTTPS/marketplace harness. No workflow or
+// snapshot is seeded: the plugin's desired state must produce both through the import worker.
+func importWorkflowForInvocation(t *testing.T) *importedWorkflowEvidence {
+ t.Helper()
+ f := setup(t)
+ // These tests prove persisted invocation identity, never a simulated node trace.
+ f.store.Simulator = nil
+ document := startPromptGraph([]any{core.Object{
+ "name": "releaseName", "displayName": "Imported release", "fieldType": "text-input",
+ "valueType": "string", "required": true, "value": "package-default",
+ }}, "Ship the imported release")
+ document["name"] = "Imported Release Evidence"
+ document["version"] = 47
+ document["globalVariables"] = []any{core.Object{"name": "release.mode", "valueType": "string", "value": "safe"}}
+ encoded, err := json.Marshal(document)
+ must(t, err)
+ pkg := workflowPackage(t, map[string]string{"release.json": string(encoded)})
+ packages := newWorkflowPackages(t, map[string][]byte{"/packages/invocation-1.0.0.orax": pkg})
+ repo := workflowMarketplace(t, f.root)
+ publishWorkflowRelease(t, repo, "invocation-kit", "1.0.0", packages.server.URL+"/packages/invocation-1.0.0.orax", pkg)
+ f.syncMarketplace(t, repo)
+ f.store.WorkflowAssets = workflowasset.NewFetcher(nil, packages.server.Client().Transport)
+ sid := f.defaultSpaceID()
+ f.installWorkflowPlugin(t, sid, "official/invocation-kit", "")
+ must(t, f.importWorkflowAssets(t))
+ item := f.assetImport(t, sid, "official/invocation-kit")
+ if item == nil || item.S("state") != "completed" {
+ t.Fatalf("import did not complete: %v", item)
+ }
+ imported := assetDocuments(item)["assets/workflows/release.json"]
+ if imported.S("outcome") != "imported" || imported.S("workflowId") == "" || imported.N("snapshotVersion") != 1 {
+ t.Fatalf("document was not imported and published: %v", imported)
+ }
+ wid := imported.S("workflowId")
+ snapshots := issueItems(f.call("GET", f.path("/workflows/"+wid+"/snapshots"), nil, "", 200))
+ if len(snapshots) != 1 || snapshots[0].S("id") == "" || snapshots[0].N("version") != 1 {
+ t.Fatalf("import must leave exactly one Cloud-assigned snapshot: %v", snapshots)
+ }
+ var expected core.Object
+ must(t, json.Unmarshal(encoded, &expected))
+ if !reflect.DeepEqual(snapshots[0].O("graph"), expected) {
+ t.Fatalf("published graph differs from the package document: %v", snapshots[0])
+ }
+ return &importedWorkflowEvidence{f: f, spaceID: sid, workflow: wid, snapshot: snapshots[0]}
+}
+
+func importedInvocationInput() core.Object {
+ return core.Object{"releaseName": "release-from-package", "release.mode": "strict"}
+}
+
+// assertRun checks both the public frozen definition and the real PostgreSQL binding.
+// IDs and timestamps differ by invocation; the returned projection names only run semantics.
+func (e *importedWorkflowEvidence) assertRun(t *testing.T, user core.Claims, id string, snapshot core.Object) core.Object {
+ t.Helper()
+ run := callAs(t, e.f, user, "GET", e.f.path("/workflows/"+e.workflow+"/runs/"+id), "", nil, 200)
+ if run.S("snapshotId") != snapshot.S("id") || run.S("workflowId") != e.workflow || run.S("status") != "pending" {
+ t.Fatalf("run did not bind a pending imported invocation: %v", run)
+ }
+ if !reflect.DeepEqual(run.O("definitionSnapshot"), snapshot.O("graph")) || !reflect.DeepEqual(run.O("input"), importedInvocationInput()) {
+ t.Fatalf("run did not freeze the selected graph and input: %v", run)
+ }
+ if n := e.f.scalar("SELECT count(*) FROM workflow_runs WHERE id=$1 AND tenant_id=$2 AND workflow_id=$3 AND snapshot_id=$4", id, e.f.tid, e.workflow, snapshot.S("id")); n != 1 {
+ t.Fatalf("stored WorkflowRun binding = %d rows, want 1", n)
+ }
+ return core.Object{"workflowId": run.S("workflowId"), "snapshotId": run.S("snapshotId"),
+ "status": run.S("status"), "input": run.O("input"), "definitionSnapshot": run.O("definitionSnapshot")}
+}
+
+func TestImportedWorkflowDirectoryAndRunsIgnoreImportTriggerUser(t *testing.T) {
+ e := importWorkflowForInvocation(t)
+ f := e.f
+ trigger := f.user
+ other, otherID := f.addUser(t, "asset-evidence-admin", "Other asset administrator")
+ f.call("PUT", f.path("/members/"+otherID), core.Object{"role": "admin", "status": "active", "version": 1}, "", 200)
+ iid := f.call("POST", f.path("/issues"), core.Object{"title": "Discover imported workflow"}, "asset-directory-issue", 200).O("resource").S("id")
+ formPath := f.path("/workflows/" + e.workflow + "/snapshots/" + e.snapshot.S("id") + "/form?issueId=" + iid)
+ initialForm := f.call("GET", formPath, nil, "", 200)
+ liveFormPath := f.path("/collaboration/forms/" + e.workflow + "?issueId=" + iid)
+ if live := f.call("GET", liveFormPath, nil, "", 200); !reflect.DeepEqual(live["fields"], initialForm["fields"]) {
+ t.Fatalf("the imported live copy and initial published form disagree: live=%v snapshot=%v", live, initialForm)
+ }
+ // Distinguish the selected snapshot's form from the mutable live form. This is the same
+ // snapshot-form endpoint the @ composer reads when the user selects a published version.
+ live := f.call("GET", f.path("/workflows/"+e.workflow), nil, "", 200)
+ f.call("PUT", f.path("/workflows/"+e.workflow), core.Object{
+ "version": live.N("version"), "graph": startPromptGraph([]any{invocationVariable("replacement")}, "Draft-only prompt"),
+ }, "", 200)
+ changed := formFields(t, f.call("GET", liveFormPath, nil, "", 200))
+ if changed["replacement"] == nil || changed["releaseName"] != nil {
+ t.Fatalf("test did not give the live draft a different form: %v", changed)
+ }
+ project := func(user core.Claims) core.Object {
+ t.Helper()
+ // This is the public directory used by useCollaborationTargets in the Issue @ picker.
+ directory := callAs(t, f, user, "GET", f.path("/collaboration/targets"), "", nil, 200)
+ var target core.Object
+ for _, item := range issueItems(directory) {
+ if item.S("type") == "workflow" && item.S("id") == e.workflow {
+ target = item
+ }
+ }
+ if target == nil || target.S("displayName") != "Imported Release Evidence" {
+ t.Fatalf("imported copy is absent from the @ directory: %v", directory)
+ }
+ interaction := target.O("interactionDescriptor")
+ if interaction.S("mode") != "form" || interaction.B("requiresTask") || interaction.S("formRef") != e.workflow {
+ t.Fatalf("imported target does not advertise its workflow form: %v", target)
+ }
+ form := callAs(t, f, user, "GET", formPath, "", nil, 200)
+ fields := formFields(t, form)
+ if form.S("snapshotId") != e.snapshot.S("id") || form.S("formRef") != interaction.S("formRef") ||
+ fields["releaseName"].S("label") != "Imported release" || fields["releaseName"].S("defaultValue") != "package-default" ||
+ !fields["releaseName"].B("required") || fields["replacement"] != nil ||
+ fields["prompt"].S("defaultValue") != "Ship the imported release" || fields["release.mode"].S("defaultValue") != "safe" {
+ t.Fatalf("selected form was not projected from the imported snapshot: %v", form)
+ }
+ if !reflect.DeepEqual(form, initialForm) {
+ t.Fatalf("editing the live copy changed its published form: initial=%v selected=%v", initialForm, form)
+ }
+ workflow := callAs(t, f, user, "GET", f.path("/workflows/"+e.workflow), "", nil, 200)
+ return core.Object{"target": target, "form": form, "workflow": workflow}
+ }
+ start := func(user core.Claims, key string) core.Object {
+ t.Helper()
+ run := callAs(t, f, user, "POST", f.path("/workflows/"+e.workflow+"/runs"), key, core.Object{
+ "snapshotId": e.snapshot.S("id"), "input": importedInvocationInput(),
+ }, 200).O("resource")
+ return e.assertRun(t, user, run.S("id"), e.snapshot)
+ }
+ baseline := project(trigger)
+ baselineRun := start(trigger, "asset-author-trigger-run")
+ if otherProjection, otherRun := project(other), start(other, "asset-author-other-run"); !reflect.DeepEqual(otherProjection, baseline) || !reflect.DeepEqual(otherRun, baselineRun) {
+ t.Fatalf("changing administrator changed imported projection/run: projection=%v run=%v", otherProjection, otherRun)
+ }
+ // Global account disable has no public admin endpoint; change the actual authorization row,
+ // then prove the old identity is refused over HTTP before comparing the surviving admin.
+ _, err := f.store.Pool.Exec("UPDATE users SET status='disabled' WHERE id=$1", f.uid)
+ must(t, err)
+ callAs(t, f, trigger, "GET", f.path("/collaboration/targets"), "", nil, 403)
+ if afterProjection, afterRun := project(other), start(other, "asset-author-after-disable-run"); !reflect.DeepEqual(afterProjection, baseline) || !reflect.DeepEqual(afterRun, baselineRun) {
+ t.Fatalf("deactivating the import trigger changed the retained copy: projection=%v run=%v", afterProjection, afterRun)
+ }
+ if n := f.scalar("SELECT count(*) FROM workflow_runs WHERE workflow_id=$1 AND snapshot_id=$2", e.workflow, e.snapshot.S("id")); n != 3 {
+ t.Fatalf("expected the same snapshot for all three administrators/account states, got %d", n)
+ }
+}
+
+func TestImportedWorkflowSnapshotBindsConfirmAndDirectRuns(t *testing.T) {
+ e := importWorkflowForInvocation(t)
+ f, wid, snapshot := e.f, e.workflow, e.snapshot
+ iid := f.call("POST", f.path("/issues"), core.Object{"title": "Use imported snapshot"}, "asset-inv-issue", 200).O("resource").S("id")
+ interaction := workflowComment(t, f, wid, iid, "asset-inv-comment", "@ imported release")
+ values := importedInvocationInput()
+ values["repository"], values["branch"], values["snapshotId"] = "https://example.invalid/repo.git", "main", snapshot.S("id")
+ issueRun := f.call("POST", confirmPath(f, iid, interaction.S("id")), core.Object{"values": values}, "asset-inv-confirm", 200).O("resource")
+ ref := issueRun.S("workflowInvocationRef")
+ if ref == "" || issueRun.O("input").S("workflowSnapshotId") != snapshot.S("id") {
+ t.Fatalf("Confirm did not link the imported snapshot: %v", issueRun)
+ }
+ e.assertRun(t, f.user, ref, snapshot)
+ if n := f.scalar(`SELECT count(*) FROM issue_interactions ix
+ JOIN issue_runs ir ON ir.id=ix.run_id
+ JOIN workflow_runs wr ON wr.id=ir.workflow_invocation_ref AND wr.tenant_id=ir.tenant_id
+ WHERE ix.id=$1 AND ir.id=$2 AND wr.id=$3 AND wr.snapshot_id=$4`, interaction.S("id"), issueRun.S("id"), ref, snapshot.S("id")); n != 1 {
+ t.Fatalf("imported Confirm did not persist the interaction/run pair: %d", n)
+ }
+ create := func(key string) string {
+ t.Helper()
+ return f.call("POST", f.path("/workflows/"+wid+"/runs"), core.Object{
+ "snapshotId": snapshot.S("id"), "input": importedInvocationInput(),
+ }, key, 200).O("resource").S("id")
+ }
+ first := create("asset-inv-direct")
+ e.assertRun(t, f.user, first, snapshot)
+ var frozen string
+ must(t, f.store.Pool.QueryRow("SELECT graph::text FROM workflow_snapshots WHERE id=$1", snapshot.S("id")).Scan(&frozen))
+ live := f.call("GET", f.path("/workflows/"+wid), nil, "", 200)
+ draft := startPromptGraph([]any{invocationVariable("replacement")}, "Changed draft prompt")
+ f.call("PUT", f.path("/workflows/"+wid), core.Object{"graph": draft, "version": live.N("version")}, "", 200)
+ // Direct admission judges the selected snapshot even when the draft has a different schema.
+ // Confirm's separate live-form compatibility gate is deliberately not bypassed here.
+ second := create("asset-inv-after-edit")
+ if first == second || second == ref || first == ref {
+ t.Fatal("distinct starts must create distinct WorkflowRun records")
+ }
+ e.assertRun(t, f.user, second, snapshot)
+ e.assertRun(t, f.user, first, snapshot)
+ e.assertRun(t, f.user, ref, snapshot)
+ var after string
+ must(t, f.store.Pool.QueryRow("SELECT graph::text FROM workflow_snapshots WHERE id=$1", snapshot.S("id")).Scan(&after))
+ if after != frozen {
+ t.Fatal("editing the live draft rewrote the imported snapshot")
+ }
+ if n := f.scalar("SELECT count(*) FROM workflow_runs WHERE workflow_id=$1 AND snapshot_id=$2", wid, snapshot.S("id")); n != 3 {
+ t.Fatalf("expected Confirm and two direct starts bound to the imported snapshot, got %d", n)
+ }
+}
diff --git a/integration/workflow_asset_retained_test.go b/integration/workflow_asset_retained_test.go
new file mode 100644
index 000000000..cedafd331
--- /dev/null
+++ b/integration/workflow_asset_retained_test.go
@@ -0,0 +1,58 @@
+package integration
+
+import (
+ "testing"
+
+ "github.com/wanglongan587/cloud/internal/core"
+)
+
+// TestImportedWorkflowRemainsPublishableAndRunnableAfterUninstall proves an imported document is
+// retained user data: removing its plugin neither removes its history nor prevents a new invocation.
+func TestImportedWorkflowRemainsPublishableAndRunnableAfterUninstall(t *testing.T) {
+ e := importWorkflowForInvocation(t)
+ f, wid, old := e.f, e.workflow, e.snapshot
+ readOldSnapshot := func() (string, string) {
+ t.Helper()
+ var graph, row string
+ must(t, f.store.Pool.QueryRow(`SELECT graph::text, row_to_json(s)::text FROM workflow_snapshots s
+ WHERE id=$1 AND tenant_id=$2 AND workflow_id=$3`, old.S("id"), f.tid, wid).Scan(&graph, &row))
+ return graph, row
+ }
+ frozenGraph, frozenRow := readOldSnapshot()
+ plugin := f.spacePlugin(e.spaceID, "official/invocation-kit")
+ if plugin == nil {
+ t.Fatal("imported workflow has no installed plugin selection to remove")
+ }
+ f.call("DELETE", f.pluginSpacePath(e.spaceID)+"/plugins", core.Object{
+ "identifier": "official/invocation-kit", "version": plugin.N("version"),
+ }, "asset-retained-uninstall", 200)
+ if removed := f.spacePlugin(e.spaceID, "official/invocation-kit"); removed == nil || removed.S("desiredState") != "removed" {
+ t.Fatalf("plugin uninstall did not persist its removed intent: %v", removed)
+ }
+ if retained := f.call("GET", f.path("/workflows/"+wid), nil, "", 200); retained.S("id") != wid {
+ t.Fatalf("uninstall did not retain the imported workflow: %v", retained)
+ }
+ if graph, row := readOldSnapshot(); graph != frozenGraph || row != frozenRow {
+ t.Fatal("uninstall changed the imported snapshot row or graph")
+ }
+
+ newSnapshot := publishWorkflow(t, f, wid, "asset-retained-publish")
+ if newSnapshot.S("id") == "" || newSnapshot.S("id") == old.S("id") || newSnapshot.N("version") != old.N("version")+1 {
+ t.Fatalf("retained workflow did not publish its next distinct snapshot: old=%v new=%v", old, newSnapshot)
+ }
+ run := f.call("POST", f.path("/workflows/"+wid+"/runs"), core.Object{
+ "snapshotId": newSnapshot.S("id"), "input": importedInvocationInput(),
+ }, "asset-retained-run", 200).O("resource")
+ e.assertRun(t, f.user, run.S("id"), newSnapshot)
+ if graph, row := readOldSnapshot(); graph != frozenGraph || row != frozenRow {
+ t.Fatal("publishing or running the retained workflow changed its old snapshot")
+ }
+ snapshots := issueItems(f.call("GET", f.path("/workflows/"+wid+"/snapshots"), nil, "", 200))
+ seen := map[string]int64{}
+ for _, snapshot := range snapshots {
+ seen[snapshot.S("id")] = snapshot.N("version")
+ }
+ if len(snapshots) != 2 || seen[old.S("id")] != old.N("version") || seen[newSnapshot.S("id")] != newSnapshot.N("version") {
+ t.Fatalf("retained workflow lost its old or new snapshot: %v", snapshots)
+ }
+}
From af0ee5612e36fc76de54e99a46da4ab1239f5a89 Mon Sep 17 00:00:00 2001
From: creegon2 <315603738+creegon2@users.noreply.github.com>
Date: Sat, 17 Oct 2026 01:22:16 -0700
Subject: [PATCH 08/12] style(test): format the imported snapshot invocation
test
Co-Authored-By: Claude Opus 5.5
---
integration/workflow_asset_invocation_test.go | 6 ++++--
1 file changed, 4 insertions(+), 2 deletions(-)
diff --git a/integration/workflow_asset_invocation_test.go b/integration/workflow_asset_invocation_test.go
index c0a1886e7..356a791ef 100644
--- a/integration/workflow_asset_invocation_test.go
+++ b/integration/workflow_asset_invocation_test.go
@@ -80,8 +80,10 @@ func (e *importedWorkflowEvidence) assertRun(t *testing.T, user core.Claims, id
if n := e.f.scalar("SELECT count(*) FROM workflow_runs WHERE id=$1 AND tenant_id=$2 AND workflow_id=$3 AND snapshot_id=$4", id, e.f.tid, e.workflow, snapshot.S("id")); n != 1 {
t.Fatalf("stored WorkflowRun binding = %d rows, want 1", n)
}
- return core.Object{"workflowId": run.S("workflowId"), "snapshotId": run.S("snapshotId"),
- "status": run.S("status"), "input": run.O("input"), "definitionSnapshot": run.O("definitionSnapshot")}
+ return core.Object{
+ "workflowId": run.S("workflowId"), "snapshotId": run.S("snapshotId"),
+ "status": run.S("status"), "input": run.O("input"), "definitionSnapshot": run.O("definitionSnapshot"),
+ }
}
func TestImportedWorkflowDirectoryAndRunsIgnoreImportTriggerUser(t *testing.T) {
From 52314da27b6b093e96a0213b0cfff9795da237fa Mon Sep 17 00:00:00 2001
From: creegon2 <315603738+creegon2@users.noreply.github.com>
Date: Sun, 18 Oct 2026 18:52:22 -0700
Subject: [PATCH 09/12] =?UTF-8?q?docs(workflow):=20=E4=BF=AE=E6=AD=A3?=
=?UTF-8?q?=E5=BF=AB=E7=85=A7=E8=A1=A8=E5=8D=95=E6=9D=A5=E6=BA=90=E4=B8=8E?=
=?UTF-8?q?=E5=AD=97=E6=AE=B5=E5=B7=AE=E5=BC=82=E9=99=90=E5=88=B6?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
Issue 协作文档仍把确认前的表单描述为始终取自当前文档，与前端选择已发布版本后重取快照表单的行为不符。
同步修正中英文说明，区分初次打开与选择快照后的表单来源，明确现有活稿字段差异 409 检查仍然生效，并在流程图显示关联创建的两类运行。
验证：对照 WorkflowInteractionComposer、useWorkflowSnapshotForm 和 confirmInteraction 的现有实现；两份文档的相对文件链接均有效；git diff --check 通过。
仅修改文档，不改变 API、字段差异门禁、迁移或调度行为。
---
docs/collaboration/interactions.en.md | 14 +++++++++-----
docs/collaboration/interactions.md | 8 +++++---
2 files changed, 14 insertions(+), 8 deletions(-)
diff --git a/docs/collaboration/interactions.en.md b/docs/collaboration/interactions.en.md
index f351fb4ba..969a37087 100644
--- a/docs/collaboration/interactions.en.md
+++ b/docs/collaboration/interactions.en.md
@@ -69,7 +69,7 @@ Cloud's own background loops, see [Agent IssueRuns](../agent-runs/issue-runs.en.
A workflow has exactly one execution boundary: an explicit **confirm**.
```text
-select workflow ──▶ read form descriptor ──▶ fill in (optional AI assist) ──▶ confirm ──▶ IssueRun
+select workflow ──▶ read form descriptor ──▶ fill in (optional AI assist) ──▶ confirm ──▶ WorkflowRun + IssueRun
no execution no execution no execution, no run the only execution boundary
```
@@ -77,6 +77,9 @@ select workflow ──▶ read form descriptor ──▶ fill in (optional AI as
Issue-facing rendering description: `formRef`, title, description and a list of fields. It is not the
workflow's schema and binds to no schema technology. Fields come in three bands: platform fields, the
variables declared by the workflow's Start node, and the graph's global variables.
+ Opening the form initially reads the current document. Selecting a published version in the platform
+ version selector reloads the chosen snapshot's form through
+ `GET /tenants/:tid/workflows/:wfid/snapshots/:snapshotId/form?issueId=…`.
- **Platform fields** are injected only when the form is opened for an Issue: repository URL and branch
(defaulting to the repository and default branch of the Issue's Project; required), a prompt, the
published version to run `snapshotId`, and additional context references `context_refs`. When the
@@ -90,10 +93,11 @@ select workflow ──▶ read form descriptor ──▶ fill in (optional AI as
- Frontend validation is only for user experience. On confirm Cloud resolves the target **again** and
validates the submitted values against the form of **the published version being confirmed**; a
workflow with no published version cannot be confirmed, because an unpublished draft cannot run.
- - The form the user saw comes from the current document; if its fields no longer match the confirmed
- version's, the confirm is refused with the changed fields listed
- (`409 workflow_version_fields_changed`), rather than validating the user's values against a schema
- they never saw.
+ - Cloud still compares the live draft's fields with the selected published version's fields. A
+ difference rejects confirm and lists the changed fields (`409 workflow_version_fields_changed`),
+ even when the frontend has rendered the selected snapshot's form. The frontend asks the user to
+ publish the current draft or choose another version; reading the snapshot's form does not remove
+ this check.
- The submitted values are kept as the interaction's input and the executed version is frozen into the
run input, so the run still knows which version it executed after the workflow is edited or
published again.
diff --git a/docs/collaboration/interactions.md b/docs/collaboration/interactions.md
index 6b548a1f5..717c41cfb 100644
--- a/docs/collaboration/interactions.md
+++ b/docs/collaboration/interactions.md
@@ -54,12 +54,14 @@ Agent 名册（`space_agents`，由已安装的 Agent 插件形成，见 [插件
工作流的执行边界只有一个：显式的**确认**。
```text
-选中工作流 ──▶ 读取表单描述 ──▶ 填写（可选 AI 辅助）──▶ 确认 ──▶ IssueRun
+选中工作流 ──▶ 读取表单描述 ──▶ 填写（可选 AI 辅助）──▶ 确认 ──▶ WorkflowRun + IssueRun
不执行 不执行 不执行、不建运行 唯一的执行边界
```
- **表单描述**（`GET /tenants/:tid/collaboration/forms/:formRef?issueId=…`）是面向 Issue 的渲染描述：`formRef`、标题、说明和
字段列表。它不是工作流的 schema，也不绑定任何 schema 技术。字段分三段：平台字段、工作流 Start 节点声明的变量、图的全局变量。
+ 初次打开时读取当前文档；在平台版本选择器中选择已发布版本后，前端通过
+ `GET /tenants/:tid/workflows/:wfid/snapshots/:snapshotId/form?issueId=…` 重新读取所选快照的表单。
- **平台字段**只在带着 Issue 打开表单时注入：仓库地址和分支（默认取 Issue 关联 Project 的仓库与默认分支，必填）、提示词、要执行
的已发布版本 `snapshotId`、附加的上下文引用 `context_refs`。工作流自己声明了同名变量时，以工作流的声明为准。
- **AI 辅助**（`POST .../issues/:iid/collaboration/assist`）只返回字段级的建议补丁和建议的上下文引用：从不创建运行，也不写入
@@ -67,8 +69,8 @@ Agent 名册（`space_agents`，由已安装的 Agent 插件形成，见 [插件
- **确认**（`POST .../issues/:iid/interactions/:ixid/confirm`）是唯一把工作流交互变成执行意图的转换：
- 前端校验只是体验，Cloud 在确认时**重新**解析目标，并按**被确认的那个已发布版本**的表单校验提交的值；没有任何已发布版本的
工作流不能确认，因为不能执行未发布的草稿。
- - 用户看到的表单来自当前文档；如果它与被确认版本的字段已经不一致，确认被拒绝并列出变化的字段
- （`409 workflow_version_fields_changed`），而不是用用户没见过的 schema 校验他填的值。
+ - Cloud 仍比较当前活稿与所选发布版本的字段；即使前端已展示所选快照的表单，两者不一致时仍拒绝确认并列出变化的字段
+ （`409 workflow_version_fields_changed`）。前端提示先发布当前活稿或选择其他版本；按快照读取表单并未取消这条检查。
- 提交的值作为交互的输入保存，被执行的版本冻结进运行输入；运行之后工作流再被编辑或发布，它仍然知道自己执行的是哪个版本。
- 同一事务里创建一个 WorkflowRun 和一个 IssueRun，IssueRun 的 `workflowInvocationRef` 指向该 WorkflowRun（与租户一起的复合外键，且唯一）；直接运行、manual 工作流 IssueRun 与确认共用同一个创建函数。已关联的 IssueRun 不交给 `ExecutionDispatcher`。
- 用 `run_id IS NULL` 的比较并交换认领交互，重复确认返回 `409 interaction_already_confirmed`；对非表单交互确认返回
From 713b19168519607b3ab7bea8914692c1eac2bda4 Mon Sep 17 00:00:00 2001
From: creegon2 <315603738+creegon2@users.noreply.github.com>
Date: Mon, 19 Oct 2026 19:26:24 -0700
Subject: [PATCH 10/12] feat(core): project workflow run states onto linked
issue runs
Project only successful workflow status writes in the same transaction. Propagate settle CAS outcomes before session dispatch or cleanup, and truncate existing output/error summaries at UTF-8 boundaries within 20000 bytes.
Add PostgreSQL coverage for Confirm lifecycle, repeat invocation after completion, deterministic failure, concurrent passes, lost CAS, summary limits, and atomic rollback. Validate with task format, go run ./cmd/checkformat, task lint, targeted DB tests, and the requested core/integration suite.
---
integration/workflow_invocation_test.go | 152 +++++++++++++++++
internal/core/workflow_issue_projection.go | 32 +++-
internal/core/workflow_run.go | 4 +-
internal/core/workflow_run_session.go | 12 +-
internal/core/workflow_scheduler.go | 29 ++--
internal/core/workflow_scheduler_db_test.go | 170 ++++++++++++++++++++
6 files changed, 381 insertions(+), 18 deletions(-)
diff --git a/integration/workflow_invocation_test.go b/integration/workflow_invocation_test.go
index 96460d9c2..cb83e357f 100644
--- a/integration/workflow_invocation_test.go
+++ b/integration/workflow_invocation_test.go
@@ -4,6 +4,7 @@ import (
"context"
"reflect"
"sort"
+ "strings"
"sync"
"testing"
@@ -256,3 +257,154 @@ func TestManualWorkflowInvocationCreatesLinkedRun(t *testing.T) {
t.Fatalf("manual start created %d valid linked pairs, want 1", n)
}
}
+
+// confirmInvocationScene crosses the public comment/form boundary. Unlike the fixture's direct
+// workflow-run POST, Confirm leaves execution pending for the real scheduler used by these tests.
+func confirmInvocationScene(t *testing.T, f *fixture, graph core.Object, key string) (wid, iid string, confirmed core.Object) {
+ t.Helper()
+ wid = f.call("POST", f.path("/workflows"), core.Object{
+ "name": "Scheduled invocation", "graph": graph,
+ }, key+"-workflow", 200).O("resource").S("id")
+ publishWorkflow(t, f, wid, key+"-publish")
+ iid = launchIssue(t, f, runProject(t, f, key+"-project"), "Scheduled invocation", key+"-issue").S("id")
+ interaction := workflowComment(t, f, wid, iid, key+"-comment", "run this workflow")
+ confirmed = f.call("POST", confirmPath(f, iid, interaction.S("id")), core.Object{"values": core.Object{
+ "repository": "ora-space/cloud", "branch": "main",
+ }}, key+"-confirm", 200).O("resource")
+ if confirmed.S("status") != "queued" || confirmed.S("workflowInvocationRef") == "" {
+ t.Fatalf("Confirm must enqueue a linked invocation before scheduling: %v", confirmed)
+ }
+ return wid, iid, confirmed
+}
+
+// assertInvocationProjection reads both public run resources and the issue timeline, so a scheduler
+// state change without its IssueRun projection or lifecycle activity cannot satisfy the test.
+func assertInvocationProjection(t *testing.T, f *fixture, wid, iid string, confirmed core.Object, workflowStatus, issueStatus string, actions []string) (core.Object, core.Object) {
+ t.Helper()
+ workflow := f.call("GET", f.path("/workflows/"+wid+"/runs/"+confirmed.S("workflowInvocationRef")), nil, "", 200)
+ var issueRun core.Object
+ for _, candidate := range issueItems(f.call("GET", f.path("/issues/"+iid+"/runs"), nil, "", 200)) {
+ if candidate.S("id") == confirmed.S("id") {
+ issueRun = candidate
+ break
+ }
+ }
+ if workflow.S("status") != workflowStatus || issueRun.S("status") != issueStatus {
+ t.Fatalf("scheduler projection disagrees: workflow=%v issue=%v; want %s/%s", workflow, issueRun, workflowStatus, issueStatus)
+ }
+ if issueRun.S("workflowInvocationRef") != workflow.S("id") || issueRun.S("executorId") != wid || issueRun.S("executorType") != "workflow" {
+ t.Fatalf("projected IssueRun lost its execution identity: %v", issueRun)
+ }
+ if workflow.S("startedAt") == "" || issueRun.S("startedAt") != workflow.S("startedAt") || issueRun.S("completedAt") != workflow.S("finishedAt") {
+ t.Fatalf("projected lifecycle timestamps disagree: workflow=%v issue=%v", workflow, issueRun)
+ }
+ if issueRun.S("failureReason") != workflow.S("error") || issueRun.S("error") != workflow.S("error") {
+ t.Fatalf("projected failure reason disagrees: workflow=%v issue=%v", workflow, issueRun)
+ }
+ var got []string
+ for _, item := range issueItems(f.call("GET", f.path("/issues/"+iid+"/timeline"), nil, "", 200)) {
+ details := item.O("details")
+ if item.S("kind") != "activity" || details.S("runId") != confirmed.S("id") {
+ continue
+ }
+ got = append(got, item.S("action"))
+ if details.S("executorType") != "workflow" || details.S("executorId") != wid {
+ t.Fatalf("activity lost its executor identity: %v", item)
+ }
+ if item.S("action") != "run.enqueued" && details.S("workflowInvocationRef") != workflow.S("id") {
+ t.Fatalf("scheduler activity lost its workflow invocation identity: %v", item)
+ }
+ if item.S("action") == "run.failed" && details.S("failureReason") != workflow.S("error") {
+ t.Fatalf("failure activity lost the authoritative reason: %v", item)
+ }
+ }
+ if !reflect.DeepEqual(got, actions) {
+ t.Fatalf("invocation lifecycle activities = %v, want %v", got, actions)
+ }
+ return workflow, issueRun
+}
+
+func TestWorkflowConfirmSchedulerProjectsRunningOnce(t *testing.T) {
+ f := setup(t)
+ seedPersonalModel(t, f)
+ wid, iid, confirmed := confirmInvocationScene(t, f, runGraph(), "inv-scheduler-running")
+ // The Project is still provisioning: session startup waits without Node or cluster execution.
+ // A second advance exercises the running-state retry, which must not emit another run.started.
+ for range 2 {
+ must(t, f.store.AdvanceWorkflowRunsOnce(t.Context()))
+ }
+ workflow, _ := assertInvocationProjection(t, f, wid, iid, confirmed, "running", "running", []string{"run.enqueued", "run.started"})
+ if len(workflow.O("nodeStates")) != 0 || workflow.S("finishedAt") != "" {
+ t.Fatalf("waiting invocation fabricated committed nodes or terminal evidence: %v", workflow)
+ }
+}
+
+func TestWorkflowConfirmSchedulerCompletesAndAllowsNextInvocation(t *testing.T) {
+ f := setup(t)
+ graph := startInputGraph(nil)
+ graph["nodes"] = append(graph["nodes"].([]any), core.Object{
+ "id": "output-1", "type": "workflow", "position": core.Object{"x": 300, "y": 0},
+ "data": core.Object{"kind": "output", "title": "Done"},
+ })
+ graph["edges"] = []any{core.Object{"id": "output-edge", "source": "start-1", "target": "output-1"}}
+ wid, iid, confirmed := confirmInvocationScene(t, f, graph, "inv-scheduler-completed")
+ for range 2 {
+ must(t, f.store.AdvanceWorkflowRunsOnce(t.Context()))
+ }
+ workflow, _ := assertInvocationProjection(t, f, wid, iid, confirmed, "succeeded", "completed", []string{"run.enqueued", "run.started", "run.completed"})
+ for _, node := range []string{"start-1", "output-1"} {
+ if workflow.O("nodeStates").O(node).S("status") != "succeeded" {
+ t.Fatalf("completed invocation did not execute node %s: %v", node, workflow)
+ }
+ }
+ if workflow.S("finishedAt") == "" {
+ t.Fatal("completed invocation has no terminal timestamp")
+ }
+
+ // workflowComment returns the first form, so find the fresh unclaimed form explicitly on this
+ // same Issue. Terminal projection must release the pending-executor dedup for this Confirm.
+ f.call("POST", f.path("/issues/"+iid+"/comments"), core.Object{
+ "body": "run this workflow again", "targets": []any{core.Object{"type": "workflow", "id": wid}},
+ }, "inv-scheduler-completed-next-comment", 200)
+ var nextInteraction core.Object
+ for _, interaction := range issueItems(f.call("GET", f.path("/issues/"+iid+"/interactions"), nil, "", 200)) {
+ if interaction.S("mode") == "form" && interaction.S("runId") == "" {
+ nextInteraction = interaction
+ break
+ }
+ }
+ if nextInteraction == nil {
+ t.Fatal("second workflow comment did not create an unclaimed form")
+ }
+ next := f.call("POST", confirmPath(f, iid, nextInteraction.S("id")), core.Object{"values": core.Object{
+ "repository": "ora-space/cloud", "branch": "main",
+ }}, "inv-scheduler-completed-next-confirm", 200).O("resource")
+ if next.S("status") != "queued" || next.S("id") == confirmed.S("id") || next.S("workflowInvocationRef") == "" || next.S("workflowInvocationRef") == confirmed.S("workflowInvocationRef") {
+ t.Fatalf("second Confirm did not create a new linked queued invocation: first=%v next=%v", confirmed, next)
+ }
+}
+
+func TestWorkflowConfirmSchedulerProjectsConditionFailure(t *testing.T) {
+ f := setup(t)
+ graph := startInputGraph(nil)
+ // input is a declared Start root and is always a string. Publishing admits the selector, but
+ // traversing a nested object path on that string deterministically fails during real execution.
+ graph["nodes"] = append(graph["nodes"].([]any), core.Object{
+ "id": "condition-1", "type": "workflow", "position": core.Object{"x": 300, "y": 0},
+ "data": core.Object{"kind": "condition", "title": "Invalid nested input", "cases": []any{core.Object{
+ "id": "matched", "logic": "and", "conditions": []any{core.Object{
+ "variableSelector": []any{"start-1", "input", "missing"}, "operator": "equals", "value": "value",
+ }},
+ }}},
+ })
+ graph["edges"] = []any{core.Object{"id": "condition-edge", "source": "start-1", "target": "condition-1"}}
+ wid, iid, confirmed := confirmInvocationScene(t, f, graph, "inv-scheduler-failed")
+ for range 2 {
+ must(t, f.store.AdvanceWorkflowRunsOnce(t.Context()))
+ }
+ workflow, _ := assertInvocationProjection(t, f, wid, iid, confirmed, "failed", "failed", []string{"run.enqueued", "run.started", "run.message", "run.failed"})
+ failed := workflow.O("nodeStates").O("condition-1")
+ if failed.S("status") != "failed" || !strings.Contains(workflow.S("error"), "start-1.input.missing") || failed.S("error") != workflow.S("error") || workflow.S("finishedAt") == "" {
+ t.Fatalf("condition failure did not preserve its node reason and terminal evidence: %v", workflow)
+ }
+}
diff --git a/internal/core/workflow_issue_projection.go b/internal/core/workflow_issue_projection.go
index c42d2696a..9931b4c5d 100644
--- a/internal/core/workflow_issue_projection.go
+++ b/internal/core/workflow_issue_projection.go
@@ -1,9 +1,37 @@
package core
-import "strings"
+import (
+ "strings"
+ "unicode/utf8"
+)
+
+// workflowIssueOutputSummary selects existing output-node results or the run error. Callers bound
+// the UTF-8 message before projection so execution settlement cannot fail on activity size alone.
+func workflowIssueOutputSummary(graph, states Object, runErr string) string {
+ message := strings.TrimSpace(runErr)
+ if message == "" {
+ outputs := Object{}
+ for _, node := range readWorkflowGraph(graph).Nodes {
+ if state := states.O(node.ID); node.Kind == "output" && state.S("status") == "succeeded" && len(state.O("output")) > 0 {
+ outputs[node.ID] = state.O("output")
+ }
+ }
+ if len(outputs) > 0 {
+ message = jsonText(outputs)
+ }
+ }
+ if len(message) > 20000 {
+ end := 20000
+ for !utf8.RuneStart(message[end]) {
+ end--
+ }
+ message = message[:end]
+ }
+ return message
+}
// projectWorkflowRunToIssueRun mirrors the committed-in-this-transaction workflow state onto its
-// optional IssueRun. The scheduler must call it after its state compare-and-set, in the same
+// optional IssueRun. Execution callers invoke it only after a successful state write, in the same
// Store.transact boundary, so a projection or activity failure rolls back the workflow transition.
// Direct workflow runs have no IssueRun and are a no-op. The caller supplies the execution's output
// summary because per-node output has no canonical summary field in workflow_runs. awaiting_input
diff --git a/internal/core/workflow_run.go b/internal/core/workflow_run.go
index 001dca868..37afd0c44 100644
--- a/internal/core/workflow_run.go
+++ b/internal/core/workflow_run.go
@@ -126,7 +126,9 @@ func createWorkflowRun(t *transaction, r *PublicRequest, uid string) Object {
id := created.S("id")
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)
+ if t.execRows("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) > 0 {
+ projectWorkflowRunToIssueRun(t, r.TenantID, id, workflowIssueOutputSummary(s.O("graph"), nodeStates, ""))
+ }
} else {
// The run-creation gate (mappings decision D2): every slot the snapshot references must
// resolve against the invoker's personal configuration before the run exists, and each
diff --git a/internal/core/workflow_run_session.go b/internal/core/workflow_run_session.go
index b35aae95d..751f8ea19 100644
--- a/internal/core/workflow_run_session.go
+++ b/internal/core/workflow_run_session.go
@@ -124,8 +124,7 @@ func commitWorkflowAgentTerminal(t *transaction, runID string, version int64, no
return states, false
default:
// The node succeeded; downstream waves continue in this same pass.
- settleWorkflowRunStates(t, runID, version, "running", "", states)
- return states, true
+ return states, settleWorkflowRunStates(t, runID, version, "running", "", states)
}
}
@@ -135,8 +134,9 @@ func commitWorkflowAgentTerminal(t *transaction, runID string, version int64, no
// Declaring delete on a busy Project returns busy — a later pass re-declares, so a terminal run
// is never left with a live Workspace forever.
func finishWorkflowRun(t *transaction, runID string, version int64, status, runErr string, states map[string]workflowNodeState) {
- settleWorkflowRunStates(t, runID, version, status, runErr, states)
- finishWorkflowRunCleanup(t, runID)
+ if settleWorkflowRunStates(t, runID, version, status, runErr, states) {
+ finishWorkflowRunCleanup(t, runID)
+ }
}
// finishWorkflowRunCleanup deletes the run's unregistered execution work and re-declares the
@@ -265,7 +265,9 @@ func startWorkflowAgentSession(t *transaction, run Object, node workflowNodeView
return
}
states[node.ID] = workflowNodeState{Status: "running", DisplayName: workflowNodeDisplayName(node), StartedAt: nowRFC3339()}
- settleWorkflowRunStates(t, runID, version, "running", "", states)
+ if !settleWorkflowRunStates(t, runID, version, "running", "", states) {
+ return
+ }
enqueueExecutionWork(t, runID, "agent_session", spec, target, node.ID, time.Time{})
}
diff --git a/internal/core/workflow_scheduler.go b/internal/core/workflow_scheduler.go
index b41a035f8..6a91ea4a1 100644
--- a/internal/core/workflow_scheduler.go
+++ b/internal/core/workflow_scheduler.go
@@ -479,6 +479,7 @@ func advanceWorkflowRun(t *transaction, runID string) {
WHERE id=$1 AND status='pending' AND deleted_at IS NULL`, runID) == 0 {
return
}
+ projectWorkflowRunToIssueRun(t, run.S("tenantId"), runID, "")
run = t.one(`
SELECT r.*, s.graph FROM workflow_runs r
JOIN workflow_snapshots s ON s.id = r.snapshot_id
@@ -591,25 +592,33 @@ func workflowInactiveStates(graph workflowGraphView, states map[string]workflowN
// the first terminal write, so a later pass re-declaring a terminal status cannot move the moment
// the run actually stopped. started_at is not this function's business: it is set once, where the
// run first leaves `pending`.
-func settleWorkflowRunStates(t *transaction, runID string, version int64, status, runErr string, states map[string]workflowNodeState) {
+func settleWorkflowRunStates(t *transaction, runID string, version int64, status, runErr string, states map[string]workflowNodeState) bool {
// The statuses a run never leaves. `pending`, `running` and `awaiting_input` are the others the
// column admits, and all three are states a run is still in.
terminal := status == "succeeded" || status == "failed" || status == "cancelled"
+ var applied int64
+ encoded := Object{}
if states == nil {
- t.exec(`UPDATE workflow_runs SET status=$2, error=$3,
+ applied = t.execRows(`UPDATE workflow_runs SET status=$2, error=$3,
finished_at=CASE WHEN $5 THEN COALESCE(finished_at,now()) ELSE finished_at END,
version=version+1, updated_at=now()
WHERE id=$1 AND version=$4 AND deleted_at IS NULL`, runID, status, runErr, version, terminal)
- return
+ } else {
+ for id, state := range states {
+ encoded[id] = state.object()
+ }
+ applied = t.execRows(`UPDATE workflow_runs SET node_states=$2, status=$3, error=$4,
+ finished_at=CASE WHEN $6 THEN COALESCE(finished_at,now()) ELSE finished_at END,
+ version=version+1, updated_at=now()
+ WHERE id=$1 AND version=$5 AND deleted_at IS NULL`, runID, jsonText(encoded), status, runErr, version, terminal)
}
- encoded := Object{}
- for id, state := range states {
- encoded[id] = state.object()
+ if applied == 0 {
+ return false
}
- t.exec(`UPDATE workflow_runs SET node_states=$2, status=$3, error=$4,
- finished_at=CASE WHEN $6 THEN COALESCE(finished_at,now()) ELSE finished_at END,
- version=version+1, updated_at=now()
- WHERE id=$1 AND version=$5 AND deleted_at IS NULL`, runID, jsonText(encoded), status, runErr, version, terminal)
+ run := t.one(`SELECT r.tenant_id,s.graph FROM workflow_runs r
+ JOIN workflow_snapshots s ON s.id=r.snapshot_id WHERE r.id=$1`, runID)
+ projectWorkflowRunToIssueRun(t, run.S("tenantId"), runID, workflowIssueOutputSummary(run.O("graph"), encoded, runErr))
+ return true
}
// workflowRunVersion re-reads the row's version after a mid-pass settle bumped it: one pass commits
diff --git a/internal/core/workflow_scheduler_db_test.go b/internal/core/workflow_scheduler_db_test.go
index 8f1c85e12..b6d65ecb5 100644
--- a/internal/core/workflow_scheduler_db_test.go
+++ b/internal/core/workflow_scheduler_db_test.go
@@ -9,8 +9,11 @@ package core
import (
"context"
"encoding/json"
+ "reflect"
"strings"
+ "sync"
"testing"
+ "unicode/utf8"
"github.com/google/uuid"
)
@@ -93,6 +96,173 @@ func TestWorkflowAdvanceSucceedsLinearRun(t *testing.T) {
if start := states["s"].(map[string]any)["output"].(map[string]any); start["input"] == nil {
t.Fatal("start node output does not carry the kickoff prompt root")
}
+ var activities, linked int
+ if err := store.Pool.QueryRow(`SELECT (SELECT count(*) FROM issue_activities),
+ (SELECT count(*) FROM issue_runs WHERE workflow_invocation_ref=$1)`, runID).Scan(&activities, &linked); err != nil || activities != 0 || linked != 0 {
+ t.Fatalf("direct run created collaboration records: activities=%d linked=%d err=%v", activities, linked, err)
+ }
+}
+
+func TestWorkflowAdvanceProjectsConcurrentPassesOnce(t *testing.T) {
+ store := dispatcherDB(t)
+ s := seedWorkflowProjection(t, store)
+ graph := wfGraph([]Object{wfNode("s", "start", ""), wfNode("o", "output", "")}, [][3]string{{"s", "o", ""}})
+ if _, err := store.Pool.Exec(`UPDATE workflow_snapshots SET graph=$2::jsonb
+ WHERE id=(SELECT snapshot_id FROM workflow_runs WHERE id=$1)`, s.workflowRun, jsonText(graph)); err != nil {
+ t.Fatal(err)
+ }
+ start := make(chan struct{})
+ var wg sync.WaitGroup
+ errs := make([]error, 2)
+ for i := range errs {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ <-start
+ errs[i] = store.AdvanceWorkflowRunsOnce(context.Background())
+ }()
+ }
+ close(start)
+ wg.Wait()
+ for _, err := range errs {
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ var status string
+ var version int
+ if err := store.Pool.QueryRow("SELECT status,version FROM issue_runs WHERE id=$1", s.issueRun).Scan(&status, &version); err != nil || status != "completed" || version != 3 {
+ t.Fatalf("linked run: status=%s version=%d err=%v", status, version, err)
+ }
+ activities := workflowProjectionActivities(t, store, s)
+ actions := make([]string, len(activities))
+ for i, activity := range activities {
+ actions[i] = activity.S("action")
+ }
+ if !reflect.DeepEqual(actions, []string{"run.enqueued", "run.started", "run.completed"}) {
+ t.Fatalf("concurrent/repeated passes emitted %v", actions)
+ }
+}
+
+func TestWorkflowSettleLostCASDoesNotProject(t *testing.T) {
+ store := dispatcherDB(t)
+ for _, withStates := range []bool{false, true} {
+ name := "without-states"
+ if withStates {
+ name = "with-states"
+ }
+ t.Run(name, func(t *testing.T) {
+ s := seedWorkflowProjection(t, store)
+ // A concurrent writer won, but its IssueRun is deliberately still queued: calling
+ // projection after our lost CAS would now emit run.started from that winner's row.
+ if _, err := store.Pool.Exec("UPDATE workflow_runs SET status='running',started_at=now(),version=version+1 WHERE id=$1", s.workflowRun); err != nil {
+ t.Fatal(err)
+ }
+ var states map[string]workflowNodeState
+ if withStates {
+ states = map[string]workflowNodeState{"s": {Status: "succeeded"}}
+ }
+ _, err := store.transact(context.Background(), func(tx *transaction) Object {
+ if settleWorkflowRunStates(tx, s.workflowRun, 1, "failed", "stale failure", states) {
+ t.Fatal("stale version unexpectedly applied")
+ }
+ return Object{}
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ status, runErr, committed, version := readWorkflowRun(t, store, s.workflowRun)
+ if status != "running" || runErr != "" || len(committed) != 0 || version != 2 {
+ t.Fatalf("lost CAS changed winner: %s %q %v version=%d", status, runErr, committed, version)
+ }
+ var issueStatus string
+ if err := store.Pool.QueryRow("SELECT status FROM issue_runs WHERE id=$1", s.issueRun).Scan(&issueStatus); err != nil || issueStatus != "queued" {
+ t.Fatalf("lost CAS projected: status=%s err=%v", issueStatus, err)
+ }
+ if activities := workflowProjectionActivities(t, store, s); len(activities) != 1 {
+ t.Fatalf("lost CAS emitted activities: %v", activities)
+ }
+ })
+ }
+}
+
+func TestWorkflowSettleTruncatesIssueSummary(t *testing.T) {
+ store := dispatcherDB(t)
+ for _, source := range []string{"error", "output"} {
+ t.Run(source, func(t *testing.T) {
+ s := seedWorkflowProjection(t, store)
+ text := strings.Repeat("界", 7000)
+ status, runErr := "failed", text
+ var states map[string]workflowNodeState
+ if source == "output" {
+ graph := wfGraph([]Object{wfNode("s", "start", ""), wfNode("o", "output", "")}, [][3]string{{"s", "o", ""}})
+ if _, err := store.Pool.Exec(`UPDATE workflow_snapshots SET graph=$2::jsonb
+ WHERE id=(SELECT snapshot_id FROM workflow_runs WHERE id=$1)`, s.workflowRun, jsonText(graph)); err != nil {
+ t.Fatal(err)
+ }
+ status, runErr = "succeeded", ""
+ states = map[string]workflowNodeState{"o": {Status: "succeeded", Output: Object{"text": text}}}
+ }
+ _, err := store.transact(context.Background(), func(tx *transaction) Object {
+ if !settleWorkflowRunStates(tx, s.workflowRun, 1, status, runErr, states) {
+ t.Fatal("settle did not apply")
+ }
+ return Object{}
+ })
+ if err != nil {
+ t.Fatalf("oversized summary rolled back settlement: %v", err)
+ }
+ activities := workflowProjectionActivities(t, store, s)
+ if len(activities) != 3 || activities[1].S("action") != "run.message" {
+ t.Fatalf("terminal summary activities: %v", activities)
+ }
+ message := activities[1].O("details").S("message")
+ if len(message) < 19997 || len(message) > 20000 || !utf8.ValidString(message) {
+ t.Fatalf("summary bytes=%d valid UTF-8=%v", len(message), utf8.ValidString(message))
+ }
+ actual, actualErr, _, version := readWorkflowRun(t, store, s.workflowRun)
+ if actual != status || actualErr != runErr || version != 2 {
+ t.Fatalf("settle did not preserve execution result: status=%s error bytes=%d version=%d", actual, len(actualErr), version)
+ }
+ var issueStatus string
+ wantIssueStatus := "failed"
+ if status == "succeeded" {
+ wantIssueStatus = "completed"
+ }
+ if err := store.Pool.QueryRow("SELECT status FROM issue_runs WHERE id=$1", s.issueRun).Scan(&issueStatus); err != nil || issueStatus != wantIssueStatus {
+ t.Fatalf("terminal IssueRun: status=%s err=%v", issueStatus, err)
+ }
+ })
+ }
+}
+
+func TestWorkflowAdvanceProjectionFailureRollsBackClaim(t *testing.T) {
+ store := dispatcherDB(t)
+ s := seedWorkflowProjection(t, store)
+ if _, err := store.Pool.Exec("ALTER TABLE issue_activities ADD CONSTRAINT reject_scheduler_start CHECK(action <> 'run.started')"); err != nil {
+ t.Fatal(err)
+ }
+ _, err := store.transact(context.Background(), func(tx *transaction) Object {
+ advanceWorkflowRun(tx, s.workflowRun)
+ return Object{}
+ })
+ if err == nil {
+ t.Fatal("failed projection must abort the scheduler transaction")
+ }
+ status, _, _, version := readWorkflowRun(t, store, s.workflowRun)
+ if status != "pending" || version != 1 {
+ t.Fatalf("projection failure left claimed workflow: %s version=%d", status, version)
+ }
+ var issueStatus string
+ if err := store.Pool.QueryRow("SELECT status FROM issue_runs WHERE id=$1", s.issueRun).Scan(&issueStatus); err != nil || issueStatus != "queued" {
+ t.Fatalf("projection failure changed IssueRun: status=%s err=%v", issueStatus, err)
+ }
+ if activities := workflowProjectionActivities(t, store, s); len(activities) != 1 {
+ t.Fatalf("rolled-back claim emitted activities: %v", activities)
+ }
}
// A ready agent node on a run with no project settles the run failed with an explicit reason: the
From fb34d8248de73b569e9d932e9dbc8bb0d19b42e3 Mon Sep 17 00:00:00 2001
From: creegon2 <315603738+creegon2@users.noreply.github.com>
Date: Mon, 19 Oct 2026 19:26:25 -0700
Subject: [PATCH 11/12] docs(workflows): describe linked issue run status
projection
Keep Chinese and English workflow, collaboration, and core documentation aligned with same-transaction projection, affected-row guards, bounded summaries, and direct-run/awaiting-input no-ops.
---
docs/collaboration/interactions.en.md | 6 ++++++
docs/collaboration/interactions.md | 4 ++++
docs/workflows/workflows.en.md | 7 ++++++-
docs/workflows/workflows.md | 5 ++++-
internal/core/README.en.md | 8 +++++---
internal/core/README.md | 5 +++--
6 files changed, 28 insertions(+), 7 deletions(-)
diff --git a/docs/collaboration/interactions.en.md b/docs/collaboration/interactions.en.md
index 969a37087..066fe12c4 100644
--- a/docs/collaboration/interactions.en.md
+++ b/docs/collaboration/interactions.en.md
@@ -126,6 +126,12 @@ Agent runs do not go through `ExecutionDispatcher`; Cloud's own scheduling carri
[Agent IssueRuns](../agent-runs/issue-runs.en.md)). Runs started directly from a workflow are a different
aggregate, WorkflowRun; see [Workflows](../workflows/workflows.en.md).
+For a linked workflow IssueRun, actual WorkflowRun status updates advance it in the same transaction:
+updates affecting no rows do not project, `awaiting_input` leaves it unchanged, and contract conflicts
+roll back the transaction. Output summaries are `system` `run.message` activities bounded to 20000
+bytes of valid UTF-8, never workflow-authored comments. Direct workflow runs without a linked IssueRun
+create none of these collaboration records.
+
**Development fixtures.** `collaboration.development_fixtures: true` (environment variable
`CLOUD_COLLABORATION_DEVELOPMENT_FIXTURES`) installs the in-process doubles from `internal/collab` on the
Store: Agent and team targets with fixed IDs, a deterministic context builder (at most the 10 latest
diff --git a/docs/collaboration/interactions.md b/docs/collaboration/interactions.md
index 717c41cfb..db7a668cf 100644
--- a/docs/collaboration/interactions.md
+++ b/docs/collaboration/interactions.md
@@ -91,6 +91,10 @@ Issue 协作通过一组端口向外部能力请求"做一件事"，而不定义
Agent 运行不走 `ExecutionDispatcher`，由 Cloud 自己的调度推进（见 [Agent IssueRun](../agent-runs/issue-runs.md)）。直接从工作流
发起的运行是另一个聚合 WorkflowRun，见 [工作流](../workflows/workflows.md)。
+关联工作流的 IssueRun 由 WorkflowRun 的实际状态更新在同一事务内推进：未更新到行时不投影，`awaiting_input` 不改变 IssueRun，
+契约冲突使事务回滚。输出摘要写成 `system` 的 `run.message` 活动，最多 20000 字节且保持合法 UTF-8，不创建工作流作者评论。
+没有关联 IssueRun 的直接工作流运行不产生这些协作记录。
+
**开发夹具。** `collaboration.development_fixtures: true`（环境变量 `CLOUD_COLLABORATION_DEVELOPMENT_FIXTURES`）在 Store 上装入
`internal/collab` 的进程内替身：固定 ID 的 Agent 和团队目标、确定性的上下文构建器（最近至多 10 条评论，不用 AI）、给出固定输出的
派发器和 AI 辅助。工作流目标仍然来自数据库。这个开关只用于开发和演示，生产默认关闭；它与认证完全独立：开启 GitHub 登录不会打开
diff --git a/docs/workflows/workflows.en.md b/docs/workflows/workflows.en.md
index d7878d128..44ca7bfe0 100644
--- a/docs/workflows/workflows.en.md
+++ b/docs/workflows/workflows.en.md
@@ -114,7 +114,12 @@ seconds):
Development fixtures (`collaboration.development_fixtures`) install a deterministic run simulator that fills in the
execution trace at run creation and says in its output that nothing really ran; production never installs it.
-A workflow IssueRun is linked through `workflow_invocation_ref` to the WorkflowRun created in the same transaction, and the scheduling loop above advances that run. Projecting WorkflowRun status back onto the IssueRun is not yet called from the scheduler, so until then the IssueRun stays `queued` (see
+A workflow IssueRun is linked through `workflow_invocation_ref` to the WorkflowRun created in the same transaction,
+and the scheduling loop above advances that run. After a WorkflowRun status update affects a row, its state,
+timestamps and errors are projected onto the linked IssueRun in that same transaction. An update affecting no rows
+does not project; contract conflicts roll back the transaction. `awaiting_input` leaves the IssueRun unchanged,
+and direct runs without a linked IssueRun create no collaboration records. Output summaries are `system`
+`run.message` activities, bounded to 20000 bytes of valid UTF-8, never workflow-authored comments (see
[Issue collaboration](../collaboration/interactions.en.md#ports-issues-own-only-the-consuming-side-contract)).
## Plugin asset import
diff --git a/docs/workflows/workflows.md b/docs/workflows/workflows.md
index bbc975671..0cdf5fc29 100644
--- a/docs/workflows/workflows.md
+++ b/docs/workflows/workflows.md
@@ -81,7 +81,10 @@ WorkflowRun 由 Cloud server 里的调度循环推进（`AdvanceWorkflowRunsOnce
这次运行的 Workspace。
开发夹具（`collaboration.development_fixtures`）会装入一个确定性的运行模拟器，在创建运行时就填好执行轨迹，并在输出里说明没有真实执行；生产环境
-不装它。工作流 IssueRun 通过 `workflow_invocation_ref` 关联同一事务里创建的 WorkflowRun，执行由上面的调度循环推进；把 WorkflowRun 状态投影回 IssueRun 的调用点还没有接入调度器，在此之前 IssueRun 保持 `queued`（见 [Issue 协作](../collaboration/interactions.md#端口issue-只拥有消费侧的契约)）。
+不装它。工作流 IssueRun 通过 `workflow_invocation_ref` 关联同一事务里创建的 WorkflowRun，执行由上面的调度循环推进。
+WorkflowRun 的实际状态更新成功后，在同一事务里将状态、时间和错误投影回关联的 IssueRun；未更新到行时不投影，契约冲突使事务回滚。
+`awaiting_input` 不改变 IssueRun，未关联 IssueRun 的直接运行也不产生协作记录。输出摘要作为 `system` 的 `run.message` 活动写入，
+最多 20000 字节且保持合法 UTF-8，不创建工作流作者评论（见 [Issue 协作](../collaboration/interactions.md#端口issue-只拥有消费侧的契约)）。
## 插件资产导入
diff --git a/internal/core/README.en.md b/internal/core/README.en.md
index bda913b2a..b393b49cb 100644
--- a/internal/core/README.en.md
+++ b/internal/core/README.en.md
@@ -76,9 +76,11 @@ whose `input.workflowSnapshotId` matches the linked execution. Linked IssueRuns
Dispatcher, including development doubles.
`projectWorkflowRunToIssueRun` projects authoritative state, timestamps and errors inside the caller's
-transaction. Output summaries are `system` `run.message` activities, never workflow-authored comments.
-PostgreSQL tests call this function directly. Scheduler calls are not yet wired, so the linked IssueRun
-still stays `queued`. Freezing execution identity and slot bindings does not establish real Node
+transaction. Every actual status-update path calls it only after affecting a row; projection contract
+conflicts roll back the whole transaction. `awaiting_input` and direct runs without a linked IssueRun
+leave collaboration records unchanged. Output summaries are `system` `run.message` activities bounded
+to 20000 bytes of valid UTF-8, never workflow-authored comments.
+Freezing execution identity and slot bindings does not establish real Node
acceptance.
## Session Dashboard reads
diff --git a/internal/core/README.md b/internal/core/README.md
index 4b18483c6..e837637d1 100644
--- a/internal/core/README.md
+++ b/internal/core/README.md
@@ -72,8 +72,9 @@ Start 与全局变量；平台字段保留在 IssueRun，`input.workflowSnapshot
有引用的 IssueRun 不进入 legacy Dispatcher；开发 mock 的完成状态不能替代 WorkflowRun 状态。
`projectWorkflowRunToIssueRun` 在调用方事务内从 WorkflowRun 投影状态、时间及错误，
-输出摘要写为 `system` 的 `run.message` 活动，不创建 workflow 作者评论。
-该函数已由 PostgreSQL 测试直接调用；调用点尚未接入调度器，因此关联的 IssueRun 仍保持 `queued`。
+各实际状态更新路径仅在更新到行后调用它；投影契约冲突使整个事务回滚。
+`awaiting_input` 和没有关联 IssueRun 的直接运行不改变协作记录。
+输出摘要写为 `system` 的 `run.message` 活动，最多 20000 字节且保持合法 UTF-8，不创建 workflow 作者评论。
创建身份和槽位绑定的接入不代表实际 Node 联合验收已经完成。
## 会话 Dashboard 读取
From 6699540325723d410aa8e98489f41bed18acf4d2 Mon Sep 17 00:00:00 2001
From: creegon2 <315603738+creegon2@users.noreply.github.com>
Date: Tue, 20 Oct 2026 06:34:34 -0700
Subject: [PATCH 12/12] docs(migrations): count 0052 and name its tenant-unique
prerequisite
The migration directory now holds 54 SQL files (main's 53 plus 0052), so both READMEs say so.
The 0052 entry also names `workflow_runs_id_tenant_uniq`, the unique (id, tenant_id) constraint
that the composite invocation foreign key requires.
Co-Authored-By: Claude Opus 5.5
---
internal/core/migrations/README.en.md | 4 ++--
internal/core/migrations/README.md | 4 ++--
2 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/internal/core/migrations/README.en.md b/internal/core/migrations/README.en.md
index 6d31d32a1..3b069125d 100644
--- a/internal/core/migrations/README.en.md
+++ b/internal/core/migrations/README.en.md
@@ -6,7 +6,7 @@ SQL is embedded in the binary through `embed.FS` in `core/store.go`. `Store.Migr
## File index
-There are currently 53 SQL files.
+There are currently 54 SQL files.
| SQL file | Schema change |
|---|---|
@@ -78,7 +78,7 @@ There are currently 53 SQL files.
- 0046–0048 backfill no old messages, Threads, history positions or pause state; `visible_text` is stored separately from the raw `record`, and old tasks are not paused automatically. Fresh, upgrade and repeated application are covered by `TestMigration0046TaskConversationFreshAndUpgrade` and the integration migration tests.
- 0049 lets `thread_commands` accept a WorkflowRun owner and requires a graph node on workflow commands; each node holds at most one `end_session`, so the scheduler can replay it safely. Every old row carries `run_id` and is not backfilled. Semantics are in [workflows](../../../docs/workflows/workflows.en.md); coverage is in the `TestWorkflowSessionEndsAfterSuccessfulTurn` DB tests.
- 0051 adds the cancellation-request instant to `workflow_runs` and backfills no old row; the request records intent only and is settled by the scheduler on later passes. Semantics are in [workflows](../../../docs/workflows/workflows.en.md#scheduling); coverage is in the `TestWorkflowRunCancelBeforeStartNeverStarts` DB tests.
-- 0052 first settles every unlinked non-terminal historical workflow IssueRun (including running and deferred) as failed with `workflow_execution_unavailable_before_invocation_linkage`, preserving its input and outcome without inventing an execution. It then adds the `(workflow_invocation_ref, tenant_id)` → `workflow_runs(id, tenant_id)` composite foreign key, a unique reference, and a CHECK that non-terminal workflow rows carry a reference while other executors carry none, all unconditionally validated. Historical terminal workflow rows may keep a NULL reference. Fresh, seeded upgrade, repeated application and constraint validation coverage is in `integration/workflow_invocation_migration_test.go`.
+- 0052 first settles every unlinked non-terminal historical workflow IssueRun (including running and deferred) as failed with `workflow_execution_unavailable_before_invocation_linkage`, preserving its input and outcome without inventing an execution. It then adds `workflow_runs_id_tenant_uniq` (unique `(id, tenant_id)` on `workflow_runs`, which the composite foreign key requires), the `(workflow_invocation_ref, tenant_id)` → `workflow_runs(id, tenant_id)` composite foreign key, a unique reference, and a CHECK that non-terminal workflow rows carry a reference while other executors carry none, all unconditionally validated. Historical terminal workflow rows may keep a NULL reference. Fresh, seeded upgrade, repeated application and constraint validation coverage is in `integration/workflow_invocation_migration_test.go`.
- New migrations must verify fresh databases, upgrades from the preceding schema, repeated application and key constraints. See [migration_upgrade_path_test.go](../../../integration/migration_upgrade_path_test.go) and domain upgrade tests in integration.
See [Persistence](../../../docs/core/persistence.en.md) for ownership/transactions, and [Workflows](../../../docs/workflows/workflows.en.md) / [External Git line](../../../docs/repositories/external-git.en.md) for runtime semantics.
diff --git a/internal/core/migrations/README.md b/internal/core/migrations/README.md
index 7256a402c..a2d7993cc 100644
--- a/internal/core/migrations/README.md
+++ b/internal/core/migrations/README.md
@@ -6,7 +6,7 @@ SQL 通过 `core/store.go` 的 `embed.FS` 编入二进制。`Store.Migrate` 在
## 文件索引
-当前共 53 个 SQL 文件。
+当前共 54 个 SQL 文件。
| SQL 文件 | schema 变化 |
|---|---|
@@ -78,7 +78,7 @@ SQL 通过 `core/store.go` 的 `embed.FS` 编入二进制。`Store.Migrate` 在
- 0046–0048 不回填旧消息、旧 Thread、历史位置或暂停状态；`visible_text` 与原始 `record` 分开保存，旧任务不自动暂停。空库、升级与重复应用覆盖见 `TestMigration0046TaskConversationFreshAndUpgrade` 及 integration 的迁移测试。
- 0049 让 `thread_commands` 接受 WorkflowRun 归属，并要求工作流命令带图节点；每个节点至多一条 `end_session`，调度器可安全重放。旧行全部带 `run_id`，不回填。语义见 [工作流](../../../docs/workflows/workflows.md)，覆盖见 `TestWorkflowSessionEndsAfterSuccessfulTurn` 等 DB 测试。
- 0051 给 `workflow_runs` 增加取消请求时刻，不回填旧行；请求只记录意图，由调度器在后续轮次结算。语义见[工作流](../../../docs/workflows/workflows.md#调度)，覆盖见 `TestWorkflowRunCancelBeforeStartNeverStarts` 等 DB 测试。
-- 0052 先把没有引用的历史非终态 workflow IssueRun（含 running、deferred）结算为 failed（`workflow_execution_unavailable_before_invocation_linkage`），保留其输入和结果、不虚构执行；再加 `(workflow_invocation_ref, tenant_id)` → `workflow_runs(id, tenant_id)` 复合外键、唯一引用，以及「非终态 workflow 行必须有引用、其他执行器不得有引用」的 CHECK，无条件 VALIDATE。历史终态 workflow 行可保留 NULL 引用。空库、带历史数据升级、重复应用与约束验证状态见 `integration/workflow_invocation_migration_test.go`。
+- 0052 先把没有引用的历史非终态 workflow IssueRun（含 running、deferred）结算为 failed（`workflow_execution_unavailable_before_invocation_linkage`），保留其输入和结果、不虚构执行；再给 `workflow_runs` 加复合外键所需的 `workflow_runs_id_tenant_uniq`（`(id, tenant_id)` 唯一），并加 `(workflow_invocation_ref, tenant_id)` → `workflow_runs(id, tenant_id)` 复合外键、唯一引用，以及「非终态 workflow 行必须有引用、其他执行器不得有引用」的 CHECK，无条件 VALIDATE。历史终态 workflow 行可保留 NULL 引用。空库、带历史数据升级、重复应用与约束验证状态见 `integration/workflow_invocation_migration_test.go`。
- 新增迁移须验证空库、前一 schema 升级、重复应用与关键约束。覆盖见 [migration_upgrade_path_test.go](../../../integration/migration_upgrade_path_test.go) 和 integration 的领域升级测试。
数据归属与事务规则见 [持久化](../../../docs/core/persistence.md)，运行语义见 [工作流](../../../docs/workflows/workflows.md) 和 [外部 Git 线](../../../docs/repositories/external-git.md)。