mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
feat(workflow): add linear runtime for the four base step modes
This commit is contained in:
matevip
parent
3065d095fd
commit
4e457c2e8b
@ -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);
|
||||
}
|
||||
@ -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) + "...";
|
||||
}
|
||||
}
|
||||
@ -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<AgentEntity>()
|
||||
.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<AgentEntity>()
|
||||
.eq(AgentEntity::getName, agentName.trim())
|
||||
.eq(AgentEntity::getEnabled, true));
|
||||
}
|
||||
return entity == null ? null : entity.getId();
|
||||
}
|
||||
}
|
||||
@ -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> T readJson(String payloadUri, Class<T> 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<WorkflowPayloadEntity>()
|
||||
.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); }
|
||||
}
|
||||
}
|
||||
@ -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);
|
||||
}
|
||||
@ -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<String, StepAdapter> adapters;
|
||||
|
||||
public StepAdapterRegistry(List<StepAdapter> adapters) {
|
||||
Map<String, StepAdapter> 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);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,43 @@
|
||||
package vip.mate.workflow.runtime;
|
||||
|
||||
/**
|
||||
* Outcome reported by a {@link StepAdapter#execute}. Records:
|
||||
* <ul>
|
||||
* <li>{@link State} — succeeded / skipped / failed; the runner translates
|
||||
* these to {@code mate_workflow_run_step.state}.</li>
|
||||
* <li>{@code outputPayloadUri} — payload URI for the step's output, or
|
||||
* {@code null} when the step produced nothing (e.g. skipped, collect).</li>
|
||||
* <li>{@code outputContentType} — resolved content type, defaults to
|
||||
* {@code text}; lets the runner persist {@code output_content_type}
|
||||
* without rebuilding the step contract.</li>
|
||||
* <li>{@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}.</li>
|
||||
* <li>{@code outputSummary} / {@code errorMessage} — short labels for the
|
||||
* step row; both optional.</li>
|
||||
* </ul>
|
||||
*/
|
||||
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);
|
||||
}
|
||||
}
|
||||
@ -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()}.
|
||||
*
|
||||
* <p>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<String, Object> inputs;
|
||||
private final Map<String, Object> outputs = new LinkedHashMap<>();
|
||||
|
||||
public WorkflowRunContext(long runId, long workspaceId, long workflowId, long revisionId,
|
||||
Map<String, Object> 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<String, Object> inputs() { return inputs; }
|
||||
|
||||
/** Mutable outputs map. Use {@link #putOutput} for writes. */
|
||||
public synchronized Map<String, Object> 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<String, Object> templateContext() {
|
||||
Map<String, Object> ctx = new LinkedHashMap<>();
|
||||
ctx.put("inputs", inputs);
|
||||
ctx.put("outputs", new LinkedHashMap<>(outputs));
|
||||
return ctx;
|
||||
}
|
||||
}
|
||||
@ -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<String, Object> inputs
|
||||
) {
|
||||
public WorkflowRunRequest {
|
||||
inputs = inputs == null ? Map.of() : Map.copyOf(inputs);
|
||||
}
|
||||
}
|
||||
@ -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
|
||||
) {
|
||||
}
|
||||
@ -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.
|
||||
*
|
||||
* <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.
|
||||
*/
|
||||
@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<WorkflowStep> 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<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) {}
|
||||
List<Branch> branches = new ArrayList<>();
|
||||
for (int i = from; i < collectIdx; i++) {
|
||||
int idx = i;
|
||||
WorkflowStep step = steps.get(i);
|
||||
Future<StepResult> 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);
|
||||
}
|
||||
|
||||
}
|
||||
@ -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");
|
||||
}
|
||||
}
|
||||
@ -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);
|
||||
}
|
||||
}
|
||||
@ -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);
|
||||
}
|
||||
}
|
||||
@ -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);
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user