From c3388c3f1e1665d2a69cd47890d65d955316ef18 Mon Sep 17 00:00:00 2001 From: matevip Date: Sun, 31 May 2026 07:56:17 +0800 Subject: [PATCH] =?UTF-8?q?feat(wiki):=20pipeline=20runtime=20schema=20?= =?UTF-8?q?=E2=80=94=20definitions,=20runs,=20step=20runs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../model/WikiPipelineDefinitionEntity.java | 55 +++++++++++++++ .../wiki/model/WikiPipelineRunEntity.java | 59 ++++++++++++++++ .../wiki/model/WikiPipelineStepRunEntity.java | 51 ++++++++++++++ .../WikiPipelineDefinitionMapper.java | 14 ++++ .../repository/WikiPipelineRunMapper.java | 14 ++++ .../repository/WikiPipelineStepRunMapper.java | 14 ++++ .../h2/V136__wiki_pipeline_runtime.sql | 68 +++++++++++++++++++ .../mysql/V136__wiki_pipeline_runtime.sql | 55 +++++++++++++++ .../WikiPipelineRunMapperE2ETest.java | 64 +++++++++++++++++ 9 files changed, 394 insertions(+) create mode 100644 mateclaw-server/src/main/java/vip/mate/wiki/model/WikiPipelineDefinitionEntity.java create mode 100644 mateclaw-server/src/main/java/vip/mate/wiki/model/WikiPipelineRunEntity.java create mode 100644 mateclaw-server/src/main/java/vip/mate/wiki/model/WikiPipelineStepRunEntity.java create mode 100644 mateclaw-server/src/main/java/vip/mate/wiki/repository/WikiPipelineDefinitionMapper.java create mode 100644 mateclaw-server/src/main/java/vip/mate/wiki/repository/WikiPipelineRunMapper.java create mode 100644 mateclaw-server/src/main/java/vip/mate/wiki/repository/WikiPipelineStepRunMapper.java create mode 100644 mateclaw-server/src/main/resources/db/migration/h2/V136__wiki_pipeline_runtime.sql create mode 100644 mateclaw-server/src/main/resources/db/migration/mysql/V136__wiki_pipeline_runtime.sql create mode 100644 mateclaw-server/src/test/java/vip/mate/wiki/repository/WikiPipelineRunMapperE2ETest.java diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/model/WikiPipelineDefinitionEntity.java b/mateclaw-server/src/main/java/vip/mate/wiki/model/WikiPipelineDefinitionEntity.java new file mode 100644 index 00000000..62dc57e8 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/wiki/model/WikiPipelineDefinitionEntity.java @@ -0,0 +1,55 @@ +package vip.mate.wiki.model; + +import com.baomidou.mybatisplus.annotation.FieldFill; +import com.baomidou.mybatisplus.annotation.IdType; +import com.baomidou.mybatisplus.annotation.TableField; +import com.baomidou.mybatisplus.annotation.TableId; +import com.baomidou.mybatisplus.annotation.TableLogic; +import com.baomidou.mybatisplus.annotation.TableName; +import lombok.Data; + +import java.time.LocalDateTime; + +/** + * A KB-scoped pipeline definition: a processing chain triggered by a pageType + * event, executed under a concrete owner agent's permissions. + * + * @author MateClaw Team + */ +@Data +@TableName("mate_wiki_pipeline_definition") +public class WikiPipelineDefinitionEntity { + + @TableId(type = IdType.ASSIGN_ID) + private Long id; + + private Long kbId; + + private String name; + + /** Agent whose identity (and RFC permissions) the steps run under. */ + private Long ownerAgentId; + + /** Trigger kind, e.g. {@code page_type_count}. */ + private String triggerType; + + /** Trigger configuration as JSON (e.g. page_type + threshold). */ + private String triggerConfigJson; + + /** Ordered step definitions as JSON. */ + private String stepsJson; + + /** Window (seconds) within which duplicate triggers collapse to one run. */ + private Integer dedupWindowSeconds; + + private Integer enabled; + + @TableField(fill = FieldFill.INSERT) + private LocalDateTime createTime; + + @TableField(fill = FieldFill.INSERT_UPDATE) + private LocalDateTime updateTime; + + @TableLogic + private Integer deleted; +} diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/model/WikiPipelineRunEntity.java b/mateclaw-server/src/main/java/vip/mate/wiki/model/WikiPipelineRunEntity.java new file mode 100644 index 00000000..5e6f85b0 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/wiki/model/WikiPipelineRunEntity.java @@ -0,0 +1,59 @@ +package vip.mate.wiki.model; + +import com.baomidou.mybatisplus.annotation.FieldFill; +import com.baomidou.mybatisplus.annotation.IdType; +import com.baomidou.mybatisplus.annotation.TableField; +import com.baomidou.mybatisplus.annotation.TableId; +import com.baomidou.mybatisplus.annotation.TableLogic; +import com.baomidou.mybatisplus.annotation.TableName; +import lombok.Data; + +import java.time.LocalDateTime; + +/** + * One execution instance of a pipeline definition. The + * (definition_id, trigger_type, trigger_subject, trigger_bucket) unique key + * makes duplicate triggers idempotent across instances. + * + * @author MateClaw Team + */ +@Data +@TableName("mate_wiki_pipeline_run") +public class WikiPipelineRunEntity { + + @TableId(type = IdType.ASSIGN_ID) + private Long id; + + private Long definitionId; + + private Long kbId; + + /** {@code pending} / {@code running} / {@code succeeded} / {@code failed}. */ + private String status; + + private String triggerType; + + /** The entity the trigger fired on, e.g. a pageType name. */ + private String triggerSubject; + + /** Dedup envelope (time/threshold bucket) for idempotency. */ + private String triggerBucket; + + private String triggerPayloadJson; + + private String inputJson; + + private String outputJson; + + private String errorMessage; + + private LocalDateTime startedAt; + + private LocalDateTime finishedAt; + + @TableField(fill = FieldFill.INSERT) + private LocalDateTime createTime; + + @TableLogic + private Integer deleted; +} diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/model/WikiPipelineStepRunEntity.java b/mateclaw-server/src/main/java/vip/mate/wiki/model/WikiPipelineStepRunEntity.java new file mode 100644 index 00000000..636f65ab --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/wiki/model/WikiPipelineStepRunEntity.java @@ -0,0 +1,51 @@ +package vip.mate.wiki.model; + +import com.baomidou.mybatisplus.annotation.FieldFill; +import com.baomidou.mybatisplus.annotation.IdType; +import com.baomidou.mybatisplus.annotation.TableField; +import com.baomidou.mybatisplus.annotation.TableId; +import com.baomidou.mybatisplus.annotation.TableLogic; +import com.baomidou.mybatisplus.annotation.TableName; +import lombok.Data; + +import java.time.LocalDateTime; + +/** + * One step invocation within a {@link WikiPipelineRunEntity}. + * + * @author MateClaw Team + */ +@Data +@TableName("mate_wiki_pipeline_step_run") +public class WikiPipelineStepRunEntity { + + @TableId(type = IdType.ASSIGN_ID) + private Long id; + + private Long runId; + + /** Step id from the definition's steps_json. */ + private String stepId; + + /** {@code llm} / {@code skill} (Python is out of MVP). */ + private String executor; + + /** {@code pending} / {@code running} / {@code succeeded} / {@code failed}. */ + private String status; + + private String inputJson; + + private String outputJson; + + private String errorMessage; + + private LocalDateTime startedAt; + + private LocalDateTime finishedAt; + + @TableField(fill = FieldFill.INSERT) + private LocalDateTime createTime; + + @TableLogic + private Integer deleted; +} diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/repository/WikiPipelineDefinitionMapper.java b/mateclaw-server/src/main/java/vip/mate/wiki/repository/WikiPipelineDefinitionMapper.java new file mode 100644 index 00000000..3c46012a --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/wiki/repository/WikiPipelineDefinitionMapper.java @@ -0,0 +1,14 @@ +package vip.mate.wiki.repository; + +import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import org.apache.ibatis.annotations.Mapper; +import vip.mate.wiki.model.WikiPipelineDefinitionEntity; + +/** + * Mapper for {@link WikiPipelineDefinitionEntity}. + * + * @author MateClaw Team + */ +@Mapper +public interface WikiPipelineDefinitionMapper extends BaseMapper { +} diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/repository/WikiPipelineRunMapper.java b/mateclaw-server/src/main/java/vip/mate/wiki/repository/WikiPipelineRunMapper.java new file mode 100644 index 00000000..f23c9dd8 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/wiki/repository/WikiPipelineRunMapper.java @@ -0,0 +1,14 @@ +package vip.mate.wiki.repository; + +import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import org.apache.ibatis.annotations.Mapper; +import vip.mate.wiki.model.WikiPipelineRunEntity; + +/** + * Mapper for {@link WikiPipelineRunEntity}. + * + * @author MateClaw Team + */ +@Mapper +public interface WikiPipelineRunMapper extends BaseMapper { +} diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/repository/WikiPipelineStepRunMapper.java b/mateclaw-server/src/main/java/vip/mate/wiki/repository/WikiPipelineStepRunMapper.java new file mode 100644 index 00000000..a8f0b34e --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/wiki/repository/WikiPipelineStepRunMapper.java @@ -0,0 +1,14 @@ +package vip.mate.wiki.repository; + +import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import org.apache.ibatis.annotations.Mapper; +import vip.mate.wiki.model.WikiPipelineStepRunEntity; + +/** + * Mapper for {@link WikiPipelineStepRunEntity}. + * + * @author MateClaw Team + */ +@Mapper +public interface WikiPipelineStepRunMapper extends BaseMapper { +} diff --git a/mateclaw-server/src/main/resources/db/migration/h2/V136__wiki_pipeline_runtime.sql b/mateclaw-server/src/main/resources/db/migration/h2/V136__wiki_pipeline_runtime.sql new file mode 100644 index 00000000..721ecf0e --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/h2/V136__wiki_pipeline_runtime.sql @@ -0,0 +1,68 @@ +-- V136: Wiki pipeline runtime — definitions, runs, and per-step runs. +-- +-- A definition is a KB-scoped processing chain triggered by a pageType event +-- (MVP: page_type_count threshold). It declares an owner_agent_id so steps run +-- under a concrete RFC-permissioned identity. A run is one execution instance; +-- its unique key (definition_id, trigger_type, trigger_subject, trigger_bucket) +-- absorbs duplicate triggers across multiple instances. Step runs record each +-- executor invocation (MVP executors: llm, skill). + +CREATE TABLE IF NOT EXISTS mate_wiki_pipeline_definition ( + id BIGINT NOT NULL PRIMARY KEY, + kb_id BIGINT NOT NULL, + name VARCHAR(128) NOT NULL, + owner_agent_id BIGINT NOT NULL, + trigger_type VARCHAR(32) NOT NULL, + trigger_config_json CLOB, + steps_json CLOB NOT NULL, + dedup_window_seconds INT NOT NULL DEFAULT 0, + enabled TINYINT NOT NULL DEFAULT 1, + create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + update_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + deleted INT NOT NULL DEFAULT 0 +); +CREATE UNIQUE INDEX IF NOT EXISTS uk_wiki_pipeline_def_name + ON mate_wiki_pipeline_definition (kb_id, name, deleted); +CREATE INDEX IF NOT EXISTS idx_wiki_pipeline_def_trigger + ON mate_wiki_pipeline_definition (kb_id, trigger_type, enabled, deleted); + +CREATE TABLE IF NOT EXISTS mate_wiki_pipeline_run ( + id BIGINT NOT NULL PRIMARY KEY, + definition_id BIGINT NOT NULL, + kb_id BIGINT NOT NULL, + status VARCHAR(16) NOT NULL, + trigger_type VARCHAR(32) NOT NULL, + trigger_subject VARCHAR(128) NOT NULL, + trigger_bucket VARCHAR(64) NOT NULL, + trigger_payload_json CLOB, + input_json CLOB, + output_json CLOB, + error_message VARCHAR(2048), + started_at TIMESTAMP, + finished_at TIMESTAMP, + create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + deleted INT NOT NULL DEFAULT 0 +); +-- Idempotency: one run per (definition, trigger envelope). Duplicate triggers +-- across instances collide here instead of spawning parallel runs. +CREATE UNIQUE INDEX IF NOT EXISTS uk_wiki_pipeline_run_dedup + ON mate_wiki_pipeline_run (definition_id, trigger_type, trigger_subject, trigger_bucket, deleted); +CREATE INDEX IF NOT EXISTS idx_wiki_pipeline_run_def + ON mate_wiki_pipeline_run (definition_id, status); + +CREATE TABLE IF NOT EXISTS mate_wiki_pipeline_step_run ( + id BIGINT NOT NULL PRIMARY KEY, + run_id BIGINT NOT NULL, + step_id VARCHAR(128) NOT NULL, + executor VARCHAR(32) NOT NULL, + status VARCHAR(16) NOT NULL, + input_json CLOB, + output_json CLOB, + error_message VARCHAR(2048), + started_at TIMESTAMP, + finished_at TIMESTAMP, + create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + deleted INT NOT NULL DEFAULT 0 +); +CREATE INDEX IF NOT EXISTS idx_wiki_pipeline_step_run + ON mate_wiki_pipeline_step_run (run_id, status); diff --git a/mateclaw-server/src/main/resources/db/migration/mysql/V136__wiki_pipeline_runtime.sql b/mateclaw-server/src/main/resources/db/migration/mysql/V136__wiki_pipeline_runtime.sql new file mode 100644 index 00000000..eb32f1fc --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/mysql/V136__wiki_pipeline_runtime.sql @@ -0,0 +1,55 @@ +-- V136: Wiki pipeline runtime — definitions, runs, and per-step runs. +-- See the H2 file for design rationale. + +CREATE TABLE IF NOT EXISTS mate_wiki_pipeline_definition ( + id BIGINT NOT NULL PRIMARY KEY, + kb_id BIGINT NOT NULL, + name VARCHAR(128) NOT NULL, + owner_agent_id BIGINT NOT NULL, + trigger_type VARCHAR(32) NOT NULL, + trigger_config_json LONGTEXT, + steps_json LONGTEXT NOT NULL, + dedup_window_seconds INT NOT NULL DEFAULT 0, + enabled TINYINT NOT NULL DEFAULT 1, + create_time DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + update_time DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + deleted INT NOT NULL DEFAULT 0, + UNIQUE KEY uk_wiki_pipeline_def_name (kb_id, name, deleted), + KEY idx_wiki_pipeline_def_trigger (kb_id, trigger_type, enabled, deleted) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + +CREATE TABLE IF NOT EXISTS mate_wiki_pipeline_run ( + id BIGINT NOT NULL PRIMARY KEY, + definition_id BIGINT NOT NULL, + kb_id BIGINT NOT NULL, + status VARCHAR(16) NOT NULL, + trigger_type VARCHAR(32) NOT NULL, + trigger_subject VARCHAR(128) NOT NULL, + trigger_bucket VARCHAR(64) NOT NULL, + trigger_payload_json LONGTEXT, + input_json LONGTEXT, + output_json LONGTEXT, + error_message VARCHAR(2048), + started_at DATETIME(3), + finished_at DATETIME(3), + create_time DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + deleted INT NOT NULL DEFAULT 0, + UNIQUE KEY uk_wiki_pipeline_run_dedup (definition_id, trigger_type, trigger_subject, trigger_bucket, deleted), + KEY idx_wiki_pipeline_run_def (definition_id, status) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + +CREATE TABLE IF NOT EXISTS mate_wiki_pipeline_step_run ( + id BIGINT NOT NULL PRIMARY KEY, + run_id BIGINT NOT NULL, + step_id VARCHAR(128) NOT NULL, + executor VARCHAR(32) NOT NULL, + status VARCHAR(16) NOT NULL, + input_json LONGTEXT, + output_json LONGTEXT, + error_message VARCHAR(2048), + started_at DATETIME(3), + finished_at DATETIME(3), + create_time DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), + deleted INT NOT NULL DEFAULT 0, + KEY idx_wiki_pipeline_step_run (run_id, status) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; diff --git a/mateclaw-server/src/test/java/vip/mate/wiki/repository/WikiPipelineRunMapperE2ETest.java b/mateclaw-server/src/test/java/vip/mate/wiki/repository/WikiPipelineRunMapperE2ETest.java new file mode 100644 index 00000000..cd833706 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/wiki/repository/WikiPipelineRunMapperE2ETest.java @@ -0,0 +1,64 @@ +package vip.mate.wiki.repository; + +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.test.annotation.DirtiesContext; +import vip.mate.wiki.model.WikiPipelineRunEntity; + +import java.time.LocalDateTime; + +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** + * Validates the pipeline run dedup invariant against H2: two runs sharing the + * same (definition, trigger envelope) collide on the unique key, so duplicate + * triggers cannot spawn parallel runs. + */ +@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 WikiPipelineRunMapperE2ETest { + + @Autowired + private WikiPipelineRunMapper mapper; + + private static final java.util.concurrent.atomic.AtomicLong SEQ = + new java.util.concurrent.atomic.AtomicLong(System.nanoTime()); + + private WikiPipelineRunEntity run(long defId, String bucket) { + WikiPipelineRunEntity r = new WikiPipelineRunEntity(); + r.setDefinitionId(defId); + r.setKbId(1L); + r.setStatus("pending"); + r.setTriggerType("page_type_count"); + r.setTriggerSubject("episode"); + r.setTriggerBucket(bucket); + r.setCreateTime(LocalDateTime.now()); + return r; + } + + @Test + void duplicateTriggerEnvelope_isRejected() { + long defId = SEQ.incrementAndGet(); + WikiPipelineRunEntity first = run(defId, "20"); + mapper.insert(first); + assertNotNull(first.getId()); + + assertThrows(Exception.class, () -> mapper.insert(run(defId, "20"))); + } + + @Test + void differentBucket_isAllowed() { + long defId = SEQ.incrementAndGet(); + mapper.insert(run(defId, "20")); + mapper.insert(run(defId, "40")); // next threshold bucket — distinct run + } +}