feat: Phase1 核心——模型/状态机 + SQLite Store(守卫+审批闸) + REST/WS API + daemon

- src/model: 复杂度分级、16 状态状态机、实体类型
- src/store: better-sqlite3 接 schema,transition 受 canTransition 守卫,
  decide 审批闸(reject 必带改进意见),事件订阅广播,nextExecutable
- src/api: Fastify REST + ws 事件广播(/ws)
- src/daemon: maestrod 入口(env 配置,默认 ~/.maestro :4517)
- test: 9 个生命周期单测全过;typecheck/build 干净;REST+WS 端到端实跑验证

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
wangjia
2026-06-12 20:44:45 +08:00
commit 3172718a94
18 changed files with 2757 additions and 0 deletions
+160
View File
@@ -0,0 +1,160 @@
import Fastify, { type FastifyInstance } from 'fastify';
import { WebSocketServer, type WebSocket } from 'ws';
import { Store, StoreError } from '../store/index.js';
import type { Complexity } from '../model/complexity.js';
import { isComplexity } from '../model/complexity.js';
import type { TaskStatus } from '../model/status.js';
export interface ApiOptions {
store: Store;
logger?: boolean;
}
/**
* REST + WebSocket API。所有写操作走 Store(带状态机守卫 / 审批闸)。
* Store 的事件经 subscribe 广播到所有 WS 客户端,驱动看板实时刷新。
*/
export function buildServer(opts: ApiOptions): FastifyInstance {
const { store } = opts;
const app = Fastify({ logger: opts.logger ?? false });
// Store 错误 → 400(业务校验),其余 → 500
app.setErrorHandler((err, _req, reply) => {
if (err instanceof StoreError) return reply.code(400).send({ error: err.message });
app.log.error(err);
return reply.code(500).send({ error: err.message ?? 'internal error' });
});
// ---------- Projects ----------
app.get('/api/projects', () => store.listProjects());
app.post('/api/projects', (req) => {
const b = req.body as Record<string, unknown>;
if (!b?.name || !b?.repoPath) throw new StoreError('name 与 repoPath 必填');
return store.createProject({
name: String(b.name), repoPath: String(b.repoPath),
defaultBranch: b.defaultBranch ? String(b.defaultBranch) : undefined,
verifyCmd: b.verifyCmd === undefined ? undefined : (b.verifyCmd === null ? null : String(b.verifyCmd)),
autonomy: b.autonomy as never, model: b.model === undefined ? undefined : (b.model === null ? null : String(b.model)),
concurrency: b.concurrency === undefined ? undefined : Number(b.concurrency),
});
});
app.get('/api/projects/:id', (req) => {
const { id } = req.params as { id: string };
const p = store.getProject(id);
if (!p) throw new StoreError(`项目不存在: ${id}`);
return p;
});
app.get('/api/projects/:id/tasks', (req) => {
const { id } = req.params as { id: string };
return store.listTasks(id);
});
app.get('/api/projects/:id/events', (req) => {
const { id } = req.params as { id: string };
return store.listEvents(id);
});
app.get('/api/projects/:id/next', (req) => {
const { id } = req.params as { id: string };
return store.nextExecutable(id) ?? { next: null };
});
// ---------- Tasks ----------
app.post('/api/projects/:id/tasks', (req) => {
const { id } = req.params as { id: string };
const b = req.body as Record<string, unknown>;
if (!b?.title) throw new StoreError('title 必填');
if (!isComplexity(b.complexity)) throw new StoreError('complexity 必须是 hard|medium|easy');
return store.createTask({
projectId: id, title: String(b.title), complexity: b.complexity as Complexity,
parentId: b.parentId ? String(b.parentId) : null,
priority: b.priority === undefined ? undefined : Number(b.priority),
deps: Array.isArray(b.deps) ? (b.deps as string[]) : undefined,
});
});
app.get('/api/tasks/:id', (req) => {
const { id } = req.params as { id: string };
const t = store.getTask(id);
if (!t) throw new StoreError(`任务不存在: ${id}`);
return t;
});
app.get('/api/tasks/:id/children', (req) => {
const { id } = req.params as { id: string };
return store.childrenOf(id);
});
app.get('/api/tasks/:id/runs', (req) => {
const { id } = req.params as { id: string };
return store.listRuns(id);
});
app.post('/api/tasks/:id/plan', (req) => {
const { id } = req.params as { id: string };
const b = req.body as { plan?: string };
if (!b?.plan) throw new StoreError('plan 必填');
return store.setPlan(id, b.plan);
});
app.post('/api/tasks/:id/spec', (req) => {
const { id } = req.params as { id: string };
const b = req.body as { spec?: string };
if (!b?.spec) throw new StoreError('spec 必填');
return store.setSpec(id, b.spec);
});
app.post('/api/tasks/:id/operations', (req) => {
const { id } = req.params as { id: string };
const b = req.body as { operations?: string };
if (!b?.operations) throw new StoreError('operations 必填');
return store.setOperations(id, b.operations);
});
app.post('/api/tasks/:id/transition', (req) => {
const { id } = req.params as { id: string };
const b = req.body as { to?: string; meta?: Record<string, unknown> };
if (!b?.to) throw new StoreError('to 必填');
return store.transition(id, b.to as TaskStatus, b.meta ?? {});
});
// 审批闸:accept / rejectreject 必带 reason
app.post('/api/tasks/:id/decide', (req) => {
const { id } = req.params as { id: string };
const b = req.body as { action?: string; actor?: string; reason?: string | null };
if (b?.action !== 'accept' && b?.action !== 'reject') throw new StoreError('action 必须是 accept|reject');
return store.decide(id, b.action, b.actor ?? 'user', b.reason ?? null);
});
// ---------- Approvals ----------
app.get('/api/approvals', (req) => {
const q = req.query as { projectId?: string };
return store.pendingApprovals(q.projectId);
});
return app;
}
/** 在 Fastify 的底层 http server 上挂 WebSocket,把 Store 事件广播给所有客户端。 */
export function attachWebSocket(app: FastifyInstance, store: Store): WebSocketServer {
const wss = new WebSocketServer({ server: app.server, path: '/ws' });
const clients = new Set<WebSocket>();
wss.on('connection', (ws) => {
clients.add(ws);
ws.on('close', () => clients.delete(ws));
ws.on('error', () => clients.delete(ws));
});
store.subscribe((evt) => {
const msg = JSON.stringify(evt);
for (const ws of clients) {
if (ws.readyState === ws.OPEN) ws.send(msg);
}
});
return wss;
}
+22
View File
@@ -0,0 +1,22 @@
import { homedir } from 'node:os';
import { join } from 'node:path';
import { mkdirSync } from 'node:fs';
export interface DaemonConfig {
dataDir: string; // 数据库 + run 转录日志所在目录
dbFile: string;
host: string;
port: number;
}
/** 解析配置:环境变量覆盖默认值。首次访问时确保数据目录存在。 */
export function loadConfig(): DaemonConfig {
const dataDir = process.env.MAESTRO_DATA_DIR ?? join(homedir(), '.maestro');
mkdirSync(dataDir, { recursive: true });
return {
dataDir,
dbFile: join(dataDir, 'maestro.sqlite'),
host: process.env.MAESTRO_HOST ?? '127.0.0.1',
port: Number(process.env.MAESTRO_PORT ?? 4517),
};
}
+29
View File
@@ -0,0 +1,29 @@
import { Store } from '../store/index.js';
import { buildServer, attachWebSocket } from '../api/server.js';
import { loadConfig } from './config.js';
/** maestrod:核心 daemon。Phase 1 = Store + REST/WS API(手动驱动;编排器在 Phase 2 接入)。 */
async function main(): Promise<void> {
const cfg = loadConfig();
const store = new Store(cfg.dbFile);
const app = buildServer({ store, logger: true });
attachWebSocket(app, store);
await app.listen({ host: cfg.host, port: cfg.port });
app.log.info(`maestrod 就绪 · db=${cfg.dbFile} · http://${cfg.host}:${cfg.port} · ws ${cfg.host}:${cfg.port}/ws`);
const shutdown = async (): Promise<void> => {
app.log.info('收到退出信号,关闭中…');
await app.close();
store.close();
process.exit(0);
};
process.on('SIGINT', shutdown);
process.on('SIGTERM', shutdown);
}
main().catch((err) => {
console.error('maestrod 启动失败:', err);
process.exit(1);
});
+23
View File
@@ -0,0 +1,23 @@
/** 任务复杂度:Hard / Medium / Easy(对应旧 todo skill 的 tier 1/2/3 */
export type Complexity = 'hard' | 'medium' | 'easy';
export const COMPLEXITY_ORDER: Record<Complexity, number> = { hard: 1, medium: 2, easy: 3 };
export const COMPLEXITY_LABEL: Record<Complexity, string> = { hard: 'Hard', medium: 'Medium', easy: 'Easy' };
/** 旧 todo skill 的 tier(1/2/3) ↔ 复杂度,用于迁移导入 */
export const TIER_TO_COMPLEXITY: Record<number, Complexity> = { 1: 'hard', 2: 'medium', 3: 'easy' };
export const COMPLEXITY_TO_TIER: Record<Complexity, number> = { hard: 1, medium: 2, easy: 3 };
/** 前置闸:执行前是否需要用户确认(Hard / Medium 需要,Easy 不需要) */
export function requiresPreGate(c: Complexity): boolean {
return c === 'hard' || c === 'medium';
}
/** Hard 必须先进 plan 分析 + 任务拆解 */
export function mustDecompose(c: Complexity): boolean {
return c === 'hard';
}
export function isComplexity(v: unknown): v is Complexity {
return v === 'hard' || v === 'medium' || v === 'easy';
}
+95
View File
@@ -0,0 +1,95 @@
import type { Complexity } from './complexity.js';
/** 任务执行状态机(设计见 DESIGN.md §5 */
export type TaskStatus =
| 'init' // 新建,未分析 / 未定方案
| 'analyzing' // Hardplan 分析 + 拆解中
| 'plan_review' // Hard)拆解完成,待 accept
| 'decomposed' // (Hard)已拆解,作为容器,进度由子任务汇总
| 'speccing' // Medium)写方案中
| 'spec_review' // Medium)方案完成,待 accept
| 'ready' // 可执行叶子
| 'blocked' // 依赖未满足
| 'queued' // 编排器已领取,等 worker 槽位
| 'executing' // agent 在 worktree 跑 headless Claude Code
| 'exec_review' // 执行完成,改动在分支,待审 / 合
| 'failed' // 执行失败
| 'needs_attention' // 失败超阈值 / 冲突 / 歧义,需人工
| 'done' // 已合并 / 已接受
| 'paused'
| 'cancelled';
export const STATUS_LABEL: Record<TaskStatus, string> = {
init: '新建',
analyzing: '分析拆解中',
plan_review: '待确认拆解',
decomposed: '已拆解',
speccing: '写方案中',
spec_review: '待确认方案',
ready: '可执行',
blocked: '被依赖阻塞',
queued: '排队中',
executing: '执行中',
exec_review: '待审/合',
failed: '失败',
needs_attention: '需人工',
done: '完成',
paused: '暂停',
cancelled: '取消',
};
/** 审批闸类型 */
export type GateKind = 'plan' | 'spec' | 'exec';
/** 哪些状态在等待用户审批,以及对应的闸类型 */
export const GATE_STATUS: Partial<Record<TaskStatus, GateKind>> = {
plan_review: 'plan',
spec_review: 'spec',
exec_review: 'exec',
};
export function gateOf(s: TaskStatus): GateKind | null {
return GATE_STATUS[s] ?? null;
}
export const TERMINAL_STATUS: ReadonlySet<TaskStatus> = new Set<TaskStatus>(['done', 'cancelled']);
/** 合法状态流转表:from → 允许的 to[] */
export const TRANSITIONS: Record<TaskStatus, TaskStatus[]> = {
init: ['analyzing', 'speccing', 'ready', 'cancelled'], // 按复杂度分流
analyzing: ['plan_review', 'cancelled', 'paused'],
plan_review: ['decomposed', 'analyzing', 'cancelled'], // accept / reject(+意见)
decomposed: ['done', 'analyzing', 'cancelled', 'paused'], // 子全 done / 重拆
speccing: ['spec_review', 'cancelled', 'paused'],
spec_review: ['ready', 'speccing', 'cancelled'], // accept / reject(+意见)
ready: ['blocked', 'queued', 'speccing', 'analyzing', 'cancelled', 'paused'],
blocked: ['ready', 'cancelled', 'paused'],
queued: ['executing', 'ready', 'cancelled', 'paused'],
executing: ['exec_review', 'failed', 'cancelled'],
exec_review: ['done', 'ready', 'failed', 'cancelled'], // accept(合并) / reject(返工)
failed: ['queued', 'needs_attention', 'cancelled'], // 重试 / 升级
needs_attention: ['ready', 'queued', 'analyzing', 'speccing', 'cancelled', 'paused'],
done: [],
paused: ['ready', 'analyzing', 'speccing', 'queued', 'cancelled'],
cancelled: [],
};
export function canTransition(from: TaskStatus, to: TaskStatus): boolean {
return TRANSITIONS[from]?.includes(to) ?? false;
}
/** 新建任务按复杂度决定从 init 进入的下一个状态 */
export function initialNextStatus(c: Complexity): TaskStatus {
if (c === 'hard') return 'analyzing'; // 强制分析 + 拆解
if (c === 'medium') return 'speccing'; // 写方案
return 'ready'; // Easy:写完 operations 即可执行
}
/** 可被执行器领取的状态(叶子 + 依赖满足由调用方另判) */
export function isExecutable(s: TaskStatus): boolean {
return s === 'ready';
}
export function isGate(s: TaskStatus): boolean {
return s in GATE_STATUS;
}
+92
View File
@@ -0,0 +1,92 @@
import type { Complexity } from './complexity.js';
import type { TaskStatus, GateKind } from './status.js';
export type Id = string;
/** 层级上限:一般 ≤3 层,极复杂 ≤4 层 */
export const DEFAULT_MAX_DEPTH = 3;
export const HARD_MAX_DEPTH = 4;
/** 自动执行边界(Easy 任务由后台执行器自动跑到什么程度) */
export type Autonomy = 'manual' | 'auto-easy' | 'auto-approved';
export interface Project {
id: Id;
name: string;
repoPath: string;
defaultBranch: string;
verifyCmd: string | null; // 执行后的校验命令(build/test/lint
autonomy: Autonomy;
model: string | null;
concurrency: number; // 每项目并发执行上限
status: 'active' | 'paused';
createdAt: string;
}
export interface ApprovalRecord {
gate: GateKind;
action: 'accept' | 'reject';
actor: string;
reason: string | null; // reject 必填(改进意见)
at: string;
}
export interface TaskResult {
branch: string | null;
worktree: string | null;
diffSummary: string | null;
commits: string[];
prUrl: string | null;
}
export interface Task {
id: Id;
projectId: Id;
parentId: Id | null;
depth: number; // 1..4
title: string;
complexity: Complexity;
status: TaskStatus;
priority: number;
deps: Id[]; // 兄弟依赖(同层)
// 三选一产出(按复杂度)
plan: string | null; // Hard:分析 + 拆解说明
spec: string | null; // Medium:改动内容 + 为什么
operations: string | null; // Easy:将执行的操作
approvals: ApprovalRecord[];
result: TaskResult | null;
assignee: 'agent' | 'human' | null;
createdAt: string;
updatedAt: string;
}
export type RunKind = 'planner' | 'executor';
export type RunStatus = 'started' | 'succeeded' | 'failed' | 'cancelled';
export interface Run {
id: Id;
taskId: Id;
kind: RunKind; // planner(产出拆解/方案)| executor(执行改动)
worktree: string | null;
branch: string | null;
status: RunStatus;
startedAt: string;
endedAt: string | null;
transcriptRef: string | null; // agent 转录日志文件路径
claudeSessionId: string | null;
error: string | null;
}
export type EventType =
| 'task.created' | 'task.updated' | 'status.changed'
| 'approval.requested' | 'approval.granted' | 'approval.rejected'
| 'run.started' | 'run.finished';
export interface Event {
id: Id;
projectId: Id;
taskId: Id | null;
type: EventType;
payload: Record<string, unknown>;
at: string;
}
+18
View File
@@ -0,0 +1,18 @@
import Database from 'better-sqlite3';
import { readFileSync } from 'node:fs';
import { fileURLToPath } from 'node:url';
import { dirname, join } from 'node:path';
const HERE = dirname(fileURLToPath(import.meta.url));
/** 打开(或新建)数据库并应用 schema。schema.sql 在 dev(src) 与 build(dist) 两处都与本文件同目录。 */
export function openDb(file: string): Database.Database {
const db = new Database(file);
db.pragma('journal_mode = WAL');
db.pragma('foreign_keys = ON');
const schema = readFileSync(join(HERE, 'schema.sql'), 'utf8');
db.exec(schema);
return db;
}
export type DB = Database.Database;
+4
View File
@@ -0,0 +1,4 @@
export { Store, StoreError } from './store.js';
export type { CreateProjectInput, CreateTaskInput } from './store.js';
export { openDb } from './db.js';
export type { DB } from './db.js';
+71
View File
@@ -0,0 +1,71 @@
import type { Project, Task, ApprovalRecord, Run, Event, TaskResult, Autonomy } from '../model/types.js';
import type { Complexity } from '../model/complexity.js';
import type { TaskStatus, GateKind } from '../model/status.js';
import type { RunKind, RunStatus, EventType } from '../model/types.js';
/** SQLite 行类型(snake_case,文本/数字原样) */
export interface ProjectRow {
id: string; name: string; repo_path: string; default_branch: string;
verify_cmd: string | null; autonomy: string; model: string | null;
concurrency: number; status: string; created_at: string;
}
export interface TaskRow {
id: string; project_id: string; parent_id: string | null; depth: number;
title: string; complexity: string; status: string; priority: number;
deps: string; plan: string | null; spec: string | null; operations: string | null;
result: string | null; assignee: string | null; created_at: string; updated_at: string;
}
export interface ApprovalRow {
id: string; task_id: string; gate: string; action: string;
actor: string; reason: string | null; at: string;
}
export interface RunRow {
id: string; task_id: string; kind: string; worktree: string | null; branch: string | null;
status: string; started_at: string; ended_at: string | null;
transcript_ref: string | null; claude_session_id: string | null; error: string | null;
}
export interface EventRow {
id: string; project_id: string; task_id: string | null; type: string; payload: string; at: string;
}
export function rowToProject(r: ProjectRow): Project {
return {
id: r.id, name: r.name, repoPath: r.repo_path, defaultBranch: r.default_branch,
verifyCmd: r.verify_cmd, autonomy: r.autonomy as Autonomy, model: r.model,
concurrency: r.concurrency, status: r.status as Project['status'], createdAt: r.created_at,
};
}
export function rowToTask(r: TaskRow, approvals: ApprovalRecord[] = []): Task {
return {
id: r.id, projectId: r.project_id, parentId: r.parent_id, depth: r.depth,
title: r.title, complexity: r.complexity as Complexity, status: r.status as TaskStatus,
priority: r.priority, deps: JSON.parse(r.deps) as string[],
plan: r.plan, spec: r.spec, operations: r.operations,
approvals, result: r.result ? (JSON.parse(r.result) as TaskResult) : null,
assignee: r.assignee as Task['assignee'], createdAt: r.created_at, updatedAt: r.updated_at,
};
}
export function rowToApproval(r: ApprovalRow): ApprovalRecord {
return {
gate: r.gate as GateKind, action: r.action as ApprovalRecord['action'],
actor: r.actor, reason: r.reason, at: r.at,
};
}
export function rowToRun(r: RunRow): Run {
return {
id: r.id, taskId: r.task_id, kind: r.kind as RunKind,
worktree: r.worktree, branch: r.branch, status: r.status as RunStatus,
startedAt: r.started_at, endedAt: r.ended_at,
transcriptRef: r.transcript_ref, claudeSessionId: r.claude_session_id, error: r.error,
};
}
export function rowToEvent(r: EventRow): Event {
return {
id: r.id, projectId: r.project_id, taskId: r.task_id,
type: r.type as EventType, payload: JSON.parse(r.payload) as Record<string, unknown>, at: r.at,
};
}
+74
View File
@@ -0,0 +1,74 @@
-- maestro · SQLite schema v0.1(设计见 DESIGN.md §6/§9
PRAGMA journal_mode = WAL;
PRAGMA foreign_keys = ON;
CREATE TABLE IF NOT EXISTS projects (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
repo_path TEXT NOT NULL UNIQUE,
default_branch TEXT NOT NULL DEFAULT 'main',
verify_cmd TEXT,
autonomy TEXT NOT NULL DEFAULT 'manual', -- manual | auto-easy | auto-approved
model TEXT,
concurrency INTEGER NOT NULL DEFAULT 1,
status TEXT NOT NULL DEFAULT 'active', -- active | paused
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS tasks (
id TEXT PRIMARY KEY,
project_id TEXT NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
parent_id TEXT REFERENCES tasks(id) ON DELETE CASCADE,
depth INTEGER NOT NULL DEFAULT 1, -- 1..4
title TEXT NOT NULL,
complexity TEXT NOT NULL, -- hard | medium | easy
status TEXT NOT NULL DEFAULT 'init',
priority INTEGER NOT NULL DEFAULT 0,
deps TEXT NOT NULL DEFAULT '[]', -- JSON array of task ids
plan TEXT, -- Hard:分析 + 拆解
spec TEXT, -- Medium:改动 + 理由
operations TEXT, -- Easy:将执行的操作
result TEXT, -- JSON TaskResult
assignee TEXT, -- agent | human
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_tasks_project ON tasks(project_id, status);
CREATE INDEX IF NOT EXISTS idx_tasks_parent ON tasks(parent_id);
CREATE TABLE IF NOT EXISTS approvals (
id TEXT PRIMARY KEY,
task_id TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
gate TEXT NOT NULL, -- plan | spec | exec
action TEXT NOT NULL, -- accept | reject
actor TEXT NOT NULL,
reason TEXT, -- reject 必填
at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_approvals_task ON approvals(task_id, at);
CREATE TABLE IF NOT EXISTS runs (
id TEXT PRIMARY KEY,
task_id TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
kind TEXT NOT NULL, -- planner | executor
worktree TEXT,
branch TEXT,
status TEXT NOT NULL, -- started | succeeded | failed | cancelled
started_at TEXT NOT NULL,
ended_at TEXT,
transcript_ref TEXT,
claude_session_id TEXT,
error TEXT
);
CREATE INDEX IF NOT EXISTS idx_runs_task ON runs(task_id, started_at);
-- 追加型事件日志:审计 + 实时看板推送源
CREATE TABLE IF NOT EXISTS events (
id TEXT PRIMARY KEY,
project_id TEXT NOT NULL,
task_id TEXT,
type TEXT NOT NULL,
payload TEXT NOT NULL DEFAULT '{}',
at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_events_project ON events(project_id, at);
+285
View File
@@ -0,0 +1,285 @@
import { nanoid } from 'nanoid';
import { openDb, type DB } from './db.js';
import {
rowToProject, rowToTask, rowToApproval, rowToRun, rowToEvent,
type ProjectRow, type TaskRow, type ApprovalRow, type RunRow, type EventRow,
} from './mappers.js';
import type { Project, Task, ApprovalRecord, Run, Event, TaskResult, Autonomy, EventType } from '../model/types.js';
import { DEFAULT_MAX_DEPTH, HARD_MAX_DEPTH } from '../model/types.js';
import type { Complexity } from '../model/complexity.js';
import {
type TaskStatus, type GateKind, canTransition, initialNextStatus, gateOf, STATUS_LABEL,
} from '../model/status.js';
const now = (): string => new Date().toISOString();
const id = (prefix: string): string => `${prefix}_${nanoid(12)}`;
export class StoreError extends Error {}
export interface CreateProjectInput {
name: string; repoPath: string; defaultBranch?: string; verifyCmd?: string | null;
autonomy?: Autonomy; model?: string | null; concurrency?: number;
}
export interface CreateTaskInput {
projectId: string; title: string; complexity: Complexity;
parentId?: string | null; priority?: number; deps?: string[];
}
type EventListener = (e: Event) => void;
/**
* 任务系统的唯一数据出入口。所有状态变更都经 transition() 守卫(canTransition),
* 审批 reject 强制带改进意见,每次变更追加 Event 并广播给监听者(供 WS 推送)。
*/
export class Store {
readonly db: DB;
private listeners = new Set<EventListener>();
constructor(file: string) {
this.db = openDb(file);
}
close(): void { this.db.close(); }
/** 订阅事件流(看板 WS)。返回取消订阅函数。 */
subscribe(fn: EventListener): () => void {
this.listeners.add(fn);
return () => this.listeners.delete(fn);
}
private emit(projectId: string, taskId: string | null, type: EventType, payload: Record<string, unknown> = {}): Event {
const row: EventRow = {
id: id('evt'), project_id: projectId, task_id: taskId, type,
payload: JSON.stringify(payload), at: now(),
};
this.db.prepare(
`INSERT INTO events (id, project_id, task_id, type, payload, at) VALUES (@id,@project_id,@task_id,@type,@payload,@at)`,
).run(row);
const evt = rowToEvent(row);
for (const l of this.listeners) { try { l(evt); } catch { /* listener 错误不影响主流程 */ } }
return evt;
}
// ---------- Projects ----------
createProject(input: CreateProjectInput): Project {
const row: ProjectRow = {
id: id('prj'), name: input.name, repo_path: input.repoPath,
default_branch: input.defaultBranch ?? 'main', verify_cmd: input.verifyCmd ?? null,
autonomy: input.autonomy ?? 'manual', model: input.model ?? null,
concurrency: input.concurrency ?? 1, status: 'active', created_at: now(),
};
this.db.prepare(
`INSERT INTO projects (id,name,repo_path,default_branch,verify_cmd,autonomy,model,concurrency,status,created_at)
VALUES (@id,@name,@repo_path,@default_branch,@verify_cmd,@autonomy,@model,@concurrency,@status,@created_at)`,
).run(row);
this.emit(row.id, null, 'task.created', { kind: 'project', name: row.name });
return rowToProject(row);
}
listProjects(): Project[] {
const rows = this.db.prepare(`SELECT * FROM projects ORDER BY created_at`).all() as ProjectRow[];
return rows.map(rowToProject);
}
getProject(projectId: string): Project | null {
const row = this.db.prepare(`SELECT * FROM projects WHERE id = ?`).get(projectId) as ProjectRow | undefined;
return row ? rowToProject(row) : null;
}
// ---------- Tasks ----------
createTask(input: CreateTaskInput): Task {
const project = this.getProject(input.projectId);
if (!project) throw new StoreError(`项目不存在: ${input.projectId}`);
let depth = 1;
if (input.parentId) {
const parent = this.getTaskRow(input.parentId);
if (!parent) throw new StoreError(`父任务不存在: ${input.parentId}`);
if (parent.project_id !== input.projectId) throw new StoreError('子任务必须与父任务同项目');
depth = parent.depth + 1;
const max = parent.complexity === 'hard' ? HARD_MAX_DEPTH : DEFAULT_MAX_DEPTH;
if (depth > max) throw new StoreError(`层级超限:最多 ${max} 层(父任务 ${parent.complexity}`);
}
const status = initialNextStatus(input.complexity);
const row: TaskRow = {
id: id('tsk'), project_id: input.projectId, parent_id: input.parentId ?? null, depth,
title: input.title, complexity: input.complexity, status, priority: input.priority ?? 0,
deps: JSON.stringify(input.deps ?? []), plan: null, spec: null, operations: null,
result: null, assignee: null, created_at: now(), updated_at: now(),
};
this.db.prepare(
`INSERT INTO tasks (id,project_id,parent_id,depth,title,complexity,status,priority,deps,plan,spec,operations,result,assignee,created_at,updated_at)
VALUES (@id,@project_id,@parent_id,@depth,@title,@complexity,@status,@priority,@deps,@plan,@spec,@operations,@result,@assignee,@created_at,@updated_at)`,
).run(row);
this.emit(input.projectId, row.id, 'task.created', { title: row.title, complexity: row.complexity, status });
return rowToTask(row);
}
private getTaskRow(taskId: string): TaskRow | undefined {
return this.db.prepare(`SELECT * FROM tasks WHERE id = ?`).get(taskId) as TaskRow | undefined;
}
getTask(taskId: string): Task | null {
const row = this.getTaskRow(taskId);
if (!row) return null;
return rowToTask(row, this.listApprovals(taskId));
}
listTasks(projectId: string): Task[] {
const rows = this.db.prepare(
`SELECT * FROM tasks WHERE project_id = ? ORDER BY depth, priority DESC, created_at`,
).all(projectId) as TaskRow[];
return rows.map((r) => rowToTask(r, this.listApprovals(r.id)));
}
childrenOf(taskId: string): Task[] {
const rows = this.db.prepare(
`SELECT * FROM tasks WHERE parent_id = ? ORDER BY priority DESC, created_at`,
).all(taskId) as TaskRow[];
return rows.map((r) => rowToTask(r, this.listApprovals(r.id)));
}
/** 写产出:Hard→plan / Medium→spec / Easy→operations。仅更新对应字段。 */
setPlan(taskId: string, plan: string): Task { return this.patchField(taskId, 'plan', plan); }
setSpec(taskId: string, spec: string): Task { return this.patchField(taskId, 'spec', spec); }
setOperations(taskId: string, ops: string): Task { return this.patchField(taskId, 'operations', ops); }
private patchField(taskId: string, field: 'plan' | 'spec' | 'operations', value: string): Task {
const row = this.getTaskRow(taskId);
if (!row) throw new StoreError(`任务不存在: ${taskId}`);
this.db.prepare(`UPDATE tasks SET ${field} = ?, updated_at = ? WHERE id = ?`).run(value, now(), taskId);
this.emit(row.project_id, taskId, 'task.updated', { field });
return this.getTask(taskId)!;
}
setResult(taskId: string, result: TaskResult): Task {
const row = this.getTaskRow(taskId);
if (!row) throw new StoreError(`任务不存在: ${taskId}`);
this.db.prepare(`UPDATE tasks SET result = ?, updated_at = ? WHERE id = ?`).run(JSON.stringify(result), now(), taskId);
this.emit(row.project_id, taskId, 'task.updated', { field: 'result' });
return this.getTask(taskId)!;
}
/** 受守卫的状态变更:非法流转抛错;记录 status.changed 事件。 */
transition(taskId: string, to: TaskStatus, meta: Record<string, unknown> = {}): Task {
const row = this.getTaskRow(taskId);
if (!row) throw new StoreError(`任务不存在: ${taskId}`);
const from = row.status as TaskStatus;
if (from === to) return this.getTask(taskId)!;
if (!canTransition(from, to)) {
throw new StoreError(`非法状态流转:${STATUS_LABEL[from]}(${from}) → ${STATUS_LABEL[to]}(${to})`);
}
this.db.prepare(`UPDATE tasks SET status = ?, updated_at = ? WHERE id = ?`).run(to, now(), taskId);
this.emit(row.project_id, taskId, 'status.changed', { from, to, ...meta });
return this.getTask(taskId)!;
}
// ---------- Approvals(审批闸) ----------
private listApprovals(taskId: string): ApprovalRecord[] {
const rows = this.db.prepare(`SELECT * FROM approvals WHERE task_id = ? ORDER BY at`).all(taskId) as ApprovalRow[];
return rows.map(rowToApproval);
}
/** 列出所有处于审批闸状态(plan_review/spec_review/exec_review)的任务。 */
pendingApprovals(projectId?: string): Task[] {
const sql = projectId
? `SELECT * FROM tasks WHERE project_id = ? AND status IN ('plan_review','spec_review','exec_review') ORDER BY updated_at`
: `SELECT * FROM tasks WHERE status IN ('plan_review','spec_review','exec_review') ORDER BY updated_at`;
const rows = (projectId
? this.db.prepare(sql).all(projectId)
: this.db.prepare(sql).all()) as TaskRow[];
return rows.map((r) => rowToTask(r, this.listApprovals(r.id)));
}
/**
* 处理一次审批。accept/reject 必须发生在闸状态上;reject 必须带 reason(改进意见)。
* acceptplan_review→decomposed, spec_review→ready, exec_review→done。
* rejectplan_review→analyzing, spec_review→speccing, exec_review→ready(返工)。
*/
decide(taskId: string, action: 'accept' | 'reject', actor: string, reason?: string | null): Task {
const row = this.getTaskRow(taskId);
if (!row) throw new StoreError(`任务不存在: ${taskId}`);
const from = row.status as TaskStatus;
const gate: GateKind | null = gateOf(from);
if (!gate) throw new StoreError(`任务不在审批闸状态:当前 ${STATUS_LABEL[from]}(${from})`);
if (action === 'reject' && !reason?.trim()) {
throw new StoreError('reject 必须填写改进意见');
}
const to: TaskStatus = action === 'accept'
? ({ plan: 'decomposed', spec: 'ready', exec: 'done' } as const)[gate]
: ({ plan: 'analyzing', spec: 'speccing', exec: 'ready' } as const)[gate];
const txn = this.db.transaction(() => {
const ap: ApprovalRow = {
id: id('apr'), task_id: taskId, gate, action, actor, reason: reason ?? null, at: now(),
};
this.db.prepare(
`INSERT INTO approvals (id,task_id,gate,action,actor,reason,at) VALUES (@id,@task_id,@gate,@action,@actor,@reason,@at)`,
).run(ap);
this.db.prepare(`UPDATE tasks SET status = ?, updated_at = ? WHERE id = ?`).run(to, now(), taskId);
});
txn();
this.emit(row.project_id, taskId, action === 'accept' ? 'approval.granted' : 'approval.rejected', { gate, from, to, reason: reason ?? null });
return this.getTask(taskId)!;
}
// ---------- Runs ----------
startRun(taskId: string, kind: Run['kind'], fields: Partial<Pick<Run, 'worktree' | 'branch'>> = {}): Run {
const row = this.getTaskRow(taskId);
if (!row) throw new StoreError(`任务不存在: ${taskId}`);
const rr: RunRow = {
id: id('run'), task_id: taskId, kind, worktree: fields.worktree ?? null, branch: fields.branch ?? null,
status: 'started', started_at: now(), ended_at: null, transcript_ref: null, claude_session_id: null, error: null,
};
this.db.prepare(
`INSERT INTO runs (id,task_id,kind,worktree,branch,status,started_at,ended_at,transcript_ref,claude_session_id,error)
VALUES (@id,@task_id,@kind,@worktree,@branch,@status,@started_at,@ended_at,@transcript_ref,@claude_session_id,@error)`,
).run(rr);
this.emit(row.project_id, taskId, 'run.started', { runId: rr.id, kind });
return rowToRun(rr);
}
finishRun(runId: string, status: Run['status'], fields: Partial<Pick<Run, 'transcriptRef' | 'claudeSessionId' | 'error'>> = {}): Run {
const row = this.db.prepare(`SELECT * FROM runs WHERE id = ?`).get(runId) as RunRow | undefined;
if (!row) throw new StoreError(`run 不存在: ${runId}`);
this.db.prepare(
`UPDATE runs SET status = ?, ended_at = ?, transcript_ref = ?, claude_session_id = ?, error = ? WHERE id = ?`,
).run(status, now(), fields.transcriptRef ?? row.transcript_ref, fields.claudeSessionId ?? row.claude_session_id, fields.error ?? row.error, runId);
const task = this.getTaskRow(row.task_id);
if (task) this.emit(task.project_id, row.task_id, 'run.finished', { runId, status });
return rowToRun(this.db.prepare(`SELECT * FROM runs WHERE id = ?`).get(runId) as RunRow);
}
listRuns(taskId: string): Run[] {
const rows = this.db.prepare(`SELECT * FROM runs WHERE task_id = ? ORDER BY started_at`).all(taskId) as RunRow[];
return rows.map(rowToRun);
}
// ---------- Events ----------
listEvents(projectId: string, limit = 200): Event[] {
const rows = this.db.prepare(
`SELECT * FROM events WHERE project_id = ? ORDER BY at DESC LIMIT ?`,
).all(projectId, limit) as EventRow[];
return rows.map(rowToEvent).reverse();
}
/**
* 取下一个可执行任务(叶子、ready、依赖全部 done)。供编排器领取。
* 被拆解的 Hard 容器任务不会是 ready(停在 decomposed),天然排除。
*/
nextExecutable(projectId: string): Task | null {
const rows = this.db.prepare(
`SELECT * FROM tasks WHERE project_id = ? AND status = 'ready' ORDER BY priority DESC, created_at`,
).all(projectId) as TaskRow[];
for (const r of rows) {
if (this.childrenOf(r.id).length > 0) continue; // 非叶子跳过
const deps = JSON.parse(r.deps) as string[];
const depsDone = deps.every((d) => this.getTaskRow(d)?.status === 'done');
if (!depsDone) continue;
return rowToTask(r, this.listApprovals(r.id));
}
return null;
}
}