diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/pipeline/WikiPipelineService.java b/mateclaw-server/src/main/java/vip/mate/wiki/pipeline/WikiPipelineService.java new file mode 100644 index 00000000..52c75aec --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/wiki/pipeline/WikiPipelineService.java @@ -0,0 +1,177 @@ +package vip.mate.wiki.pipeline; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import lombok.extern.slf4j.Slf4j; +import org.springframework.dao.DuplicateKeyException; +import org.springframework.stereotype.Service; +import vip.mate.wiki.model.WikiPipelineDefinitionEntity; +import vip.mate.wiki.model.WikiPipelineRunEntity; +import vip.mate.wiki.model.WikiPipelineStepRunEntity; +import vip.mate.wiki.repository.WikiPipelineRunMapper; +import vip.mate.wiki.repository.WikiPipelineStepRunMapper; + +import java.time.LocalDateTime; +import java.util.List; +import java.util.Map; + +/** + * Runs a wiki pipeline definition: creates a dedup-guarded run, executes each + * step through the matching {@link WikiStepExecutor} under the definition's + * owner agent, and records run / step status. + * + *

Run creation is idempotent: a duplicate trigger envelope collides on the + * run table's unique key and is skipped, so concurrent instances cannot spawn + * parallel runs for the same trigger. + * + * @author MateClaw Team + */ +@Slf4j +@Service +public class WikiPipelineService { + + private final WikiPipelineRunMapper runMapper; + private final WikiPipelineStepRunMapper stepRunMapper; + private final ObjectMapper objectMapper; + private final Map executors; + + public WikiPipelineService(WikiPipelineRunMapper runMapper, + WikiPipelineStepRunMapper stepRunMapper, + ObjectMapper objectMapper, + List executorBeans) { + this.runMapper = runMapper; + this.stepRunMapper = stepRunMapper; + this.objectMapper = objectMapper; + this.executors = new java.util.HashMap<>(); + for (WikiStepExecutor e : executorBeans) { + this.executors.put(e.type(), e); + } + } + + /** Outcome of an attempted run. {@code run} is null when skipped as a duplicate. */ + public record RunOutcome(WikiPipelineRunEntity run, boolean duplicate) {} + + /** + * Execute a definition for one trigger. Returns {@code duplicate=true} with + * a null run when the trigger envelope was already handled. + */ + public RunOutcome execute(WikiPipelineDefinitionEntity def, String triggerSubject, + String triggerBucket, String inputJson) { + if (def.getEnabled() != null && def.getEnabled() == 0) { + return new RunOutcome(null, false); + } + if (def.getOwnerAgentId() == null) { + throw new IllegalStateException("Pipeline definition " + def.getId() + " has no owner agent"); + } + + WikiPipelineRunEntity run = new WikiPipelineRunEntity(); + run.setDefinitionId(def.getId()); + run.setKbId(def.getKbId()); + run.setStatus("running"); + run.setTriggerType(def.getTriggerType()); + run.setTriggerSubject(triggerSubject); + run.setTriggerBucket(triggerBucket); + run.setInputJson(inputJson); + run.setStartedAt(LocalDateTime.now()); + run.setCreateTime(LocalDateTime.now()); + try { + runMapper.insert(run); + } catch (DuplicateKeyException dup) { + // Another instance / earlier trigger already created this run. + log.info("[WikiPipeline] duplicate trigger for def={} subject={} bucket={} — skipped", + def.getId(), triggerSubject, triggerBucket); + return new RunOutcome(null, true); + } + + List> steps = parseSteps(def.getStepsJson()); + String previousOutput = null; + try { + for (Map step : steps) { + previousOutput = runStep(def, run.getId(), step, previousOutput); + } + finishRun(run, "succeeded", previousOutput, null); + } catch (StepFailure f) { + finishRun(run, "failed", previousOutput, f.getMessage()); + } + return new RunOutcome(run, false); + } + + private String runStep(WikiPipelineDefinitionEntity def, Long runId, Map step, + String previousOutput) throws StepFailure { + String stepId = String.valueOf(step.getOrDefault("id", "step")); + String executorType = String.valueOf(step.getOrDefault("executor", "")); + + WikiPipelineStepRunEntity stepRun = new WikiPipelineStepRunEntity(); + stepRun.setRunId(runId); + stepRun.setStepId(stepId); + stepRun.setExecutor(executorType); + stepRun.setStatus("running"); + stepRun.setStartedAt(LocalDateTime.now()); + stepRun.setCreateTime(LocalDateTime.now()); + stepRunMapper.insert(stepRun); + + WikiStepExecutor executor = executors.get(executorType); + if (executor == null) { + String msg = "No executor registered for type '" + executorType + "'"; + failStep(stepRun, msg); + throw new StepFailure(msg); + } + try { + @SuppressWarnings("unchecked") + Map config = step.get("config") instanceof Map + ? (Map) step.get("config") : Map.of(); + String output = executor.execute(new WikiStepContext( + def.getKbId(), def.getOwnerAgentId(), stepId, config, previousOutput)); + stepRun.setStatus("succeeded"); + stepRun.setOutputJson(output); + stepRun.setFinishedAt(LocalDateTime.now()); + stepRunMapper.updateById(stepRun); + return output; + } catch (Exception e) { + String msg = "Step '" + stepId + "' failed: " + e.getMessage(); + failStep(stepRun, msg); + throw new StepFailure(msg); + } + } + + private void failStep(WikiPipelineStepRunEntity stepRun, String message) { + stepRun.setStatus("failed"); + stepRun.setErrorMessage(truncate(message)); + stepRun.setFinishedAt(LocalDateTime.now()); + stepRunMapper.updateById(stepRun); + } + + private void finishRun(WikiPipelineRunEntity run, String status, String output, String error) { + run.setStatus(status); + run.setOutputJson(output); + run.setErrorMessage(truncate(error)); + run.setFinishedAt(LocalDateTime.now()); + runMapper.updateById(run); + } + + private List> parseSteps(String stepsJson) { + if (stepsJson == null || stepsJson.isBlank()) { + return List.of(); + } + try { + return objectMapper.readValue(stepsJson, new TypeReference>>() {}); + } catch (Exception e) { + log.warn("[WikiPipeline] unparseable steps_json: {}", e.getMessage()); + return List.of(); + } + } + + private static String truncate(String s) { + if (s == null) { + return null; + } + return s.length() > 2000 ? s.substring(0, 2000) : s; + } + + /** Internal control-flow signal that a step failed and the run should stop. */ + private static final class StepFailure extends Exception { + StepFailure(String message) { + super(message); + } + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/pipeline/WikiStepContext.java b/mateclaw-server/src/main/java/vip/mate/wiki/pipeline/WikiStepContext.java new file mode 100644 index 00000000..b67e2022 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/wiki/pipeline/WikiStepContext.java @@ -0,0 +1,18 @@ +package vip.mate.wiki.pipeline; + +import java.util.Map; + +/** + * Inputs available to a {@link WikiStepExecutor}: the KB, the owner agent the + * step runs under (for permission checks), the step's declared config, and the + * output of the previous step. + * + * @author MateClaw Team + */ +public record WikiStepContext( + Long kbId, + Long ownerAgentId, + String stepId, + Map stepConfig, + String previousOutput) { +} diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/pipeline/WikiStepExecutor.java b/mateclaw-server/src/main/java/vip/mate/wiki/pipeline/WikiStepExecutor.java new file mode 100644 index 00000000..a28ef541 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/wiki/pipeline/WikiStepExecutor.java @@ -0,0 +1,24 @@ +package vip.mate.wiki.pipeline; + +/** + * Executes one pipeline step of a given kind. Implementations register their + * {@link #type()} (e.g. {@code llm}, {@code skill}); the pipeline service + * dispatches each step to the matching executor. + * + *

Python execution is intentionally not provided here — it requires a real + * OS sandbox and a separate security review, and is out of the MVP. + * + * @author MateClaw Team + */ +public interface WikiStepExecutor { + + /** The executor kind this handles, matched against a step's {@code executor}. */ + String type(); + + /** + * Run the step and return its textual output. + * + * @throws Exception on failure — the pipeline records the step as failed + */ + String execute(WikiStepContext context) throws Exception; +} diff --git a/mateclaw-server/src/test/java/vip/mate/wiki/pipeline/WikiPipelineServiceE2ETest.java b/mateclaw-server/src/test/java/vip/mate/wiki/pipeline/WikiPipelineServiceE2ETest.java new file mode 100644 index 00000000..b97dc7a1 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/wiki/pipeline/WikiPipelineServiceE2ETest.java @@ -0,0 +1,135 @@ +package vip.mate.wiki.pipeline; + +import com.baomidou.mybatisplus.core.toolkit.Wrappers; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.context.TestConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.test.annotation.DirtiesContext; +import vip.mate.wiki.model.WikiPipelineDefinitionEntity; +import vip.mate.wiki.model.WikiPipelineStepRunEntity; +import vip.mate.wiki.repository.WikiPipelineStepRunMapper; + +import java.time.LocalDateTime; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Orchestration test for {@link WikiPipelineService} against H2 with stub + * executors: success path records succeeded run + steps, a failing step fails + * the run and stops, and a duplicate trigger envelope is skipped. + */ +@SpringBootTest( + webEnvironment = SpringBootTest.WebEnvironment.NONE, + properties = { + "spring.flyway.enabled=true", + "spring.flyway.locations=classpath:db/migration/h2", + "mateclaw.feature-flag.refresh-ms=999999" + } +) +@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS) +class WikiPipelineServiceE2ETest { + + /** Stub executors: 'echo' returns a marker; 'boom' always throws. */ + @TestConfiguration + static class StubExecutors { + @Bean + WikiStepExecutor echoExecutor() { + return new WikiStepExecutor() { + public String type() { return "echo"; } + public String execute(WikiStepContext c) { + return "echo:" + c.stepId() + ":" + (c.previousOutput() == null ? "" : c.previousOutput()); + } + }; + } + @Bean + WikiStepExecutor boomExecutor() { + return new WikiStepExecutor() { + public String type() { return "boom"; } + public String execute(WikiStepContext c) throws Exception { throw new RuntimeException("kaboom"); } + }; + } + } + + @Autowired + private WikiPipelineService pipelineService; + @Autowired + private WikiPipelineStepRunMapper stepRunMapper; + @Autowired + private ObjectMapper objectMapper; + + private static final java.util.concurrent.atomic.AtomicLong SEQ = + new java.util.concurrent.atomic.AtomicLong(System.nanoTime()); + + private WikiPipelineDefinitionEntity def(String stepsJson) { + WikiPipelineDefinitionEntity d = new WikiPipelineDefinitionEntity(); + d.setId(SEQ.incrementAndGet()); + d.setKbId(1L); + d.setName("p" + d.getId()); + d.setOwnerAgentId(42L); + d.setTriggerType("page_type_count"); + d.setStepsJson(stepsJson); + d.setEnabled(1); + return d; + } + + @Test + void successPath_runsAllSteps_chainingOutput() { + WikiPipelineDefinitionEntity d = def( + "[{\"id\":\"a\",\"executor\":\"echo\"},{\"id\":\"b\",\"executor\":\"echo\"}]"); + WikiPipelineService.RunOutcome outcome = pipelineService.execute(d, "episode", "20", null); + + assertFalse(outcome.duplicate()); + assertNotNull(outcome.run()); + assertEquals("succeeded", outcome.run().getStatus()); + // step b sees step a's output (chaining) + assertEquals("echo:b:echo:a:", outcome.run().getOutputJson()); + + List steps = stepRunMapper.selectList( + Wrappers.lambdaQuery() + .eq(WikiPipelineStepRunEntity::getRunId, outcome.run().getId())); + assertEquals(2, steps.size()); + assertTrue(steps.stream().allMatch(s -> s.getStatus().equals("succeeded"))); + } + + @Test + void failingStep_failsRun_andStops() { + WikiPipelineDefinitionEntity d = def( + "[{\"id\":\"a\",\"executor\":\"echo\"},{\"id\":\"b\",\"executor\":\"boom\"},{\"id\":\"c\",\"executor\":\"echo\"}]"); + WikiPipelineService.RunOutcome outcome = pipelineService.execute(d, "episode", "20", null); + + assertEquals("failed", outcome.run().getStatus()); + assertTrue(outcome.run().getErrorMessage().contains("kaboom")); + // step c must NOT have run (pipeline stopped at b) + List steps = stepRunMapper.selectList( + Wrappers.lambdaQuery() + .eq(WikiPipelineStepRunEntity::getRunId, outcome.run().getId())); + assertEquals(2, steps.size()); + } + + @Test + void duplicateTrigger_isSkipped() { + WikiPipelineDefinitionEntity d = def("[{\"id\":\"a\",\"executor\":\"echo\"}]"); + WikiPipelineService.RunOutcome first = pipelineService.execute(d, "episode", "20", null); + assertFalse(first.duplicate()); + + WikiPipelineService.RunOutcome second = pipelineService.execute(d, "episode", "20", null); + assertTrue(second.duplicate()); + assertNull(second.run()); + } + + @Test + void unknownExecutor_failsRun() { + WikiPipelineDefinitionEntity d = def("[{\"id\":\"a\",\"executor\":\"nope\"}]"); + WikiPipelineService.RunOutcome outcome = pipelineService.execute(d, "episode", "20", null); + assertEquals("failed", outcome.run().getStatus()); + assertTrue(outcome.run().getErrorMessage().contains("No executor")); + } +}