mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
feat(wiki): pipeline runtime schema — definitions, runs, step runs
This commit is contained in:
parent
66e4788226
commit
c3388c3f1e
@ -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;
|
||||
}
|
||||
@ -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;
|
||||
}
|
||||
@ -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;
|
||||
}
|
||||
@ -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<WikiPipelineDefinitionEntity> {
|
||||
}
|
||||
@ -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<WikiPipelineRunEntity> {
|
||||
}
|
||||
@ -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<WikiPipelineStepRunEntity> {
|
||||
}
|
||||
@ -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);
|
||||
@ -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;
|
||||
@ -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
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user