feat: bootstrap boss control plane prototype
This commit is contained in:
Codex
656
src/engine.ts
Normal file
656
src/engine.ts
Normal file
@@ -0,0 +1,656 @@
|
||||
import { resolve } from "node:path";
|
||||
import type {
|
||||
ApprovalRequest,
|
||||
AppState,
|
||||
BossEvent,
|
||||
Message,
|
||||
Session,
|
||||
SessionDetails,
|
||||
Task,
|
||||
WorkerNode,
|
||||
} from "./types.js";
|
||||
import { EventBroker } from "./event-broker.js";
|
||||
import { createPlan, buildPlannerMessage, materializeTasks } from "./planner.js";
|
||||
import { chooseAssignmentCandidates } from "./scheduler.js";
|
||||
import { FileStore } from "./store.js";
|
||||
import { createId, now } from "./utils.js";
|
||||
|
||||
const DATA_FILE = resolve(process.cwd(), ".boss-data", "store.json");
|
||||
|
||||
function isActiveTask(task: Task): boolean {
|
||||
return !["completed", "failed", "cancelled"].includes(task.status);
|
||||
}
|
||||
|
||||
export class BossEngine {
|
||||
readonly store = new FileStore(DATA_FILE);
|
||||
readonly events = new EventBroker();
|
||||
|
||||
getState(): AppState {
|
||||
return this.store.snapshot;
|
||||
}
|
||||
|
||||
bootstrap(): AppState {
|
||||
return this.getState();
|
||||
}
|
||||
|
||||
createSession(title?: string): SessionDetails {
|
||||
const timestamp = now();
|
||||
const session: Session = {
|
||||
id: createId("session"),
|
||||
title: title?.trim() || "未命名项目",
|
||||
status: "active",
|
||||
activeObjective: "",
|
||||
lastPlannerSummary: "",
|
||||
createdAt: timestamp,
|
||||
updatedAt: timestamp,
|
||||
};
|
||||
|
||||
this.store.mutate((state) => {
|
||||
state.sessions.unshift(session);
|
||||
state.events.push(
|
||||
this.makeEvent({
|
||||
sessionId: session.id,
|
||||
taskId: null,
|
||||
source: "system",
|
||||
type: "session.created",
|
||||
payload: { title: session.title },
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
this.publishLatestEvent();
|
||||
return this.getSession(session.id);
|
||||
}
|
||||
|
||||
getSession(sessionId: string): SessionDetails {
|
||||
const state = this.getState();
|
||||
const session = state.sessions.find((candidate) => candidate.id === sessionId);
|
||||
if (!session) {
|
||||
throw new Error(`Session not found: ${sessionId}`);
|
||||
}
|
||||
|
||||
return {
|
||||
session,
|
||||
messages: state.messages.filter((message) => message.sessionId === sessionId),
|
||||
tasks: state.tasks.filter((task) => task.sessionId === sessionId),
|
||||
approvals: state.approvals.filter((approval) => approval.sessionId === sessionId),
|
||||
};
|
||||
}
|
||||
|
||||
listSessions(): Session[] {
|
||||
return this.getState().sessions;
|
||||
}
|
||||
|
||||
addMessage(sessionId: string, content: string, channel = "web"): SessionDetails {
|
||||
const session = this.getSession(sessionId).session;
|
||||
const message: Message = {
|
||||
id: createId("msg"),
|
||||
sessionId,
|
||||
role: "user",
|
||||
channel,
|
||||
content: content.trim(),
|
||||
createdAt: now(),
|
||||
};
|
||||
|
||||
if (!message.content) {
|
||||
throw new Error("Message content is required.");
|
||||
}
|
||||
|
||||
this.store.mutate((state) => {
|
||||
const mutableSession = state.sessions.find((candidate) => candidate.id === sessionId);
|
||||
if (!mutableSession) {
|
||||
throw new Error(`Session not found: ${sessionId}`);
|
||||
}
|
||||
|
||||
mutableSession.activeObjective = message.content;
|
||||
mutableSession.updatedAt = message.createdAt;
|
||||
if (!mutableSession.title || mutableSession.title === "未命名项目") {
|
||||
mutableSession.title = message.content.slice(0, 32);
|
||||
}
|
||||
state.messages.push(message);
|
||||
state.events.push(
|
||||
this.makeEvent({
|
||||
sessionId,
|
||||
taskId: null,
|
||||
source: "user",
|
||||
type: "session.message.added",
|
||||
payload: {
|
||||
channel,
|
||||
content: message.content,
|
||||
},
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
this.publishLatestEvent();
|
||||
this.applyPlan(session, message.content);
|
||||
return this.getSession(sessionId);
|
||||
}
|
||||
|
||||
registerWorker(input: {
|
||||
name: string;
|
||||
os: WorkerNode["os"];
|
||||
capabilities: string[];
|
||||
}): WorkerNode {
|
||||
const timestamp = now();
|
||||
const existing = this.getState().workers.find((worker) => worker.name === input.name);
|
||||
if (existing) {
|
||||
return this.updateWorker(existing.id, {
|
||||
os: input.os,
|
||||
capabilities: input.capabilities,
|
||||
status: "idle",
|
||||
load: 0,
|
||||
});
|
||||
}
|
||||
|
||||
const worker: WorkerNode = {
|
||||
id: createId("worker"),
|
||||
name: input.name,
|
||||
os: input.os,
|
||||
capabilities: Array.from(new Set(input.capabilities)),
|
||||
status: "idle",
|
||||
currentTaskId: null,
|
||||
load: 0,
|
||||
lastSeenAt: timestamp,
|
||||
createdAt: timestamp,
|
||||
updatedAt: timestamp,
|
||||
};
|
||||
|
||||
this.store.mutate((state) => {
|
||||
state.workers.push(worker);
|
||||
state.events.push(
|
||||
this.makeEvent({
|
||||
sessionId: null,
|
||||
taskId: null,
|
||||
source: "system",
|
||||
type: "worker.registered",
|
||||
payload: {
|
||||
workerId: worker.id,
|
||||
name: worker.name,
|
||||
os: worker.os,
|
||||
capabilities: worker.capabilities,
|
||||
},
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
this.publishLatestEvent();
|
||||
this.syncAssignments();
|
||||
return worker;
|
||||
}
|
||||
|
||||
updateWorker(
|
||||
workerId: string,
|
||||
input: Partial<Pick<WorkerNode, "os" | "capabilities" | "status" | "load">>,
|
||||
): WorkerNode {
|
||||
let updated!: WorkerNode;
|
||||
this.store.mutate((state) => {
|
||||
const worker = state.workers.find((candidate) => candidate.id === workerId);
|
||||
if (!worker) {
|
||||
throw new Error(`Worker not found: ${workerId}`);
|
||||
}
|
||||
|
||||
if (input.os) worker.os = input.os;
|
||||
if (input.capabilities) worker.capabilities = Array.from(new Set(input.capabilities));
|
||||
if (input.status) worker.status = input.status;
|
||||
if (typeof input.load === "number") worker.load = input.load;
|
||||
worker.updatedAt = now();
|
||||
worker.lastSeenAt = worker.updatedAt;
|
||||
updated = { ...worker };
|
||||
});
|
||||
return updated;
|
||||
}
|
||||
|
||||
heartbeat(workerId: string, load = 0): WorkerNode {
|
||||
let updated!: WorkerNode;
|
||||
this.store.mutate((state) => {
|
||||
const worker = state.workers.find((candidate) => candidate.id === workerId);
|
||||
if (!worker) {
|
||||
throw new Error(`Worker not found: ${workerId}`);
|
||||
}
|
||||
|
||||
worker.lastSeenAt = now();
|
||||
worker.updatedAt = worker.lastSeenAt;
|
||||
worker.load = load;
|
||||
if (!worker.currentTaskId && worker.status !== "offline") {
|
||||
worker.status = "idle";
|
||||
}
|
||||
|
||||
state.events.push(
|
||||
this.makeEvent({
|
||||
sessionId: null,
|
||||
taskId: worker.currentTaskId,
|
||||
source: "worker",
|
||||
type: "worker.heartbeat",
|
||||
payload: {
|
||||
workerId: worker.id,
|
||||
status: worker.status,
|
||||
load: worker.load,
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
updated = { ...worker };
|
||||
});
|
||||
|
||||
this.publishLatestEvent();
|
||||
this.syncAssignments();
|
||||
return updated;
|
||||
}
|
||||
|
||||
claimNextTask(workerId: string): Task | null {
|
||||
let claimedTask: Task | null = null;
|
||||
|
||||
this.store.mutate((state) => {
|
||||
const worker = state.workers.find((candidate) => candidate.id === workerId);
|
||||
if (!worker) {
|
||||
throw new Error(`Worker not found: ${workerId}`);
|
||||
}
|
||||
|
||||
const task = state.tasks.find(
|
||||
(candidate) => candidate.assignedWorkerId === workerId && candidate.status === "assigned",
|
||||
);
|
||||
if (!task) {
|
||||
return;
|
||||
}
|
||||
|
||||
task.status = "running";
|
||||
task.currentStep = "start";
|
||||
task.summary = "任务已被 worker 接收。";
|
||||
task.updatedAt = now();
|
||||
worker.status = "busy";
|
||||
worker.currentTaskId = task.id;
|
||||
worker.updatedAt = task.updatedAt;
|
||||
worker.lastSeenAt = task.updatedAt;
|
||||
|
||||
claimedTask = { ...task };
|
||||
state.events.push(
|
||||
this.makeEvent({
|
||||
sessionId: task.sessionId,
|
||||
taskId: task.id,
|
||||
source: "worker",
|
||||
type: "task.started",
|
||||
payload: {
|
||||
workerId,
|
||||
title: task.title,
|
||||
},
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
if (claimedTask) {
|
||||
this.publishLatestEvent();
|
||||
}
|
||||
|
||||
return claimedTask;
|
||||
}
|
||||
|
||||
reportProgress(
|
||||
taskId: string,
|
||||
workerId: string,
|
||||
input: {
|
||||
progressPercent: number;
|
||||
summary: string;
|
||||
currentStep: string;
|
||||
nextStep: string;
|
||||
},
|
||||
): Task {
|
||||
let updated!: Task;
|
||||
this.store.mutate((state) => {
|
||||
const task = state.tasks.find((candidate) => candidate.id === taskId);
|
||||
if (!task) {
|
||||
throw new Error(`Task not found: ${taskId}`);
|
||||
}
|
||||
|
||||
task.progressPercent = Math.max(0, Math.min(100, input.progressPercent));
|
||||
task.summary = input.summary;
|
||||
task.currentStep = input.currentStep;
|
||||
task.nextStep = input.nextStep;
|
||||
task.updatedAt = now();
|
||||
if (task.status === "assigned") {
|
||||
task.status = "running";
|
||||
}
|
||||
|
||||
state.events.push(
|
||||
this.makeEvent({
|
||||
sessionId: task.sessionId,
|
||||
taskId: task.id,
|
||||
source: "worker",
|
||||
type: "task.progress",
|
||||
payload: {
|
||||
workerId,
|
||||
progressPercent: task.progressPercent,
|
||||
summary: task.summary,
|
||||
currentStep: task.currentStep,
|
||||
nextStep: task.nextStep,
|
||||
},
|
||||
}),
|
||||
);
|
||||
updated = { ...task };
|
||||
});
|
||||
|
||||
this.publishLatestEvent();
|
||||
return updated;
|
||||
}
|
||||
|
||||
completeTask(taskId: string, workerId: string, summary: string): Task {
|
||||
let updated!: Task;
|
||||
|
||||
this.store.mutate((state) => {
|
||||
const task = state.tasks.find((candidate) => candidate.id === taskId);
|
||||
const worker = state.workers.find((candidate) => candidate.id === workerId);
|
||||
if (!task) {
|
||||
throw new Error(`Task not found: ${taskId}`);
|
||||
}
|
||||
if (!worker) {
|
||||
throw new Error(`Worker not found: ${workerId}`);
|
||||
}
|
||||
|
||||
task.status = "completed";
|
||||
task.progressPercent = 100;
|
||||
task.summary = summary;
|
||||
task.currentStep = "done";
|
||||
task.nextStep = "";
|
||||
task.updatedAt = now();
|
||||
worker.status = "idle";
|
||||
worker.currentTaskId = null;
|
||||
worker.updatedAt = task.updatedAt;
|
||||
worker.lastSeenAt = task.updatedAt;
|
||||
|
||||
state.events.push(
|
||||
this.makeEvent({
|
||||
sessionId: task.sessionId,
|
||||
taskId: task.id,
|
||||
source: "worker",
|
||||
type: "task.completed",
|
||||
payload: {
|
||||
workerId,
|
||||
summary,
|
||||
},
|
||||
}),
|
||||
);
|
||||
updated = { ...task };
|
||||
});
|
||||
|
||||
this.publishLatestEvent();
|
||||
this.syncAssignments();
|
||||
return updated;
|
||||
}
|
||||
|
||||
failTask(taskId: string, workerId: string, errorMessage: string): Task {
|
||||
let updated!: Task;
|
||||
|
||||
this.store.mutate((state) => {
|
||||
const task = state.tasks.find((candidate) => candidate.id === taskId);
|
||||
const worker = state.workers.find((candidate) => candidate.id === workerId);
|
||||
if (!task) {
|
||||
throw new Error(`Task not found: ${taskId}`);
|
||||
}
|
||||
if (!worker) {
|
||||
throw new Error(`Worker not found: ${workerId}`);
|
||||
}
|
||||
|
||||
task.status = "failed";
|
||||
task.summary = errorMessage;
|
||||
task.currentStep = "failed";
|
||||
task.nextStep = "";
|
||||
task.updatedAt = now();
|
||||
worker.status = "idle";
|
||||
worker.currentTaskId = null;
|
||||
worker.updatedAt = task.updatedAt;
|
||||
worker.lastSeenAt = task.updatedAt;
|
||||
|
||||
state.events.push(
|
||||
this.makeEvent({
|
||||
sessionId: task.sessionId,
|
||||
taskId: task.id,
|
||||
source: "worker",
|
||||
type: "task.failed",
|
||||
payload: {
|
||||
workerId,
|
||||
errorMessage,
|
||||
},
|
||||
}),
|
||||
);
|
||||
updated = { ...task };
|
||||
});
|
||||
|
||||
this.publishLatestEvent();
|
||||
this.syncAssignments();
|
||||
return updated;
|
||||
}
|
||||
|
||||
pauseTask(taskId: string): Task {
|
||||
return this.transitionTask(taskId, "paused", "system", "task.paused", {
|
||||
summary: "任务已被暂停。",
|
||||
});
|
||||
}
|
||||
|
||||
cancelTask(taskId: string): Task {
|
||||
return this.transitionTask(taskId, "cancelled", "system", "task.cancelled", {
|
||||
summary: "任务已被取消。",
|
||||
});
|
||||
}
|
||||
|
||||
respondApproval(approvalId: string, approved: boolean, responder: string): ApprovalRequest {
|
||||
let updatedApproval!: ApprovalRequest;
|
||||
|
||||
this.store.mutate((state) => {
|
||||
const approval = state.approvals.find((candidate) => candidate.id === approvalId);
|
||||
if (!approval) {
|
||||
throw new Error(`Approval not found: ${approvalId}`);
|
||||
}
|
||||
|
||||
const task = state.tasks.find((candidate) => candidate.id === approval.taskId);
|
||||
if (!task) {
|
||||
throw new Error(`Task not found for approval: ${approval.taskId}`);
|
||||
}
|
||||
|
||||
const timestamp = now();
|
||||
approval.status = approved ? "approved" : "rejected";
|
||||
approval.responder = responder;
|
||||
approval.updatedAt = timestamp;
|
||||
task.approvalStatus = approval.status;
|
||||
task.updatedAt = timestamp;
|
||||
task.status = approved ? "queued" : "cancelled";
|
||||
task.summary = approved ? "审批已通过，重新进入队列。" : "审批被拒绝，任务已取消。";
|
||||
|
||||
state.events.push(
|
||||
this.makeEvent({
|
||||
sessionId: approval.sessionId,
|
||||
taskId: approval.taskId,
|
||||
source: "system",
|
||||
type: approved ? "approval.approved" : "approval.rejected",
|
||||
payload: {
|
||||
approvalId,
|
||||
responder,
|
||||
},
|
||||
}),
|
||||
);
|
||||
updatedApproval = { ...approval };
|
||||
});
|
||||
|
||||
this.publishLatestEvent();
|
||||
this.syncAssignments();
|
||||
return updatedApproval;
|
||||
}
|
||||
|
||||
private applyPlan(session: Session, content: string): void {
|
||||
const sessionDetails = this.getSession(session.id);
|
||||
const result = createPlan(sessionDetails.session, content, sessionDetails.tasks.filter(isActiveTask));
|
||||
const tasks = materializeTasks(session.id, result);
|
||||
const plannerMessage = buildPlannerMessage(result.summary);
|
||||
const timestamp = now();
|
||||
|
||||
this.store.mutate((state) => {
|
||||
const mutableSession = state.sessions.find((candidate) => candidate.id === session.id);
|
||||
if (!mutableSession) {
|
||||
throw new Error(`Session not found: ${session.id}`);
|
||||
}
|
||||
|
||||
mutableSession.activeObjective = content;
|
||||
mutableSession.lastPlannerSummary = plannerMessage;
|
||||
mutableSession.updatedAt = timestamp;
|
||||
|
||||
if (result.pauseExistingTasks) {
|
||||
for (const task of state.tasks.filter(
|
||||
(candidate) => candidate.sessionId === session.id && isActiveTask(candidate),
|
||||
)) {
|
||||
if (["running", "assigned", "queued", "planning", "blocked"].includes(task.status)) {
|
||||
task.status = "paused";
|
||||
task.updatedAt = timestamp;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
state.tasks.push(...tasks);
|
||||
for (const task of tasks) {
|
||||
if (task.approvalStatus === "pending") {
|
||||
state.approvals.push({
|
||||
id: createId("approval"),
|
||||
sessionId: session.id,
|
||||
taskId: task.id,
|
||||
kind: "dangerous_action",
|
||||
summary: `任务 "${task.title}" 包含高风险关键词，需要审批后才能执行。`,
|
||||
riskLevel: "high",
|
||||
status: "pending",
|
||||
requester: "manager",
|
||||
responder: null,
|
||||
createdAt: timestamp,
|
||||
updatedAt: timestamp,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
const managerMessage: Message = {
|
||||
id: createId("msg"),
|
||||
sessionId: session.id,
|
||||
role: "manager",
|
||||
channel: "system",
|
||||
content: plannerMessage,
|
||||
createdAt: timestamp,
|
||||
};
|
||||
state.messages.push(managerMessage);
|
||||
|
||||
state.events.push(
|
||||
this.makeEvent({
|
||||
sessionId: session.id,
|
||||
taskId: null,
|
||||
source: "manager",
|
||||
type: "plan.created",
|
||||
payload: {
|
||||
summary: result.summary,
|
||||
taskIds: tasks.map((task) => task.id),
|
||||
},
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
this.publishLatestEvent();
|
||||
this.syncAssignments();
|
||||
}
|
||||
|
||||
private transitionTask(
|
||||
taskId: string,
|
||||
status: Task["status"],
|
||||
source: BossEvent["source"],
|
||||
eventType: string,
|
||||
payload: Record<string, unknown>,
|
||||
): Task {
|
||||
let updated!: Task;
|
||||
this.store.mutate((state) => {
|
||||
const task = state.tasks.find((candidate) => candidate.id === taskId);
|
||||
if (!task) {
|
||||
throw new Error(`Task not found: ${taskId}`);
|
||||
}
|
||||
|
||||
task.status = status;
|
||||
task.updatedAt = now();
|
||||
task.summary = typeof payload.summary === "string" ? payload.summary : task.summary;
|
||||
|
||||
if (task.assignedWorkerId) {
|
||||
const worker = state.workers.find((candidate) => candidate.id === task.assignedWorkerId);
|
||||
if (worker && worker.currentTaskId === task.id) {
|
||||
worker.currentTaskId = null;
|
||||
worker.status = "idle";
|
||||
worker.updatedAt = task.updatedAt;
|
||||
worker.lastSeenAt = task.updatedAt;
|
||||
}
|
||||
}
|
||||
|
||||
state.events.push(
|
||||
this.makeEvent({
|
||||
sessionId: task.sessionId,
|
||||
taskId: task.id,
|
||||
source,
|
||||
type: eventType,
|
||||
payload,
|
||||
}),
|
||||
);
|
||||
updated = { ...task };
|
||||
});
|
||||
|
||||
this.publishLatestEvent();
|
||||
this.syncAssignments();
|
||||
return updated;
|
||||
}
|
||||
|
||||
private syncAssignments(): void {
|
||||
const candidates = chooseAssignmentCandidates(this.getState());
|
||||
if (candidates.length === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.store.mutate((state) => {
|
||||
for (const candidate of candidates) {
|
||||
const task = state.tasks.find((item) => item.id === candidate.taskId);
|
||||
const worker = state.workers.find((item) => item.id === candidate.workerId);
|
||||
if (!task || !worker || task.status !== "queued" || worker.status !== "idle") {
|
||||
continue;
|
||||
}
|
||||
|
||||
const timestamp = now();
|
||||
task.status = "assigned";
|
||||
task.assignedWorkerId = worker.id;
|
||||
task.updatedAt = timestamp;
|
||||
task.summary = `已分配给 ${worker.name}`;
|
||||
worker.status = "busy";
|
||||
worker.currentTaskId = task.id;
|
||||
worker.updatedAt = timestamp;
|
||||
worker.lastSeenAt = timestamp;
|
||||
|
||||
state.events.push(
|
||||
this.makeEvent({
|
||||
sessionId: task.sessionId,
|
||||
taskId: task.id,
|
||||
source: "system",
|
||||
type: "task.assigned",
|
||||
payload: {
|
||||
workerId: worker.id,
|
||||
workerName: worker.name,
|
||||
},
|
||||
}),
|
||||
);
|
||||
}
|
||||
});
|
||||
|
||||
this.publishLatestEvent();
|
||||
}
|
||||
|
||||
private publishLatestEvent(): void {
|
||||
const state = this.getState();
|
||||
const latestEvent = state.events[state.events.length - 1];
|
||||
if (latestEvent) {
|
||||
this.events.publish(latestEvent);
|
||||
}
|
||||
}
|
||||
|
||||
private makeEvent(input: Omit<BossEvent, "id" | "timestamp">): BossEvent {
|
||||
return {
|
||||
id: createId("evt"),
|
||||
timestamp: now(),
|
||||
...input,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user