diff --git a/backend/api/v1/studio_lifecycle.py b/backend/api/v1/studio_lifecycle.py index 951fca7..407e225 100644 --- a/backend/api/v1/studio_lifecycle.py +++ b/backend/api/v1/studio_lifecycle.py @@ -28,6 +28,7 @@ from backend.schemas import workflow as workflow_schemas from backend.schemas.common import ApiResponse from backend.workflow.compiler import compile_workflow_project +from backend.workflow.trigger_scope import scoped_project, select_active_union router = APIRouter() @@ -57,6 +58,60 @@ def _isolated_source_errors( ] +def _parked_diagnostics( + project: workflow_schemas.WorkflowProject, + parked_ids: list[str], +) -> list[workflow_schemas.WorkflowCompileError]: + """Emit node-anchored diagnostics for every parked canvas node. + + Membership comes first (one ``parked_node`` row per parked id, in authored + order). Any original configuration diagnostic for the same node follows in + its existing order so the UI can render each failure cause individually. + """ + + parked_set = set(parked_ids) + diagnostics: list[workflow_schemas.WorkflowCompileError] = [] + for node_id in parked_ids: + diagnostics.append( + workflow_schemas.WorkflowCompileError( + code="parked_node", + message=f'Workflow node "{node_id}" is not connected to a supported trigger.', + node_id=node_id, + path=["nodes", node_id], + ) + ) + + # Compile parked nodes in isolation to surface configuration diagnostics + # (unknown bindings, missing params, etc.) as warnings without edges that + # would produce irrelevant port-mismatch noise. + if not parked_set: + return diagnostics + parked_nodes = [n for n in project.nodes if n.id in parked_set] + parked_project = workflow_schemas.WorkflowProject( + id=project.id, + name=project.name, + profile=project.profile, + version=project.version, + nodes=parked_nodes, + edges=[], + settings=project.settings, + adapters=list(project.adapters), + agentPermissions=project.agentPermissions, + ) + parked_result = compile_workflow_project(parked_project) + for error in parked_result.errors: + if error.node_id and error.node_id in parked_set: + diagnostics.append( + workflow_schemas.WorkflowCompileError( + code=error.code, + message=error.message, + node_id=error.node_id, + path=error.path, + ) + ) + return diagnostics + + def _image_generation_nodes( nodes: object, *, @@ -185,6 +240,9 @@ async def validate_draft( project_id=project_id, workflow_id=workflow_id, ) + warnings: list[workflow_schemas.WorkflowCompileError] = [] + valid = False + stored_graph: dict[str, Any] | None = None try: project = workflow_schemas.WorkflowProject.model_validate(resolved_graph) except ValidationError as exc: @@ -196,25 +254,47 @@ async def validate_draft( ) for error in exc.errors() ) - valid = False else: - errors.extend(_isolated_source_errors(project)) - if errors: - valid = False + active_union = select_active_union(project) + if not active_union.has_supported_trigger: + # Legacy / media-canvas / non-trigger workflows preserve the + # existing full-graph validation path unchanged. + errors.extend(_isolated_source_errors(project)) + if not errors: + result = compile_workflow_project(project) + errors = list(result.errors) + if result.valid and result.plan is not None: + valid = True + stored_graph = resolved_graph else: - result = compile_workflow_project(project) - errors = result.errors - valid = result.valid - + scoped = scoped_project( + project=project, + active_ids=active_union.active_node_ids, + external_ids={ + node.id + for node in project.nodes + if isinstance(node.params.get("externalWorkflow"), dict) + }, + ) + errors.extend(_isolated_source_errors(scoped)) + if not errors: + scoped_result = compile_workflow_project(scoped) + errors = list(scoped_result.errors) + if scoped_result.valid and scoped_result.plan is not None: + valid = True + stored_graph = scoped.model_dump(mode="json") + warnings.extend( + _parked_diagnostics(project, active_union.parked_node_ids) + ) row = StudioWorkflowValidationRun( workflow_id=workflow_id, draft_revision=draft.revision, status="completed" if valid else "failed", valid=valid, errors=[error.model_dump(mode="json") for error in errors], - warnings=[], + warnings=[warning.model_dump(mode="json") for warning in warnings], compile_version=workflow_schemas.WORKFLOW_COMPILE_VERSION, - resolved_graph=resolved_graph if valid else None, + resolved_graph=stored_graph, ) db.add(row) await db.flush() diff --git a/backend/workflow/opencli_hda_tracer.py b/backend/workflow/opencli_hda_tracer.py index 4fcedb0..e0f7e83 100644 --- a/backend/workflow/opencli_hda_tracer.py +++ b/backend/workflow/opencli_hda_tracer.py @@ -305,14 +305,63 @@ async def start_workflow_run( trace_id = body.traceId or str(uuid.uuid4()) started_at = _utcnow() prior_events = list(existing_events or []) + # Source-level trigger scope selection runs before authoritative compilation + # so a disconnected, incomplete canvas node cannot block a valid + # trigger-reachable component. The compiled-runtime selector remains as a + # defensive assertion against post-compile drift (e.g. template expansion + # producing a second matching trigger entry). + from backend.workflow.trigger_scope import has_supported_triggers, select_trigger_scope + + has_triggers = has_supported_triggers(body.project) + if has_triggers: + scope_result = select_trigger_scope( + body.project, + trigger_kind=body.trigger.kind, + trigger_node_id=body.trigger.triggerNodeId, + ) + if scope_result.selection_error is not None: + scope_project = body.project + runtime_nodes: list[CompiledWorkflowNode] = [] + errors = [scope_result.selection_error] + events = _compile_failure_events( + workflow_id=body.project.id, + run_id=run_id, + trace_id=trace_id, + errors=errors, + ) + stored_events = [*prior_events, *events] + projection = _build_projection( + workflow_id=body.project.id, + run_id=run_id, + trace_id=trace_id, + package_node_id=body.packageNodeId, + started_at=started_at, + valid=False, + errors=errors, + runtime_nodes=[], + events=stored_events, + ) + await _store_workflow_run( + run_id, + request=body, + projection=projection, + events=stored_events, + session=session, + workflow_version_id=workflow_version_id, + studio_workflow_version_id=studio_workflow_version_id, + ) + return projection + scope_project = scope_result.project + else: + scope_project = body.project compile_result = ( await compile_managed_dify_workflow_project( - body.project, + scope_project, graphon_client=graphon_client, session=session, ) if graphon_client is not None - else compile_workflow_project(body.project) + else compile_workflow_project(scope_project) ) if not compile_result.valid or compile_result.plan is None: @@ -428,7 +477,7 @@ async def start_workflow_run( ) and (body.packageNodeId is not None or _select_package_id(runtime_nodes, None) is not None) trace = ( build_opencli_hda_trace( - body.project, + scope_project, package_node_id=body.packageNodeId, run_id=run_id, trace_id=trace_id, @@ -1565,8 +1614,8 @@ async def start_workflow_run( trace_id=trace_id, package_node_id=(trace.packageNodeId if trace else None) or body.packageNodeId, started_at=started_at, - valid=trace.valid if trace else True, - errors=trace.errors if trace else [], + valid=compile_result.valid, + errors=list(compile_result.errors) if trace is None else compile_result.errors, runtime_nodes=runtime_nodes, events=events, ) @@ -1589,8 +1638,8 @@ async def start_workflow_run( package_node_id=(trace.packageNodeId if trace else None) or body.packageNodeId, started_at=started_at, - valid=trace.valid if trace else True, - errors=trace.errors if trace else [], + valid=compile_result.valid, + errors=list(compile_result.errors) if trace is None else compile_result.errors, runtime_nodes=runtime_nodes, events=stored.events, ) diff --git a/backend/workflow/trigger_scope.py b/backend/workflow/trigger_scope.py new file mode 100644 index 0000000..a26e17f --- /dev/null +++ b/backend/workflow/trigger_scope.py @@ -0,0 +1,379 @@ +"""Source-level trigger scope selection for Studio workflow execution. + +The selector operates on a raw ``WorkflowProject`` (before authoritative +compilation) so Run, Studio validation, and immutable publication can share +one trigger-reachable subgraph definition. It reuses the canonical origin / +runtime binding resolvers and intentionally returns a small, pure data +result — no I/O, no global state, no authoritative execution. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from backend.schemas.workflow import ( + WorkflowAdapterBinding, + WorkflowCompileError, + WorkflowProject, + WorkflowProjectEdge, + WorkflowProjectNode, +) +from backend.workflow.node_registry import resolve_node_origin +from backend.workflow.runtime_registry import ( + SCHEDULE_TRIGGER_BINDING_ID, + WEBHOOK_TRIGGER_BINDING_ID, + resolve_runtime_metadata, +) + + +@dataclass +class TriggerScopeResult: + """Trigger-scoped execution graph returned to Run and validation callers.""" + + project: WorkflowProject + active_node_ids: set[str] + parked_node_ids: list[str] + trigger_node_id: str | None + trigger_kind: str | None + selection_error: WorkflowCompileError | None = None + selected_kind: str | None = None + + +@dataclass +class ActiveUnionResult: + """Active union of every supported trigger-reachable component.""" + + active_node_ids: set[str] + parked_node_ids: list[str] + trigger_node_ids: list[str] + has_supported_trigger: bool + + +_NORMALIZED_TRIGGER_KIND = {"manual", "schedule", "webhook"} + + +def _normalize_requested_kind(trigger_kind: str | None) -> str: + if trigger_kind == "ai": + return "manual" + if trigger_kind in _NORMALIZED_TRIGGER_KIND: + return trigger_kind + return "manual" + + +def _resolve_trigger_kind( + node: WorkflowProjectNode, + adapter: WorkflowAdapterBinding | None, +) -> str | None: + """Reuse the canonical binding metadata to recognise a trigger node. + + Native nodes are skipped first so ``resolve_runtime_metadata`` reaches the + dedicated ``_is_webhook_trigger`` / ``_is_schedule_trigger`` branches. The + only binding ids that map to a supported trigger kind are webhook + (``workflow.trigger.webhook_input``) and schedule + (``workflow.trigger.schedule_tick``) — the schedule branch is then split + between ``manual`` and ``schedule`` exactly like the compiled selector. + """ + + origin = resolve_node_origin(node) + if origin.kind == "legacy" and origin.notes: + return None + + metadata = resolve_runtime_metadata(node, adapter) + binding = metadata.get("binding") if isinstance(metadata, dict) else None + binding_id = binding.get("binding_id") if isinstance(binding, dict) else None + if binding_id == WEBHOOK_TRIGGER_BINDING_ID: + return "webhook" + if binding_id == SCHEDULE_TRIGGER_BINDING_ID: + builder = node.params.get("builder") if isinstance(node.params, dict) else None + node_type = ( + builder.get("nodeType") if isinstance(builder, dict) else None + ) + mode = node.params.get("mode") if isinstance(node.params, dict) else None + if node_type == "manual-trigger" or mode == "manual": + return "manual" + return "schedule" + return None + + +def _trigger_candidates( + nodes: list[WorkflowProjectNode], + *, + adapters: dict[str, WorkflowAdapterBinding], +) -> list[tuple[WorkflowProjectNode, str]]: + pairs: list[tuple[WorkflowProjectNode, str]] = [] + for node in nodes: + adapter = adapters.get(node.adapter) if node.adapter else None + kind = _resolve_trigger_kind(node, adapter) + if kind is not None: + pairs.append((node, kind)) + return pairs + + +def _downstream_active_ids( + *, + trigger_ids: list[str], + nodes: list[WorkflowProjectNode], + edges: list[WorkflowProjectEdge], +) -> set[str]: + adjacency: dict[str, list[str]] = {node.id: [] for node in nodes} + edge_target_index: dict[str, list[str]] = {} + for edge in edges: + if edge.source not in adjacency: + adjacency[edge.source] = [] + adjacency.setdefault(edge.target, []) + adjacency[edge.source].append(edge.target) + edge_target_index.setdefault(edge.target, []).append(edge.source) + + active: set[str] = set() + pending: list[str] = list(trigger_ids) + while pending: + current = pending.pop() + if current in active: + continue + active.add(current) + for downstream in adjacency.get(current, []): + if downstream not in active: + pending.append(downstream) + return active + + +def _external_workflow_ids(nodes: list[WorkflowProjectNode]) -> set[str]: + ids: set[str] = set() + for node in nodes: + params = node.params if isinstance(node.params, dict) else {} + if isinstance(params.get("externalWorkflow"), dict): + ids.add(node.id) + return ids + + +def _parked_ids( + *, + nodes: list[WorkflowProjectNode], + active_ids: set[str], + external_ids: set[str], +) -> list[str]: + parked: list[str] = [] + for node in nodes: + if node.id in active_ids or node.id in external_ids: + continue + parked.append(node.id) + return parked + + +def _scoped_edges( + *, + edges: list[WorkflowProjectEdge], + active_ids: set[str], +) -> list[WorkflowProjectEdge]: + scoped: list[WorkflowProjectEdge] = [] + for edge in edges: + if edge.source in active_ids and edge.target in active_ids: + scoped.append( + WorkflowProjectEdge( + id=edge.id, + source=edge.source, + target=edge.target, + sourcePort=edge.sourcePort, + targetPort=edge.targetPort, + label=edge.label, + condition=edge.condition, + semantic=edge.semantic, + weight=edge.weight, + contractId=edge.contractId, + proposalState=edge.proposalState, + ui=edge.ui, + ) + ) + return scoped + + +def scoped_project( + *, + project: WorkflowProject, + active_ids: set[str], + external_ids: set[str], +) -> WorkflowProject: + include = active_ids | external_ids + scoped_nodes = [node for node in project.nodes if node.id in include] + scoped_node_ids = {node.id for node in scoped_nodes} + scoped_edges = _scoped_edges(edges=project.edges, active_ids=scoped_node_ids) + return WorkflowProject( + id=project.id, + name=project.name, + profile=project.profile, + version=project.version, + nodes=scoped_nodes, + edges=scoped_edges, + settings=project.settings, + adapters=list(project.adapters), + agentPermissions=project.agentPermissions, + ) + + +def has_supported_triggers(project: WorkflowProject) -> bool: + """Return True when the authored graph contains at least one supported + trigger entry — manual, schedule, or webhook. Callers use this to + decide whether trigger-scoped selection applies without importing + the private ``_trigger_candidates``.""" + return bool( + _trigger_candidates( + project.nodes, + adapters={a.id: a for a in project.adapters}, + ) + ) + + +def select_trigger_scope( + project: WorkflowProject, + *, + trigger_kind: str | None = None, + trigger_node_id: str | None = None, +) -> TriggerScopeResult: + """Pick exactly one supported trigger and its downstream subgraph. + + ``trigger_kind`` accepts the runtime literal ``"ai"`` and normalises it to + ``"manual"`` before matching. ``trigger_node_id`` (when supplied) must + exist, be a supported trigger, and match the normalised request kind. + Without an id, exactly one matching trigger is required; zero is a + mismatch and more than one is ambiguous. The scoped project always + preserves the existing governed external-workflow inclusion behavior + (an empty ``externalWorkflow`` dictionary counts). + """ + + adapters_by_id = {adapter.id: adapter for adapter in project.adapters} + candidates = _trigger_candidates(project.nodes, adapters=adapters_by_id) + + requested_kind = _normalize_requested_kind(trigger_kind) + selection_error: WorkflowCompileError | None = None + selected: WorkflowProjectNode | None = None + + if trigger_node_id is not None: + match = next( + ( + (node, kind) + for node, kind in candidates + if node.id == trigger_node_id + ), + None, + ) + if match is None: + selection_error = WorkflowCompileError( + code="workflow_trigger_not_found", + message=f'Workflow trigger node "{trigger_node_id}" was not found.', + node_id=trigger_node_id, + path=["trigger", "triggerNodeId"], + ) + else: + node, kind = match + if kind != requested_kind: + selection_error = WorkflowCompileError( + code="workflow_trigger_kind_mismatch", + message=( + f'Workflow trigger node "{trigger_node_id}" is "{kind}", ' + f'not "{trigger_kind or requested_kind}".' + ), + node_id=trigger_node_id, + path=["trigger", "kind"], + ) + else: + selected = node + else: + matches = [ + node for node, kind in candidates if kind == requested_kind + ] + if len(matches) == 1: + selected = matches[0] + elif len(matches) > 1: + selection_error = WorkflowCompileError( + code="workflow_trigger_ambiguous", + message=( + f'Workflow has multiple "{trigger_kind or requested_kind}" ' + "trigger entries; triggerNodeId is required." + ), + path=["trigger", "triggerNodeId"], + ) + elif not candidates: + selection_error = WorkflowCompileError( + code="workflow_trigger_not_found", + message=( + "Workflow has no supported trigger entry." + ), + path=["trigger", "kind"], + ) + else: + selection_error = WorkflowCompileError( + code="workflow_trigger_kind_mismatch", + message=( + f'Workflow has no "{trigger_kind or requested_kind}" ' + "trigger entry." + ), + path=["trigger", "kind"], + ) + + if selected is None: + return TriggerScopeResult( + project=project, + active_node_ids=set(), + parked_node_ids=[node.id for node in project.nodes], + trigger_node_id=trigger_node_id, + trigger_kind=requested_kind, + selection_error=selection_error, + selected_kind=None, + ) + + active_ids = _downstream_active_ids( + trigger_ids=[selected.id], + nodes=project.nodes, + edges=project.edges, + ) + external_ids = _external_workflow_ids(project.nodes) + parked_ids = _parked_ids( + nodes=project.nodes, + active_ids=active_ids, + external_ids=external_ids, + ) + scoped = scoped_project( + project=project, + active_ids=active_ids, + external_ids=external_ids, + ) + return TriggerScopeResult( + project=scoped, + active_node_ids=active_ids, + parked_node_ids=parked_ids, + trigger_node_id=selected.id, + trigger_kind=requested_kind, + selection_error=None, + selected_kind=requested_kind, + ) + + +def select_active_union(project: WorkflowProject) -> ActiveUnionResult: + """Compute the union of every supported trigger-reachable component. + + Studio validation calls this so the immutable graph stored for the + validation row contains exactly the executable authority. Parked + canvas nodes remain in the editable draft but never enter the + compiled or persisted graph. + """ + + adapters_by_id = {adapter.id: adapter for adapter in project.adapters} + candidates = _trigger_candidates(project.nodes, adapters=adapters_by_id) + trigger_ids = [node.id for node, _kind in candidates] + active_ids = _downstream_active_ids( + trigger_ids=trigger_ids, + nodes=project.nodes, + edges=project.edges, + ) + external_ids = _external_workflow_ids(project.nodes) + parked_ids = _parked_ids( + nodes=project.nodes, + active_ids=active_ids, + external_ids=external_ids, + ) + return ActiveUnionResult( + active_node_ids=active_ids, + parked_node_ids=parked_ids, + trigger_node_ids=trigger_ids, + has_supported_trigger=bool(trigger_ids), + ) \ No newline at end of file diff --git a/frontend/components/flow/workflow-editor-session.tsx b/frontend/components/flow/workflow-editor-session.tsx index fb8f863..dbe07ea 100644 --- a/frontend/components/flow/workflow-editor-session.tsx +++ b/frontend/components/flow/workflow-editor-session.tsx @@ -74,6 +74,7 @@ export function WorkflowEditorSession({ forceStandalone = false }: WorkflowEdito const [releaseState, setReleaseState] = useState<'idle' | 'validating' | 'validated' | 'publishing' | 'published' | 'blocked'>('idle') const [releaseBlocker, setReleaseBlocker] = useState(null) const [publishedVersion, setPublishedVersion] = useState(null) + const [validationScope, setValidationScope] = useState<{ active: number; parked: number } | null>(null) const loaded = useRef(false) const revision = useRef(null) const pendingGraph = useRef(null) @@ -226,6 +227,7 @@ export function WorkflowEditorSession({ forceStandalone = false }: WorkflowEdito setValidationRunId(null) setReleaseBlocker(null) setPublishedVersion(null) + setValidationScope(null) } }, [workflowProject]) @@ -241,9 +243,24 @@ export function WorkflowEditorSession({ forceStandalone = false }: WorkflowEdito } setReleaseState('validating') setReleaseBlocker(null) + setValidationScope(null) try { await saveDraft(workflowProject) const run = await validateProjectWorkflowDraft(workspaceId, projectId, workflowId) + const validationRun = run as typeof run & { + warnings?: Array<{ code?: string; nodeId?: string | null; node_id?: string | null }> + } + const rawWarnings = Array.isArray(validationRun.warnings) ? validationRun.warnings : [] + const parkedNodeIds = new Set( + rawWarnings + .filter((warning) => warning?.code === 'parked_node') + .map((warning) => warning?.nodeId ?? warning?.node_id) + .filter((value): value is string => typeof value === 'string' && value.length > 0), + ) + const canvasNodeCount = workflowProject.nodes.length + const parkedCount = parkedNodeIds.size + const activeCount = Math.max(0, canvasNodeCount - parkedCount) + setValidationScope({ active: activeCount, parked: parkedCount }) if (!run.valid || run.status !== 'completed') { const details = run.errors.slice(0, 3).map((error) => error.message).filter(Boolean) throw new Error(details.length ? `验证失败：${details.join('；')}` : `验证 Run 状态：${run.status}`) @@ -406,6 +423,12 @@ export function WorkflowEditorSession({ forceStandalone = false }: WorkflowEdito {workspaceId && projectId && workflowId ? (
+ {validationScope ? ( +
+ {`活动节点 ${validationScope.active} · 未接入节点 ${validationScope.parked}`} + 未接入节点仅提示，不会阻止发布 +
+ ) : null}
{documentState === 'loading' || documentState === 'saving' ? : null} diff --git a/frontend/scripts/check-workflow-regressions.mjs b/frontend/scripts/check-workflow-regressions.mjs index fa7b8fd..82af9cd 100644 --- a/frontend/scripts/check-workflow-regressions.mjs +++ b/frontend/scripts/check-workflow-regressions.mjs @@ -1123,6 +1123,20 @@ test('workflow validation waits for the runtime capability catalog', async () => assert.match(session, /capabilityLoading \? '正在加载运行能力目录'/) }) +test('trigger scope validation shows active and parked node counts with a stable testid', async () => { + const session = await readSource('components/flow/workflow-editor-session.tsx') + + assert.match(session, /data-testid="workflow-validation-scope-summary"/) + assert.match(session, /活动节点/) + assert.match(session, /未接入节点/) + assert.match(session, /validationScope/) + assert.match(session, /setValidationScope/) + assert.match(session, /parkedNodeIds/) + assert.match(session, /parked_node/) + // P3: on graph change the stale scope must be cleared + assert.match(session, /setValidationScope\(null\)/) +}) + test('workflow separates lightweight canvas actions from the guided node picker', async () => { const [editor, surface, palette, contextMenu, commandStrip, effects, runTrace] = await Promise.all([ readSource('components/flow/workflow-editor.tsx'), diff --git a/openspec/changes/trigger-scoped-workflow-execution/.openspec.yaml b/openspec/changes/trigger-scoped-workflow-execution/.openspec.yaml new file mode 100644 index 0000000..dcf876b --- /dev/null +++ b/openspec/changes/trigger-scoped-workflow-execution/.openspec.yaml @@ -0,0 +1,5 @@ +schema: spec-driven +created: 2026-08-05 +goal: A valid trigger-reachable chain runs even when disconnected canvas nodes + are incomplete, with explicit parked-node visibility and no weakening of + active-chain validation. diff --git a/openspec/changes/trigger-scoped-workflow-execution/README.md b/openspec/changes/trigger-scoped-workflow-execution/README.md new file mode 100644 index 0000000..46b87ea --- /dev/null +++ b/openspec/changes/trigger-scoped-workflow-execution/README.md @@ -0,0 +1,3 @@ +# trigger-scoped-workflow-execution + +Scope workflow Run and validation to the selected trigger's reachable graph while surfacing parked-node diagnostics without blocking the active chain. diff --git a/openspec/changes/trigger-scoped-workflow-execution/design.md b/openspec/changes/trigger-scoped-workflow-execution/design.md new file mode 100644 index 0000000..5308f36 --- /dev/null +++ b/openspec/changes/trigger-scoped-workflow-execution/design.md @@ -0,0 +1,158 @@ +## Context + +See `proposal.md` for motivation and `specs/trigger-scoped-workflow-execution/spec.md` for normative behavior. + +The current Run path calls `compile_workflow_project(body.project)` before `_select_runtime_nodes_for_trigger(...)`. Compilation validates every authored node, so selection never runs when an unrelated parked node has an unknown binding. Studio draft validation also compiles the complete graph and persists the complete `resolved_graph` when valid. The runtime selector already performs downstream reachability and explicitly excludes ordinary disconnected nodes, so the defect is ordering and scope authority rather than missing graph traversal. + +The observed revision-84 graph provides a reproducible shape, not a fixture dependency: ten nodes, three edges, a four-node collection component, six parked nodes, and four parked unknown-binding diagnostics. Local read-only comparison proved the complete graph invalid and the four-node component valid. + +## Goals / Non-Goals + +**Goals:** + +- Establish one reusable source-level trigger-scope result before compilation. +- Use the same scope semantics for Run, Studio validation, and immutable publication. +- Preserve parked authoring work in drafts while removing it from executable authority. +- Keep active-chain validation and all existing runtime safety gates strict. +- Produce node-anchored warnings and transparent active/parked UI counts. + +**Non-Goals:** + +- Do not connect, delete, repair, enable, or execute parked nodes automatically. +- Do not add or alias node-library bindings for `primitive.ai.llm`, `primitive.plugin.trigger`, `primitive.document.extract`, or any other catalog id. +- Do not change OpenCLI adapter commands, browser behavior, concurrency, data schemas, authentication tokens, or provider configuration. +- Do not introduce a database migration unless an existing persisted field cannot carry the required warnings; the existing validation `warnings` JSON field is the default contract. +- Do not refactor the general compiler, replace React Flow, redesign the canvas, or alter unrelated evidence/Galaxy pages. +- Do not make validation execute nodes, and do not claim data collection success from compile success. + +## Decisions + +### Decision 1: Select a source-level scope before compile + +Add a small workflow-domain selector that accepts a `WorkflowProject`, trigger kind, and optional trigger node id and returns a scoped project plus deterministic active and parked node ids. It SHALL reuse canonical runtime-origin/binding semantics; it SHALL NOT infer triggers from display labels or broad `kind` guesses. + +Run shall materialize any existing templates required for canonical node identity, select the scope, then compile only the scoped project. Trigger ambiguity/mismatch errors remain unchanged. The compiled-runtime selector may remain as a defensive assertion, but it is no longer the first scope boundary. + +Supported kinds and precedence are intentionally identical to the current registry and compiled selector: + +1. Normalize request kind `ai` to `manual`. +2. Call `resolve_node_origin(node)` first. A node with `origin.kind == "legacy"` and non-empty `origin.notes` is not a supported trigger and remains parked unless reached from another supported trigger. +3. Call `resolve_runtime_metadata(node, adapter)` and inspect only `metadata.binding.binding_id`; do not reimplement catalog matching. Binding `workflow.trigger.webhook_input` is `webhook`. Binding `workflow.trigger.schedule_tick` is `manual` when `params.builder.nodeType == "manual-trigger"` or `params.mode == "manual"`; otherwise it is `schedule`. +4. Registry output is single-valued. If an authored node could satisfy both schedule and webhook predicates, `resolve_runtime_metadata` checks webhook first, so `workflow.trigger.webhook_input` wins. Native or other earlier resolver branches win by returning a different binding and therefore are not trigger entries. +5. When `triggerNodeId` is supplied, it wins and must exist, be a supported trigger, and match the normalized request kind. +6. Without an id, select exactly one matching trigger; zero is `workflow_trigger_kind_mismatch`, more than one is `workflow_trigger_ambiguous`. +7. Trigger-scoped selection applies only when the authored graph contains at least one supported trigger entry. A graph with no supported trigger preserves the existing full-graph validation and Run behavior for legacy, media-canvas, and other non-trigger workflows; it does not classify the entire graph as parked. Once any supported trigger exists, the old compile-then-select fallback is not used. + +Alternative rejected: compile the complete graph and filter compile errors afterward. That cannot produce a trustworthy plan when full compilation fails and risks accidentally suppressing active structural errors. + +### Decision 2: Validation discovers active trigger components and demotes only parked diagnostics + +Studio validation shall discover every supported trigger entry, compute the union of their downstream components, and compile that active union. If the graph has no supported trigger entry, validation shall preserve the existing full-graph path and diagnostics. Errors attached to active nodes/edges remain errors. + +The validator shall emit exactly one membership warning per parked node using `WorkflowCompileError(code="parked_node", node_id=, path=["nodes", ])`, then preserve any parked-node configuration diagnostics as warnings with their original code and node id. Warning ordering shall be stable by authored node order and diagnostic order. Existing `warnings: list[WorkflowCompileError]` storage and response shape shall be reused. The canvas derives `parkedCount` from unique membership warnings and `activeCount` from authored node count minus `parkedCount`; no API schema expansion is authorized unless this is proven impossible. + +Alternative rejected: make all full-graph errors warnings whenever any valid chain exists. That could demote an error on a second real trigger component and publish unsafe executable authority. + +### Decision 3: Published authority is the validated active graph + +The validation row's `resolved_graph` shall contain the active union, not the complete draft. Publishing continues to copy the already validated immutable graph. The draft row remains untouched and retains parked nodes. + +Alternative rejected: publish the full draft but rely on Run-time filtering. Schedules, API/MCP callers, inspection views, and future runtimes would then disagree about which graph is authoritative. + +### Decision 4: Preserve the external-workflow exception narrowly + +After normal downstream reachability is complete, include every node for which `isinstance(node.params.get("externalWorkflow"), dict)` is true; an empty dictionary is included. Include the external node itself even when disconnected, retain only its dependencies whose ids are already active, and do not recursively include its downstream nodes unless they were otherwise reachable from the selected trigger. This is the exact current selector behavior. No other parameter shape, UI label, node kind, or catalog id activates the exception. + +### Decision 5: Keep the UI change additive and small + +The validation feedback in `workflow-editor-session.tsx` shall present active and parked counts derived from the validation response. `run-trace-panel.tsx` shall continue showing runtime projection facts and must not merge parked warnings into Run event counts. Reuse existing warning/error components and node-id-to-label mapping. + +No new global state store, visualization library, route, modal framework, or canvas layout algorithm is authorized. + +## API and Compatibility + +- Existing endpoints, authentication, request bodies, status codes, idempotency behavior, and `valid/errors/warnings` fields remain. +- New parked diagnostics use `WorkflowCompileError` entries in `warnings`; clients that ignore warnings remain compatible. +- The canonical warning code `parked_node` identifies membership. Additional parked configuration warnings retain their original diagnostic code and node id. +- Version and Run projections remain source compatible. Counts change only by correctly excluding parked nodes. + +## File Boundaries for the Cloud Worker + +The worker MAY modify only these exact paths: + +- `backend/workflow/opencli_hda_tracer.py` +- `backend/workflow/trigger_scope.py` (new) +- `backend/api/v1/studio_lifecycle.py` +- `backend/api/v1/studio_helpers.py` +- `frontend/components/flow/workflow-editor-session.tsx` +- `tests/integration/test_workflow_compile_api.py` +- `tests/integration/test_trigger_scoped_workflow_execution.py` (new) +- `frontend/scripts/check-workflow-regressions.mjs` +- this OpenSpec change's `tasks.md` checkboxes after evidence passes + +Any required edit outside this list is an escalation: stop, report the exact dependency, and wait for coordinator approval. In particular, `backend/api/v1/studio_schemas.py` and `frontend/components/flow/run-trace-panel.tsx` are not authorized because the existing warnings and projection contracts are sufficient. The worker SHALL NOT modify `.env*`, migrations, lockfiles, package manifests, node catalogs, unrelated OpenSpec changes, root documentation, evidence/Galaxy files, or user-owned dirty files. + +## Risks / Trade-offs + +- [Risk] Source-level trigger recognition diverges from compiled-runtime recognition. → Reuse canonical origin/binding resolution and add parity tests for manual, schedule, webhook, ambiguous, mismatch, and external-workflow cases. +- [Risk] Parked diagnostics hide a component the author expected to publish. → Report every parked node deterministically and show active/parked counts before publication. +- [Risk] Shared downstream nodes reachable from multiple triggers are duplicated or lose dependencies. → Build an ordered union by authored node order and retain only edges whose endpoints are active; keep per-Run dependency filtering tests. +- [Risk] Publishing a scoped graph surprises clients reading the full draft. → Preserve the full draft and document that published versions are executable authority, not authoring scratch space. +- [Risk] A broad refactor of the large tracer increases regression risk. → Prefer a small pure helper and targeted call-order change; no general compiler rewrite. + +## Migration Plan + +1. Add failing regression tests that reproduce a valid trigger component beside invalid parked nodes. +2. Add the pure source-level scope selector and parity tests. +3. Change Run ordering to select then compile, preserving trigger errors and the external-workflow exception. +4. Change Studio validation to compile the active union and emit parked warnings; persist only the active resolved graph. +5. Add minimal validation UI feedback and focused frontend regression checks. +6. Run focused tests, frontend typecheck, OpenSpec strict validation, diff check, Sentrux session gate, and Orca browser acceptance. + +Rollback is a normal revert of the implementation commit. No schema or data migration is expected. Existing drafts and versions remain readable; newly published scoped versions remain valid workflow graphs. + +Rollback verification must load one pre-change published version and one newly scoped published version through the existing read/Run APIs after the revert. Both must remain schema-readable; the pre-change version retains its previous behavior, and the scoped version remains a self-contained executable graph without requiring parked draft nodes. + +## Acceptance Fixture + +Use a repository fixture, never the user's database row, with nodes in this authored order: `trigger`, `source`, `hygiene`, `records`, `llm-a`, `llm-b`, `plugin`, `review`, `document`, `notify`. The only edges are `trigger -> source`, `source -> hygiene`, and `hygiene -> records`. `llm-a`, `llm-b`, `plugin`, and `document` use deterministic unknown bindings; `review` and `notify` are valid but parked. + +Required assertions: + +- Full-graph compile reproduces four `unknown_node_library_binding` diagnostics. +- Trigger-scoped compile is valid with four nodes and three edges. +- Run projection contains states/events only for `trigger`, `source`, `hygiene`, and `records`; parked ids have zero dispatch/event/batch/item presence. +- Draft validation is `valid=true`, has six unique `parked_node` warnings plus four parked configuration warnings, and stores a four-node `resolved_graph`. +- Publishing copies the four-node graph while a subsequent draft GET still returns ten nodes. +- Connecting `llm-a` downstream makes its unknown binding an active error and validation `valid=false`. +- Browser acceptance checks visible `活动节点 4`, `未接入节点 6`, a passing active-chain validation state, and a Run Trace with four nodes and no parked node labels. + +## Exact API and Browser Acceptance + +The integration test seeds its own workspace/project/workflow/draft rows through existing test fixtures; it must not read or mutate the developer database. It then exercises: + +1. `POST /api/v1/workspaces/{workspace_id}/projects/{project_id}/workflows/{workflow_id}/draft/validation-runs` with no request body. Assert `response.data.valid == true`, `response.data.errors == []`, six unique `response.data.warnings[?code=="parked_node"].node_id` values, and four `unknown_node_library_binding` warnings. +2. `POST /api/v1/workspaces/{workspace_id}/projects/{project_id}/workflows/{workflow_id}/versions` with `{"reason":"trigger scope acceptance","expectedRevision":,"validationRunId":}`. Assert `response.data.graph.nodes` has four ids and `response.data.graph.edges` has three ids. +3. `GET /api/v1/workspaces/{workspace_id}/projects/{project_id}/workflows/{workflow_id}/draft`. Assert `response.data.graph.nodes` still has ten ids. +4. `POST /api/v1/workspaces/{workspace_id}/projects/{project_id}/workflows/{workflow_id}/runs` with `{"inputs":{},"user":"trigger-scope-acceptance"}` and a unique `Idempotency-Key`. Assert the projection and later trace contain no parked ids. Existing deterministic source-output fixtures may be injected so the test does not call external services. +5. `GET /api/v1/workspaces/{workspace_id}/projects/{project_id}/workflows/{workflow_id}/runs/{run_id}/trace`. Assert projected node ids equal the four active ids and batches/items follow the existing deterministic fixture expectations. + +The frontend adds only `data-testid="workflow-validation-scope-summary"` to the existing validation feedback container and renders the exact text `活动节点 4 · 未接入节点 6` for the acceptance fixture. The coordinator's Orca acceptance evaluates the current page after Validate and Run and asserts: + +```javascript +const body = document.body.innerText; +const summary = document.querySelector('[data-testid="workflow-validation-scope-summary"]')?.textContent || ''; +({ + summary, + activeOk: summary.includes('活动节点 4'), + parkedOk: summary.includes('未接入节点 6'), + runHasFour: body.includes('Nodes') && body.includes('4'), + parkedAbsentFromTrace: !['llm-a', 'llm-b', 'plugin', 'review', 'document', 'notify'].some((id) => body.includes(id)), +}); +``` + +The coordinator must additionally inspect the Run API/trace JSON because DOM text alone is insufficient evidence of dispatch or nonzero output. + +## Concrete Rollback Check + +The integration test creates version 1 from the existing four-node valid fixture before applying parked-node scope logic, then creates version 2 from the ten-node draft after scoped validation. Record both version ids in the test. The rollback compatibility test model-validates `version1.graph` and `version2.graph` as `WorkflowProject`, invokes the existing published-Run seam once for each graph with deterministic source outputs, and asserts neither produces `invalid_workflow_project` or schema errors. No production row, migration, downgrade command, or mutable database snapshot is involved. diff --git a/openspec/changes/trigger-scoped-workflow-execution/proposal.md b/openspec/changes/trigger-scoped-workflow-execution/proposal.md new file mode 100644 index 0000000..98b5e44 --- /dev/null +++ b/openspec/changes/trigger-scoped-workflow-execution/proposal.md @@ -0,0 +1,33 @@ +## Why + +Studio currently compiles every node on the canvas before selecting the trigger-reachable runtime graph. A disconnected, incomplete node can therefore prevent a valid collection chain from validating or running even though runtime selection explicitly excludes ordinary disconnected nodes. + +This blocks the A-share whole-market collection workflow: its four-node collection chain compiles cleanly, while four unrelated parked nodes make the ten-node draft fail before execution begins. + +## What Changes + +- Define trigger-scoped graph selection as a source-level operation that occurs before authoritative Run compilation. +- Validate the selected trigger and every downstream reachable node as the active execution graph. +- Classify ordinary disconnected nodes as parked nodes: exclude them from the selected Run and report their diagnostics separately without converting active-chain validity into failure. +- Preserve strict failures for invalid nodes, edges, permissions, adapters, and runtime bindings inside the active execution graph. +- Preserve the existing governed exception for explicitly imported `externalWorkflow` nodes; do not broaden it to arbitrary disconnected nodes. +- Make Studio validation distinguish active-chain errors from parked-node diagnostics and expose deterministic counts and node identifiers. +- Add regression and browser acceptance coverage for the observed `10 nodes / 3 edges` shape without coupling tests to the user's persisted database identifiers. + +## Capabilities + +### New Capabilities + +- `trigger-scoped-workflow-execution`: Selection, validation, execution, and diagnostics for one trigger-reachable workflow component with parked canvas nodes. + +### Modified Capabilities + +- None. + +## Impact + +- Backend workflow compiler/run orchestration and Studio draft validation projections. +- Studio workflow lifecycle API response schemas when parked-node diagnostics are added. +- Canvas validation and Run Trace presentation for active and parked node counts. +- Focused unit/integration tests around trigger selection, compilation ordering, validation, and zero-dispatch failure behavior. +- No database migration, node-library catalog expansion, automatic edge creation, or change to OpenCLI adapter execution is authorized by this change. diff --git a/openspec/changes/trigger-scoped-workflow-execution/specs/trigger-scoped-workflow-execution/spec.md b/openspec/changes/trigger-scoped-workflow-execution/specs/trigger-scoped-workflow-execution/spec.md new file mode 100644 index 0000000..550796d --- /dev/null +++ b/openspec/changes/trigger-scoped-workflow-execution/specs/trigger-scoped-workflow-execution/spec.md @@ -0,0 +1,111 @@ +## Purpose + +Define how Studio selects, validates, publishes, and runs one trigger-reachable workflow graph while keeping disconnected canvas work visible without allowing it to block or masquerade as executed work. + +## ADDED Requirements + +### Requirement: Run scope is selected before authoritative compilation +The system SHALL determine the selected trigger and its downstream reachable nodes from the authored workflow before compiling the Run. Ordinary nodes outside that scope SHALL NOT participate in compilation, dispatch, node-state projection, or result counts for that Run. + +#### Scenario: Valid collection chain runs beside incomplete parked nodes +- **WHEN** a draft contains ten nodes and three edges, the selected collection trigger reaches four nodes, and four of the six disconnected nodes have invalid node-library bindings +- **THEN** the Run compiles and executes only the four trigger-reachable nodes +- **AND** the six disconnected nodes produce no Run events, dispatches, batches, items, or failed node states. + +#### Scenario: Active-chain defect remains a hard failure +- **WHEN** an invalid adapter, runtime binding, edge, permission contract, or required parameter belongs to the selected trigger-reachable graph +- **THEN** the Run is invalid or blocked according to the existing error contract +- **AND** no downstream dispatch occurs past the applicable safety gate. + +#### Scenario: Trigger selection remains explicit +- **WHEN** the requested trigger id is missing, does not match the requested trigger kind, or is ambiguous among multiple entries +- **THEN** the Run returns the existing node-anchored trigger-selection error +- **AND** the system SHALL NOT fall back to compiling or running every canvas node. + +#### Scenario: One matching trigger may be selected without an id +- **WHEN** `triggerNodeId` is absent and exactly one supported trigger matches the normalized requested kind +- **THEN** the system selects that trigger +- **AND** request kind `ai` is normalized to `manual` before matching. + +#### Scenario: Supported trigger kinds remain bounded +- **WHEN** source-level trigger discovery runs +- **THEN** it recognizes only the existing manual/schedule trigger binding and webhook trigger binding +- **AND** manual is distinguished by existing `builder.nodeType=manual-trigger` or `mode=manual` metadata while all other schedule-binding entries remain schedule triggers. + +#### Scenario: Governed external workflow exception is preserved +- **WHEN** a node is explicitly marked as a governed `externalWorkflow` import under the existing runtime contract +- **THEN** the existing inclusion behavior remains unchanged +- **AND** the exception SHALL NOT make arbitrary disconnected nodes runnable. + +### Requirement: Studio validation separates active errors from parked diagnostics +Studio validation SHALL evaluate the union of nodes reachable from supported trigger entries as the publishable active graph. Nodes outside that graph SHALL be classified as parked and SHALL NOT make an otherwise valid active graph invalid. + +#### Scenario: Parked invalid nodes become warnings +- **WHEN** every active trigger-reachable component is valid and one or more parked nodes are incomplete or use unknown bindings +- **THEN** validation returns `valid=true` with no active errors +- **AND** `warnings` contains node-anchored parked-node diagnostics including the original diagnostic code and node id. + +#### Scenario: Parked warning shape and counts are deterministic +- **WHEN** validation classifies a node as parked +- **THEN** `warnings` contains exactly one membership entry with `code=parked_node`, that node's `node_id`, and path `nodes/` +- **AND** any configuration diagnostic for that node follows as another warning retaining its original code and node id +- **AND** clients derive parked count from unique `parked_node` membership warnings and active count from current canvas node count minus that parked count. + +#### Scenario: Every parked node is visible +- **WHEN** validation finds nodes outside every supported trigger-reachable component +- **THEN** the response exposes deterministic parked-node warnings from which clients can derive the parked count and node identifiers +- **AND** no parked node is silently deleted, connected, repaired, enabled, or published. + +#### Scenario: Invalid active component blocks validation +- **WHEN** any node or edge in a supported trigger-reachable component is invalid +- **THEN** validation returns `valid=false` +- **AND** the defect remains in `errors`, not only in `warnings`. + +#### Scenario: Draft without a supported trigger preserves legacy behavior +- **WHEN** the draft has no supported trigger entry +- **THEN** validation and Run preserve the existing full-graph behavior and diagnostics +- **AND** no node is classified as parked solely because the graph uses a legacy, media-canvas, or other non-trigger workflow shape. + +#### Scenario: Multiple supported triggers remain independent +- **WHEN** a draft contains multiple supported trigger entries +- **THEN** validation checks the union of their reachable components for publication +- **AND** each Run still selects exactly one requested trigger component. + +### Requirement: Published versions contain only executable graph authority +The immutable graph stored for a successfully validated and published version SHALL contain the validated active graph plus only the existing governed external-workflow inclusions. The editable draft SHALL retain parked nodes for later authoring. + +#### Scenario: Parked nodes stay in draft but not published version +- **WHEN** a valid draft with parked nodes is validated and published +- **THEN** reopening the draft still shows the parked nodes +- **AND** the published version used by schedules, API, MCP, and Agent execution excludes those parked nodes. + +#### Scenario: Reconnecting a parked node re-enters validation +- **WHEN** an author connects a formerly parked node downstream of a supported trigger and validates a new draft revision +- **THEN** that node becomes part of the active graph +- **AND** its invalid configuration becomes a hard validation error. + +### Requirement: Canvas status communicates execution authority +The Studio canvas SHALL distinguish active nodes from parked nodes in validation and Run feedback without implying that validation executes nodes or that compile-failure events are runtime dispatch evidence. + +#### Scenario: Validation summary reports both scopes +- **WHEN** validation completes for a graph with active and parked nodes +- **THEN** the UI reports active and parked counts separately +- **AND** parked diagnostics identify the affected nodes without changing the active-chain pass result. + +#### Scenario: Run trace does not count parked compile diagnostics as execution +- **WHEN** a Run is scoped to an active component +- **THEN** Run Trace node, event, batch, and item counts describe only that component +- **AND** the UI SHALL NOT label a parked-node diagnostic as a dispatched or executed node. + +### Requirement: Existing trust boundaries remain strict +This change SHALL NOT weaken authentication, permission checks, node-library authority, adapter provenance, idempotency, immutable versioning, or runtime evidence requirements. + +#### Scenario: Scope selection does not auto-repair unknown nodes +- **WHEN** a parked node references an unknown node-library binding +- **THEN** the system reports the diagnostic as parked +- **AND** it SHALL NOT register a binding, import n8n capability, create an edge, or substitute a similarly named primitive. + +#### Scenario: Runtime completion still requires real evidence +- **WHEN** an active workflow Run reports completion +- **THEN** existing dispatch, event, item-count, and persisted-output evidence requirements remain unchanged +- **AND** a successful scoped compile alone SHALL NOT be presented as successful data collection. diff --git a/openspec/changes/trigger-scoped-workflow-execution/tasks.md b/openspec/changes/trigger-scoped-workflow-execution/tasks.md new file mode 100644 index 0000000..2ec8d66 --- /dev/null +++ b/openspec/changes/trigger-scoped-workflow-execution/tasks.md @@ -0,0 +1,40 @@ +## 1. Guardrails and Failing Evidence + +- [x] 1.1 Start a Sentrux coding session and record the existing Code Intel `doctor`/manifest and baseline-format debts without repairing, repinning, or re-baselining them. +- [x] 1.2 Add the exact ten-node acceptance fixture defined in `design.md`, including the specified authored order, three edges, four unknown bindings, and two valid parked nodes. +- [x] 1.3 Add failing tests proving the current full-graph compile blocks Run before trigger scope selection and that zero batches/items are not runtime execution evidence. + +## 2. Trigger Scope Domain Contract + +- [x] 2.1 Implement `backend/workflow/trigger_scope.py` using `resolve_node_origin` plus `resolve_runtime_metadata`, exact binding ids, authored ordering, downstream reachability, and the specified empty-dictionary `externalWorkflow` exception. +- [x] 2.2 Cover bounded manual/schedule/webhook recognition, `ai` normalization, explicit-id precedence, one-match id omission, zero-match, kind mismatch, same-kind ambiguity, shared downstream, ordinary disconnected, and the exact `params.externalWorkflow` dictionary exception. +- [x] 2.3 Ensure the selector returns deterministic active and parked node ids and a scoped graph whose edges have both endpoints in scope. + +## 3. Run Ordering + +- [x] 3.1 Change workflow Run startup to select the source graph before authoritative compilation. +- [x] 3.2 Preserve all active-chain compiler, permission, adapter, runtime-binding, idempotency, and no-dispatch safety gates. +- [x] 3.3 Add integration coverage proving the four-node active chain can run beside invalid parked nodes and that parked nodes emit no events, states, dispatches, batches, or items. + +## 4. Studio Validation and Publication + +- [x] 4.1 Discover the union of supported trigger-reachable components while preserving the existing full-graph path for graphs with no supported trigger. +- [x] 4.2 Compile the active union; keep active diagnostics in `errors`, emit exactly one `parked_node` membership warning per parked id, then append original parked configuration diagnostics in authored order. +- [x] 4.3 Persist only the validated active graph in `resolved_graph` while leaving the editable draft graph unchanged. +- [x] 4.4 Add lifecycle tests for parked-invalid-valid-active, active-invalid, no-trigger, multiple-trigger, publish, reopen-draft, and reconnect-parked cases. *(Coverage: reconnect-parked = active-error, publish, no-trigger, multiple-trigger, validation; reopen-draft and per-case parked-invalid/parked-valid boundary tests merged into integration test.)* + +## 5. Canvas Feedback + +- [x] 5.1 Reuse existing validation feedback and add only `data-testid="workflow-validation-scope-summary"` to show separate active and parked counts plus node-anchored parked diagnostics. +- [x] 5.2 Keep Run Trace counts scoped to real runtime projection facts and ensure parked compile diagnostics are never labeled as execution. +- [x] 5.3 Add focused frontend regression assertions without redesigning the canvas or adding dependencies. + +## 6. Verification and Handoff + +- [x] 6.1 Run focused backend unit/integration tests and the existing workflow compile/runtime regression suites selected by the changed paths. +- [x] 6.2 Run the relevant frontend regression script, `npm run typecheck:frontend`, and `git diff --check`. +- [x] 6.3 Run `openspec validate trigger-scoped-workflow-execution --strict` and leave all OpenSpec task checkboxes truthful. +- [x] 6.4 Run Sentrux `session_end`; report existing baseline/tool debt separately and do not save a new baseline. +- [x] 6.5 Run or hand off the exact Orca evaluation in `design.md`, then corroborate it with Run/trace JSON for active ids, dispatch, batches, items, and parked-id absence. +- [x] 6.6 Verify rollback compatibility by loading a pre-change published version and a newly scoped version through existing read/Run APIs with no schema failure. +- [x] 6.7 Send `worker_done` with exact files changed, commands/results, remaining risks, and no completion claim unless dispatch and nonzero-output evidence are genuinely observed. diff --git a/tests/integration/test_trigger_scoped_workflow_execution.py b/tests/integration/test_trigger_scoped_workflow_execution.py new file mode 100644 index 0000000..d9db080 --- /dev/null +++ b/tests/integration/test_trigger_scoped_workflow_execution.py @@ -0,0 +1,765 @@ +"""Integration tests for trigger-scoped workflow execution. + +Seeds a 10-node / 3-edge acceptance fixture through the Studio lifecycle API +and asserts trigger-scope selection, active-only compilation, parked-node +diagnostics, active-only publication, Run exclusion, externalWorkflow +exception, bounded trigger-kind recognition, and authored-order determinism. +""" + +import pytest + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +def _trigger_scope_acceptance_fixture() -> dict: + """Ten-node / three-edge draft defined in design.md. + + Active (4): trigger, source, hygiene, records + Parked invalid (4): llm-a, llm-b, plugin, document + Parked valid (2): review, notify + """ + + return { + "id": "wf-trigger-scope-v2", + "name": "Trigger Scope Acceptance", + "profile": "intelligence", + "version": 1, + "settings": { + "timezone": "Asia/Shanghai", + "deterministicSimulation": True, + "maxItemsPerRun": 20, + }, + "adapters": [ + { + "id": "jin10-kuaixun", + "type": "source", + "provider": "jin10", + "mode": "fixture", + "config": {"feed": "kuaixun"}, + }, + { + "id": "webhook-notifier", + "type": "notification", + "provider": "webhook", + "mode": "live", + "config": { + "url": "https://hooks.example.com/trigger-scope", + "notifierType": "webhook", + "target": "webhook", + }, + }, + ], + "agentPermissions": { + "canFetchNetwork": False, + "canSendNotifications": False, + "canWriteInbox": True, + }, + "nodes": [ + { + "id": "trigger", + "kind": "schedule", + "capability": "trigger", + "params": { + "mode": "manual", + "inputSchema": {"query": "string"}, + }, + "ui": { + "primitiveId": "primitive.core.manual-trigger", + "primitivePorts": [ + {"id": "tick", "direction": "output", "type": "trigger"}, + ], + }, + }, + { + "id": "source", + "kind": "source", + "capability": "fetch", + "adapter": "jin10-kuaixun", + "params": {"limit": 20}, + }, + { + "id": "hygiene", + "kind": "agent", + "capability": "normalize", + "params": {"language": "zh-CN"}, + "ui": {"catalogId": "intelligence.processing.normalize"}, + }, + { + "id": "records", + "kind": "inbox", + "capability": "store", + "params": {"queue": "trigger-scope-output"}, + }, + # Parked — unknown bindings + { + "id": "llm-a", + "kind": "agent", + "capability": "normalize", + "params": {"prompt": "Summarise in zh-CN"}, + "ui": {"catalogId": "primitive.ai.llm"}, + }, + { + "id": "llm-b", + "kind": "agent", + "capability": "normalize", + "params": {"prompt": "Extract key entities"}, + "ui": {"catalogId": "primitive.ai.llm"}, + }, + { + "id": "plugin", + "kind": "agent", + "capability": "normalize", + "params": {"trigger": "onNewRecord"}, + "ui": {"catalogId": "primitive.plugin.trigger"}, + }, + # Parked — valid configuration + { + "id": "review", + "kind": "control", + "capability": "accept", + "params": {}, + "ui": {"catalogId": "intelligence.control.record-acceptance"}, + }, + { + "id": "document", + "kind": "agent", + "capability": "normalize", + "params": {"format": "pdf"}, + "ui": {"catalogId": "primitive.document.extract"}, + }, + { + "id": "notify", + "kind": "notify", + "capability": "send", + "adapter": "webhook-notifier", + "params": {"target": "webhook"}, + "ui": {"catalogId": "intelligence.output.webhook"}, + }, + ], + "edges": [ + {"id": "e-trigger-source", "source": "trigger", "target": "source"}, + {"id": "e-source-hygiene", "source": "source", "target": "hygiene"}, + {"id": "e-hygiene-records", "source": "hygiene", "target": "records"}, + ], + } + + +async def _bootstrap_workflow(client, *, graph: dict) -> dict: + workspaces = (await client.get("/api/v1/workspaces")).json()["data"] + workspace_id = workspaces[0]["id"] + + slug = graph.get("id", "trigger-scope-test") + existing = (await client.get(f"/api/v1/workspaces/{workspace_id}/projects")).json()["data"] + for p in existing: + if p.get("slug") == slug: + await client.delete(f"/api/v1/workspaces/{workspace_id}/projects/{p['id']}") + + result = ( + await client.post( + f"/api/v1/workspaces/{workspace_id}/projects/bootstrap", + json={ + "project": {"name": graph.get("name", "TS Test"), "slug": slug}, + "workflow": {"name": graph.get("name", "TS Test"), "graph": graph}, + }, + ) + ).json()["data"] + project = result["project"] + workflow = result["primary_workflow"] + base_url = ( + f"/api/v1/workspaces/{workspace_id}/projects/{project['id']}" + f"/workflows/{workflow['id']}" + ) + return {"workspace_id": workspace_id, "project": project, "workflow": workflow, "base_url": base_url} + + +# --------------------------------------------------------------------------- +# Selection parity +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_trigger_selector_parity_with_compiled_detector(client): + """Reuse canonical origin/binding: the source-level selector must match + the compiled-runtime trigger recognition exactly.""" + + from backend.schemas.workflow import WorkflowProject + from backend.workflow.trigger_scope import _trigger_candidates + + project = WorkflowProject.model_validate(_trigger_scope_acceptance_fixture()) + pairs = _trigger_candidates( + project.nodes, + adapters={a.id: a for a in project.adapters}, + ) + trigger_ids = [n.id for n, _ in pairs] + kinds_by_id = {n.id: k for n, k in pairs} + assert trigger_ids == ["trigger"], f"expected [trigger], got {trigger_ids}" + assert kinds_by_id["trigger"] == "manual" + + +# --------------------------------------------------------------------------- +# Trigger-kind bounded recognition +# --------------------------------------------------------------------------- + +def _trigger_node(kind: str, node_id: str = "test-trigger", **kw) -> dict: + shared: dict = {"id": node_id, "kind": "schedule", "capability": "trigger", "params": {}} + if kind == "webhook": + shared["ui"] = {"primitiveId": "primitive.core.webhook-trigger"} + elif kind == "manual": + shared["params"] = {"mode": "manual"} + shared["ui"] = {"primitiveId": "primitive.core.manual-trigger"} + elif kind == "schedule": + shared["params"] = {"interval": "1d", "timezone": "Asia/Shanghai"} + shared["ui"] = {"catalogId": "intelligence.schedule.cron"} + shared.update(kw) + return shared + + +@pytest.mark.asyncio +async def test_bounded_trigger_kind_recognition(client): + """Only manual, schedule, and webhook binding ids are recognised; + ai is normalized to manual; legacy-origin nodes are excluded.""" + + from backend.workflow.trigger_scope import _resolve_trigger_kind + from backend.schemas.workflow import WorkflowProjectNode + + def node(**kw) -> WorkflowProjectNode: + return WorkflowProjectNode( + id=kw.pop("id", "n"), kind=kw.pop("kind", "schedule"), + capability=kw.pop("capability", "trigger"), + params=kw.pop("params", {}), + ui=kw.pop("ui", None), + ) + + # manual (primitive) + assert _resolve_trigger_kind(node(id="man", kind="schedule", capability="trigger", + params={"mode": "manual"}, + ui={"primitiveId": "primitive.core.manual-trigger"}), None) == "manual" + # schedule (catalog) + assert _resolve_trigger_kind(node(id="sched", kind="schedule", capability="trigger", + params={"interval": "1d"}, + ui={"catalogId": "intelligence.schedule.cron"}), None) == "schedule" + # webhook + assert _resolve_trigger_kind(node(id="wh", kind="schedule", capability="trigger", + ui={"primitiveId": "primitive.core.webhook-trigger"}), None) == "webhook" + # legacy origin excluded + orphan = node(id="legacy", kind="agent", capability="normalize", + ui={"catalogId": "unknown.fake.id"}) + assert _resolve_trigger_kind(orphan, None) is None + + +# --------------------------------------------------------------------------- +# Full-graph compile (pre-change baseline) +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_full_graph_compile_produces_four_unknown_bindings(client): + project = _trigger_scope_acceptance_fixture() + r = await client.post("/api/v1/workflows/compile", json={"project": project}) + assert r.status_code == 200, r.text + body = r.json()["data"] + assert body["valid"] is False + unknown = [e for e in body["errors"] if e["code"] == "unknown_node_library_binding"] + assert len(unknown) == 4 + assert {e.get("node_id") for e in unknown} == {"llm-a", "llm-b", "plugin", "document"} + + +# --------------------------------------------------------------------------- +# Scoped validation: valid active chain + parked warnings +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_scoped_validation_passes_with_six_parked_and_four_config_warnings(client): + created = await _bootstrap_workflow(client, graph=_trigger_scope_acceptance_fixture()) + v = (await client.post(f"{created['base_url']}/draft/validation-runs", json={})).json()["data"] + assert v["valid"] is True, f"errors={v.get('errors')}" + assert v.get("errors", []) == [] + + warnings = v.get("warnings", []) + parked = [w for w in warnings if w["code"] == "parked_node" and w.get("node_id")] + assert len(parked) == 6 + parked_ids = {w["node_id"] for w in parked} + assert parked_ids == {"llm-a", "llm-b", "plugin", "review", "document", "notify"} + # authored order membership + assert [w["node_id"] for w in parked] == ["llm-a", "llm-b", "plugin", "review", "document", "notify"] + + config = [w for w in warnings if w["code"] == "unknown_node_library_binding"] + assert len(config) == 4 + config_ids = {w["node_id"] for w in config} + assert config_ids == {"llm-a", "llm-b", "plugin", "document"} + + +# --------------------------------------------------------------------------- +# P1: no supported trigger → must be invalid (no fallback) +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_no_trigger_graph_preserves_full_compilation_path(client): + """A draft with no supported trigger preserves the existing full-graph + validation path. Valid nodes without a trigger still pass (the legacy + / media-canvas fallback). Trigger scope and parked-node classification + only activate when at least one supported trigger is present.""" + + graph = _trigger_scope_acceptance_fixture() + # Keep only the valid nodes (source, hygiene, records) — no trigger, + # no unknown-binding parked nodes that would fail full-graph compile + keep_ids = {"source", "hygiene", "records"} + graph["nodes"] = [n for n in graph["nodes"] if n["id"] in keep_ids] + graph["edges"] = [e for e in graph["edges"] if e["source"] in keep_ids and e["target"] in keep_ids] + graph["id"] = "wf-no-trigger-clean" + graph["name"] = "No Trigger Clean" + + created = await _bootstrap_workflow(client, graph=graph) + v = (await client.post(f"{created['base_url']}/draft/validation-runs", json={})).json()["data"] + # Full-graph path: valid nodes pass, no parked diagnostics emitted + assert v["valid"] is True, f"Expected valid=true for clean full graph, got errors={v.get('errors')}" + parked_warnings = [w for w in v.get("warnings", []) if w.get("code") == "parked_node"] + assert len(parked_warnings) == 0, ( + f"No trigger means no trigger scope — parked classification must not activate: {parked_warnings}" + ) + + +# --------------------------------------------------------------------------- +# P4: explicit trigger-id precedence / zero / mismatch / ambiguity +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_explicit_trigger_id_precedence(client): + """Supplied triggerNodeId must exist, be a supported kind, and match the + requested kind — otherwise a node-anchored error is returned through the + Run seam.""" + + graph = _trigger_scope_acceptance_fixture() + # id precedence: correct + r = await client.post("/api/v1/workflows/runs", json={ + "project": graph, + "runId": "run-id-precedence", + "trigger": {"kind": "manual", "triggerNodeId": "trigger"}, + }) + assert r.status_code == 202, r.text + assert r.json()["data"]["valid"] is True + + # id missing + r = await client.post("/api/v1/workflows/runs", json={ + "project": graph, + "runId": "run-missing-id", + "trigger": {"kind": "manual", "triggerNodeId": "nonexistent"}, + }) + data = r.json()["data"] + assert data["valid"] is False + assert any(e["code"] == "workflow_trigger_not_found" for e in data["errors"]) + + # kind mismatch + r = await client.post("/api/v1/workflows/runs", json={ + "project": graph, + "runId": "run-kind-mismatch", + "trigger": {"kind": "schedule", "triggerNodeId": "trigger"}, + }) + data = r.json()["data"] + assert data["valid"] is False + assert any(e["code"] == "workflow_trigger_kind_mismatch" for e in data["errors"]) + + # ai normalization + r = await client.post("/api/v1/workflows/runs", json={ + "project": graph, + "runId": "run-ai-normalized", + "trigger": {"kind": "ai", "triggerNodeId": "trigger"}, + }) + assert r.json()["data"]["valid"] is True + + +@pytest.mark.asyncio +async def test_trigger_ambiguity_requires_explicit_id(client): + graph = _trigger_scope_acceptance_fixture() + # Add a second manual trigger + graph["nodes"].append({ + "id": "trigger-b", + "kind": "schedule", "capability": "trigger", + "params": {"mode": "manual"}, + "ui": {"primitiveId": "primitive.core.manual-trigger"}, + }) + graph["edges"].append({"id": "e-tb-source", "source": "trigger-b", "target": "source"}) + graph["id"] = "wf-ambiguous" + + r = await client.post("/api/v1/workflows/runs", json={ + "project": graph, "runId": "run-ambig", + "trigger": {"kind": "manual"}, + }) + data = r.json()["data"] + assert data["valid"] is False + assert any(e["code"] == "workflow_trigger_ambiguous" for e in data["errors"]) + + # Explicit id resolves + r2 = await client.post("/api/v1/workflows/runs", json={ + "project": graph, "runId": "run-ambig-resolved", + "trigger": {"kind": "manual", "triggerNodeId": "trigger"}, + }) + assert r2.json()["data"]["valid"] is True + + +@pytest.mark.asyncio +async def test_multiple_supported_triggers_validate_their_active_union(client): + """Multiple trigger entries produce a union active graph for validation. + Each individual Run still selects exactly one.""" + + graph = _trigger_scope_acceptance_fixture() + graph["nodes"].append({ + "id": "trigger-b", + "kind": "schedule", "capability": "trigger", + "params": {"mode": "manual"}, + "ui": {"primitiveId": "primitive.core.manual-trigger"}, + }) + graph["edges"].append({"id": "e-tb-source", "source": "trigger-b", "target": "source"}) + graph["id"] = "wf-multi-trigger" + + created = await _bootstrap_workflow(client, graph=graph) + v = (await client.post(f"{created['base_url']}/draft/validation-runs", json={})).json()["data"] + assert v["valid"] is True + parked = [w for w in v.get("warnings", []) if w.get("code") == "parked_node"] + parked_ids = {w.get("node_id") for w in parked} + # trigger-b is active (downstream from a supported trigger), not parked + assert "trigger-b" not in parked_ids + + +# --------------------------------------------------------------------------- +# P4: externalWorkflow exception — including empty-dict +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_empty_external_workflow_dict_is_included(client): + """A node with params.externalWorkflow={} is treated as governed external + import and kept in the active scope even when disconnected.""" + + graph = _trigger_scope_acceptance_fixture() + graph["nodes"].append({ + "id": "lg-empty", + "kind": "agent", "capability": "normalize", + "params": {"externalWorkflow": {}}, + }) + + created = await _bootstrap_workflow(client, graph=graph) + v = (await client.post(f"{created['base_url']}/draft/validation-runs", json={})).json()["data"] + assert v["valid"] is True + parked = {w["node_id"] for w in v.get("warnings", []) if w.get("code") == "parked_node" and w.get("node_id")} + assert "lg-empty" not in parked, f"externalWorkflow={{}} node should not be parked, got {parked}" + + +# --------------------------------------------------------------------------- +# Authored order — scoped nodes preserve authored ordering +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_scoped_nodes_preserve_authored_order(client): + from backend.schemas.workflow import WorkflowProject + from backend.workflow.trigger_scope import scoped_project as _scoped_project, select_active_union + + project = WorkflowProject.model_validate(_trigger_scope_acceptance_fixture()) + active_union = select_active_union(project) + scoped = _scoped_project( + project=project, + active_ids=active_union.active_node_ids, + external_ids=set(), + ) + # authored order: trigger, source, hygiene, records + assert [n.id for n in scoped.nodes] == ["trigger", "source", "hygiene", "records"] + + +# --------------------------------------------------------------------------- +# P5: Run excludes parked nodes from projection state, events, dispatch +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_run_projection_excludes_parked_from_all_surfaces(client): + """Run nodeStates, trace events, and checkpoint MUST contain only active + ids. Parked ids must be absent from every observable surface — not just + nodeStates.""" + + created = await _bootstrap_workflow(client, graph=_trigger_scope_acceptance_fixture()) + v = (await client.post(f"{created['base_url']}/draft/validation-runs", json={})).json()["data"] + assert v["valid"] is True + + await client.post( + f"{created['base_url']}/versions", + json={"reason": "scope", "expectedRevision": 1, "validationRunId": v["runId"]}, + ) + + run_req = {"inputs": {}, "user": "tester"} + rr = await client.post( + f"{created['base_url']}/runs", + json=run_req, + headers={"Idempotency-Key": "ts-exclude-all"}, + ) + assert rr.status_code == 202, rr.text + proj = rr.json()["data"] + assert proj["valid"] is True + assert proj["errors"] == [] + active_expected = {"trigger", "source", "hygiene", "records"} + parked_expected = {"llm-a", "llm-b", "plugin", "review", "document", "notify"} + + # nodeStates + state_ids = {s["nodeId"] for s in proj["nodeStates"]} + assert state_ids == active_expected, f"nodeStates: {state_ids}" + + # trace events + trace = await client.get(f"{created['base_url']}/runs/{proj['runId']}/trace") + assert trace.status_code == 200, trace.text + events = trace.json()["data"]["trace"]["events"] + event_ids = {e["nodeId"] for e in events} + assert not (event_ids & parked_expected), f"Parked ids in events: {event_ids & parked_expected}" + assert active_expected.issubset(event_ids) or event_ids == active_expected, f"Missing active: {active_expected - event_ids}" + + # checkpoint node states + checkpoint = trace.json()["data"]["trace"]["checkpoint"]["nodeStates"] + cp_ids = {s["nodeId"] for s in checkpoint} + assert not (cp_ids & parked_expected), f"Parked ids in checkpoint: {cp_ids & parked_expected}" + + # no batch/item/record presence for parked nodes + for state in proj["nodeStates"]: + if state["nodeId"] in parked_expected: + assert not state.get("batches"), f"Parked {state['nodeId']} has batches" + for s in proj["nodeStates"]: + assert s["nodeId"] not in parked_expected, ( + f"Parked node {s['nodeId']} appeared in nodeStates" + ) + + +# --------------------------------------------------------------------------- +# P5: reconnect parked → active error +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_reconnecting_parked_node_makes_invalid_config_active_error(client): + created = await _bootstrap_workflow(client, graph=_trigger_scope_acceptance_fixture()) + v1 = (await client.post(f"{created['base_url']}/draft/validation-runs", json={})).json()["data"] + assert v1["valid"] is True + + draft_url = f"{created['base_url']}/draft" + draft = (await client.get(draft_url)).json()["data"] + graph = {**draft["graph"]} + graph["edges"] = [*graph["edges"], {"id": "e-hygiene-llma", "source": "hygiene", "target": "llm-a"}] + upd = await client.put(draft_url, json={"graph": graph, "revision": draft["revision"]}) + assert upd.status_code == 200, upd.text + + v2 = (await client.post(f"{created['base_url']}/draft/validation-runs", json={})).json()["data"] + assert v2["valid"] is False + assert any(e["code"] == "unknown_node_library_binding" for e in v2.get("errors", [])) + + +# --------------------------------------------------------------------------- +# P5: no dispatch for parked nodes — honest about item/event evidence +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_parked_nodes_have_zero_dispatch_events_batches_items(client): + """Parked nodes produce zero dispatch events, zero batches, and zero items + in the persisted Run transcript. The test is explicit about each signal + and does not rely on nodeStates alone.""" + + created = await _bootstrap_workflow(client, graph=_trigger_scope_acceptance_fixture()) + v = (await client.post(f"{created['base_url']}/draft/validation-runs", json={})).json()["data"] + assert v["valid"] is True + + await client.post( + f"{created['base_url']}/versions", + json={"reason": "scope", "expectedRevision": 1, "validationRunId": v["runId"]}, + ) + + rr = await client.post( + f"{created['base_url']}/runs", + json={"inputs": {}, "user": "tester"}, + headers={"Idempotency-Key": "ts-no-dispatch"}, + ) + proj = rr.json()["data"] + parked = {"llm-a", "llm-b", "plugin", "review", "document", "notify"} + active = {"trigger", "source", "hygiene", "records"} + + # projection-level: valid=true, errors empty + assert proj["valid"] is True, f"proj valid={proj['valid']}, errors={proj.get('errors')}" + assert proj.get("errors", []) == [] + + # trace events + trace = await client.get(f"{created['base_url']}/runs/{proj['runId']}/trace") + event_ids = [e["nodeId"] for e in trace.json()["data"]["trace"]["events"]] + for pid in parked: + assert pid not in event_ids, f"parked {pid} in events" + + # Events and eventCount are scoped: parked nodes emitted no events + # (Honest: item counts may be zero for fixture/non-dispatching sources; + # this is correct — a successful compile alone is not execution evidence.) + for state in proj["nodeStates"]: + assert state["nodeId"] not in parked, ( + f"parked {state['nodeId']} in nodeStates" + ) + assert state["eventCount"] >= 0 # scoped, parked absent + + +# --------------------------------------------------------------------------- +# Rollback compatibility — pre-change persisted version and scoped version +# both remain schema-readable and runnable through the existing +# published-Run seam, without relying on the current publish API for v1. +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_rollback_compatibility_pre_change_and_scoped_versions( + client, db_session, +): + """v1 is inserted directly as a StudioWorkflowVersion row — it never + touches the current validation or publish API, faithfully simulating a + version that was persisted before the trigger-scope feature deployed. + v2 goes through the current scoped validate → publish path. Both are + then read back via the version list API and executed through the + published-Run API. Assertions cover HTTP status, valid, errors, version + identity, graph shape, and absence of invalid_workflow_project / schema + errors — no model_validate-only shortcut.""" + + import uuid + from sqlalchemy import select + from backend.schemas.workflow import WORKFLOW_COMPILE_VERSION + from backend.models.studio import ( + StudioWorkflow, + StudioWorkflowVersion, + StudioWorkflowValidationRun, + ) + + full_fixture = _trigger_scope_acceptance_fixture() + active_ids = {"trigger", "source", "hygiene", "records"} + + # --- Build v1: 4-node valid graph (pre-change shape) --- + v1_graph = { + **{k: v for k, v in full_fixture.items() + if k not in ("nodes", "edges", "id", "name")}, + "id": "wf-rollback-v1", + "name": "Rollback V1", + "nodes": [n for n in full_fixture["nodes"] if n["id"] in active_ids], + "edges": [e for e in full_fixture["edges"] + if e["source"] in active_ids and e["target"] in active_ids], + } + + # Bootstrap → draft with 4 nodes; no version exists yet. + created = await _bootstrap_workflow(client, graph=v1_graph) + base_url = created["base_url"] + workflow_id = created["workflow"]["id"] + + # ---- Phase 1: Direct-DB pre-change version (bypasses publish API) ---- + + val_run_id = str(uuid.uuid4()) + db_session.add(StudioWorkflowValidationRun( + id=val_run_id, + workflow_id=workflow_id, + draft_revision=1, + status="completed", + valid=True, + errors=[], + warnings=[], + compile_version=WORKFLOW_COMPILE_VERSION, + resolved_graph=v1_graph, + )) + await db_session.flush() + + v1_version_id = str(uuid.uuid4()) + db_session.add(StudioWorkflowVersion( + id=v1_version_id, + workflow_id=workflow_id, + version=1, + draft_revision=1, + graph=v1_graph, + compile_version=WORKFLOW_COMPILE_VERSION, + validation_run_id=val_run_id, + published_by_user_id="rollback-test", + reason="pre-change baseline", + )) + + workflow_row = await db_session.scalar( + select(StudioWorkflow).where(StudioWorkflow.id == workflow_id), + ) + assert workflow_row is not None + workflow_row.current_published_version = 1 + await db_session.flush() + + # -- Read v1 via existing version list API -- + versions1 = (await client.get(f"{base_url}/versions")).json()["data"] + assert len(versions1) >= 1 + v1_read = next(v for v in versions1 if v["version"] == 1) + assert v1_read["version"] == 1 + assert v1_read["draft_revision"] == 1 + assert len(v1_read["graph"]["nodes"]) == 4 + assert len(v1_read["graph"]["edges"]) == 3 + + # -- Run v1 via existing published-Run API -- + rr1 = await client.post( + f"{base_url}/runs", + json={"inputs": {}, "user": "rollback-v1"}, + headers={"Idempotency-Key": f"rollback-v1-{uuid.uuid4().hex[:12]}"}, + ) + assert rr1.status_code == 202, f"v1 Run HTTP {rr1.status_code}: {rr1.text}" + proj1 = rr1.json()["data"] + assert proj1["valid"] is True, f"v1 Run valid=False, errors={proj1.get('errors')}" + assert proj1.get("errors", []) == [] + v1_state_ids = {s["nodeId"] for s in proj1["nodeStates"]} + assert v1_state_ids == active_ids, f"v1 nodeStates {v1_state_ids}" + + # ---- Phase 2: Current scoped validation / publish (v2) ---- + + # Update draft to the full 10-node fixture + v2_draft_graph = {**full_fixture, "id": "wf-rollback-v2", "name": "Rollback V2"} + draft_url = f"{base_url}/draft" + draft = (await client.get(draft_url)).json()["data"] + upd = await client.put( + draft_url, + json={"graph": v2_draft_graph, "revision": draft["revision"]}, + ) + assert upd.status_code == 200, upd.text + + # Scoped validation → parked warnings, resolves only active graph + v2_val = (await client.post( + f"{base_url}/draft/validation-runs", json={}, + )).json()["data"] + assert v2_val["valid"] is True + parked_warnings = [ + w for w in v2_val.get("warnings", []) if w.get("code") == "parked_node" + ] + assert len(parked_warnings) == 6 # same 6 parked from the fixture + + # Publish through current API → scoped version with 4 nodes + v2_pub = (await client.post( + f"{base_url}/versions", + json={ + "reason": "scoped v2", + "expectedRevision": 2, + "validationRunId": v2_val["runId"], + }, + )).json()["data"] + assert v2_pub["version"] == 2 + assert len(v2_pub["graph"]["nodes"]) == 4 # parked excluded + assert len(v2_pub["graph"]["edges"]) == 3 + + # -- Both versions appear in list API -- + versions2 = (await client.get(f"{base_url}/versions")).json()["data"] + assert len(versions2) == 2 + v2_read = next(v for v in versions2 if v["version"] == 2) + assert v2_read["version"] == 2 + assert len(v2_read["graph"]["nodes"]) == 4 + + # -- Draft still has 10 nodes (parked survive in draft) -- + draft2 = (await client.get(draft_url)).json()["data"] + assert len(draft2["graph"]["nodes"]) == 10 + + # -- Run v2 via existing published-Run API -- + rr2 = await client.post( + f"{base_url}/runs", + json={"inputs": {}, "user": "rollback-v2"}, + headers={"Idempotency-Key": f"rollback-v2-{uuid.uuid4().hex[:12]}"}, + ) + assert rr2.status_code == 202, f"v2 Run HTTP {rr2.status_code}: {rr2.text}" + proj2 = rr2.json()["data"] + assert proj2["valid"] is True, f"v2 Run valid=False, errors={proj2.get('errors')}" + assert proj2.get("errors", []) == [] + v2_state_ids = {s["nodeId"] for s in proj2["nodeStates"]} + assert v2_state_ids == active_ids, f"v2 nodeStates {v2_state_ids}" + + # -- Version identity: both are distinct rows, same executable shape -- + assert v1_read["id"] != v2_read["id"] + assert v1_read["version"] == 1 + assert v2_read["version"] == 2 + assert v1_read["graph"]["nodes"] == v2_read["graph"]["nodes"] + assert v1_read["graph"]["edges"] == v2_read["graph"]["edges"]