krisolo/boss
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: ship usable local v1 with demo, workers, approvals, and docker support

Browse Source
This commit is contained in:
Codex
2026-03-23 12:59:41 +08:00
parent 0ab83990b2
commit 515ce72d0d
14 changed files with 1015 additions and 215 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.dockerignore Normal file
View File

@@ -0,0 +1,5 @@
node_modules
dist
.git
.boss-data

23
Dockerfile Normal file
View File

@@ -0,0 +1,23 @@
FROM node:22-bookworm-slim AS deps
WORKDIR /app
COPY package.json package-lock.json ./
RUN npm ci
FROM deps AS build
COPY tsconfig.json ./
COPY src ./src
COPY public ./public
COPY docs ./docs
COPY README.md ./
RUN npm run build
FROM node:22-bookworm-slim AS runtime
WORKDIR /app
ENV NODE_ENV=production
COPY --from=deps /app/node_modules ./node_modules
COPY --from=build /app/dist ./dist
COPY --from=build /app/public ./public
COPY package.json package-lock.json README.md ./
EXPOSE 43210
CMD ["node", "dist/server.js"]

46
README.md
View File

@@ -23,6 +23,8 @@ Boss 是一个面向多设备开发协作的 agent control plane。
- SSE 实时事件流
- Web 控制台
- `boss-worker` 模拟执行器
- `npm run smoke` 自动跑端到端验证
- `Dockerfile` + `compose.yaml` 支持容器启动
## 当前推荐方向
@@ -45,7 +47,7 @@ Boss 是一个面向多设备开发协作的 agent control plane。
```bash
npm install
npm run dev
npm run demo
```
浏览器打开：
@@ -54,10 +56,50 @@ npm run dev
http://127.0.0.1:43210
```
另开终端启动 worker：
如果你只想单独启动服务端：
```bash
npm run dev
```
如果你要手工启动 worker：
```bash
npm run worker -- --name win-a --os windows --capability terminal --capability browser
npm run worker -- --name win-b --os windows --capability terminal --capability test
npm run worker -- --name mac-a --os macos --capability terminal --capability test --capability browser
```
一键本地 demo：
```bash
npm install
npm run demo
```
这会拉起：
- Web/API 服务
- 2 台 Windows 模拟 worker
- 1 台 Mac 模拟 worker
自动 smoke test：
```bash
npm run smoke
```
容器启动：
```bash
docker compose up --build
```
## 当前 v1 能力
- 创建项目会话并持续对话
- 自动生成任务树并调度到不同 worker
- worker 心跳、掉线回收、任务重排
- 审批、暂停、恢复、取消、重排
- SSE 实时事件流和 Web 控制台
- 一键 demo 启动

28
compose.yaml Normal file
View File

@@ -0,0 +1,28 @@
services:
boss:
build: .
ports:
- "43210:43210"
volumes:
- boss-data:/app/.boss-data
worker-win-a:
build: .
depends_on:
- boss
command: ["node", "dist/worker.js", "--name", "win-a", "--os", "windows", "--capability", "terminal", "--capability", "browser", "--server", "http://boss:43210"]
worker-win-b:
build: .
depends_on:
- boss
command: ["node", "dist/worker.js", "--name", "win-b", "--os", "windows", "--capability", "terminal", "--capability", "test", "--server", "http://boss:43210"]
worker-mac-a:
build: .
depends_on:
- boss
command: ["node", "dist/worker.js", "--name", "mac-a", "--os", "macos", "--capability", "terminal", "--capability", "test", "--capability", "browser", "--server", "http://boss:43210"]
volumes:
boss-data:

5
package.json
View File

@@ -8,7 +8,9 @@
"build": "tsc -p tsconfig.json",
"start": "node dist/server.js",
"check": "tsc --noEmit -p tsconfig.json",
"worker": "tsx src/worker.ts"
"worker": "tsx src/worker.ts",
"demo": "tsx src/demo.ts",
"smoke": "tsx src/smoke.ts"
},
"dependencies": {
"@fastify/static": "^8.2.0",
@@ -20,4 +22,3 @@
"typescript": "^5.9.2"
}
}

111
public/app.js
View File

@@ -15,17 +15,23 @@ const elements = {
taskList: document.querySelector("#task-list"),
approvalList: document.querySelector("#approval-list"),
eventList: document.querySelector("#event-list"),
planHint: document.querySelector("#plan-hint"),
sessionTitleDisplay: document.querySelector("#session-title-display"),
sessionSummary: document.querySelector("#session-summary"),
createSessionForm: document.querySelector("#create-session-form"),
sessionTitleInput: document.querySelector("#session-title"),
createWorkerForm: document.querySelector("#create-worker-form"),
workerName: document.querySelector("#worker-name"),
workerOs: document.querySelector("#worker-os"),
workerCapabilities: document.querySelector("#worker-capabilities"),
messageForm: document.querySelector("#message-form"),
messageInput: document.querySelector("#message-input"),
resetDemo: document.querySelector("#reset-demo"),
archiveSession: document.querySelector("#archive-session"),
};
function escapeHtml(input) {
return input
return String(input)
.replaceAll("&", "&")
.replaceAll("<", "&lt;")
.replaceAll(">", "&gt;")
@@ -39,11 +45,18 @@ async function request(url, options = {}) {
...options,
});
if (!response.ok) {
throw new Error(`${response.status} ${response.statusText}`);
let payload = {};
try {
payload = await response.json();
} catch {
payload = {};
}
return response.json();
if (!response.ok) {
throw new Error(payload.error || `${response.status} ${response.statusText}`);
}
return payload;
}
function selectedSession() {
@@ -70,13 +83,20 @@ function eventsForSelectedSession() {
}
function renderSessions() {
if (state.sessions.length === 0) {
elements.sessionList.innerHTML = `<p class="muted">先创建一个项目会话。</p>`;
return;
}
elements.sessionList.innerHTML = state.sessions
.map((session) => {
const active = session.id === state.selectedSessionId ? "active" : "";
const archived = session.status === "archived" ? "archived" : "";
return `
<button class="session-item ${active}" data-session-id="${session.id}">
<button class="session-item ${active} ${archived}" data-session-id="${session.id}">
<strong>${escapeHtml(session.title)}</strong>
<span>${escapeHtml(session.activeObjective || "暂无目标")}</span>
<span class="muted">${escapeHtml(session.status)}</span>
</button>
`;
})
@@ -92,6 +112,11 @@ function renderSessions() {
}
function renderWorkers() {
if (state.workers.length === 0) {
elements.workerList.innerHTML = `<p class="muted">还没有 worker。可以手动注册，或直接运行 \`npm run demo\`。</p>`;
return;
}
elements.workerList.innerHTML = state.workers
.map(
(worker) => `
@@ -101,6 +126,8 @@ function renderWorkers() {
<span class="badge ${worker.status}">${escapeHtml(worker.status)}</span>
</div>
<div class="muted">${escapeHtml(worker.os)}</div>
<div class="muted">current: ${escapeHtml(worker.currentTaskId || "idle")}</div>
<div class="muted">last seen: ${new Date(worker.lastSeenAt).toLocaleTimeString()}</div>
<div class="tags">
${worker.capabilities.map((capability) => `<span>${escapeHtml(capability)}</span>`).join("")}
</div>
@@ -115,12 +142,21 @@ function renderSessionHeader() {
if (!session) {
elements.sessionTitleDisplay.textContent = "选择一个项目会话";
elements.sessionSummary.textContent = "创建会话后，在这里持续对话并观察任务状态。";
elements.archiveSession.disabled = true;
elements.messageInput.disabled = true;
elements.messageInput.placeholder = "先创建或选择一个项目会话。";
return;
}
elements.sessionTitleDisplay.textContent = session.title;
elements.sessionTitleDisplay.textContent = `${session.title} (${session.status})`;
elements.sessionSummary.textContent =
session.lastPlannerSummary || session.activeObjective || "等待用户输入。";
elements.archiveSession.disabled = session.status === "archived";
elements.messageInput.disabled = session.status === "archived";
elements.messageInput.placeholder =
session.status === "archived"
? "当前会话已归档，不能继续发送消息。"
: "输入需求。示例：先调研登录失败根因，不要急着改代码。";
}
function renderMessages() {
@@ -143,6 +179,18 @@ function renderMessages() {
}
function renderTasks() {
const latestPlan = [...state.events]
.reverse()
.find((event) => event.sessionId === state.selectedSessionId && event.type === "plan.created");
if (latestPlan) {
const pausedCount = Array.isArray(latestPlan.payload.pausedTaskIds)
? latestPlan.payload.pausedTaskIds.length
: 0;
elements.planHint.textContent = `最新计划创建了 ${latestPlan.payload.taskIds.length} 个任务，自动暂停旧任务 ${pausedCount} 个。`;
} else {
elements.planHint.textContent = "";
}
const tasks = tasksForSelectedSession();
elements.taskList.innerHTML = tasks.length
? tasks
@@ -157,9 +205,23 @@ function renderTasks() {
<div class="muted">worker: ${escapeHtml(task.assignedWorkerId || "未分配")}</div>
<div class="muted">progress: ${task.progressPercent}%</div>
<div class="muted">summary: ${escapeHtml(task.summary || "暂无")}</div>
<div class="muted">next: ${escapeHtml(task.nextStep || "暂无")}</div>
<div class="actions">
<button data-action="pause" data-task-id="${task.id}" class="ghost">暂停</button>
<button data-action="cancel" data-task-id="${task.id}" class="ghost danger">取消</button>
${
["queued", "assigned", "running"].includes(task.status)
? `<button data-action="pause" data-task-id="${task.id}" class="ghost">暂停</button>`
: ""
}
${
["paused", "blocked"].includes(task.status)
? `<button data-action="resume" data-task-id="${task.id}" class="ghost">续跑</button>`
: ""
}
${
!["completed", "cancelled", "failed"].includes(task.status)
? `<button data-action="cancel" data-task-id="${task.id}" class="ghost danger">取消</button>`
: ""
}
</div>
</div>
`,
@@ -173,6 +235,7 @@ function renderTasks() {
const action = button.dataset.action;
await request(`/api/tasks/${taskId}/${action}`, { method: "POST", body: "{}" });
await loadSession(state.selectedSessionId);
await loadBootstrap();
render();
});
});
@@ -209,6 +272,7 @@ function renderApprovals() {
body: JSON.stringify({ approved, responder: "web-user" }),
});
await loadSession(state.selectedSessionId);
await loadBootstrap();
render();
});
});
@@ -288,6 +352,25 @@ elements.createSessionForm.addEventListener("submit", async (event) => {
render();
});
elements.createWorkerForm.addEventListener("submit", async (event) => {
event.preventDefault();
const name = elements.workerName.value.trim();
if (!name) return;
const os = elements.workerOs.value;
const capabilities = elements.workerCapabilities.value
.split(",")
.map((item) => item.trim())
.filter(Boolean);
await request("/api/workers/register", {
method: "POST",
body: JSON.stringify({ name, os, capabilities }),
});
elements.workerName.value = "";
elements.workerCapabilities.value = "terminal";
await loadBootstrap();
render();
});
elements.messageForm.addEventListener("submit", async (event) => {
event.preventDefault();
if (!state.selectedSessionId) return;
@@ -303,6 +386,17 @@ elements.messageForm.addEventListener("submit", async (event) => {
render();
});
elements.archiveSession.addEventListener("click", async () => {
if (!state.selectedSessionId) return;
await request(`/api/sessions/${state.selectedSessionId}/archive`, {
method: "POST",
body: "{}",
});
await loadSession(state.selectedSessionId);
await loadBootstrap();
render();
});
elements.resetDemo.addEventListener("click", async () => {
await request("/api/demo/reset", { method: "POST", body: "{}" });
state.sessions = [];
@@ -330,4 +424,3 @@ loadBootstrap().then(render).catch((error) => {
console.error(error);
elements.sessionSummary.textContent = error.message;
});

13
public/index.html
View File

@@ -32,6 +32,16 @@
<div class="panel-header">
<h2>设备</h2>
</div>
<form id="create-worker-form" class="stack compact">
<input id="worker-name" placeholder="worker 名称" />
<select id="worker-os">
<option value="windows">windows</option>
<option value="macos">macos</option>
<option value="linux">linux</option>
</select>
<input id="worker-capabilities" placeholder="capabilities，逗号分隔" value="terminal" />
<button type="submit">注册 worker</button>
</form>
<div id="worker-list" class="list"></div>
</div>
</aside>
@@ -48,6 +58,7 @@
<div class="panel">
<div class="panel-header">
<h2>对话</h2>
<button id="archive-session" class="ghost">归档会话</button>
</div>
<div id="message-list" class="timeline"></div>
<form id="message-form" class="stack">
@@ -64,6 +75,7 @@
<div class="panel-header">
<h2>任务树</h2>
</div>
<div id="plan-hint" class="hint"></div>
<div id="task-list" class="list"></div>
</div>
</section>
@@ -89,4 +101,3 @@
<script type="module" src="/app.js"></script>
</body>
</html>

30
public/styles.css
View File

@@ -25,7 +25,8 @@ body {
button,
input,
textarea {
textarea,
select {
font: inherit;
}
@@ -38,6 +39,11 @@ button {
cursor: pointer;
}
button:disabled {
opacity: 0.55;
cursor: default;
}
button.ghost {
background: transparent;
color: var(--text);
@@ -49,7 +55,8 @@ button.danger {
}
input,
textarea {
textarea,
select {
width: 100%;
border-radius: 12px;
border: 1px solid var(--line);
@@ -108,6 +115,10 @@ textarea {
gap: 0.75rem;
}
.stack.compact {
margin-bottom: 1rem;
}
.list,
.timeline {
display: grid;
@@ -143,6 +154,10 @@ textarea {
outline: 2px solid rgba(31, 111, 235, 0.2);
}
.session-item.archived {
opacity: 0.7;
}
.message header,
.event header {
display: flex;
@@ -171,7 +186,8 @@ textarea {
.badge.failed,
.badge.cancelled,
.badge.rejected {
.badge.rejected,
.badge.offline {
color: var(--danger);
}
@@ -202,6 +218,14 @@ textarea {
color: var(--muted);
}
.hint {
margin-bottom: 0.75rem;
padding: 0.75rem 0.875rem;
border-radius: 14px;
background: rgba(31, 111, 235, 0.08);
color: var(--accent);
}
pre {
margin: 0;
white-space: pre-wrap;

129
src/demo.ts Normal file
View File

@@ -0,0 +1,129 @@
import { spawn, type ChildProcess } from "node:child_process";
import { setTimeout as delay } from "node:timers/promises";
const children: ChildProcess[] = [];
function run(command: string, args: string[]) {
const child = spawn(command, args, {
stdio: "inherit",
shell: false,
env: {
...process.env,
PORT: process.env.PORT ?? "43210",
},
});
child.on("exit", (code, signal) => {
console.log(`[demo] child exited`, { code, signal, command, args: args.join(" ") });
if (code && code !== 0 && signal !== "SIGINT" && signal !== "SIGTERM") {
shutdown("SIGTERM");
}
});
children.push(child);
return child;
}
async function waitForHealth(url: string) {
for (let attempt = 0; attempt < 30; attempt += 1) {
try {
const response = await fetch(`${url}/api/health`);
if (response.ok) {
return;
}
} catch {
// wait and retry
}
await delay(1_000);
}
throw new Error(`Server did not become healthy: ${url}`);
}
async function isHealthy(url: string) {
try {
const response = await fetch(`${url}/api/health`);
if (!response.ok) {
return false;
}
const payload = await response.json();
return payload?.status === "ok";
} catch {
return false;
}
}
function shutdown(signal: NodeJS.Signals) {
for (const child of children) {
if (!child.killed) {
child.kill(signal);
}
}
process.exit(0);
}
async function main() {
const serverUrl = `http://127.0.0.1:${process.env.PORT ?? "43210"}`;
const hasExisting = await isHealthy(serverUrl);
if (!hasExisting) {
run(process.execPath, ["./node_modules/tsx/dist/cli.mjs", "src/server.ts"]);
await waitForHealth(serverUrl);
} else {
console.log(`Boss server already running at ${serverUrl}, reusing it.`);
}
run(process.execPath, [
"./node_modules/tsx/dist/cli.mjs",
"src/worker.ts",
"--name",
"win-a",
"--os",
"windows",
"--capability",
"terminal",
"--capability",
"browser",
"--server",
serverUrl,
]);
run(process.execPath, [
"./node_modules/tsx/dist/cli.mjs",
"src/worker.ts",
"--name",
"win-b",
"--os",
"windows",
"--capability",
"terminal",
"--capability",
"test",
"--server",
serverUrl,
]);
run(process.execPath, [
"./node_modules/tsx/dist/cli.mjs",
"src/worker.ts",
"--name",
"mac-a",
"--os",
"macos",
"--capability",
"terminal",
"--capability",
"browser",
"--capability",
"test",
"--server",
serverUrl,
]);
console.log(`Boss demo running at ${serverUrl}`);
console.log("Press Ctrl+C to stop server and workers.");
await new Promise(() => {});
}
process.on("SIGINT", () => shutdown("SIGINT"));
process.on("SIGTERM", () => shutdown("SIGTERM"));
main().catch((error) => {
console.error(error);
shutdown("SIGTERM");
});

455
src/engine.ts
View File

@@ -26,6 +26,7 @@ export class BossEngine {
readonly events = new EventBroker();
getState(): AppState {
this.reconcileState();
return this.store.snapshot;
}
@@ -45,20 +46,17 @@ export class BossEngine {
updatedAt: timestamp,
};
this.store.mutate((state) => {
this.commit((state, addEvent) => {
state.sessions.unshift(session);
state.events.push(
this.makeEvent({
addEvent({
sessionId: session.id,
taskId: null,
source: "system",
type: "session.created",
payload: { title: session.title },
}),
);
});
});
this.publishLatestEvent();
return this.getSession(session.id);
}
@@ -71,18 +69,76 @@ export class BossEngine {
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),
messages: state.messages
.filter((message) => message.sessionId === sessionId)
.sort((left, right) => left.createdAt.localeCompare(right.createdAt)),
tasks: state.tasks
.filter((task) => task.sessionId === sessionId)
.sort((left, right) => left.createdAt.localeCompare(right.createdAt)),
approvals: state.approvals
.filter((approval) => approval.sessionId === sessionId)
.sort((left, right) => left.createdAt.localeCompare(right.createdAt)),
};
}
listSessions(): Session[] {
return this.getState().sessions;
return this.getState().sessions.sort((left, right) => right.updatedAt.localeCompare(left.updatedAt));
}
getTask(taskId: string): Task {
const task = this.getState().tasks.find((candidate) => candidate.id === taskId);
if (!task) {
throw new Error(`Task not found: ${taskId}`);
}
return task;
}
archiveSession(sessionId: string): SessionDetails {
this.commit((state, addEvent) => {
const session = state.sessions.find((candidate) => candidate.id === sessionId);
if (!session) {
throw new Error(`Session not found: ${sessionId}`);
}
const timestamp = now();
session.status = "archived";
session.updatedAt = timestamp;
for (const task of state.tasks.filter(
(candidate) => candidate.sessionId === sessionId && isActiveTask(candidate),
)) {
if (!["completed", "failed", "cancelled"].includes(task.status)) {
this.detachTaskFromWorker(state, task, timestamp);
task.status = "paused";
task.summary = "会话已归档，任务暂停。";
task.updatedAt = timestamp;
addEvent({
sessionId,
taskId: task.id,
source: "system",
type: "task.paused",
payload: { reason: "session_archived" },
});
}
}
addEvent({
sessionId,
taskId: null,
source: "system",
type: "session.archived",
payload: { sessionId },
});
});
return this.getSession(sessionId);
}
addMessage(sessionId: string, content: string, channel = "web"): SessionDetails {
const session = this.getSession(sessionId).session;
if (session.status === "archived") {
throw new Error(`Session ${sessionId} is archived`);
}
const message: Message = {
id: createId("msg"),
sessionId,
@@ -96,7 +152,7 @@ export class BossEngine {
throw new Error("Message content is required.");
}
this.store.mutate((state) => {
this.commit((state, addEvent) => {
const mutableSession = state.sessions.find((candidate) => candidate.id === sessionId);
if (!mutableSession) {
throw new Error(`Session not found: ${sessionId}`);
@@ -108,8 +164,7 @@ export class BossEngine {
mutableSession.title = message.content.slice(0, 32);
}
state.messages.push(message);
state.events.push(
this.makeEvent({
addEvent({
sessionId,
taskId: null,
source: "user",
@@ -118,11 +173,9 @@ export class BossEngine {
channel,
content: message.content,
},
}),
);
});
});
this.publishLatestEvent();
this.applyPlan(session, message.content);
return this.getSession(sessionId);
}
@@ -156,10 +209,9 @@ export class BossEngine {
updatedAt: timestamp,
};
this.store.mutate((state) => {
this.commit((state, addEvent) => {
state.workers.push(worker);
state.events.push(
this.makeEvent({
addEvent({
sessionId: null,
taskId: null,
source: "system",
@@ -170,11 +222,9 @@ export class BossEngine {
os: worker.os,
capabilities: worker.capabilities,
},
}),
);
});
});
this.publishLatestEvent();
this.syncAssignments();
return worker;
}
@@ -201,9 +251,68 @@ export class BossEngine {
return updated;
}
getWorker(workerId: string): WorkerNode {
const worker = this.getState().workers.find((candidate) => candidate.id === workerId);
if (!worker) {
throw new Error(`Worker not found: ${workerId}`);
}
return worker;
}
markWorkerOffline(workerId: string): WorkerNode {
let updated!: WorkerNode;
this.commit((state, addEvent) => {
const worker = state.workers.find((candidate) => candidate.id === workerId);
if (!worker) {
throw new Error(`Worker not found: ${workerId}`);
}
const timestamp = now();
for (const task of state.tasks.filter(
(candidate) =>
candidate.assignedWorkerId === worker.id &&
["assigned", "running"].includes(candidate.status),
)) {
this.requeueTaskState(state, task, timestamp, `${worker.name} 被手动下线`);
addEvent({
sessionId: task.sessionId,
taskId: task.id,
source: "system",
type: "task.requeued",
payload: {
workerId: worker.id,
workerName: worker.name,
reason: "manual_worker_offline",
},
});
}
worker.status = "offline";
worker.currentTaskId = null;
worker.updatedAt = timestamp;
worker.lastSeenAt = timestamp;
addEvent({
sessionId: null,
taskId: null,
source: "system",
type: "worker.offline",
payload: {
workerId: worker.id,
workerName: worker.name,
reason: "manual",
},
});
updated = { ...worker };
});
this.syncAssignments();
return updated;
}
heartbeat(workerId: string, load = 0): WorkerNode {
let updated!: WorkerNode;
this.store.mutate((state) => {
this.commit((state, addEvent) => {
const worker = state.workers.find((candidate) => candidate.id === workerId);
if (!worker) {
throw new Error(`Worker not found: ${workerId}`);
@@ -216,8 +325,7 @@ export class BossEngine {
worker.status = "idle";
}
state.events.push(
this.makeEvent({
addEvent({
sessionId: null,
taskId: worker.currentTaskId,
source: "worker",
@@ -227,13 +335,11 @@ export class BossEngine {
status: worker.status,
load: worker.load,
},
}),
);
});
updated = { ...worker };
});
this.publishLatestEvent();
this.syncAssignments();
return updated;
}
@@ -241,7 +347,7 @@ export class BossEngine {
claimNextTask(workerId: string): Task | null {
let claimedTask: Task | null = null;
this.store.mutate((state) => {
this.commit((state, addEvent) => {
const worker = state.workers.find((candidate) => candidate.id === workerId);
if (!worker) {
throw new Error(`Worker not found: ${workerId}`);
@@ -264,8 +370,7 @@ export class BossEngine {
worker.lastSeenAt = task.updatedAt;
claimedTask = { ...task };
state.events.push(
this.makeEvent({
addEvent({
sessionId: task.sessionId,
taskId: task.id,
source: "worker",
@@ -274,13 +379,8 @@ export class BossEngine {
workerId,
title: task.title,
},
}),
);
});
if (claimedTask) {
this.publishLatestEvent();
}
});
return claimedTask;
}
@@ -296,11 +396,16 @@ export class BossEngine {
},
): Task {
let updated!: Task;
this.store.mutate((state) => {
this.commit((state, addEvent) => {
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}`);
}
this.assertTaskOwnership(task, workerId, worker.currentTaskId);
task.progressPercent = Math.max(0, Math.min(100, input.progressPercent));
task.summary = input.summary;
@@ -311,8 +416,7 @@ export class BossEngine {
task.status = "running";
}
state.events.push(
this.makeEvent({
addEvent({
sessionId: task.sessionId,
taskId: task.id,
source: "worker",
@@ -324,19 +428,17 @@ export class BossEngine {
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) => {
this.commit((state, addEvent) => {
const task = state.tasks.find((candidate) => candidate.id === taskId);
const worker = state.workers.find((candidate) => candidate.id === workerId);
if (!task) {
@@ -345,6 +447,7 @@ export class BossEngine {
if (!worker) {
throw new Error(`Worker not found: ${workerId}`);
}
this.assertTaskOwnership(task, workerId, worker.currentTaskId);
task.status = "completed";
task.progressPercent = 100;
@@ -357,8 +460,7 @@ export class BossEngine {
worker.updatedAt = task.updatedAt;
worker.lastSeenAt = task.updatedAt;
state.events.push(
this.makeEvent({
addEvent({
sessionId: task.sessionId,
taskId: task.id,
source: "worker",
@@ -367,12 +469,10 @@ export class BossEngine {
workerId,
summary,
},
}),
);
});
updated = { ...task };
});
this.publishLatestEvent();
this.syncAssignments();
return updated;
}
@@ -380,7 +480,7 @@ export class BossEngine {
failTask(taskId: string, workerId: string, errorMessage: string): Task {
let updated!: Task;
this.store.mutate((state) => {
this.commit((state, addEvent) => {
const task = state.tasks.find((candidate) => candidate.id === taskId);
const worker = state.workers.find((candidate) => candidate.id === workerId);
if (!task) {
@@ -389,6 +489,7 @@ export class BossEngine {
if (!worker) {
throw new Error(`Worker not found: ${workerId}`);
}
this.assertTaskOwnership(task, workerId, worker.currentTaskId);
task.status = "failed";
task.summary = errorMessage;
@@ -400,8 +501,7 @@ export class BossEngine {
worker.updatedAt = task.updatedAt;
worker.lastSeenAt = task.updatedAt;
state.events.push(
this.makeEvent({
addEvent({
sessionId: task.sessionId,
taskId: task.id,
source: "worker",
@@ -410,12 +510,10 @@ export class BossEngine {
workerId,
errorMessage,
},
}),
);
});
updated = { ...task };
});
this.publishLatestEvent();
this.syncAssignments();
return updated;
}
@@ -432,10 +530,70 @@ export class BossEngine {
});
}
resumeTask(taskId: string): Task {
let updated!: Task;
this.commit((state, addEvent) => {
const task = state.tasks.find((candidate) => candidate.id === taskId);
if (!task) {
throw new Error(`Task not found: ${taskId}`);
}
if (task.status !== "paused") {
updated = { ...task };
return;
}
task.assignedWorkerId = null;
task.status = task.approvalStatus === "pending" ? "waiting_approval" : "queued";
task.summary = "任务已恢复，等待重新调度。";
task.updatedAt = now();
addEvent({
sessionId: task.sessionId,
taskId: task.id,
source: "system",
type: "task.resumed",
payload: {
summary: task.summary,
},
});
updated = { ...task };
});
this.syncAssignments();
return updated;
}
requeueTask(taskId: string): Task {
let updated!: Task;
this.commit((state, addEvent) => {
const task = state.tasks.find((candidate) => candidate.id === taskId);
if (!task) {
throw new Error(`Task not found: ${taskId}`);
}
this.requeueTaskState(state, task, now(), "手动重排");
addEvent({
sessionId: task.sessionId,
taskId: task.id,
source: "system",
type: "task.requeued",
payload: {
summary: task.summary,
},
});
updated = { ...task };
});
this.syncAssignments();
return updated;
}
respondApproval(approvalId: string, approved: boolean, responder: string): ApprovalRequest {
let updatedApproval!: ApprovalRequest;
this.store.mutate((state) => {
this.commit((state, addEvent) => {
const approval = state.approvals.find((candidate) => candidate.id === approvalId);
if (!approval) {
throw new Error(`Approval not found: ${approvalId}`);
@@ -455,8 +613,7 @@ export class BossEngine {
task.status = approved ? "queued" : "cancelled";
task.summary = approved ? "审批已通过，重新进入队列。" : "审批被拒绝，任务已取消。";
state.events.push(
this.makeEvent({
addEvent({
sessionId: approval.sessionId,
taskId: approval.taskId,
source: "system",
@@ -465,16 +622,25 @@ export class BossEngine {
approvalId,
responder,
},
}),
);
});
updatedApproval = { ...approval };
});
this.publishLatestEvent();
this.syncAssignments();
return updatedApproval;
}
listEvents(limit = 100): BossEvent[] {
const events = this.getState().events;
return events.slice(Math.max(0, events.length - limit));
}
reconcileNow(): AppState {
this.reconcileState();
this.syncAssignments();
return this.getState();
}
private applyPlan(session: Session, content: string): void {
const sessionDetails = this.getSession(session.id);
const result = createPlan(sessionDetails.session, content, sessionDetails.tasks.filter(isActiveTask));
@@ -482,7 +648,7 @@ export class BossEngine {
const plannerMessage = buildPlannerMessage(result.summary);
const timestamp = now();
this.store.mutate((state) => {
this.commit((state, addEvent) => {
const mutableSession = state.sessions.find((candidate) => candidate.id === session.id);
if (!mutableSession) {
throw new Error(`Session not found: ${session.id}`);
@@ -497,8 +663,19 @@ export class BossEngine {
(candidate) => candidate.sessionId === session.id && isActiveTask(candidate),
)) {
if (["running", "assigned", "queued", "planning", "blocked"].includes(task.status)) {
this.detachTaskFromWorker(state, task, timestamp);
task.status = "paused";
task.summary = "检测到新需求，旧任务已暂停。";
task.updatedAt = timestamp;
addEvent({
sessionId: task.sessionId,
taskId: task.id,
source: "manager",
type: "task.paused",
payload: {
reason: "replan",
},
});
}
}
}
@@ -532,8 +709,7 @@ export class BossEngine {
};
state.messages.push(managerMessage);
state.events.push(
this.makeEvent({
addEvent({
sessionId: session.id,
taskId: null,
source: "manager",
@@ -542,11 +718,9 @@ export class BossEngine {
summary: result.summary,
taskIds: tasks.map((task) => task.id),
},
}),
);
});
});
this.publishLatestEvent();
this.syncAssignments();
}
@@ -558,7 +732,7 @@ export class BossEngine {
payload: Record<string, unknown>,
): Task {
let updated!: Task;
this.store.mutate((state) => {
this.commit((state, addEvent) => {
const task = state.tasks.find((candidate) => candidate.id === taskId);
if (!task) {
throw new Error(`Task not found: ${taskId}`);
@@ -567,41 +741,29 @@ export class BossEngine {
task.status = status;
task.updatedAt = now();
task.summary = typeof payload.summary === "string" ? payload.summary : task.summary;
this.detachTaskFromWorker(state, task, task.updatedAt);
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({
addEvent({
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());
const candidates = chooseAssignmentCandidates(this.store.snapshot);
if (candidates.length === 0) {
return;
}
this.store.mutate((state) => {
this.commit((state, addEvent) => {
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);
@@ -619,8 +781,7 @@ export class BossEngine {
worker.updatedAt = timestamp;
worker.lastSeenAt = timestamp;
state.events.push(
this.makeEvent({
addEvent({
sessionId: task.sessionId,
taskId: task.id,
source: "system",
@@ -629,19 +790,98 @@ export class BossEngine {
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 reconcileState(): void {
const staleAfterMs = 20_000;
const currentTime = Date.now();
this.commit((state, addEvent) => {
for (const worker of state.workers) {
const age = currentTime - new Date(worker.lastSeenAt).getTime();
if (age <= staleAfterMs || worker.status === "offline") {
continue;
}
const timestamp = now();
worker.status = "offline";
worker.updatedAt = timestamp;
for (const task of state.tasks.filter(
(candidate) =>
candidate.assignedWorkerId === worker.id &&
["assigned", "running"].includes(candidate.status),
)) {
this.requeueTaskState(state, task, timestamp, `worker ${worker.name} 离线`);
addEvent({
sessionId: task.sessionId,
taskId: task.id,
source: "system",
type: "task.requeued",
payload: {
workerId: worker.id,
workerName: worker.name,
reason: "heartbeat_timeout",
},
});
}
worker.currentTaskId = null;
addEvent({
sessionId: null,
taskId: null,
source: "system",
type: "worker.offline",
payload: {
workerId: worker.id,
workerName: worker.name,
reason: "heartbeat_timeout",
},
});
}
});
}
private detachTaskFromWorker(state: AppState, task: Task, timestamp: string): void {
if (!task.assignedWorkerId) {
return;
}
const worker = state.workers.find((candidate) => candidate.id === task.assignedWorkerId);
if (worker && worker.currentTaskId === task.id) {
worker.currentTaskId = null;
if (worker.status !== "offline") {
worker.status = "idle";
}
worker.updatedAt = timestamp;
worker.lastSeenAt = timestamp;
}
task.assignedWorkerId = null;
}
private requeueTaskState(state: AppState, task: Task, timestamp: string, reason: string): void {
this.detachTaskFromWorker(state, task, timestamp);
task.status = task.approvalStatus === "pending" ? "waiting_approval" : "queued";
task.summary = `${reason}，任务已重新排队。`;
task.currentStep = "requeued";
task.nextStep = "等待新 worker 认领";
task.updatedAt = timestamp;
}
private assertTaskOwnership(task: Task, workerId: string, workerCurrentTaskId: string | null): void {
if (task.assignedWorkerId !== workerId) {
throw new Error(`Task ${task.id} is not assigned to worker ${workerId}`);
}
if (workerCurrentTaskId !== task.id) {
throw new Error(`Worker ${workerId} is not currently executing task ${task.id}`);
}
if (!["assigned", "running"].includes(task.status)) {
throw new Error(`Task ${task.id} does not accept worker updates in status ${task.status}`);
}
}
@@ -652,5 +892,26 @@ export class BossEngine {
...input,
};
}
}
private commit<T>(
mutator: (
state: AppState,
addEvent: (input: Omit<BossEvent, "id" | "timestamp">) => void,
) => T,
): T {
const published: BossEvent[] = [];
const result = this.store.mutate((state) =>
mutator(state, (input) => {
const event = this.makeEvent(input);
state.events.push(event);
published.push(event);
}),
);
for (const event of published) {
this.events.publish(event);
}
return result;
}
}

70
src/server.ts
View File

@@ -4,7 +4,37 @@ import fastifyStatic from "@fastify/static";
import { BossEngine } from "./engine.js";
const engine = new BossEngine();
const app = Fastify({ logger: true });
const app = Fastify({ logger: process.env.BOSS_DEBUG === "1" });
app.setErrorHandler((error, request, reply) => {
const message =
typeof error === "object" && error !== null && "message" in error
? String(error.message)
: "Internal Server Error";
const normalized = message.toLowerCase();
if (normalized.includes("not found")) {
return reply.status(404).send({ error: "Not Found", message });
}
if (
normalized.includes("is not assigned to worker") ||
normalized.includes("is not currently executing") ||
normalized.includes("does not accept worker updates")
) {
return reply.status(409).send({ error: "Conflict", message });
}
if (
normalized.includes("required") ||
normalized.includes("archived")
) {
return reply.status(400).send({ error: "Bad Request", message });
}
request.log.error(error);
return reply.status(500).send({ error: "Internal Server Error", message });
});
await app.register(fastifyStatic, {
root: path.resolve(process.cwd(), "public"),
@@ -23,6 +53,12 @@ app.get("/api/health", async () => ({
app.get("/api/bootstrap", async () => engine.bootstrap());
app.get("/api/events", async (request) => {
const query = request.query as { limit?: string };
const limit = Number(query.limit ?? 100);
return engine.listEvents(Number.isFinite(limit) ? limit : 100);
});
app.get("/api/sessions", async () => engine.listSessions());
app.post("/api/sessions", async (request) => {
@@ -30,11 +66,21 @@ app.post("/api/sessions", async (request) => {
return engine.createSession(body.title);
});
app.post("/api/sessions/:sessionId/archive", async (request) => {
const params = request.params as { sessionId: string };
return engine.archiveSession(params.sessionId);
});
app.get("/api/sessions/:sessionId", async (request) => {
const params = request.params as { sessionId: string };
return engine.getSession(params.sessionId);
});
app.get("/api/tasks/:taskId", async (request) => {
const params = request.params as { taskId: string };
return engine.getTask(params.taskId);
});
app.post("/api/sessions/:sessionId/messages", async (request) => {
const params = request.params as { sessionId: string };
const body = (request.body ?? {}) as { content?: string; channel?: string };
@@ -64,6 +110,11 @@ app.get("/api/events/stream", async (_request, reply) => {
app.get("/api/workers", async () => engine.getState().workers);
app.get("/api/workers/:workerId", async (request) => {
const params = request.params as { workerId: string };
return engine.getWorker(params.workerId);
});
app.post("/api/workers/register", async (request) => {
const body = request.body as {
name?: string;
@@ -83,6 +134,11 @@ app.post("/api/workers/:workerId/heartbeat", async (request) => {
return engine.heartbeat(params.workerId, body.load ?? 0);
});
app.post("/api/workers/:workerId/offline", async (request) => {
const params = request.params as { workerId: string };
return engine.markWorkerOffline(params.workerId);
});
app.post("/api/workers/:workerId/claim-next", async (request) => {
const params = request.params as { workerId: string };
return {
@@ -135,6 +191,16 @@ app.post("/api/tasks/:taskId/cancel", async (request) => {
return engine.cancelTask(params.taskId);
});
app.post("/api/tasks/:taskId/resume", async (request) => {
const params = request.params as { taskId: string };
return engine.resumeTask(params.taskId);
});
app.post("/api/tasks/:taskId/requeue", async (request) => {
const params = request.params as { taskId: string };
return engine.requeueTask(params.taskId);
});
app.post("/api/approvals/:approvalId/respond", async (request) => {
const params = request.params as { approvalId: string };
const body = request.body as {
@@ -149,5 +215,7 @@ app.post("/api/demo/reset", async () => {
return ok();
});
app.post("/api/reconcile", async () => engine.reconcileNow());
const port = Number(process.env.PORT ?? 43210);
await app.listen({ port, host: "0.0.0.0" });

84
src/smoke.ts Normal file
View File

@@ -0,0 +1,84 @@
import { setTimeout as delay } from "node:timers/promises";
const baseUrl = process.env.BOSS_BASE_URL || "http://127.0.0.1:43210";
async function request(path: string, options: RequestInit = {}) {
const response = await fetch(`${baseUrl}${path}`, {
headers: { "Content-Type": "application/json" },
...options,
});
if (!response.ok) {
throw new Error(`${response.status} ${response.statusText}`);
}
return response.json();
}
async function waitForSessionSettled(sessionId: string, timeoutMs = 30_000) {
const start = Date.now();
while (Date.now() - start < timeoutMs) {
const session = await request(`/api/sessions/${sessionId}`);
const pendingApproval = session.approvals.find((approval: { status: string }) => approval.status === "pending");
if (pendingApproval) {
await request(`/api/approvals/${pendingApproval.id}/respond`, {
method: "POST",
body: JSON.stringify({ approved: true, responder: "smoke-script" }),
});
}
const unsettled = session.tasks.filter((task: { status: string }) =>
["planning", "queued", "assigned", "running", "waiting_approval"].includes(task.status),
);
if (unsettled.length === 0) {
return session;
}
await delay(1_000);
}
throw new Error("Session did not settle in time.");
}
async function main() {
await request("/api/demo/reset", { method: "POST", body: "{}" });
const created = await request("/api/sessions", {
method: "POST",
body: JSON.stringify({ title: "Smoke Test" }),
});
const sessionId = created.session.id;
await request(`/api/sessions/${sessionId}/messages`, {
method: "POST",
body: JSON.stringify({
content: "先调研登录失败根因，然后 delete 旧缓存目录并做验证。",
channel: "smoke",
}),
});
const settled = await waitForSessionSettled(sessionId);
console.log(
JSON.stringify(
{
sessionId,
taskStatuses: settled.tasks.map((task: { title: string; status: string }) => ({
title: task.title,
status: task.status,
})),
approvalStatuses: settled.approvals.map((approval: { id: string; status: string }) => ({
id: approval.id,
status: approval.status,
})),
},
null,
2,
),
);
}
main().catch((error) => {
console.error(error);
process.exit(1);
});

26
src/store.ts
View File

@@ -1,4 +1,4 @@
import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs";
import { copyFileSync, existsSync, mkdirSync, readFileSync, renameSync, writeFileSync } from "node:fs";
import { dirname } from "node:path";
import type { AppState } from "./types.js";
@@ -15,8 +15,12 @@ function defaultState(): AppState {
export class FileStore {
private state: AppState;
private readonly backupFilePath: string;
private readonly tempFilePath: string;
constructor(private readonly filePath: string) {
this.backupFilePath = `${filePath}.bak`;
this.tempFilePath = `${filePath}.tmp`;
this.ensureDirectory();
this.state = this.load();
}
@@ -42,20 +46,28 @@ export class FileStore {
}
private load(): AppState {
if (!existsSync(this.filePath)) {
return defaultState();
for (const path of [this.filePath, this.backupFilePath]) {
if (!existsSync(path)) {
continue;
}
try {
const raw = readFileSync(this.filePath, "utf8");
const raw = readFileSync(path, "utf8");
return { ...defaultState(), ...(JSON.parse(raw) as AppState) };
} catch {
return defaultState();
// try backup/default
}
}
return defaultState();
}
private save(): void {
writeFileSync(this.filePath, `${JSON.stringify(this.state, null, 2)}\n`, "utf8");
const content = `${JSON.stringify(this.state, null, 2)}\n`;
writeFileSync(this.tempFilePath, content, "utf8");
if (existsSync(this.filePath)) {
copyFileSync(this.filePath, this.backupFilePath);
}
renameSync(this.tempFilePath, this.filePath);
}
}

19
src/worker.ts
View File

@@ -54,6 +54,19 @@ async function postJson(url: string, body: unknown) {
return response.json();
}
async function getJson(url: string) {
const response = await fetch(url);
if (!response.ok) {
throw new Error(`Request failed: ${response.status} ${response.statusText}`);
}
return response.json();
}
async function taskStillRunnable(server: string, taskId: string) {
const task = await getJson(`${server}/api/tasks/${taskId}`);
return ["assigned", "running"].includes(task.status);
}
async function simulateTask(server: string, workerId: string, task: Task) {
const steps = [
{
@@ -77,6 +90,9 @@ async function simulateTask(server: string, workerId: string, task: Task) {
];
for (const step of steps) {
if (!(await taskStillRunnable(server, task.id))) {
return;
}
await delay(1_500);
await postJson(`${server}/api/tasks/${task.id}/progress`, {
workerId,
@@ -84,6 +100,9 @@ async function simulateTask(server: string, workerId: string, task: Task) {
});
}
if (!(await taskStillRunnable(server, task.id))) {
return;
}
await delay(1_000);
await postJson(`${server}/api/tasks/${task.id}/complete`, {
workerId,