feat(workflow,trigger): wire workflow_completion as a real event source

This commit is contained in:
matevip 2026-05-08 15:06:53 +08:00
parent e3af03645b
commit 87727414d5
6 changed files with 161 additions and 8 deletions

View File

@ -0,0 +1,63 @@
package vip.mate.trigger.dispatch;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;
import vip.mate.trigger.ingest.TriggerEventEnvelope;
import vip.mate.trigger.ingest.TriggerEventIngestService;
import vip.mate.workflow.runtime.WorkflowCompletionEvent;
import java.util.HashMap;
import java.util.Map;
/**
* Bridges {@link WorkflowCompletionEvent} from the workflow module into the
* trigger ingest pipeline. Lives in the trigger module so the workflow
* runtime stays free of trigger / ingest dependencies that's how we
* dodge the Runner Dispatcher Ingest Runner cycle Spring would
* otherwise refuse to construct.
*
* <p>Each terminal-state run is translated into a {@code workflow_completion}
* envelope with a deterministic {@code wf-run-{runId}} eventId, so the
* {@code mate_trigger_event} unique constraint dedups any re-publish
* (e.g. a runner crash + retry). Failures inside the ingest pipeline are
* logged and swallowed a bad downstream trigger MUST NOT corrupt the
* just-completed run.
*
* <p>The listener fires in the runner thread by default; if a downstream
* ingest does heavy work, switch to {@code @Async} once a dedicated
* executor is wired.
*/
@Slf4j
@Component
@RequiredArgsConstructor
public class WorkflowCompletionEventBridge {
private final TriggerEventIngestService ingestService;
@EventListener
public void onCompletion(WorkflowCompletionEvent event) {
if (event == null) return;
try {
Map<String, Object> data = new HashMap<>();
data.put("sourceWorkflowId", event.workflowId());
data.put("revisionId", event.revisionId());
data.put("runId", event.runId());
data.put("state", event.state());
if (event.finalOutputRef() != null) data.put("finalOutputRef", event.finalOutputRef());
if (event.errorMessage() != null) data.put("errorMessage", event.errorMessage());
TriggerEventEnvelope envelope = new TriggerEventEnvelope(
event.workspaceId(),
"workflow_completion",
"wf-run-" + event.runId(),
"system",
data);
ingestService.ingest(envelope);
} catch (Exception e) {
log.warn("[WorkflowCompletionBridge] forwarding run {} completion failed: {}",
event.runId(), e.getMessage());
}
}
}

View File

@ -140,8 +140,17 @@ public class TriggerPatternMatcher {
}
String wantState = textOrNull(pattern, "stateFilter");
if (wantState != null && !"any".equalsIgnoreCase(wantState)) {
Object state = data.get("state");
if (!(state instanceof String s) || !wantState.equalsIgnoreCase(s)) return false;
Object stateObj = data.get("state");
if (!(stateObj instanceof String actualState)) return false;
// The runtime emits "succeeded" / "failed"; pattern authors
// commonly type "completed" to mean "non-failed terminal".
// Treat the two as equivalent so authors don't have to care
// which vocabulary the runner happens to use today.
if ("completed".equalsIgnoreCase(wantState)) {
if (!"succeeded".equalsIgnoreCase(actualState)) return false;
} else if (!wantState.equalsIgnoreCase(actualState)) {
return false;
}
}
return true;
}

View File

@ -0,0 +1,24 @@
package vip.mate.workflow.runtime;
/**
* Spring application event fired when a workflow run reaches a terminal
* state ({@code succeeded} / {@code failed}). The trigger module
* subscribes via {@code @EventListener} and pushes the payload through
* {@link vip.mate.trigger.ingest.TriggerEventIngestService} so downstream
* triggers (e.g. {@code workflow_completion} pattern) can chain off the
* outcome.
*
* <p>Going through the event bus instead of injecting the trigger
* service directly into the workflow runner breaks the
* Runner Dispatcher Ingest Runner circular dependency that Spring
* would otherwise refuse to construct.
*/
public record WorkflowCompletionEvent(
long runId,
long workflowId,
long revisionId,
long workspaceId,
String state,
String finalOutputRef,
String errorMessage
) {}

View File

@ -105,6 +105,12 @@ public class WorkflowResumer {
runRow.setErrorMessage("paused step '" + stepRow.getStepName() + "' " + outcome.token());
runRow.setCompletedAt(LocalDateTime.now());
runMapper.updateById(runRow);
// Publish the workflow_completion event downstream same as the
// runner's finishFailed path. Without this, runs that end on a
// rejected / timed-out approval would never fire their
// completion trigger because the resumer skips
// runner.continueFromIndex on the failure branch.
runner.publishCompletionEvent(runRow, STATE_FAILED, null, runRow.getErrorMessage());
return Outcome.failed(runRow.getId(), runRow.getErrorMessage());
}

View File

@ -1,6 +1,8 @@
package vip.mate.workflow.runtime;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;
import vip.mate.workflow.compiler.ir.StepMode;
import vip.mate.workflow.compiler.ir.WorkflowGraph;
@ -26,11 +28,20 @@ import java.util.concurrent.TimeUnit;
* marks the row {@code failed}. The last non-skipped step's output payload
* is recorded as {@code final_output_ref} on success.
*
* <p>StateGraph is intentionally not used here the four base modes
* (sequential / fan_out / collect / conditional) are linear plus one bounded
* parallel section, which a small executor handles more directly. When
* {@code await_approval} lands the runtime will switch to a graph-backed
* scheduler so pause / resume can survive a JVM restart.
* <p><b>v0 runtime decision:</b> StateGraph is intentionally not used here.
* The seven v0 modes (sequential / fan_out / collect / conditional +
* await_approval / dispatch_channel / write_memory) are linear plus one
* bounded parallel section, which this small executor handles more
* directly than wrapping a graph DSL. {@code await_approval} pause / resume
* is implemented via {@link WorkflowResumer} reading the persisted
* {@code mate_workflow_run_pause} row, so a JVM restart still recovers the
* run. v1 will reassess whether to graduate to a graph-backed scheduler
* once {@code loop} / {@code invoke_skill} land until then, "linear
* executor" is the explicit, supported runtime.
*
* <p>StateGraph remains in use elsewhere for agent-internal control flow
* (ReAct / Plan-Execute) that's the runtime owned by
* {@link vip.mate.agent agent module}, not this workflow module.
*/
@Slf4j
@Service
@ -49,6 +60,10 @@ public class WorkflowRunner {
private final WorkflowRunStepMapper stepMapper;
private final StepAdapterRegistry adapters;
private final PayloadStore payloadStore;
/** Optional wired in production, may be null in narrow test contexts.
* Spring's stock publisher is always available in a full context. */
@Autowired(required = false)
private ApplicationEventPublisher events;
public WorkflowRunner(WorkflowRunMapper runMapper,
WorkflowRunStepMapper stepMapper,
@ -241,6 +256,7 @@ public class WorkflowRunner {
runRow.setFinalOutputRef(finalOutputRef);
runRow.setCompletedAt(LocalDateTime.now());
runMapper.updateById(runRow);
publishCompletionEvent(runRow, STATE_SUCCEEDED, finalOutputRef, null);
return new WorkflowRunResult(runRow.getId(), STATE_SUCCEEDED, finalOutputRef, null);
}
@ -249,9 +265,44 @@ public class WorkflowRunner {
runRow.setErrorMessage(errorMessage);
runRow.setCompletedAt(LocalDateTime.now());
runMapper.updateById(runRow);
publishCompletionEvent(runRow, STATE_FAILED, null, errorMessage);
return new WorkflowRunResult(runRow.getId(), STATE_FAILED, null, errorMessage);
}
/**
* Fire a {@code workflow_completion} event into the trigger pipeline so
* downstream workflows (or workflows reacting to upstream success /
* failure) can chain off this run. Synchronous and best-effort: a
* fan-out failure here MUST NOT corrupt the just-completed run state.
*
* <p>The eventId is keyed on {@code wf-run-{runId}} so a retry of the
* same run never duplicate-fires its completion downstream the
* mate_trigger_event UNIQUE(trigger_id, dedup_key) constraint catches
* any redundant publish at insert time.
*
* <p>Package-private so {@link WorkflowResumer} can publish the same
* event for resumed runs that end on a rejected / timed-out approval
* (those don't go through {@link #finishFailed} since the resumer
* writes terminal state directly).
*/
void publishCompletionEvent(WorkflowRunEntity runRow, String state,
String finalOutputRef, String errorMessage) {
if (events == null || runRow == null) return;
try {
events.publishEvent(new WorkflowCompletionEvent(
runRow.getId(),
runRow.getWorkflowId() == null ? 0L : runRow.getWorkflowId(),
runRow.getRevisionId() == null ? 0L : runRow.getRevisionId(),
runRow.getWorkspaceId() == null ? 0L : runRow.getWorkspaceId(),
state,
finalOutputRef,
errorMessage));
} catch (Exception e) {
log.warn("Workflow run {} completion event publish failed: {}",
runRow.getId(), e.getMessage());
}
}
private WorkflowRunResult finishPaused(WorkflowRunEntity runRow, String pauseToken) {
runRow.setState(STATE_PAUSED);
// Pause leaves the run open completedAt stays null until resume settles it.

View File

@ -3,7 +3,7 @@
class="step-node"
:class="`mode-${data.modeType}`"
:data-selected="props.selected"
@click="handleClick"
@click.stop="handleClick"
>
<Handle type="target" :position="targetPosition" />
<div class="step-band" />