diff --git a/mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java b/mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java index 70fc7322..343b5d62 100644 --- a/mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java +++ b/mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java @@ -8,8 +8,10 @@ import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.ApplicationArguments; import org.springframework.boot.ApplicationRunner; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.event.EventListener; import org.springframework.core.annotation.Order; import org.springframework.stereotype.Service; @@ -18,6 +20,7 @@ import org.springframework.transaction.support.TransactionSynchronization; import org.springframework.transaction.support.TransactionSynchronizationManager; import vip.mate.agent.context.ChatOrigin; import vip.mate.agent.context.ChatOriginHolder; +import vip.mate.approval.event.WorkflowApprovalResolvedEvent; import vip.mate.approval.model.ToolApprovalEntity; import vip.mate.approval.repository.ToolApprovalMapper; import vip.mate.tool.guard.model.GuardEvaluation; @@ -52,6 +55,12 @@ public class ApprovalWorkflowService implements ApplicationRunner { private final ToolApprovalMapper approvalMapper; private final ObjectMapper objectMapper; private final ConversationService conversationService; + /** Optional — injected only in full Spring context. The workflow + * module listens for {@link WorkflowApprovalResolvedEvent}; in tests + * that don't wire the workflow runtime this stays null and the + * publish is a no-op. */ + @Autowired(required = false) + private ApplicationEventPublisher events; /** * GC scheduler — owns the 5-minute clock for the entire approval state machine @@ -578,6 +587,42 @@ public class ApprovalWorkflowService implements ApplicationRunner { if (removeFromMap) approvalService.removeFromMap(snapshot.getPendingId()); }); + // Phase 4 — workflow bridge. Workflow-scoped approval rows + // (pendingId starting with "wf-") are linked to a paused workflow + // run via {@code mate_workflow_run_pause.external_approval_id}. + // Publishing the resolve here lets the workflow module's listener + // call WorkflowResumer with the matching outcome, so an operator + // approving in the inbox actually advances the workflow instead + // of leaving it paused forever. We publish AFTER commit so a tx + // rollback can't fire a stale resume; the row id is stable + // because the row already lived in DB. + if (events != null && snapshot.getPendingId() != null + && snapshot.getPendingId().startsWith("wf-")) { + // Look up the row id since the snapshot only carries the string + // pendingId, not the long primary key. One quick equality query. + try { + ToolApprovalEntity row = approvalMapper.selectOne( + new LambdaQueryWrapper() + .eq(ToolApprovalEntity::getPendingId, snapshot.getPendingId())); + if (row != null && row.getId() != null) { + final long rowId = row.getId(); + final String pendingId = snapshot.getPendingId(); + afterCommit(() -> { + try { + events.publishEvent(new WorkflowApprovalResolvedEvent( + rowId, pendingId, snapshotStatus, /* workspaceId */ null)); + } catch (Exception e) { + log.warn("[ApprovalWorkflow] failed to publish workflow-resolved event for {}: {}", + pendingId, e.getMessage()); + } + }); + } + } catch (Exception e) { + log.warn("[ApprovalWorkflow] approval row lookup for resolve event failed for {}: {}", + snapshot.getPendingId(), e.getMessage()); + } + } + boolean consumed = "consumed".equals(snapshotStatus); ResolveOutcome outcome = consumed ? ResolveOutcome.consumed(snapshot, true, rewritten) diff --git a/mateclaw-server/src/main/java/vip/mate/approval/event/WorkflowApprovalResolvedEvent.java b/mateclaw-server/src/main/java/vip/mate/approval/event/WorkflowApprovalResolvedEvent.java new file mode 100644 index 00000000..6c4d03b1 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/approval/event/WorkflowApprovalResolvedEvent.java @@ -0,0 +1,36 @@ +package vip.mate.approval.event; + +/** + * Spring application event fired when a workflow-scoped approval row is + * resolved (approved / denied / timed out / superseded). The workflow + * module subscribes via {@code @EventListener}, looks up the pause row + * by {@code mate_workflow_run_pause.external_approval_id == approvalRowId}, + * and idempotently calls {@code WorkflowResumer.resume} so the pause + * actually advances. + * + *

Without this bridge, an operator who clicks "approve" in the + * approval inbox would only flip the {@code mate_tool_approval} row to + * APPROVED — the workflow run would stay paused forever until someone + * separately POSTed the pause token to the resume endpoint. That's the + * "approval is just a visibility surface, not an actual approval" + * trap RFC §3.4 calls out. + * + *

{@code approvalRowId} is the {@code mate_tool_approval.id} long key, + * NOT the {@code pendingId} string. Pause rows store the long id in + * {@code external_approval_id}, so the listener can find the right + * pause with a single equality query. + * + *

{@code decision} mirrors the resolve vocabulary so the listener + * can route to the right {@code WorkflowResumer.ResumeOutcome}: + *

+ */ +public record WorkflowApprovalResolvedEvent( + long approvalRowId, + String pendingId, + String decision, + Long workspaceId +) {} diff --git a/mateclaw-server/src/main/java/vip/mate/trigger/api/TriggerController.java b/mateclaw-server/src/main/java/vip/mate/trigger/api/TriggerController.java index 5493c714..89f78112 100644 --- a/mateclaw-server/src/main/java/vip/mate/trigger/api/TriggerController.java +++ b/mateclaw-server/src/main/java/vip/mate/trigger/api/TriggerController.java @@ -79,16 +79,27 @@ public class TriggerController { /** * Ingest one event envelope through the dedup / rate-limit / bot-self - * pipeline. The endpoint is open to operators; production deployments - * SHOULD gate it behind the workspace interceptor / a service token - * before exposing it externally. + * pipeline. The endpoint is the operator-facing surface — workspace + * is taken from the trusted {@code X-Workspace-Id} header. Body + * {@code workspaceId} is intentionally ignored so a caller in + * workspace A can't fan-fire triggers in workspace B by hand-rolling + * a JSON body. + * + *

External webhooks should NOT use this endpoint directly — + * production deployments wire their own signed-token webhook + * (e.g. Feishu / DingTalk adapters) which authenticates first and + * publishes a {@link vip.mate.channel.event.ChannelMessageReceivedEvent} + * with a workspace fixed by the channel-token mapping. The + * {@code ChannelMessageEventBridge} then forwards into ingest. */ @Operation(summary = "Ingest one event envelope; returns per-trigger fire / drop summary.") @PostMapping("/events") public R> ingestEvent( - @RequestBody EventIngestRequest body) { + @RequestBody EventIngestRequest body, + @RequestHeader("X-Workspace-Id") long workspaceId) { TriggerEventEnvelope env = new TriggerEventEnvelope( - body.workspaceId(), + // Header wins — body.workspaceId is dropped on purpose. + workspaceId, body.patternType(), body.eventId(), body.senderId(), @@ -96,6 +107,9 @@ public class TriggerController { return R.ok(ingestService.ingest(env)); } + /** {@code workspaceId} is retained on the request shape for backwards + * compatibility but ignored at the controller — the trusted header + * is the source of truth. */ public record EventIngestRequest( long workspaceId, String patternType, diff --git a/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/DefaultWorkflowGraphLoader.java b/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/DefaultWorkflowGraphLoader.java index 9a4bbe6f..bbb18411 100644 --- a/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/DefaultWorkflowGraphLoader.java +++ b/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/DefaultWorkflowGraphLoader.java @@ -33,12 +33,21 @@ public class DefaultWorkflowGraphLoader implements WorkflowGraphLoader { } @Override - public Loaded load(long workflowId) { + public Loaded load(long workflowId, long workspaceId) { WorkflowEntity workflow = workflowMapper.selectById(workflowId); if (workflow == null || Boolean.FALSE.equals(workflow.getEnabled()) || workflow.getLatestRevisionId() == null) { return Loaded.missing(); } + // Workspace ownership check — the trigger must live in the same + // workspace as the workflow. Without this gate, fixture data / + // manual imports / a service-bypass code path could let a + // workspace A trigger fire a workspace B workflow. + if (workflow.getWorkspaceId() == null || workflow.getWorkspaceId() != workspaceId) { + log.warn("Trigger graph load: workflow {} is in workspace {}, caller asked for {}", + workflowId, workflow.getWorkspaceId(), workspaceId); + return Loaded.missing(); + } WorkflowRevisionEntity revision = revisionMapper.selectById(workflow.getLatestRevisionId()); if (revision == null) return Loaded.missing(); try { @@ -49,4 +58,20 @@ public class DefaultWorkflowGraphLoader implements WorkflowGraphLoader { return Loaded.missing(); } } + + /** + * @deprecated production callers MUST use the workspace-scoped overload + * {@link #load(long, long)}. Kept available for legacy test + * stubs that bind a fake workspace context. Returns + * {@code missing()} unconditionally so a production code + * path that accidentally hits this overload doesn't silently + * cross workspaces. + */ + @Override + @Deprecated + public Loaded load(long workflowId) { + log.warn("Workspace-blind WorkflowGraphLoader.load({}) called — refusing. " + + "Use load(workflowId, workspaceId) instead.", workflowId); + return Loaded.missing(); + } } diff --git a/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/TriggerDispatcher.java b/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/TriggerDispatcher.java index d038b1f2..473a2450 100644 --- a/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/TriggerDispatcher.java +++ b/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/TriggerDispatcher.java @@ -56,10 +56,15 @@ public class TriggerDispatcher { return DispatchResult.skipped( "unsupported target_type: " + trigger.getTargetType()); } - WorkflowGraphLoader.Loaded loaded = graphLoader.load(trigger.getTargetId()); + // Workspace-scoped lookup so a workspace A trigger can never fire + // a workspace B workflow even if fixture data / manual imports / + // a service-bypass path somehow planted a cross-workspace + // targetId. The loader returns missing() on mismatch. + long workspaceId = trigger.getWorkspaceId() == null ? 0L : trigger.getWorkspaceId(); + WorkflowGraphLoader.Loaded loaded = graphLoader.load(trigger.getTargetId(), workspaceId); if (loaded.graph() == null) { - log.info("Trigger {} dispatch skipped: no published revision for workflow {}", - trigger.getId(), trigger.getTargetId()); + log.info("Trigger {} dispatch skipped: no published revision for workflow {} in workspace {}", + trigger.getId(), trigger.getTargetId(), workspaceId); return DispatchResult.skipped( "no published revision for workflow " + trigger.getTargetId()); } @@ -98,22 +103,39 @@ public class TriggerDispatcher { } } + /** + * Render the trigger's payload template into the workflow's input map. + * + *

Failure mode is strict. If the template fails to parse, + * fails to render, or produces output that isn't a JSON object, this + * method throws and {@link #dispatch} returns + * {@link DispatchResult#failed(String)} so the trigger row records a + * non-null {@code last_error} and the operator can see why this fire + * didn't run. The previous "fall back to raw event" behaviour is the + * exact silent-failure trap the design forbade — a typo'd template + * would keep firing the workflow with the wrong inputs and lastError + * would stay clean. + * + *

An empty / null {@code payloadTemplate} is the explicit + * opt-in to "use the raw event as inputs" — that path stays + * supported because it's intentional, not accidental. + */ private Map renderInputs(TriggerEntity trigger, Map event) { if (trigger.getPayloadTemplate() == null || trigger.getPayloadTemplate().isBlank()) { return event == null ? Map.of() : event; } + var compiled = pebble.parseTemplate(trigger.getPayloadTemplate()); + String rendered = pebble.evaluateAsString(compiled, + Map.of("event", event == null ? Map.of() : event, + "trigger", Map.of( + "id", trigger.getId(), + "name", trigger.getName() == null ? "" : trigger.getName()))); try { - var compiled = pebble.parseTemplate(trigger.getPayloadTemplate()); - String rendered = pebble.evaluateAsString(compiled, - Map.of("event", event == null ? Map.of() : event, - "trigger", Map.of( - "id", trigger.getId(), - "name", trigger.getName() == null ? "" : trigger.getName()))); return objectMapper.readValue(rendered, MAP_REF); } catch (Exception e) { - log.warn("Trigger {} payload template render failed; falling back to raw event: {}", - trigger.getId(), e.getMessage()); - return event == null ? Map.of() : event; + // Wrap so the dispatcher's catch surfaces the JSON parse failure + // distinctly from a Pebble parse / evaluate failure. + throw new RuntimeException("payloadTemplate produced non-JSON output: " + e.getMessage(), e); } } } diff --git a/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/WorkflowGraphLoader.java b/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/WorkflowGraphLoader.java index 5f73f7e3..7de8e9bb 100644 --- a/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/WorkflowGraphLoader.java +++ b/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/WorkflowGraphLoader.java @@ -19,5 +19,27 @@ public interface WorkflowGraphLoader { public static Loaded missing() { return new Loaded(null, null); } } + /** + * Workspace-scoped lookup. Production callers MUST use this overload + * so a trigger in workspace A can never resolve to a workflow in + * workspace B (e.g. via fixture data, manual DB import, or a + * service-bypass code path). The default binding validates that + * {@code mate_workflow.workspace_id == workspaceId}; tests that don't + * care override this to delegate to the workspace-blind overload. + */ + default Loaded load(long workflowId, long workspaceId) { + // Default: fall through to the single-arg lookup. The production + // {@link DefaultWorkflowGraphLoader} overrides this to enforce + // ownership; test stubs that don't care inherit the lenient + // default. + return load(workflowId); + } + + /** + * @deprecated workspace-blind lookup; only kept for legacy test stubs + * and the deprecated path inside the production binding. + * New callers must use {@link #load(long, long)}. + */ + @Deprecated Loaded load(long workflowId); } diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/ApprovalResumeBridge.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/ApprovalResumeBridge.java new file mode 100644 index 00000000..49ae4aff --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/ApprovalResumeBridge.java @@ -0,0 +1,137 @@ +package vip.mate.workflow.runtime; + +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.context.event.EventListener; +import org.springframework.stereotype.Component; +import vip.mate.approval.event.WorkflowApprovalResolvedEvent; +import vip.mate.workflow.compiler.PublishContext; +import vip.mate.workflow.compiler.WorkflowAclPort; +import vip.mate.workflow.compiler.WorkflowCompiler; +import vip.mate.workflow.model.WorkflowRevisionEntity; +import vip.mate.workflow.model.WorkflowRunEntity; +import vip.mate.workflow.model.WorkflowRunPauseEntity; +import vip.mate.workflow.repository.WorkflowRevisionMapper; +import vip.mate.workflow.repository.WorkflowRunMapper; +import vip.mate.workflow.repository.WorkflowRunPauseMapper; + +/** + * Bridges {@link WorkflowApprovalResolvedEvent} from the approval module + * into {@link WorkflowResumer}. Without this listener an operator who + * clicks "approve" in the approval inbox would only flip the + * {@code mate_tool_approval} row terminal — the workflow run stays + * paused forever until someone separately POSTs the pause token to the + * resume endpoint. + * + *

The listener: + *

    + *
  1. Looks up the pause row by {@code external_approval_id} matching + * the resolved approval row's id. If no pause row references this + * approval (operator path already resumed, or the approval wasn't + * linked to a workflow), we silently no-op.
    2. + *
  2. Re-loads the workflow revision's graph and recompiles it under + * the run's workspace ACL — same code path the resume controller + * uses, so an ACL change after publish doesn't sneak past.
    3. + *
  3. Maps the approval decision to a {@link WorkflowResumer.ResumeOutcome}: + * {@code approved}/{@code consumed} → {@code APPROVED}; + * {@code denied}/{@code superseded} → {@code REJECTED}; + * {@code timeout} → {@code TIMEOUT}.
    4. + *
  4. Calls {@code WorkflowResumer.resume} with the pause token. The + * resumer's idempotency check handles the race where the operator + * resumed the run via the REST endpoint a fraction of a second + * before the approval row resolved — second resume returns + * ALREADY_RESOLVED and the listener swallows it.
    5. + *
+ * + *

Lives in the workflow runtime module so the approval module stays + * free of workflow / runner dependencies, mirroring the workflow ↔ + * trigger event-bridge pattern. + */ +@Slf4j +@Component +@RequiredArgsConstructor +public class ApprovalResumeBridge { + + private final WorkflowRunPauseMapper pauseMapper; + private final WorkflowRunMapper runMapper; + private final WorkflowRevisionMapper revisionMapper; + private final WorkflowCompiler compiler; + private final WorkflowAclPort aclPort; + private final WorkflowResumer resumer; + + @EventListener + public void onApprovalResolved(WorkflowApprovalResolvedEvent event) { + if (event == null || event.approvalRowId() <= 0) return; + WorkflowRunPauseEntity pause = pauseMapper.selectOne( + new LambdaQueryWrapper() + .eq(WorkflowRunPauseEntity::getExternalApprovalId, event.approvalRowId()) + .isNull(WorkflowRunPauseEntity::getResumedAt) + .last("LIMIT 1")); + if (pause == null) { + // Either there's no workflow pause linked to this approval + // (chat-driven approval), or the operator path already resumed + // it. Both are fine. + log.debug("[ApprovalResumeBridge] no open pause for approval row {} (pendingId={})", + event.approvalRowId(), event.pendingId()); + return; + } + + WorkflowResumer.ResumeOutcome outcome = mapDecision(event.decision()); + if (outcome == null) { + log.info("[ApprovalResumeBridge] decision '{}' on approval row {} is not a workflow-resume " + + "trigger; pause {} stays open", + event.decision(), event.approvalRowId(), pause.getId()); + return; + } + + // Re-load the revision graph through the same compiler the resume + // controller uses, so ACL changes after publish don't sneak past. + WorkflowRunEntity run = runMapper.selectById(pause.getRunId()); + if (run == null) { + log.warn("[ApprovalResumeBridge] pause {} references missing run {}", + pause.getId(), pause.getRunId()); + return; + } + WorkflowRevisionEntity revision = revisionMapper.selectById(run.getRevisionId()); + if (revision == null) { + log.warn("[ApprovalResumeBridge] run {} references missing revision {}", + run.getId(), run.getRevisionId()); + return; + } + // PublishContext is (workspaceId, publisherId) — mind the order. + WorkflowCompiler.Result compiled = compiler.compile(revision.getGraphJson(), + new PublishContext(run.getWorkspaceId(), 0L), aclPort); + if (!compiled.ok()) { + log.warn("[ApprovalResumeBridge] revision {} failed to recompile on approval-driven resume", + revision.getId()); + return; + } + + try { + WorkflowResumer.Outcome result = resumer.resume( + compiled.graph(), pause.getPauseToken(), outcome, /* resumePayloadBody */ null); + log.info("[ApprovalResumeBridge] resumed run {} via approval row {}: kind={}", + run.getId(), event.approvalRowId(), result.kind()); + } catch (Exception e) { + // Idempotency is the resumer's job — this catch only triggers + // on actual runtime failures during resume. Don't rethrow: + // the approval row already moved off PENDING and we don't + // want a transient resume failure to look like an + // approval-side bug to upstream observers. + log.warn("[ApprovalResumeBridge] resume failed for run {}: {}", + run.getId(), e.getMessage()); + } + } + + private static WorkflowResumer.ResumeOutcome mapDecision(String decision) { + if (decision == null) return null; + return switch (decision.toLowerCase()) { + case "approved", "consumed" -> WorkflowResumer.ResumeOutcome.APPROVED; + case "denied", "superseded" -> WorkflowResumer.ResumeOutcome.REJECTED; + case "timeout" -> WorkflowResumer.ResumeOutcome.TIMEOUT; + // pending / running / unknown — no terminal outcome to map to. + default -> null; + }; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowRunContext.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowRunContext.java index 3714d323..0f88013b 100644 --- a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowRunContext.java +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowRunContext.java @@ -61,4 +61,41 @@ public class WorkflowRunContext { ctx.put("outputs", new LinkedHashMap<>(outputs)); return ctx; } + + /** + * Build a child context for one fan_out branch. The child shares + * {@code inputs} with the parent (immutable already) and gets a + * deep-copied snapshot of the parent's outputs at branch-entry time + * — writes via the child's {@link #putOutput} do NOT propagate back + * to this context until the runner explicitly merges them after the + * group completes. That snapshot isolation is what stops branch B's + * Pebble template from observing branch A's mid-flight write + * (or vice-versa) when they race on the executor. + * + *

The merge step is owned by the runner — see + * {@code WorkflowRunner.executeFanOutGroup}. The branch's own + * {@code outputVar} write IS still visible inside the branch, which + * is what the schema validator promises authors: a branch can see + * its own value but never its sibling branches'. + */ + public synchronized WorkflowRunContext branchSnapshot() { + WorkflowRunContext child = new WorkflowRunContext(runId, workspaceId, + workflowId, revisionId, inputs); + // Seed the child with a snapshot of the parent's outputs so the + // branch can read everything that completed before the fan_out + // group started, but its own writes stay local. + child.outputs.putAll(this.outputs); + return child; + } + + /** + * Merge a single key/value into the outputs map. Used by the runner + * after a fan_out group completes to apply each branch's + * {@code outputVar} to the master context in deterministic + * step-index order. + */ + public synchronized void mergeOutput(String name, Object value) { + if (name == null || name.isBlank()) return; + outputs.put(name, value); + } } diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowRunner.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowRunner.java index 672c55c3..9e2f8199 100644 --- a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowRunner.java +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowRunner.java @@ -171,17 +171,37 @@ public class WorkflowRunner { private GroupOutcome executeFanOutGroup(List steps, int from, int collectIdx, WorkflowRunContext ctx) { // Steps from..collectIdx-1 are fan_out branches; collectIdx is the join. - record Branch(int stepIndex, WorkflowStep step, Future future) {} + // + // RFC §2.4 requires every branch to render expressions / prompts + // against the SAME context snapshot taken at group entry, with + // collect doing the merge. To honour that we hand each branch its + // own isolated WorkflowRunContext via branchSnapshot() — writes + // inside a branch (via ctx.putOutput from executeStep) land in + // that local copy and stay invisible to siblings until merge + // time. Without this, a branch racing ahead would mutate the + // shared outputs map and the slower branch's Pebble template + // would observe a mid-flight value, making rendering + // schedule-dependent. + record Branch(int stepIndex, WorkflowStep step, + WorkflowRunContext branchCtx, Future future) {} List branches = new ArrayList<>(); for (int i = from; i < collectIdx; i++) { int idx = i; WorkflowStep step = steps.get(i); + WorkflowRunContext branchCtx = ctx.branchSnapshot(); Future future = FAN_OUT_EXECUTOR.submit( - () -> executeStep(step, idx, idx - from, ctx)); - branches.add(new Branch(idx, step, future)); + () -> executeStep(step, idx, idx - from, branchCtx)); + branches.add(new Branch(idx, step, branchCtx, future)); } - String lastOutputRef = null; + // Collect succeeded branch results in step-index order. The + // result list lets us merge outputs into the master context + // deterministically below — a branch's outputVar always wins + // over a smaller-index branch's outputVar with the same name, + // so the conflict policy is "later step wins" and is independent + // of completion order. + record Settled(int stepIndex, WorkflowStep step, StepResult result) {} + List settled = new ArrayList<>(branches.size()); for (Branch branch : branches) { try { StepResult result = branch.future.get(resolveTimeoutSecs(branch.step), TimeUnit.SECONDS); @@ -190,7 +210,7 @@ public class WorkflowRunner { "fan_out branch '" + branch.step.name() + "' failed: " + result.errorMessage(), null); } - if (result.outputPayloadUri() != null) lastOutputRef = result.outputPayloadUri(); + settled.add(new Settled(branch.stepIndex, branch.step, result)); } catch (Exception e) { return new GroupOutcome(true, "fan_out branch '" + branch.step.name() + "' threw: " + e.getMessage(), @@ -198,6 +218,20 @@ public class WorkflowRunner { } } + // Merge phase — the master context only learns about a branch's + // outputVar value here, so collect (and any subsequent step) + // sees a stable, schedule-independent view. + String lastOutputRef = null; + settled.sort((a, b) -> Integer.compare(a.stepIndex, b.stepIndex)); + for (Settled s : settled) { + if (s.result.state() != StepResult.State.SUCCEEDED) continue; + if (s.step.outputVar() != null && !s.step.outputVar().isBlank() + && s.result.outputValue() != null) { + ctx.mergeOutput(s.step.outputVar(), s.result.outputValue()); + } + if (s.result.outputPayloadUri() != null) lastOutputRef = s.result.outputPayloadUri(); + } + // Run the collect adapter so the join is captured as its own row. StepResult collectResult = executeStep(steps.get(collectIdx), collectIdx, null, ctx); if (collectResult.state() == StepResult.State.FAILED) { diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/service/WorkflowService.java b/mateclaw-server/src/main/java/vip/mate/workflow/service/WorkflowService.java index f98be7f0..8fa5592e 100644 --- a/mateclaw-server/src/main/java/vip/mate/workflow/service/WorkflowService.java +++ b/mateclaw-server/src/main/java/vip/mate/workflow/service/WorkflowService.java @@ -155,6 +155,17 @@ public class WorkflowService { revisionMapper.insert(revision); workflow.setLatestRevisionId(revision.getId()); + // RFC v0 contract: publishing clears the inline draft on the + // workflow row. The published revision is now the canonical + // graph; keeping the draft would let the UI show "draft + v3" + // when in fact the draft has just become v3, which confuses + // operators ("did my changes go in?"). Authors who want a + // continuing-edit flow can re-save a fresh draft after publish; + // it'll show up as "draft modified after publish" naturally. + workflow.setDraftJson(null); + workflow.setDraftSchemaVersion(null); + workflow.setDraftUpdatedBy(null); + workflow.setDraftUpdatedAt(null); workflowMapper.updateById(workflow); return new PublishOutcome(workflow, revision); }