mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
fix(workflow,trigger): fan_out isolation + workspace spoof + approval→resume bridge + payload strict + draft clear
This commit is contained in:
matevip
parent
48f87066bd
commit
dcf1f56944
@ -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<ToolApprovalEntity>()
|
||||
.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)
|
||||
|
||||
@ -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.
|
||||
*
|
||||
* <p>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.
|
||||
*
|
||||
* <p>{@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.
|
||||
*
|
||||
* <p>{@code decision} mirrors the resolve vocabulary so the listener
|
||||
* can route to the right {@code WorkflowResumer.ResumeOutcome}:
|
||||
* <ul>
|
||||
* <li>{@code approved} / {@code consumed} → APPROVED</li>
|
||||
* <li>{@code denied} / {@code superseded} → REJECTED</li>
|
||||
* <li>{@code timeout} → TIMEOUT</li>
|
||||
* </ul>
|
||||
*/
|
||||
public record WorkflowApprovalResolvedEvent(
|
||||
long approvalRowId,
|
||||
String pendingId,
|
||||
String decision,
|
||||
Long workspaceId
|
||||
) {}
|
||||
@ -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.
|
||||
*
|
||||
* <p>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<List<TriggerEventIngestService.IngestResult>> 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,
|
||||
|
||||
@ -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();
|
||||
}
|
||||
}
|
||||
|
||||
@ -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.
|
||||
*
|
||||
* <p><b>Failure mode is strict.</b> 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.
|
||||
*
|
||||
* <p>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<String, Object> renderInputs(TriggerEntity trigger, Map<String, Object> 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -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);
|
||||
}
|
||||
|
||||
@ -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.
|
||||
*
|
||||
* <p>The listener:
|
||||
* <ol>
|
||||
* <li>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.</li>
|
||||
* <li>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.</li>
|
||||
* <li>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}.</li>
|
||||
* <li>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.</li>
|
||||
* </ol>
|
||||
*
|
||||
* <p>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<WorkflowRunPauseEntity>()
|
||||
.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;
|
||||
};
|
||||
}
|
||||
}
|
||||
@ -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.
|
||||
*
|
||||
* <p>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);
|
||||
}
|
||||
}
|
||||
|
||||
@ -171,17 +171,37 @@ public class WorkflowRunner {
|
||||
private GroupOutcome executeFanOutGroup(List<WorkflowStep> 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<StepResult> 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<StepResult> future) {}
|
||||
List<Branch> branches = new ArrayList<>();
|
||||
for (int i = from; i < collectIdx; i++) {
|
||||
int idx = i;
|
||||
WorkflowStep step = steps.get(i);
|
||||
WorkflowRunContext branchCtx = ctx.branchSnapshot();
|
||||
Future<StepResult> 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> 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) {
|
||||
|
||||
@ -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);
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user