From 7363ee86681c71c08bc6a234cf1fa3c7e9a322fe Mon Sep 17 00:00:00 2001 From: matevip Date: Sun, 31 May 2026 07:59:41 +0800 Subject: [PATCH] feat(wiki): pipeline definition CRUD/YAML API, run query API, page-created trigger --- .../mate/wiki/controller/WikiController.java | 83 +++++++++ .../mate/wiki/event/WikiPageCreatedEvent.java | 2 +- .../WikiPipelineDefinitionService.java | 164 ++++++++++++++++++ .../pipeline/WikiPipelineTriggerListener.java | 1 + .../pipeline/WikiPipelineTriggerService.java | 30 ++++ .../wiki/service/WikiProcessingService.java | 2 +- .../WikiPipelineDefinitionServiceE2ETest.java | 103 +++++++++++ .../WikiPipelineTriggerListenerTest.java | 4 +- 8 files changed, 385 insertions(+), 4 deletions(-) create mode 100644 mateclaw-server/src/main/java/vip/mate/wiki/pipeline/WikiPipelineDefinitionService.java create mode 100644 mateclaw-server/src/test/java/vip/mate/wiki/pipeline/WikiPipelineDefinitionServiceE2ETest.java diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/controller/WikiController.java b/mateclaw-server/src/main/java/vip/mate/wiki/controller/WikiController.java index a65e41e7..01c522e4 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/controller/WikiController.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/controller/WikiController.java @@ -62,6 +62,9 @@ public class WikiController { private final WikiPageTypeProfileService pageTypeProfileService; private final WikiSourcePathValidator pathValidator; private final vip.mate.wiki.service.WikiSourceWatcherService sourceWatcherService; + private final vip.mate.wiki.pipeline.WikiPipelineDefinitionService pipelineDefinitionService; + private final vip.mate.wiki.repository.WikiPipelineRunMapper pipelineRunMapper; + private final vip.mate.wiki.repository.WikiPipelineStepRunMapper pipelineStepRunMapper; private final ObjectMapper objectMapper; // ==================== Knowledge Base ==================== @@ -344,6 +347,86 @@ public class WikiController { return R.ok(out); } + // ==================== Pipeline ==================== + + @RequireWorkspaceRole("viewer") + @Operation(summary = "列出知识库的 pipeline 定义") + @GetMapping("/knowledge-bases/{kbId}/pipelines") + public R> listPipelines( + @PathVariable Long kbId, @RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId) { + verifyKBWorkspace(kbId, workspaceId); + return R.ok(pipelineDefinitionService.list(kbId)); + } + + @RequireWorkspaceRole("admin") + @Operation(summary = "保存(创建/更新)pipeline 定义(YAML/JSON)") + @PostMapping("/knowledge-bases/{kbId}/pipelines") + public R savePipeline( + @PathVariable Long kbId, @RequestBody Map body, + @RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId) { + verifyKBWorkspace(kbId, workspaceId); + boolean yaml = !"json".equalsIgnoreCase(body.getOrDefault("format", "yaml")); + try { + return R.ok(pipelineDefinitionService.saveFromConfig(kbId, body.get("config"), yaml)); + } catch (IllegalArgumentException e) { + return R.fail(400, e.getMessage()); + } + } + + @RequireWorkspaceRole("member") + @Operation(summary = "校验 pipeline 配置(不保存)") + @PostMapping("/knowledge-bases/{kbId}/pipelines/validate") + public R> validatePipeline( + @PathVariable Long kbId, @RequestBody Map body, + @RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId) { + verifyKBWorkspace(kbId, workspaceId); + boolean yaml = !"json".equalsIgnoreCase(body.getOrDefault("format", "yaml")); + List issues = pipelineDefinitionService.validateConfig(body.getOrDefault("config", ""), yaml); + Map out = new LinkedHashMap<>(); + out.put("valid", issues.isEmpty()); + out.put("issues", issues); + return R.ok(out); + } + + @RequireWorkspaceRole("admin") + @Operation(summary = "删除 pipeline 定义") + @DeleteMapping("/knowledge-bases/{kbId}/pipelines/{id}") + public R deletePipeline(@PathVariable Long kbId, @PathVariable Long id, + @RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId) { + verifyKBWorkspace(kbId, workspaceId); + pipelineDefinitionService.delete(id); + return R.ok(); + } + + @RequireWorkspaceRole("viewer") + @Operation(summary = "查询 pipeline 运行记录") + @GetMapping("/knowledge-bases/{kbId}/pipelines/{id}/runs") + public R> listPipelineRuns( + @PathVariable Long kbId, @PathVariable Long id, + @RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId) { + verifyKBWorkspace(kbId, workspaceId); + return R.ok(pipelineRunMapper.selectList( + com.baomidou.mybatisplus.core.toolkit.Wrappers.lambdaQuery() + .eq(vip.mate.wiki.model.WikiPipelineRunEntity::getDefinitionId, id) + .orderByDesc(vip.mate.wiki.model.WikiPipelineRunEntity::getCreateTime))); + } + + @RequireWorkspaceRole("viewer") + @Operation(summary = "查询单次 run 的步骤明细") + @GetMapping("/knowledge-bases/{kbId}/pipeline-runs/{runId}") + public R> getPipelineRun( + @PathVariable Long kbId, @PathVariable Long runId, + @RequestHeader(value = "X-Workspace-Id", required = false) Long workspaceId) { + verifyKBWorkspace(kbId, workspaceId); + Map out = new LinkedHashMap<>(); + out.put("run", pipelineRunMapper.selectById(runId)); + out.put("steps", pipelineStepRunMapper.selectList( + com.baomidou.mybatisplus.core.toolkit.Wrappers.lambdaQuery() + .eq(vip.mate.wiki.model.WikiPipelineStepRunEntity::getRunId, runId) + .orderByAsc(vip.mate.wiki.model.WikiPipelineStepRunEntity::getCreateTime))); + return R.ok(out); + } + // ==================== Raw Materials ==================== @RequireWorkspaceRole("viewer") diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/event/WikiPageCreatedEvent.java b/mateclaw-server/src/main/java/vip/mate/wiki/event/WikiPageCreatedEvent.java index 56fd907c..f7ad8fd0 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/event/WikiPageCreatedEvent.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/event/WikiPageCreatedEvent.java @@ -7,5 +7,5 @@ package vip.mate.wiki.event; * * @author MateClaw Team */ -public record WikiPageCreatedEvent(Long kbId, String pageType) { +public record WikiPageCreatedEvent(Long kbId, String pageType, Long pageId) { } diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/pipeline/WikiPipelineDefinitionService.java b/mateclaw-server/src/main/java/vip/mate/wiki/pipeline/WikiPipelineDefinitionService.java new file mode 100644 index 00000000..587db918 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/wiki/pipeline/WikiPipelineDefinitionService.java @@ -0,0 +1,164 @@ +package vip.mate.wiki.pipeline; + +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.fasterxml.jackson.databind.ObjectMapper; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; +import vip.mate.wiki.model.WikiPipelineDefinitionEntity; +import vip.mate.wiki.repository.WikiPipelineDefinitionMapper; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +/** + * CRUD + YAML/JSON parsing for user-defined pipelines. A definition's config + * (YAML or JSON) carries {@code name}, {@code owner_agent}, a {@code trigger} + * object and a {@code steps} array; this service parses it into the persisted + * entity (trigger / steps stored as JSON). + * + * @author MateClaw Team + */ +@Slf4j +@Service +public class WikiPipelineDefinitionService { + + private static final java.util.Set KNOWN_EXECUTORS = java.util.Set.of("llm", "skill", "python"); + private static final java.util.Set KNOWN_TRIGGERS = + java.util.Set.of("page_type_count", "page_created", "stale_marked"); + + private final WikiPipelineDefinitionMapper definitionMapper; + private final ObjectMapper objectMapper; + + public WikiPipelineDefinitionService(WikiPipelineDefinitionMapper definitionMapper, ObjectMapper objectMapper) { + this.definitionMapper = definitionMapper; + this.objectMapper = objectMapper; + } + + public List list(Long kbId) { + return definitionMapper.selectList(new LambdaQueryWrapper() + .eq(WikiPipelineDefinitionEntity::getKbId, kbId) + .orderByDesc(WikiPipelineDefinitionEntity::getCreateTime)); + } + + public WikiPipelineDefinitionEntity get(Long id) { + return definitionMapper.selectById(id); + } + + public void delete(Long id) { + definitionMapper.deleteById(id); + } + + /** + * Parse a YAML/JSON pipeline config and upsert it (by kb + name). Throws + * {@link IllegalArgumentException} on a structural problem. + */ + @SuppressWarnings("unchecked") + public WikiPipelineDefinitionEntity saveFromConfig(Long kbId, String config, boolean yaml) { + Map root = parse(config, yaml); + List issues = validateParsed(root); + if (!issues.isEmpty()) { + throw new IllegalArgumentException("Invalid pipeline config: " + String.join("; ", issues)); + } + String name = String.valueOf(root.get("name")); + Object owner = root.get("owner_agent"); + Map trigger = (Map) root.get("trigger"); + Object steps = root.get("steps"); + + WikiPipelineDefinitionEntity entity = definitionMapper.selectOne( + new LambdaQueryWrapper() + .eq(WikiPipelineDefinitionEntity::getKbId, kbId) + .eq(WikiPipelineDefinitionEntity::getName, name) + .last("LIMIT 1")); + boolean isNew = entity == null; + if (isNew) { + entity = new WikiPipelineDefinitionEntity(); + entity.setKbId(kbId); + entity.setName(name); + entity.setEnabled(1); + } + entity.setOwnerAgentId(Long.valueOf(String.valueOf(owner))); + entity.setTriggerType(String.valueOf(trigger.get("type"))); + try { + entity.setTriggerConfigJson(objectMapper.writeValueAsString(trigger)); + entity.setStepsJson(objectMapper.writeValueAsString(steps)); + } catch (Exception e) { + throw new IllegalArgumentException("Failed to serialize pipeline config: " + e.getMessage()); + } + Object dedup = trigger.get("dedup_window_seconds"); + entity.setDedupWindowSeconds(dedup instanceof Number ? ((Number) dedup).intValue() : 0); + + if (isNew) { + definitionMapper.insert(entity); + } else { + definitionMapper.updateById(entity); + } + return entity; + } + + /** Validate a config without saving; returns human-readable issues (empty = valid). */ + public List validateConfig(String config, boolean yaml) { + try { + return validateParsed(parse(config, yaml)); + } catch (Exception e) { + return List.of("Unparseable config: " + e.getMessage()); + } + } + + @SuppressWarnings("unchecked") + private Map parse(String config, boolean yaml) { + if (config == null || config.isBlank()) { + throw new IllegalArgumentException("config is empty"); + } + try { + if (yaml) { + Object loaded = new org.yaml.snakeyaml.Yaml().load(config); + if (!(loaded instanceof Map)) { + throw new IllegalArgumentException("YAML root must be a mapping"); + } + return (Map) loaded; + } + return objectMapper.readValue(config, Map.class); + } catch (IllegalArgumentException e) { + throw e; + } catch (Exception e) { + throw new IllegalArgumentException(e.getMessage()); + } + } + + @SuppressWarnings("unchecked") + private List validateParsed(Map root) { + List issues = new ArrayList<>(); + if (root.get("name") == null || String.valueOf(root.get("name")).isBlank()) { + issues.add("missing 'name'"); + } + if (root.get("owner_agent") == null) { + issues.add("missing 'owner_agent' (steps run under this agent)"); + } + Object trig = root.get("trigger"); + if (!(trig instanceof Map)) { + issues.add("missing 'trigger' object"); + } else { + String type = String.valueOf(((Map) trig).get("type")); + if (!KNOWN_TRIGGERS.contains(type)) { + issues.add("unknown trigger type '" + type + "' (expected one of " + KNOWN_TRIGGERS + ")"); + } + } + Object steps = root.get("steps"); + if (!(steps instanceof List) || ((List) steps).isEmpty()) { + issues.add("'steps' must be a non-empty array"); + } else { + for (Object s : (List) steps) { + if (!(s instanceof Map)) { issues.add("each step must be an object"); continue; } + String ex = String.valueOf(((Map) s).get("executor")); + if (!KNOWN_EXECUTORS.contains(ex)) { + issues.add("step executor '" + ex + "' unknown (expected " + KNOWN_EXECUTORS + ")"); + } + if ("python".equals(ex)) { + issues.add("python executor needs a sandbox and is not enabled in this build"); + } + } + } + return issues; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/pipeline/WikiPipelineTriggerListener.java b/mateclaw-server/src/main/java/vip/mate/wiki/pipeline/WikiPipelineTriggerListener.java index cf9d250e..945b637e 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/pipeline/WikiPipelineTriggerListener.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/pipeline/WikiPipelineTriggerListener.java @@ -30,6 +30,7 @@ public class WikiPipelineTriggerListener { public void onPageCreated(WikiPageCreatedEvent event) { try { triggerService.onPageTypeCount(event.kbId(), event.pageType()); + triggerService.onPageCreated(event.kbId(), event.pageType(), event.pageId()); } catch (Exception e) { log.warn("[WikiPipeline] trigger evaluation failed for kb={} pageType={}: {}", event.kbId(), event.pageType(), e.getMessage()); diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/pipeline/WikiPipelineTriggerService.java b/mateclaw-server/src/main/java/vip/mate/wiki/pipeline/WikiPipelineTriggerService.java index 4f29ba73..6df3fbf1 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/pipeline/WikiPipelineTriggerService.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/pipeline/WikiPipelineTriggerService.java @@ -49,6 +49,36 @@ public class WikiPipelineTriggerService { * number of runs actually started (0 when no threshold bucket was newly * crossed). Safe to call after every page create. */ + /** + * Fire {@code page_created} definitions once per matching page creation + * (deduped by page id). Optional {@code page_type} in the trigger config + * narrows which page types fire. Returns the number of runs started. + */ + public int onPageCreated(Long kbId, String pageType, Long pageId) { + if (kbId == null || pageType == null || pageId == null) { + return 0; + } + List defs = definitionMapper.selectList( + new LambdaQueryWrapper() + .eq(WikiPipelineDefinitionEntity::getKbId, kbId) + .eq(WikiPipelineDefinitionEntity::getTriggerType, "page_created") + .eq(WikiPipelineDefinitionEntity::getEnabled, 1)); + int started = 0; + for (WikiPipelineDefinitionEntity def : defs) { + TriggerConfig cfg = parseConfig(def.getTriggerConfigJson()); + if (cfg != null && cfg.pageType != null && !pageType.equalsIgnoreCase(cfg.pageType)) { + continue; // type filter set and doesn't match + } + String input = "{\"pageType\":\"" + pageType + "\",\"pageId\":\"" + pageId + "\"}"; + WikiPipelineService.RunOutcome outcome = + pipelineService.execute(def, pageType, "page:" + pageId, input); + if (!outcome.duplicate() && outcome.run() != null) { + started++; + } + } + return started; + } + public int onPageTypeCount(Long kbId, String pageType) { if (kbId == null || pageType == null || pageType.isBlank()) { return 0; diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java index ae26c78e..ab1ef38f 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java @@ -1508,7 +1508,7 @@ public class WikiProcessingService { // transactions), so the count is accurate. Idempotent + dedup-guarded // downstream, so firing on update paths is safe. if (eventPublisher != null && pageType != null && !pageType.isBlank()) { - eventPublisher.publishEvent(new vip.mate.wiki.event.WikiPageCreatedEvent(kbId, pageType)); + eventPublisher.publishEvent(new vip.mate.wiki.event.WikiPageCreatedEvent(kbId, pageType, pageId)); } // When an existing fact page is updated, propagate staleness to the // experience pages depending on it (async, off the ingest thread). diff --git a/mateclaw-server/src/test/java/vip/mate/wiki/pipeline/WikiPipelineDefinitionServiceE2ETest.java b/mateclaw-server/src/test/java/vip/mate/wiki/pipeline/WikiPipelineDefinitionServiceE2ETest.java new file mode 100644 index 00000000..6032b517 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/wiki/pipeline/WikiPipelineDefinitionServiceE2ETest.java @@ -0,0 +1,103 @@ +package vip.mate.wiki.pipeline; + +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import vip.mate.wiki.model.WikiPipelineDefinitionEntity; + +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; + +/** + * E2E for {@link WikiPipelineDefinitionService}: YAML/JSON parsing, structural + * validation, and upsert/list/delete against H2. + */ +@SpringBootTest( + webEnvironment = SpringBootTest.WebEnvironment.NONE, + properties = { + "spring.flyway.enabled=true", + "spring.flyway.locations=classpath:db/migration/h2", + "mateclaw.feature-flag.refresh-ms=999999" + } +) +class WikiPipelineDefinitionServiceE2ETest { + + @Autowired + private WikiPipelineDefinitionService service; + + private static final java.util.concurrent.atomic.AtomicLong SEQ = + new java.util.concurrent.atomic.AtomicLong(System.nanoTime()); + + private static final String YAML = """ + name: episode-to-pattern + owner_agent: 2055137662148763649 + trigger: + type: page_type_count + page_type: episode + threshold: 20 + dedup_window_seconds: 3600 + steps: + - id: summarize + executor: llm + prompt: pattern-analysis + - id: enrich + executor: skill + skill: wiki-link-enrich + """; + + @Test + void parsesYamlAndUpserts() { + long kb = SEQ.incrementAndGet(); + WikiPipelineDefinitionEntity def = service.saveFromConfig(kb, YAML, true); + assertNotNull(def.getId()); + assertEquals("episode-to-pattern", def.getName()); + assertEquals(2055137662148763649L, def.getOwnerAgentId()); + assertEquals("page_type_count", def.getTriggerType()); + assertEquals(3600, def.getDedupWindowSeconds()); + assertTrue(def.getTriggerConfigJson().contains("episode")); + assertTrue(def.getStepsJson().contains("wiki-link-enrich")); + + // upsert: same name → update in place, not a second row + service.saveFromConfig(kb, YAML.replace("threshold: 20", "threshold: 40"), true); + List all = service.list(kb); + assertEquals(1, all.size()); + assertTrue(all.get(0).getTriggerConfigJson().contains("40")); + + service.delete(def.getId()); + assertNull(service.get(def.getId())); + } + + @Test + void parsesJson() { + long kb = SEQ.incrementAndGet(); + String json = "{\"name\":\"p\",\"owner_agent\":42,\"trigger\":{\"type\":\"page_created\"}," + + "\"steps\":[{\"id\":\"s\",\"executor\":\"llm\",\"prompt\":\"x\"}]}"; + WikiPipelineDefinitionEntity def = service.saveFromConfig(kb, json, false); + assertEquals("page_created", def.getTriggerType()); + } + + @Test + void validation_reportsIssues() { + assertTrue(service.validateConfig(YAML, true).isEmpty()); + + // missing name + unknown trigger + python step + String bad = """ + owner_agent: 1 + trigger: + type: nope + steps: + - id: x + executor: python + """; + List issues = service.validateConfig(bad, true); + assertFalse(issues.isEmpty()); + assertTrue(issues.stream().anyMatch(s -> s.contains("name"))); + assertTrue(issues.stream().anyMatch(s -> s.contains("trigger type"))); + assertTrue(issues.stream().anyMatch(s -> s.contains("python"))); + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/wiki/pipeline/WikiPipelineTriggerListenerTest.java b/mateclaw-server/src/test/java/vip/mate/wiki/pipeline/WikiPipelineTriggerListenerTest.java index d663ad2a..be9efbd8 100644 --- a/mateclaw-server/src/test/java/vip/mate/wiki/pipeline/WikiPipelineTriggerListenerTest.java +++ b/mateclaw-server/src/test/java/vip/mate/wiki/pipeline/WikiPipelineTriggerListenerTest.java @@ -20,7 +20,7 @@ class WikiPipelineTriggerListenerTest { WikiPipelineTriggerService trigger = mock(WikiPipelineTriggerService.class); when(trigger.onPageTypeCount(7L, "episode")).thenReturn(1); - new WikiPipelineTriggerListener(trigger).onPageCreated(new WikiPageCreatedEvent(7L, "episode")); + new WikiPipelineTriggerListener(trigger).onPageCreated(new WikiPageCreatedEvent(7L, "episode", 100L)); verify(trigger).onPageTypeCount(7L, "episode"); } @@ -31,6 +31,6 @@ class WikiPipelineTriggerListenerTest { doThrow(new RuntimeException("boom")).when(trigger).onPageTypeCount(7L, "episode"); // Must not throw — ingest must be unaffected by a pipeline failure. - new WikiPipelineTriggerListener(trigger).onPageCreated(new WikiPageCreatedEvent(7L, "episode")); + new WikiPipelineTriggerListener(trigger).onPageCreated(new WikiPageCreatedEvent(7L, "episode", 100L)); } }