JianMiauandClaude Fable 5
b12627a9ef
摘要： 側邊面板從單向任務派發升級為聊天室:每則訊息 spawn claude -p --output-format stream-json,捕捉 session_id 後 下一則自動 --resume 接力,對話得以連續;回覆同步顯示在 聊天串與小人頭上的對話氣泡。 根本原因： 單向 /task 發出後看不到回覆內容,也無法追問;claudecodeui 證明 headless + resume 接力即可實現連續對話,無需常駐 runtime。 影響： watcher 新增 POST /chat(每專案一個 office thread、busy 檢查、 reset 會終止進行中的執行);SSE 新增 chat_start/chat_text/chat_done。 /task API 保留。已實測兩輪對話,第二輪正確記得第一輪內容。 修法： - watch.mjs:launchChat 解析 stream-json(init 抓 session_id、 assistant 文字轉發、result 收尾),chatThreads 記憶 thread - store:chats Map + chatAppend/chatSetBusy/chatClear - live.ts:sendChat/resetChat + chat_* 事件對映 - SidePanel:ChatPanel(氣泡串、自動捲底、Enter 送出、新對話) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
583 lines
17 KiB
JavaScript
583 lines
17 KiB
JavaScript
/**
|
||
* Claude Office live watcher.
|
||
*
|
||
* Tails the .jsonl session transcripts under ~/.claude/projects/ and
|
||
* translates appended lines into office events, streamed to the frontend
|
||
* over Server-Sent Events at http://localhost:5179/events.
|
||
*
|
||
* Zero dependencies — plain Node 18+.
|
||
*/
|
||
import { createServer } from "node:http";
|
||
import { createReadStream, promises as fs } from "node:fs";
|
||
import { spawn } from "node:child_process";
|
||
import { homedir } from "node:os";
|
||
import path from "node:path";
|
||
|
||
const PORT = Number(process.env.PORT || 5179);
|
||
const POLL_MS = 1500;
|
||
const ACTIVE_WINDOW_DAYS = 7;
|
||
|
||
/** Origins allowed to dispatch tasks (browser CSRF guard; server is localhost-only). */
|
||
const ALLOWED_ORIGINS = new Set([
|
||
"http://localhost:5173",
|
||
"http://127.0.0.1:5173",
|
||
"http://localhost:5180",
|
||
"http://127.0.0.1:5180",
|
||
]);
|
||
const TASK_TIMEOUT_MS = 15 * 60_000;
|
||
const MAX_QUEUE = 3;
|
||
/** Headless permission mode — change to taste (acceptEdits / plan / bypassPermissions). */
|
||
const PERMISSION_MODE = process.env.OFFICE_PERMISSION_MODE || "acceptEdits";
|
||
|
||
const PROJECTS_ROOT = path.join(homedir(), ".claude", "projects");
|
||
|
||
/** slug → { slug, name, cwd, lastActivity } */
|
||
const projects = new Map();
|
||
/** absolute file path → { offset, remainder, slug } */
|
||
const tails = new Map();
|
||
/** tool_use_id → { slug, desc, kind, startedAt } — pending Agent (subagent) calls */
|
||
const pendingSubs = new Map();
|
||
|
||
const clients = new Set();
|
||
|
||
function broadcast(event) {
|
||
const data = `data: ${JSON.stringify(event)}\n\n`;
|
||
for (const res of clients) res.write(data);
|
||
}
|
||
|
||
function prettifyName(slug, cwd) {
|
||
if (cwd) {
|
||
const seg = cwd.split(/[\\/]/).filter(Boolean);
|
||
if (seg.length > 0) return seg[seg.length - 1];
|
||
}
|
||
const parts = slug.split("-").filter(Boolean);
|
||
return parts[parts.length - 1] || slug;
|
||
}
|
||
|
||
/** Read the tail of the newest transcript to discover the project cwd. */
|
||
async function sniffCwd(file) {
|
||
try {
|
||
const stat = await fs.stat(file);
|
||
const start = Math.max(0, stat.size - 16384);
|
||
const buf = await readRange(file, start, stat.size);
|
||
const lines = buf.toString("utf8").split("\n");
|
||
for (let i = lines.length - 1; i >= 0; i--) {
|
||
const m = lines[i].match(/"cwd":"((?:[^"\\]|\\.)*)"/);
|
||
if (m) return JSON.parse(`"${m[1]}"`);
|
||
}
|
||
} catch {
|
||
/* ignore */
|
||
}
|
||
return null;
|
||
}
|
||
|
||
function readRange(file, start, end) {
|
||
return new Promise((resolve, reject) => {
|
||
const chunks = [];
|
||
createReadStream(file, { start, end: Math.max(start, end - 1) })
|
||
.on("data", (c) => chunks.push(c))
|
||
.on("end", () => resolve(Buffer.concat(chunks)))
|
||
.on("error", reject);
|
||
});
|
||
}
|
||
|
||
function textOf(content) {
|
||
if (typeof content === "string") return content;
|
||
if (!Array.isArray(content)) return "";
|
||
return content
|
||
.filter((c) => c && c.type === "text" && typeof c.text === "string")
|
||
.map((c) => c.text)
|
||
.join(" ");
|
||
}
|
||
|
||
function handleLine(slug, line) {
|
||
let o;
|
||
try {
|
||
o = JSON.parse(line);
|
||
} catch {
|
||
return;
|
||
}
|
||
const proj = projects.get(slug);
|
||
if (proj) {
|
||
proj.lastActivity = Date.now();
|
||
if (!proj.cwd && typeof o.cwd === "string") {
|
||
proj.cwd = o.cwd;
|
||
proj.name = prettifyName(slug, o.cwd);
|
||
}
|
||
}
|
||
|
||
const msg = o.message;
|
||
const content = msg?.content;
|
||
|
||
// Subagent completion: any tool_result whose id matches a pending Agent call.
|
||
if (Array.isArray(content)) {
|
||
for (const c of content) {
|
||
if (c && c.type === "tool_result" && c.tool_use_id && pendingSubs.has(c.tool_use_id)) {
|
||
const sub = pendingSubs.get(c.tool_use_id);
|
||
pendingSubs.delete(c.tool_use_id);
|
||
broadcast({ type: "subagent_end", slug: sub.slug, subId: c.tool_use_id });
|
||
}
|
||
}
|
||
}
|
||
|
||
// Skip subagent-internal lines for the main pawn's status.
|
||
if (o.isSidechain) return;
|
||
|
||
if (o.type === "assistant" && msg?.role === "assistant") {
|
||
if (Array.isArray(content)) {
|
||
for (const c of content) {
|
||
if (!c || c.type !== "tool_use") continue;
|
||
if (c.name === "Agent" || c.name === "Task") {
|
||
const input = c.input ?? {};
|
||
const sub = {
|
||
slug,
|
||
desc: input.description || input.prompt?.slice(0, 40) || "subagent",
|
||
kind: input.subagent_type || "Task",
|
||
startedAt: Date.now(),
|
||
};
|
||
pendingSubs.set(c.id, sub);
|
||
broadcast({ type: "subagent_spawn", slug, subId: c.id, desc: sub.desc, kind: sub.kind });
|
||
} else {
|
||
broadcast({ type: "tool", slug, tool: c.name });
|
||
}
|
||
}
|
||
}
|
||
const text = textOf(content).trim();
|
||
if (text) broadcast({ type: "speech", slug, text: text.slice(0, 80) });
|
||
return;
|
||
}
|
||
|
||
if (o.type === "user" && msg?.role === "user" && !o.isMeta) {
|
||
const hasToolResult =
|
||
Array.isArray(content) && content.some((c) => c && c.type === "tool_result");
|
||
if (!hasToolResult) {
|
||
const text = textOf(content).trim();
|
||
if (text) broadcast({ type: "prompt", slug });
|
||
}
|
||
}
|
||
}
|
||
|
||
async function pollOnce(initial) {
|
||
let dirs;
|
||
try {
|
||
dirs = await fs.readdir(PROJECTS_ROOT, { withFileTypes: true });
|
||
} catch {
|
||
return;
|
||
}
|
||
|
||
for (const d of dirs) {
|
||
if (!d.isDirectory()) continue;
|
||
const slug = d.name;
|
||
const dir = path.join(PROJECTS_ROOT, slug);
|
||
|
||
let files;
|
||
try {
|
||
files = (await fs.readdir(dir)).filter((f) => f.endsWith(".jsonl"));
|
||
} catch {
|
||
continue;
|
||
}
|
||
|
||
let newestMtime = 0;
|
||
let newestFile = null;
|
||
|
||
for (const f of files) {
|
||
const full = path.join(dir, f);
|
||
let stat;
|
||
try {
|
||
stat = await fs.stat(full);
|
||
} catch {
|
||
continue;
|
||
}
|
||
if (stat.mtimeMs > newestMtime) {
|
||
newestMtime = stat.mtimeMs;
|
||
newestFile = full;
|
||
}
|
||
|
||
let tail = tails.get(full);
|
||
if (!tail) {
|
||
// Initial scan: skip history. Files born while we're running
|
||
// (e.g. a freshly dispatched task session) are read from the top.
|
||
const bornNow = !initial && Date.now() - stat.mtimeMs < 120_000;
|
||
tail = { offset: bornNow ? 0 : stat.size, remainder: "", slug };
|
||
tails.set(full, tail);
|
||
if (!bornNow) continue;
|
||
}
|
||
if (stat.size < tail.offset) {
|
||
tail.offset = stat.size; // truncated / rotated
|
||
tail.remainder = "";
|
||
}
|
||
if (stat.size > tail.offset) {
|
||
const buf = await readRange(full, tail.offset, stat.size);
|
||
tail.offset = stat.size;
|
||
const chunk = tail.remainder + buf.toString("utf8");
|
||
const lines = chunk.split("\n");
|
||
tail.remainder = lines.pop() ?? "";
|
||
for (const line of lines) {
|
||
if (line.trim()) handleLine(slug, line);
|
||
}
|
||
}
|
||
}
|
||
|
||
// Register the project if recently active.
|
||
const ageDays = (Date.now() - newestMtime) / 86400000;
|
||
if (newestFile && ageDays <= ACTIVE_WINDOW_DAYS && !projects.has(slug)) {
|
||
const cwd = await sniffCwd(newestFile);
|
||
projects.set(slug, {
|
||
slug,
|
||
name: prettifyName(slug, cwd),
|
||
cwd,
|
||
lastActivity: newestMtime,
|
||
});
|
||
if (!initial) {
|
||
broadcast({ type: "project", slug, name: projects.get(slug).name, lastActivity: newestMtime });
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/* ── task dispatch: turn the watcher into a tiny gateway ── */
|
||
|
||
/** slug → { child, startedAt, queue: [{prompt, resume}] } */
|
||
const taskRuns = new Map();
|
||
|
||
/** Newest transcript's basename = session id, for --resume. */
|
||
async function newestSessionId(slug) {
|
||
try {
|
||
const dir = path.join(PROJECTS_ROOT, slug);
|
||
const files = (await fs.readdir(dir)).filter((f) => f.endsWith(".jsonl"));
|
||
let best = null;
|
||
let bestM = 0;
|
||
for (const f of files) {
|
||
const st = await fs.stat(path.join(dir, f));
|
||
if (st.mtimeMs > bestM) {
|
||
bestM = st.mtimeMs;
|
||
best = f;
|
||
}
|
||
}
|
||
return best ? best.replace(/\.jsonl$/, "") : null;
|
||
} catch {
|
||
return null;
|
||
}
|
||
}
|
||
|
||
async function launchTask(slug, prompt, resume) {
|
||
const proj = projects.get(slug);
|
||
if (!proj?.cwd) return;
|
||
|
||
const args = ["-p", "--permission-mode", PERMISSION_MODE];
|
||
if (resume) {
|
||
const sessionId = await newestSessionId(slug);
|
||
if (sessionId && /^[0-9a-fA-F-]+$/.test(sessionId)) args.push("--resume", sessionId);
|
||
}
|
||
|
||
// Prompt goes through stdin — never through shell arguments.
|
||
const child = spawn("claude", args, {
|
||
cwd: proj.cwd,
|
||
shell: true,
|
||
stdio: ["pipe", "ignore", "pipe"],
|
||
windowsHide: true,
|
||
});
|
||
const run = { child, startedAt: Date.now(), queue: taskRuns.get(slug)?.queue ?? [] };
|
||
taskRuns.set(slug, run);
|
||
|
||
let stderrTail = "";
|
||
child.stderr.on("data", (c) => {
|
||
stderrTail = (stderrTail + c.toString("utf8")).slice(-400);
|
||
});
|
||
child.stdin.write(prompt);
|
||
child.stdin.end();
|
||
|
||
const killer = setTimeout(() => child.kill(), TASK_TIMEOUT_MS);
|
||
broadcast({ type: "task_accepted", slug, resume: Boolean(resume) });
|
||
|
||
child.on("close", (code) => {
|
||
clearTimeout(killer);
|
||
broadcast({
|
||
type: "task_done",
|
||
slug,
|
||
code,
|
||
error: code === 0 ? undefined : stderrTail.trim().slice(-200) || undefined,
|
||
});
|
||
const next = run.queue.shift();
|
||
if (next) {
|
||
launchTask(slug, next.prompt, next.resume);
|
||
} else {
|
||
taskRuns.delete(slug);
|
||
}
|
||
});
|
||
child.on("error", (err) => {
|
||
clearTimeout(killer);
|
||
taskRuns.delete(slug);
|
||
broadcast({ type: "task_done", slug, code: -1, error: String(err).slice(0, 200) });
|
||
});
|
||
}
|
||
|
||
/* ── two-way chat: one office thread per project, stitched with --resume ── */
|
||
|
||
/** slug → { sessionId, child } */
|
||
const chatThreads = new Map();
|
||
|
||
async function launchChat(slug, message, fromProject) {
|
||
const proj = projects.get(slug);
|
||
if (!proj?.cwd) return { ok: false, error: "unknown project" };
|
||
|
||
let thread = chatThreads.get(slug);
|
||
if (!thread) {
|
||
thread = { sessionId: null, child: null };
|
||
chatThreads.set(slug, thread);
|
||
}
|
||
if (thread.child) return { ok: false, error: "busy" };
|
||
|
||
let resumeId = thread.sessionId;
|
||
if (!resumeId && fromProject) {
|
||
const sid = await newestSessionId(slug);
|
||
if (sid && /^[0-9a-fA-F-]+$/.test(sid)) resumeId = sid;
|
||
}
|
||
|
||
const args = [
|
||
"-p",
|
||
"--output-format",
|
||
"stream-json",
|
||
"--verbose",
|
||
"--permission-mode",
|
||
PERMISSION_MODE,
|
||
];
|
||
if (resumeId) args.push("--resume", resumeId);
|
||
|
||
// Message goes through stdin — never through shell arguments.
|
||
const child = spawn("claude", args, {
|
||
cwd: proj.cwd,
|
||
shell: true,
|
||
stdio: ["pipe", "pipe", "pipe"],
|
||
windowsHide: true,
|
||
});
|
||
thread.child = child;
|
||
|
||
let stderrTail = "";
|
||
let stdoutRemainder = "";
|
||
let sawResult = false;
|
||
|
||
child.stderr.on("data", (c) => {
|
||
stderrTail = (stderrTail + c.toString("utf8")).slice(-400);
|
||
});
|
||
|
||
child.stdout.on("data", (c) => {
|
||
const chunk = stdoutRemainder + c.toString("utf8");
|
||
const lines = chunk.split("\n");
|
||
stdoutRemainder = lines.pop() ?? "";
|
||
for (const line of lines) {
|
||
if (!line.trim()) continue;
|
||
let o;
|
||
try {
|
||
o = JSON.parse(line);
|
||
} catch {
|
||
continue;
|
||
}
|
||
if (o.session_id) thread.sessionId = o.session_id;
|
||
if (o.type === "system" && o.subtype === "init") {
|
||
broadcast({ type: "chat_start", slug });
|
||
} else if (o.type === "assistant") {
|
||
const text = (o.message?.content ?? [])
|
||
.filter((b) => b && b.type === "text" && b.text)
|
||
.map((b) => b.text)
|
||
.join("\n")
|
||
.trim();
|
||
if (text) broadcast({ type: "chat_text", slug, text });
|
||
} else if (o.type === "result") {
|
||
sawResult = true;
|
||
broadcast({
|
||
type: "chat_done",
|
||
slug,
|
||
ok: o.subtype === "success",
|
||
text: typeof o.result === "string" ? o.result : "",
|
||
});
|
||
}
|
||
}
|
||
});
|
||
|
||
child.stdin.write(message);
|
||
child.stdin.end();
|
||
|
||
const killer = setTimeout(() => child.kill(), TASK_TIMEOUT_MS);
|
||
child.on("close", (code) => {
|
||
clearTimeout(killer);
|
||
thread.child = null;
|
||
if (!sawResult) {
|
||
broadcast({
|
||
type: "chat_done",
|
||
slug,
|
||
ok: false,
|
||
text: "",
|
||
error: stderrTail.trim().slice(-200) || `exit ${code}`,
|
||
});
|
||
}
|
||
});
|
||
child.on("error", (err) => {
|
||
clearTimeout(killer);
|
||
thread.child = null;
|
||
broadcast({ type: "chat_done", slug, ok: false, text: "", error: String(err).slice(0, 200) });
|
||
});
|
||
|
||
return { ok: true, resumed: Boolean(resumeId) };
|
||
}
|
||
|
||
async function handleChatRequest(req, res) {
|
||
const cors = corsFor(req);
|
||
if (req.method === "OPTIONS") {
|
||
res.writeHead(Object.keys(cors).length ? 204 : 403, cors);
|
||
res.end();
|
||
return;
|
||
}
|
||
if (!Object.keys(cors).length) {
|
||
res.writeHead(403, { "Content-Type": "application/json" });
|
||
res.end(JSON.stringify({ ok: false, error: "origin not allowed" }));
|
||
return;
|
||
}
|
||
const reply = (status, payload) => {
|
||
res.writeHead(status, { ...cors, "Content-Type": "application/json" });
|
||
res.end(JSON.stringify(payload));
|
||
};
|
||
try {
|
||
const body = JSON.parse(await readBody(req));
|
||
const slug = String(body.slug ?? "");
|
||
if (!projects.has(slug)) return reply(404, { ok: false, error: "unknown project" });
|
||
|
||
if (body.reset) {
|
||
const thread = chatThreads.get(slug);
|
||
if (thread?.child) thread.child.kill();
|
||
chatThreads.delete(slug);
|
||
return reply(200, { ok: true });
|
||
}
|
||
|
||
const message = String(body.message ?? "").trim();
|
||
if (!message || message.length > 8000) {
|
||
return reply(400, { ok: false, error: "message must be 1–8000 chars" });
|
||
}
|
||
const result = await launchChat(slug, message, Boolean(body.fromProject));
|
||
reply(result.ok ? 200 : 409, result);
|
||
} catch {
|
||
reply(400, { ok: false, error: "bad request" });
|
||
}
|
||
}
|
||
|
||
function corsFor(req) {
|
||
const origin = req.headers.origin;
|
||
if (origin && ALLOWED_ORIGINS.has(origin)) {
|
||
return {
|
||
"Access-Control-Allow-Origin": origin,
|
||
"Access-Control-Allow-Headers": "content-type",
|
||
"Access-Control-Allow-Methods": "POST, OPTIONS",
|
||
Vary: "Origin",
|
||
};
|
||
}
|
||
return {};
|
||
}
|
||
|
||
function readBody(req, limit = 100_000) {
|
||
return new Promise((resolve, reject) => {
|
||
let size = 0;
|
||
const chunks = [];
|
||
req.on("data", (c) => {
|
||
size += c.length;
|
||
if (size > limit) {
|
||
reject(new Error("too large"));
|
||
req.destroy();
|
||
return;
|
||
}
|
||
chunks.push(c);
|
||
});
|
||
req.on("end", () => resolve(Buffer.concat(chunks).toString("utf8")));
|
||
req.on("error", reject);
|
||
});
|
||
}
|
||
|
||
async function handleTaskRequest(req, res) {
|
||
const cors = corsFor(req);
|
||
if (req.method === "OPTIONS") {
|
||
res.writeHead(Object.keys(cors).length ? 204 : 403, cors);
|
||
res.end();
|
||
return;
|
||
}
|
||
if (!Object.keys(cors).length) {
|
||
res.writeHead(403, { "Content-Type": "application/json" });
|
||
res.end(JSON.stringify({ ok: false, error: "origin not allowed" }));
|
||
return;
|
||
}
|
||
const fail = (status, error) => {
|
||
res.writeHead(status, { ...cors, "Content-Type": "application/json" });
|
||
res.end(JSON.stringify({ ok: false, error }));
|
||
};
|
||
try {
|
||
const body = JSON.parse(await readBody(req));
|
||
const slug = String(body.slug ?? "");
|
||
const prompt = String(body.prompt ?? "").trim();
|
||
const resume = Boolean(body.resume);
|
||
if (!projects.has(slug)) return fail(404, "unknown project");
|
||
if (!prompt || prompt.length > 8000) return fail(400, "prompt must be 1–8000 chars");
|
||
|
||
const running = taskRuns.get(slug);
|
||
if (running) {
|
||
if (running.queue.length >= MAX_QUEUE) return fail(429, "queue full");
|
||
running.queue.push({ prompt, resume });
|
||
res.writeHead(202, { ...cors, "Content-Type": "application/json" });
|
||
res.end(JSON.stringify({ ok: true, queued: running.queue.length }));
|
||
return;
|
||
}
|
||
await launchTask(slug, prompt, resume);
|
||
res.writeHead(200, { ...cors, "Content-Type": "application/json" });
|
||
res.end(JSON.stringify({ ok: true, queued: 0 }));
|
||
} catch {
|
||
fail(400, "bad request");
|
||
}
|
||
}
|
||
|
||
function snapshot() {
|
||
return {
|
||
type: "snapshot",
|
||
projects: [...projects.values()]
|
||
.sort((a, b) => b.lastActivity - a.lastActivity)
|
||
.map((p) => ({ slug: p.slug, name: p.name, lastActivity: p.lastActivity })),
|
||
subs: [...pendingSubs.entries()].map(([subId, s]) => ({
|
||
subId,
|
||
slug: s.slug,
|
||
desc: s.desc,
|
||
kind: s.kind,
|
||
})),
|
||
};
|
||
}
|
||
|
||
const server = createServer((req, res) => {
|
||
if (req.url === "/task") {
|
||
handleTaskRequest(req, res);
|
||
return;
|
||
}
|
||
if (req.url === "/chat") {
|
||
handleChatRequest(req, res);
|
||
return;
|
||
}
|
||
if (req.url === "/events") {
|
||
res.writeHead(200, {
|
||
"Content-Type": "text/event-stream",
|
||
"Cache-Control": "no-cache",
|
||
Connection: "keep-alive",
|
||
"Access-Control-Allow-Origin": "*",
|
||
});
|
||
res.write(`data: ${JSON.stringify(snapshot())}\n\n`);
|
||
clients.add(res);
|
||
req.on("close", () => clients.delete(res));
|
||
return;
|
||
}
|
||
res.writeHead(404, { "Access-Control-Allow-Origin": "*" });
|
||
res.end("not found");
|
||
});
|
||
|
||
await pollOnce(true);
|
||
// Transcripts are sensitive — never expose beyond this machine.
|
||
server.listen(PORT, "127.0.0.1", () => {
|
||
console.log(`✳ Claude Office watcher`);
|
||
console.log(` watching ${PROJECTS_ROOT}`);
|
||
console.log(` projects ${projects.size} active in last ${ACTIVE_WINDOW_DAYS} days`);
|
||
console.log(` SSE http://localhost:${PORT}/events`);
|
||
});
|
||
setInterval(() => pollOnce(false).catch(() => {}), POLL_MS);
|