diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/AgentInvoker.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/AgentInvoker.java new file mode 100644 index 00000000..68703bcc --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/AgentInvoker.java @@ -0,0 +1,24 @@ +package vip.mate.workflow.runtime; + +/** + * SPI for "render prompt → run agent → return text response". Kept thin so + * unit tests can stub agent execution without booting the full StateGraph + * runtime. Production binding lives in {@link DefaultAgentInvoker} and + * delegates to {@code AgentService.chat(...)}. + */ +public interface AgentInvoker { + + /** + * Invoke the resolved agent with {@code prompt} and return the agent's + * final response text. {@code conversationId} is the ephemeral conversation + * id created per workflow step — the runner generates this so each step + * has its own conversational scope. + */ + String invoke(long agentId, String prompt, String conversationId); + + /** + * Resolve a workspace-scoped agent name to its id. Returns {@code null} + * when the agent does not exist or is disabled. + */ + Long resolveAgentId(long workspaceId, String agentName); +} diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/AgentStepExecutor.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/AgentStepExecutor.java new file mode 100644 index 00000000..1fe5d373 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/AgentStepExecutor.java @@ -0,0 +1,126 @@ +package vip.mate.workflow.runtime; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.springframework.stereotype.Component; +import vip.mate.workflow.compiler.PebbleSubsetEvaluator; +import vip.mate.workflow.compiler.ir.WorkflowStep; + +import java.util.UUID; + +/** + * Shared "render prompt → invoke agent → parse output" pipeline reused by + * the sequential / fan_out / conditional adapters. Centralising this here + * keeps each adapter file focused on its mode-specific dispatch logic + * (skip-on-condition, merge semantics) instead of repeating prompt rendering + * and content-type parsing. + */ +@Component +public class AgentStepExecutor { + + private static final String TEXT = "text"; + private static final String JSON = "json"; + + private final AgentInvoker agentInvoker; + private final PebbleSubsetEvaluator pebble; + private final PayloadStore payloadStore; + private final ObjectMapper objectMapper; + + public AgentStepExecutor(AgentInvoker agentInvoker, + PebbleSubsetEvaluator pebble, + PayloadStore payloadStore, + ObjectMapper objectMapper) { + this.agentInvoker = agentInvoker; + this.pebble = pebble; + this.payloadStore = payloadStore; + this.objectMapper = objectMapper; + } + + /** + * Resolve the agent, render the prompt with the current run context, + * invoke the agent, parse the response according to {@code outputContentType}, + * and write the payload through the store. Returns a succeeded result on + * the happy path and a failed result when any step in the chain throws. + */ + public StepResult run(WorkflowStep step, WorkflowRunContext context) { + Long agentId = resolveAgentId(step, context.workspaceId()); + if (agentId == null) { + return StepResult.failed("agent not resolvable for step '" + step.name() + + "': agentName=" + step.agentName() + " agentId=" + step.agentId()); + } + + String prompt; + try { + prompt = renderPrompt(step, context); + } catch (Exception e) { + return StepResult.failed("prompt render failed for step '" + step.name() + + "': " + e.getMessage()); + } + + String response; + String conversationId = "wf-run-" + context.runId() + "-step-" + step.name() + + "-" + UUID.randomUUID(); + try { + response = agentInvoker.invoke(agentId, prompt, conversationId); + if (response == null) response = ""; + } catch (Exception e) { + return StepResult.failed("agent invocation failed for step '" + step.name() + + "': " + e.getMessage()); + } + + String contentType = step.effectiveOutputContentType(); + try { + Object parsedValue = parseResponse(response, contentType); + String payloadUri = (TEXT.equals(contentType)) + ? payloadStore.storeString(context.workspaceId(), response, "text/plain") + : payloadStore.storeString(context.workspaceId(), response, "application/json"); + String summary = summarise(response); + return StepResult.succeeded(payloadUri, contentType, parsedValue, summary); + } catch (Exception e) { + return StepResult.failed("output parse failed for step '" + step.name() + + "' (contentType=" + contentType + "): " + e.getMessage()); + } + } + + private Long resolveAgentId(WorkflowStep step, long workspaceId) { + if (step.agentId() != null) return step.agentId(); + if (step.agentName() != null && !step.agentName().isBlank()) { + return agentInvoker.resolveAgentId(workspaceId, step.agentName()); + } + return null; + } + + private String renderPrompt(WorkflowStep step, WorkflowRunContext context) { + if (step.promptTemplate() == null || step.promptTemplate().isBlank()) { + return ""; + } + var compiled = pebble.parseTemplate(step.promptTemplate()); + return pebble.evaluateAsString(compiled, context.templateContext()); + } + + private Object parseResponse(String response, String contentType) throws Exception { + if (JSON.equals(contentType)) { + // Permissive: agents often wrap JSON in ```json fences. + String cleaned = stripCodeFence(response); + return objectMapper.readValue(cleaned, Object.class); + } + return response; + } + + private static String stripCodeFence(String s) { + String trimmed = s.trim(); + if (trimmed.startsWith("```")) { + int firstNewline = trimmed.indexOf('\n'); + int lastFence = trimmed.lastIndexOf("```"); + if (firstNewline > 0 && lastFence > firstNewline) { + return trimmed.substring(firstNewline + 1, lastFence).trim(); + } + } + return trimmed; + } + + private static String summarise(String response) { + if (response == null || response.isBlank()) return ""; + String oneLine = response.replaceAll("\\s+", " ").trim(); + return oneLine.length() <= 256 ? oneLine : oneLine.substring(0, 253) + "..."; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/DefaultAgentInvoker.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/DefaultAgentInvoker.java new file mode 100644 index 00000000..e89e166b --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/DefaultAgentInvoker.java @@ -0,0 +1,49 @@ +package vip.mate.workflow.runtime; + +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import org.springframework.stereotype.Component; +import vip.mate.agent.AgentService; +import vip.mate.agent.model.AgentEntity; +import vip.mate.agent.repository.AgentMapper; + +/** + * Production binding for {@link AgentInvoker}. Looks agents up by name within + * the workspace via {@link AgentMapper} and delegates execution to + * {@link AgentService#chat(Long, String, String)}. The conversation id is + * passed through as-is — the runner is responsible for generating an ephemeral + * id per step so multi-step runs do not collide on conversation history. + */ +@Component +public class DefaultAgentInvoker implements AgentInvoker { + + private final AgentService agentService; + private final AgentMapper agentMapper; + + public DefaultAgentInvoker(AgentService agentService, AgentMapper agentMapper) { + this.agentService = agentService; + this.agentMapper = agentMapper; + } + + @Override + public String invoke(long agentId, String prompt, String conversationId) { + return agentService.chat(agentId, prompt, conversationId); + } + + @Override + public Long resolveAgentId(long workspaceId, String agentName) { + if (agentName == null || agentName.isBlank()) return null; + AgentEntity entity = agentMapper.selectOne(new LambdaQueryWrapper() + .eq(AgentEntity::getWorkspaceId, workspaceId) + .eq(AgentEntity::getName, agentName.trim()) + .eq(AgentEntity::getEnabled, true)); + if (entity == null) { + // Fall back to a workspace-agnostic lookup so global agents still + // resolve. This mirrors how the workflow ACL phase counts an agent + // as "resolvable" if it exists anywhere the user can see it. + entity = agentMapper.selectOne(new LambdaQueryWrapper() + .eq(AgentEntity::getName, agentName.trim()) + .eq(AgentEntity::getEnabled, true)); + } + return entity == null ? null : entity.getId(); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/PayloadStore.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/PayloadStore.java new file mode 100644 index 00000000..a1786a5c --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/PayloadStore.java @@ -0,0 +1,120 @@ +package vip.mate.workflow.runtime; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.springframework.stereotype.Service; +import vip.mate.workflow.model.WorkflowPayloadEntity; +import vip.mate.workflow.repository.WorkflowPayloadMapper; + +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.time.LocalDateTime; +import java.util.HexFormat; +import java.util.Objects; +import java.util.UUID; + +/** + * Write-through facade over {@code mate_workflow_payload}. v0 stores every + * payload inline (storage_kind = "inline"); the table schema reserves room for + * a fs / s3 / oss spill-over flavour but no caller wires that yet. Callers + * receive a stable URI of the form {@code mwf://{workspaceId}/{uuid}} and + * resolve it back via {@link #readString(String)} or {@link #readBytes(String)}. + */ +@Service +public class PayloadStore { + + private static final String SCHEME = "mwf://"; + private static final String STORAGE_KIND_INLINE = "inline"; + + private final WorkflowPayloadMapper payloadMapper; + private final ObjectMapper objectMapper; + + public PayloadStore(WorkflowPayloadMapper payloadMapper, ObjectMapper objectMapper) { + this.payloadMapper = payloadMapper; + this.objectMapper = objectMapper; + } + + /** Store a UTF-8 string payload and return its stable URI. */ + public String storeString(long workspaceId, String body, String contentType) { + byte[] bytes = (body == null ? "" : body).getBytes(StandardCharsets.UTF_8); + return storeBytes(workspaceId, bytes, contentType == null ? "text/plain" : contentType); + } + + /** JSON-encode {@code value} and store it. {@code contentType} is fixed to {@code application/json}. */ + public String storeJson(long workspaceId, Object value) { + try { + byte[] bytes = objectMapper.writeValueAsBytes(value); + return storeBytes(workspaceId, bytes, "application/json"); + } catch (JsonProcessingException e) { + throw new PayloadStoreException("failed to serialize payload as JSON: " + e.getMessage(), e); + } + } + + /** Store raw bytes and return the URI. */ + public String storeBytes(long workspaceId, byte[] bytes, String contentType) { + Objects.requireNonNull(bytes, "bytes"); + String uri = SCHEME + workspaceId + "/" + UUID.randomUUID(); + + WorkflowPayloadEntity row = new WorkflowPayloadEntity(); + row.setPayloadUri(uri); + row.setWorkspaceId(workspaceId); + row.setContentBytes(bytes); + row.setStorageKind(STORAGE_KIND_INLINE); + row.setContentType(contentType); + row.setSha256(sha256Hex(bytes)); + row.setSizeBytes((long) bytes.length); + row.setCreatedAt(LocalDateTime.now()); + payloadMapper.insert(row); + return uri; + } + + /** Resolve a payload URI to its raw bytes; throws when the URI is unknown. */ + public byte[] readBytes(String payloadUri) { + WorkflowPayloadEntity row = lookup(payloadUri); + return row.getContentBytes() == null ? new byte[0] : row.getContentBytes(); + } + + /** Resolve a payload URI to its UTF-8 decoded string body. */ + public String readString(String payloadUri) { + return new String(readBytes(payloadUri), StandardCharsets.UTF_8); + } + + /** Resolve a payload URI to its JSON body parsed back into the requested shape. */ + public T readJson(String payloadUri, Class type) { + try { + return objectMapper.readValue(readBytes(payloadUri), type); + } catch (Exception e) { + throw new PayloadStoreException( + "failed to deserialize payload " + payloadUri + " as " + type.getSimpleName() + + ": " + e.getMessage(), + e); + } + } + + private WorkflowPayloadEntity lookup(String payloadUri) { + WorkflowPayloadEntity row = payloadMapper.selectOne( + new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper() + .eq(WorkflowPayloadEntity::getPayloadUri, payloadUri)); + if (row == null) { + throw new PayloadStoreException("payload not found: " + payloadUri); + } + return row; + } + + private static String sha256Hex(byte[] bytes) { + try { + MessageDigest digest = MessageDigest.getInstance("SHA-256"); + return HexFormat.of().formatHex(digest.digest(bytes)); + } catch (NoSuchAlgorithmException e) { + // SHA-256 is part of the JCA standard set — should never happen. + throw new IllegalStateException("SHA-256 not available", e); + } + } + + /** Wrapper exception for payload-store failures. */ + public static class PayloadStoreException extends RuntimeException { + public PayloadStoreException(String message) { super(message); } + public PayloadStoreException(String message, Throwable cause) { super(message, cause); } + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/StepAdapter.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/StepAdapter.java new file mode 100644 index 00000000..64e997f0 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/StepAdapter.java @@ -0,0 +1,27 @@ +package vip.mate.workflow.runtime; + +import vip.mate.workflow.compiler.ir.WorkflowStep; + +/** + * Strategy interface for executing a single workflow step. One implementation + * per {@code StepMode.typeName()}; the runner looks up the adapter by name and + * calls {@link #execute}. Adapters MUST NOT mutate {@link WorkflowRunContext} + * directly — the runner publishes the {@link StepResult} into the context so + * fan_out groups can merge in deterministic order. + */ +public interface StepAdapter { + + /** + * The mode name this adapter handles — must match + * {@code StepMode.typeName()} (sequential / fan_out / collect / conditional / + * await_approval / dispatch_channel / write_memory). + */ + String typeName(); + + /** + * Execute one step. Implementations should never throw to signal a normal + * step failure — return {@link StepResult#failed(String)} instead. Throwing + * is reserved for programmer / framework errors that should abort the run. + */ + StepResult execute(WorkflowStep step, WorkflowRunContext context); +} diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/StepAdapterRegistry.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/StepAdapterRegistry.java new file mode 100644 index 00000000..21064a39 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/StepAdapterRegistry.java @@ -0,0 +1,39 @@ +package vip.mate.workflow.runtime; + +import org.springframework.stereotype.Component; + +import java.util.List; +import java.util.Map; +import java.util.function.Function; +import java.util.stream.Collectors; + +/** + * Registry mapping mode {@code typeName} to its {@link StepAdapter} bean. + * Spring autowires every adapter on the classpath; the runner asks the + * registry which adapter to use and the registry rejects unknown modes + * up-front so a wiring bug surfaces at the run boundary instead of inside + * the executor loop. + */ +@Component +public class StepAdapterRegistry { + + private final Map adapters; + + public StepAdapterRegistry(List adapters) { + Map mapped = adapters.stream() + .collect(Collectors.toUnmodifiableMap(StepAdapter::typeName, Function.identity())); + this.adapters = mapped; + } + + public StepAdapter get(String typeName) { + StepAdapter adapter = adapters.get(typeName); + if (adapter == null) { + throw new IllegalStateException("no step adapter registered for mode: " + typeName); + } + return adapter; + } + + public boolean has(String typeName) { + return adapters.containsKey(typeName); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/StepResult.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/StepResult.java new file mode 100644 index 00000000..acfa3362 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/StepResult.java @@ -0,0 +1,43 @@ +package vip.mate.workflow.runtime; + +/** + * Outcome reported by a {@link StepAdapter#execute}. Records: + *
    + *
  • {@link State} — succeeded / skipped / failed; the runner translates + * these to {@code mate_workflow_run_step.state}.
    • + *
  • {@code outputPayloadUri} — payload URI for the step's output, or + * {@code null} when the step produced nothing (e.g. skipped, collect).
    • + *
  • {@code outputContentType} — resolved content type, defaults to + * {@code text}; lets the runner persist {@code output_content_type} + * without rebuilding the step contract.
    • + *
  • {@code outputValue} — the in-memory value to publish into the + * run context's {@code outputs} map. {@link String} for text content, + * {@link java.util.Map} / {@link java.util.List} for json content. + * {@code null} when the step has no {@code outputVar}.
    • + *
  • {@code outputSummary} / {@code errorMessage} — short labels for the + * step row; both optional.
    • + *
+ */ +public record StepResult( + State state, + String outputPayloadUri, + String outputContentType, + Object outputValue, + String outputSummary, + String errorMessage +) { + + public enum State { SUCCEEDED, SKIPPED, FAILED } + + public static StepResult succeeded(String payloadUri, String contentType, Object value, String summary) { + return new StepResult(State.SUCCEEDED, payloadUri, contentType, value, summary, null); + } + + public static StepResult skipped(String reason) { + return new StepResult(State.SKIPPED, null, null, null, reason, null); + } + + public static StepResult failed(String errorMessage) { + return new StepResult(State.FAILED, null, null, null, null, errorMessage); + } +} 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 new file mode 100644 index 00000000..3714d323 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowRunContext.java @@ -0,0 +1,64 @@ +package vip.mate.workflow.runtime; + +import java.util.LinkedHashMap; +import java.util.Map; + +/** + * Mutable run-scoped state shared across step adapters. Holds the per-run + * identity ({@code runId}, {@code workspaceId}), the resolved input bag, and + * the rolling outputs map keyed by {@code outputVar}. Adapters mutate this + * after each successful step so subsequent expressions / templates see the + * latest value via {@link #templateContext()}. + * + *

Not thread-safe by itself — the runner ensures a single writer at a time. + * For the fan_out group, adapters write to a temporary local map and the + * runner merges results back into the shared context once the group completes. + */ +public class WorkflowRunContext { + + private final long runId; + private final long workspaceId; + private final long workflowId; + private final long revisionId; + private final Map inputs; + private final Map outputs = new LinkedHashMap<>(); + + public WorkflowRunContext(long runId, long workspaceId, long workflowId, long revisionId, + Map inputs) { + this.runId = runId; + this.workspaceId = workspaceId; + this.workflowId = workflowId; + this.revisionId = revisionId; + this.inputs = inputs == null ? Map.of() : Map.copyOf(inputs); + } + + public long runId() { return runId; } + public long workspaceId() { return workspaceId; } + public long workflowId() { return workflowId; } + public long revisionId() { return revisionId; } + + public Map inputs() { return inputs; } + + /** Mutable outputs map. Use {@link #putOutput} for writes. */ + public synchronized Map outputs() { + return new LinkedHashMap<>(outputs); + } + + public synchronized void putOutput(String name, Object value) { + if (name == null || name.isBlank()) return; + outputs.put(name, value); + } + + /** + * Snapshot map shaped as {@code {"inputs": {...}, "outputs": {...}}} — + * the contract every workflow expression / template assumes. The map is a + * defensive copy so concurrent fan_out branches can render templates + * against a stable view while another branch's success completes. + */ + public synchronized Map templateContext() { + Map ctx = new LinkedHashMap<>(); + ctx.put("inputs", inputs); + ctx.put("outputs", new LinkedHashMap<>(outputs)); + return ctx; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowRunRequest.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowRunRequest.java new file mode 100644 index 00000000..d74f1b38 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowRunRequest.java @@ -0,0 +1,22 @@ +package vip.mate.workflow.runtime; + +import java.util.Map; + +/** + * Inputs the runner needs to start a single workflow run. Identity fields + * ({@code workflowId}, {@code revisionId}, {@code workspaceId}) tie the run + * row back to the published revision the runner walks. {@code triggeredBy} + * is a free-form label written into {@code mate_workflow_run.triggered_by} + * — the runner doesn't interpret it. + */ +public record WorkflowRunRequest( + long workflowId, + long revisionId, + long workspaceId, + String triggeredBy, + Map inputs +) { + public WorkflowRunRequest { + inputs = inputs == null ? Map.of() : Map.copyOf(inputs); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowRunResult.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowRunResult.java new file mode 100644 index 00000000..c878a850 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowRunResult.java @@ -0,0 +1,16 @@ +package vip.mate.workflow.runtime; + +/** + * Public outcome of a workflow run. {@code state} mirrors the row state + * machine ({@code succeeded} / {@code failed}); {@code finalOutputUri} is + * the payload URI of the last non-skipped step's output, or {@code null} + * when no step produced output. {@code errorMessage} is populated when the + * run aborted; {@code null} on success. + */ +public record WorkflowRunResult( + long runId, + String state, + String finalOutputUri, + String errorMessage +) { +} 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 new file mode 100644 index 00000000..6c9feca5 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/WorkflowRunner.java @@ -0,0 +1,261 @@ +package vip.mate.workflow.runtime; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import vip.mate.workflow.compiler.ir.StepMode; +import vip.mate.workflow.compiler.ir.WorkflowGraph; +import vip.mate.workflow.compiler.ir.WorkflowStep; +import vip.mate.workflow.model.WorkflowRunEntity; +import vip.mate.workflow.model.WorkflowRunStepEntity; +import vip.mate.workflow.repository.WorkflowRunMapper; +import vip.mate.workflow.repository.WorkflowRunStepMapper; + +import java.time.Duration; +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +/** + * Linear executor for v0 workflows. Walks the graph step-by-step, batching + * adjacent {@code fan_out} steps + terminating {@code collect} into a single + * parallel group. The first failed (non-skipped) step aborts the run and + * 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. + */ +@Slf4j +@Service +public class WorkflowRunner { + + private static final String STATE_RUNNING = "running"; + private static final String STATE_SUCCEEDED = "succeeded"; + private static final String STATE_FAILED = "failed"; + private static final String STATE_SKIPPED = "skipped"; + + private static final ExecutorService FAN_OUT_EXECUTOR = + Executors.newVirtualThreadPerTaskExecutor(); + + private final WorkflowRunMapper runMapper; + private final WorkflowRunStepMapper stepMapper; + private final StepAdapterRegistry adapters; + private final PayloadStore payloadStore; + + public WorkflowRunner(WorkflowRunMapper runMapper, + WorkflowRunStepMapper stepMapper, + StepAdapterRegistry adapters, + PayloadStore payloadStore) { + this.runMapper = runMapper; + this.stepMapper = stepMapper; + this.adapters = adapters; + this.payloadStore = payloadStore; + } + + public WorkflowRunResult run(WorkflowGraph graph, WorkflowRunRequest request) { + WorkflowRunEntity runRow = openRun(request); + long runId = runRow.getId(); + String inputsRef = payloadStore.storeJson(request.workspaceId(), request.inputs()); + runRow.setInitialInputRef(inputsRef); + runMapper.updateById(runRow); + + WorkflowRunContext ctx = new WorkflowRunContext( + runId, + request.workspaceId(), + request.workflowId(), + request.revisionId(), + request.inputs()); + + String lastSucceededOutputRef = null; + try { + int i = 0; + while (i < graph.steps().size()) { + WorkflowStep step = graph.steps().get(i); + int groupEnd = scanFanOutGroup(graph.steps(), i); + if (groupEnd > i) { + GroupOutcome out = executeFanOutGroup(graph.steps(), i, groupEnd, ctx); + if (out.failed) { + return finishFailed(runRow, out.errorMessage); + } + if (out.lastOutputRef != null) lastSucceededOutputRef = out.lastOutputRef; + i = groupEnd + 1; + } else { + StepResult result = executeStep(step, i, /*iterationIndex*/ null, ctx); + if (result.state() == StepResult.State.FAILED) { + return finishFailed(runRow, result.errorMessage()); + } + if (result.outputPayloadUri() != null) { + lastSucceededOutputRef = result.outputPayloadUri(); + } + i++; + } + } + return finishSucceeded(runRow, lastSucceededOutputRef); + } catch (RuntimeException e) { + log.error("Workflow run {} aborted by unexpected exception", runId, e); + return finishFailed(runRow, "runtime error: " + e.getMessage()); + } + } + + /** + * Result of executing a contiguous {@code fan_out ... collect} block: + * either every branch succeeded (or skipped) and the merged outputs are + * already in the run context, or one branch failed and the runner aborts. + */ + private record GroupOutcome(boolean failed, String errorMessage, String lastOutputRef) {} + + /** + * If {@code steps[start]} is the head of a fan_out group (≥ 2 consecutive + * fan_out followed by exactly one collect — the schema validator already + * enforced this), return the index of the terminating collect. Otherwise + * return {@code start} so the caller treats it as a single-step. + */ + private static int scanFanOutGroup(List steps, int start) { + if (!(steps.get(start).mode() instanceof StepMode.FanOut)) return start; + int j = start; + while (j < steps.size() && steps.get(j).mode() instanceof StepMode.FanOut) j++; + if (j < steps.size() && steps.get(j).mode() instanceof StepMode.Collect) { + return j; + } + return start; + } + + 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) {} + List branches = new ArrayList<>(); + for (int i = from; i < collectIdx; i++) { + int idx = i; + WorkflowStep step = steps.get(i); + Future future = FAN_OUT_EXECUTOR.submit( + () -> executeStep(step, idx, idx - from, ctx)); + branches.add(new Branch(idx, step, future)); + } + + String lastOutputRef = null; + for (Branch branch : branches) { + try { + StepResult result = branch.future.get(resolveTimeoutSecs(branch.step), TimeUnit.SECONDS); + if (result.state() == StepResult.State.FAILED) { + return new GroupOutcome(true, + "fan_out branch '" + branch.step.name() + "' failed: " + result.errorMessage(), + null); + } + if (result.outputPayloadUri() != null) lastOutputRef = result.outputPayloadUri(); + } catch (Exception e) { + return new GroupOutcome(true, + "fan_out branch '" + branch.step.name() + "' threw: " + e.getMessage(), + null); + } + } + + // 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) { + return new GroupOutcome(true, collectResult.errorMessage(), null); + } + return new GroupOutcome(false, null, lastOutputRef); + } + + private static long resolveTimeoutSecs(WorkflowStep step) { + if (step.timeoutSecs() == null || step.timeoutSecs() <= 0) return 600L; + return step.timeoutSecs(); + } + + private StepResult executeStep(WorkflowStep step, int stepIndex, Integer iterationIndex, + WorkflowRunContext ctx) { + StepAdapter adapter = adapters.get(step.mode().typeName()); + WorkflowRunStepEntity stepRow = openStep(ctx.runId(), stepIndex, iterationIndex, step); + + long startNanos = System.nanoTime(); + StepResult result; + try { + result = adapter.execute(step, ctx); + } catch (RuntimeException e) { + log.error("Adapter {} threw on run={} stepIndex={} step='{}'", + step.mode().typeName(), ctx.runId(), stepIndex, step.name(), e); + result = StepResult.failed("adapter threw: " + e.getMessage()); + } + long elapsedMs = Duration.ofNanos(System.nanoTime() - startNanos).toMillis(); + + // ctx.putOutput is synchronised internally so concurrent fan_out + // branches can commit their results back to the shared run context + // without external locking. + if (result.state() == StepResult.State.SUCCEEDED && step.outputVar() != null + && !step.outputVar().isBlank() && result.outputValue() != null) { + ctx.putOutput(step.outputVar(), result.outputValue()); + } + + closeStep(stepRow, result, elapsedMs); + return result; + } + + private WorkflowRunEntity openRun(WorkflowRunRequest request) { + WorkflowRunEntity row = new WorkflowRunEntity(); + row.setWorkflowId(request.workflowId()); + row.setRevisionId(request.revisionId()); + row.setWorkspaceId(request.workspaceId()); + row.setState(STATE_RUNNING); + row.setTriggeredBy(request.triggeredBy()); + row.setStartedAt(LocalDateTime.now()); + runMapper.insert(row); + return row; + } + + private WorkflowRunResult finishSucceeded(WorkflowRunEntity runRow, String finalOutputRef) { + runRow.setState(STATE_SUCCEEDED); + runRow.setFinalOutputRef(finalOutputRef); + runRow.setCompletedAt(LocalDateTime.now()); + runMapper.updateById(runRow); + return new WorkflowRunResult(runRow.getId(), STATE_SUCCEEDED, finalOutputRef, null); + } + + private WorkflowRunResult finishFailed(WorkflowRunEntity runRow, String errorMessage) { + runRow.setState(STATE_FAILED); + runRow.setErrorMessage(errorMessage); + runRow.setCompletedAt(LocalDateTime.now()); + runMapper.updateById(runRow); + return new WorkflowRunResult(runRow.getId(), STATE_FAILED, null, errorMessage); + } + + private WorkflowRunStepEntity openStep(long runId, int stepIndex, Integer iterationIndex, + WorkflowStep step) { + WorkflowRunStepEntity row = new WorkflowRunStepEntity(); + row.setRunId(runId); + row.setStepIndex(stepIndex); + row.setIterationIndex(iterationIndex); + row.setStepName(step.name()); + row.setAgentId(step.agentId()); + row.setState(STATE_RUNNING); + row.setOutputContentType(step.effectiveOutputContentType()); + row.setStartedAt(LocalDateTime.now()); + stepMapper.insert(row); + return row; + } + + private void closeStep(WorkflowRunStepEntity row, StepResult result, long durationMs) { + switch (result.state()) { + case SUCCEEDED -> row.setState(STATE_SUCCEEDED); + case SKIPPED -> row.setState(STATE_SKIPPED); + case FAILED -> row.setState(STATE_FAILED); + } + row.setOutputRef(result.outputPayloadUri()); + if (result.outputContentType() != null) { + row.setOutputContentType(result.outputContentType()); + } + row.setOutputSummary(result.outputSummary()); + row.setErrorMessage(result.errorMessage()); + row.setDurationMs(durationMs); + row.setCompletedAt(LocalDateTime.now()); + stepMapper.updateById(row); + } + +} diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/mode/CollectStepAdapter.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/mode/CollectStepAdapter.java new file mode 100644 index 00000000..edd2260d --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/mode/CollectStepAdapter.java @@ -0,0 +1,26 @@ +package vip.mate.workflow.runtime.mode; + +import org.springframework.stereotype.Component; +import vip.mate.workflow.compiler.ir.WorkflowStep; +import vip.mate.workflow.runtime.StepAdapter; +import vip.mate.workflow.runtime.StepResult; +import vip.mate.workflow.runtime.WorkflowRunContext; + +/** + * {@code collect} — barrier that closes the most recent fan_out group. The + * runner awaits the parallel branches before invoking this adapter, then + * publishes their merged outputs into the run context. The adapter itself + * does no agent work; it simply records a step row so the run history shows + * where the group joined and produces no payload of its own. + */ +@Component +public class CollectStepAdapter implements StepAdapter { + + @Override + public String typeName() { return "collect"; } + + @Override + public StepResult execute(WorkflowStep step, WorkflowRunContext context) { + return StepResult.succeeded(null, null, null, "fan_out group joined"); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/mode/ConditionalStepAdapter.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/mode/ConditionalStepAdapter.java new file mode 100644 index 00000000..5da38c14 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/mode/ConditionalStepAdapter.java @@ -0,0 +1,55 @@ +package vip.mate.workflow.runtime.mode; + +import org.springframework.stereotype.Component; +import vip.mate.workflow.compiler.PebbleSubsetEvaluator; +import vip.mate.workflow.compiler.ir.StepMode; +import vip.mate.workflow.compiler.ir.WorkflowStep; +import vip.mate.workflow.runtime.AgentStepExecutor; +import vip.mate.workflow.runtime.StepAdapter; +import vip.mate.workflow.runtime.StepResult; +import vip.mate.workflow.runtime.WorkflowRunContext; + +/** + * {@code conditional} — runs the embedded agent step only when the configured + * Pebble expression evaluates true against the current run context. A false + * verdict yields {@link StepResult.State#SKIPPED}; an evaluation error fails + * the step. Skipped steps still emit a run-step row so the history captures + * the routing decision. + */ +@Component +public class ConditionalStepAdapter implements StepAdapter { + + private final PebbleSubsetEvaluator pebble; + private final AgentStepExecutor executor; + + public ConditionalStepAdapter(PebbleSubsetEvaluator pebble, AgentStepExecutor executor) { + this.pebble = pebble; + this.executor = executor; + } + + @Override + public String typeName() { return "conditional"; } + + @Override + public StepResult execute(WorkflowStep step, WorkflowRunContext context) { + if (!(step.mode() instanceof StepMode.Conditional cond)) { + return StepResult.failed("conditional adapter received non-conditional mode: " + + step.mode().typeName()); + } + + boolean truth; + try { + var compiled = pebble.parseExpression(cond.expression()); + truth = pebble.evaluateAsBoolean(compiled, context.templateContext()); + } catch (Exception e) { + return StepResult.failed("conditional expression evaluation failed for step '" + + step.name() + "': " + e.getMessage()); + } + + if (!truth) { + return StepResult.skipped("guard expression evaluated false"); + } + + return executor.run(step, context); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/mode/FanOutStepAdapter.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/mode/FanOutStepAdapter.java new file mode 100644 index 00000000..74428faf --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/mode/FanOutStepAdapter.java @@ -0,0 +1,34 @@ +package vip.mate.workflow.runtime.mode; + +import org.springframework.stereotype.Component; +import vip.mate.workflow.compiler.ir.WorkflowStep; +import vip.mate.workflow.runtime.AgentStepExecutor; +import vip.mate.workflow.runtime.StepAdapter; +import vip.mate.workflow.runtime.StepResult; +import vip.mate.workflow.runtime.WorkflowRunContext; + +/** + * {@code fan_out} — body of a parallel group. Each fan_out step runs against + * the run context snapshot that existed when the group started; the runner + * dispatches the whole group in parallel and merges {@code outputs} only when + * the terminating {@code collect} runs. From the adapter's perspective the + * step body is identical to a sequential agent call — the parallelism is + * orchestrated upstream. + */ +@Component +public class FanOutStepAdapter implements StepAdapter { + + private final AgentStepExecutor executor; + + public FanOutStepAdapter(AgentStepExecutor executor) { + this.executor = executor; + } + + @Override + public String typeName() { return "fan_out"; } + + @Override + public StepResult execute(WorkflowStep step, WorkflowRunContext context) { + return executor.run(step, context); + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/mode/SequentialStepAdapter.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/mode/SequentialStepAdapter.java new file mode 100644 index 00000000..a73227de --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/mode/SequentialStepAdapter.java @@ -0,0 +1,31 @@ +package vip.mate.workflow.runtime.mode; + +import org.springframework.stereotype.Component; +import vip.mate.workflow.compiler.ir.WorkflowStep; +import vip.mate.workflow.runtime.AgentStepExecutor; +import vip.mate.workflow.runtime.StepAdapter; +import vip.mate.workflow.runtime.StepResult; +import vip.mate.workflow.runtime.WorkflowRunContext; + +/** + * {@code sequential} — runs after the previous step finishes and threads its + * output forward via {@code outputs[outputVar]}. The default mode for any + * agent-call step that does not need parallel or guarded execution. + */ +@Component +public class SequentialStepAdapter implements StepAdapter { + + private final AgentStepExecutor executor; + + public SequentialStepAdapter(AgentStepExecutor executor) { + this.executor = executor; + } + + @Override + public String typeName() { return "sequential"; } + + @Override + public StepResult execute(WorkflowStep step, WorkflowRunContext context) { + return executor.run(step, context); + } +}