diff --git a/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/WorkflowCompletionEventBridge.java b/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/WorkflowCompletionEventBridge.java new file mode 100644 index 00000000..7aa0e69d --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/trigger/dispatch/WorkflowCompletionEventBridge.java @@ -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. + * + *

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. + * + *

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 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()); + } + } + +} diff --git a/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerPatternMatcher.java b/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerPatternMatcher.java index cbdbc84c..ea1caea4 100644 --- a/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerPatternMatcher.java +++ b/mateclaw-server/src/main/java/vip/mate/trigger/ingest/TriggerPatternMatcher.java @@ -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; } diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowCompletionEvent.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowCompletionEvent.java new file mode 100644 index 00000000..c8ced26a --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowCompletionEvent.java @@ -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. + * + *

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 +) {} diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowResumer.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowResumer.java index 836a7bbb..116d62a4 100644 --- a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowResumer.java +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowResumer.java @@ -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()); } 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 1a7b8c54..672c55c3 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 @@ -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. * - *

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. + *

v0 runtime decision: 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. + * + *

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. + * + *

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. + * + *

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. diff --git a/mateclaw-ui/src/components/workflow/StepNode.vue b/mateclaw-ui/src/components/workflow/StepNode.vue index 74737a2d..cf168973 100644 --- a/mateclaw-ui/src/components/workflow/StepNode.vue +++ b/mateclaw-ui/src/components/workflow/StepNode.vue @@ -3,7 +3,7 @@ class="step-node" :class="`mode-${data.modeType}`" :data-selected="props.selected" - @click="handleClick" + @click.stop="handleClick" >