diff --git a/package.json b/package.json index abcedc10..56ce2f70 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "moshcode", - "version": "0.121.2", + "version": "0.122.0", "type": "module", "description": "moshcode — a metal wrapper for coding engines and native UGig/CoinPay workflow CLIs, with OpenPRD and moshscript", "repository": { @@ -37,7 +37,7 @@ "@moshcoder/moshpit-dns": "^0.5.0" }, "dependencies": { - "@profullstack/ai-providers": "^0.2.0", + "@profullstack/ai-providers": "^0.3.0", "@profullstack/cli": "^0.2.0", "@profullstack/hqtui": "^0.7.0", "@profullstack/synconfig": "^0.1.3", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 52c5269a..049ce126 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -9,8 +9,8 @@ importers: .: dependencies: '@profullstack/ai-providers': - specifier: ^0.2.0 - version: 0.2.0 + specifier: ^0.3.0 + version: 0.3.0 '@profullstack/cli': specifier: ^0.2.0 version: 0.2.0 @@ -35,8 +35,8 @@ packages: engines: {node: '>=20'} hasBin: true - '@profullstack/ai-providers@0.2.0': - resolution: {integrity: sha512-xY7a4gHThur7uTUnmzft47TnqVK+NVA57K60LThSAeHhITbSNLhkt9Sp7uNKCSg0wC55b5wLFEFGsGzblCiT6Q==} + '@profullstack/ai-providers@0.3.0': + resolution: {integrity: sha512-OPfCv10f4UqtM2034Vwqj8X0iQbmapQ4dDvTcQ5umEjJurKXRyDVdlsClH6zojhv/VZ4z4BtWE0bo9b1vvXdtw==} engines: {node: '>=20'} hasBin: true @@ -61,7 +61,7 @@ snapshots: '@moshcoder/moshpit-dns@0.5.0': {} - '@profullstack/ai-providers@0.2.0': {} + '@profullstack/ai-providers@0.3.0': {} '@profullstack/cli@0.2.0': {} diff --git a/src/c0agent-repl.mjs b/src/c0agent-repl.mjs index 855b70ee..a2d94d62 100644 --- a/src/c0agent-repl.mjs +++ b/src/c0agent-repl.mjs @@ -136,18 +136,69 @@ function createSpinner(stream = process.stderr) { }; } +/** "anthony@profullstack-dev-vienna:e98c6ec2" -> "profullstack-dev-vienna". */ +export function shortNode(name) { + if (!name) return "a node"; + return String(name).replace(/^[^@\s]+@/, "").replace(/:[0-9a-f]{6,}$/i, ""); +} + +const kTokens = (n) => (n >= 1000 ? `${(n / 1000).toFixed(1)}k` : String(n)); + +/** + * The spinner label for one c0mpute job status. `job` carries what earlier + * statuses said (node, when writing began) so a later one can use it. + */ +export function statusLabel(status, job, { promptTokens = 0, agent = null, now = Date.now() } = {}) { + const who = agent ? `${agent} · ` : ""; + const node = shortNode(job.node); + switch (status.stage) { + case "queued": + // The node only reports in with its first token, so this covers its + // queue AND it reading the prompt (minutes on a CPU node). + return `${who}${node} has the job · waiting for the first token (queue + reading ~${kTokens(promptTokens)} prompt tokens)`; + case "started": + return `${who}${node} is reading the prompt (~${kTokens(promptTokens)} tokens; a CPU node can take minutes)`; + case "writing": { + const secs = (now - (job.writingSince ?? now)) / 1000; + const rate = secs >= 2 ? ` · ${(status.tokens / secs).toFixed(1)} tok/s` : ""; + return `${who}${node} writing · ${status.tokens} token${status.tokens === 1 ? "" : "s"}${rate}`; + } + case "retry": + return `${who}${status.model ?? "model"} unavailable (${status.reason}); trying the next model`; + default: + return null; + } +} + /** The event sink shared by the REPL and print mode. */ function createRenderer({ out, log, spinner, quiet = false }) { let atLineStart = true; let wroteText = false; const line = (s) => { spinner.stop(); if (!atLineStart) { out("\n"); atLineStart = true; } log(s); }; + const jobs = new Map(); // agent@depth -> {node, writingSince} return { reset() { wroteText = false; }, get wroteText() { return wroteText; }, finish() { spinner.stop(); if (!atLineStart) { out("\n"); atLineStart = true; } }, event(e) { if (e.type === "waiting") { - spinner.start(e.depth ? `${e.agent} working…` : "Waiting for a c0mpute node…"); + jobs.set(`${e.agent}@${e.depth}`, {}); + const size = e.promptTokens ? ` (~${kTokens(e.promptTokens)} token prompt)` : ""; + spinner.start(e.depth ? `${e.agent} · sending to c0mpute…` : `Sending to c0mpute${size}…`); + return; + } + if (e.type === "status") { + const key = `${e.agent}@${e.depth}`; + const job = jobs.get(key) ?? {}; + jobs.set(key, job); + const s = e.status; + if (s.node) job.node = s.node; + if (s.stage === "writing" && job.writingSince == null) job.writingSince = Date.now(); + const label = statusLabel(s, job, { promptTokens: e.promptTokens, agent: e.depth ? e.agent : null }); + if (!label) return; + // Text is mid-line on stdout: a spinner redraw would wipe it. + if (!spinner.active && !atLineStart) return; + if (spinner.active) spinner.label(label); else spinner.start(label); return; } if (e.type === "thinking" && !e.depth) { spinner.label("Thinking…"); return; } diff --git a/src/c0agent.mjs b/src/c0agent.mjs index d5c40788..9656561e 100644 --- a/src/c0agent.mjs +++ b/src/c0agent.mjs @@ -697,7 +697,8 @@ export async function runAgent({ session, agent = LEAD_AGENT, messages, depth = for (let step = 0; step < limit; step++) { if (signal?.aborted) return { text: last, toolUses, error: "interrupted" }; trimContext(messages); - onEvent({ type: "waiting", agent: agent.name, depth }); + const promptTokens = Math.ceil((system.length + messages.reduce((n, m) => n + String(m.content).length, 0)) / 4); + onEvent({ type: "waiting", agent: agent.name, depth, promptTokens }); const filter = createStreamFilter( (t) => onEvent({ type: "text", text: t, agent: agent.name, depth }), { onThinking: () => onEvent({ type: "thinking", agent: agent.name, depth }) }, @@ -705,7 +706,17 @@ export async function runAgent({ session, agent = LEAD_AGENT, messages, depth = let res; try { res = await session.provider.stream( - { system, messages: messages.map(({ role, content }) => ({ role, content })), maxTokens: 2048, temperature: 0.2, signal, models: { c0mpute: model } }, + { + system, + messages: messages.map(({ role, content }) => ({ role, content })), + maxTokens: 2048, + temperature: 0.2, + signal, + models: { c0mpute: model }, + // Where the job is (queued on a node, reading the prompt, writing): + // a CPU node can sit on a long prompt for minutes before token one. + onStatus: (status) => onEvent({ type: "status", status, promptTokens, agent: agent.name, depth }), + }, (t) => filter.push(t), ); } catch (e) { diff --git a/test/c0agent-status.test.mjs b/test/c0agent-status.test.mjs new file mode 100644 index 00000000..0fb281a7 --- /dev/null +++ b/test/c0agent-status.test.mjs @@ -0,0 +1,38 @@ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { shortNode, statusLabel } from "../src/c0agent-repl.mjs"; + +test("shortNode drops the account and the node hash", () => { + assert.equal(shortNode("anthony@profullstack-dev-vienna:e98c6ec2"), "profullstack-dev-vienna"); + assert.equal(shortNode("provider@bonita"), "bonita"); + assert.equal(shortNode(null), "a node"); +}); + +test("statusLabel says where a c0mpute job is", () => { + const job = { node: "anthony@dev-vienna:e98c6ec2" }; + assert.match(statusLabel({ stage: "queued", source: "p2p" }, job, { promptTokens: 640 }), /^dev-vienna has the job · waiting for the first token \(queue \+ reading ~640 prompt tokens\)$/); + assert.match(statusLabel({ stage: "started" }, job, { promptTokens: 1400 }), /dev-vienna is reading the prompt \(~1\.4k tokens/); + job.writingSince = 0; + assert.equal(statusLabel({ stage: "writing", tokens: 40 }, job, { now: 10_000 }), "dev-vienna writing · 40 tokens · 4.0 tok/s"); + assert.match(statusLabel({ stage: "queued" }, job, { agent: "explore" }), /^explore · dev-vienna has the job/); + assert.match(statusLabel({ stage: "retry", model: "m", reason: "no node" }, job), /m unavailable \(no node\)/); +}); + +test("runAgent forwards provider status as status events", async () => { + const { createSession, runAgent } = await import("../src/c0agent.mjs"); + const provider = { + async stream(req, onToken) { + req.onStatus({ stage: "queued", jobId: "j", node: "n@x:abcdef12", source: "p2p", model: "m" }); + req.onStatus({ stage: "started", node: "n@x:abcdef12", engine: "ollama", model: "m" }); + onToken("hi"); + req.onStatus({ stage: "writing", tokens: 1 }); + return { text: "hi", model: "m" }; + }, + }; + const session = createSession({ provider, model: "m", cwd: process.cwd() }); + const events = []; + await runAgent({ session, messages: [{ role: "user", content: "yo" }], onEvent: (e) => events.push(e) }); + const status = events.filter((e) => e.type === "status").map((e) => e.status.stage); + assert.deepEqual(status, ["queued", "started", "writing"]); + assert.ok(events.find((e) => e.type === "waiting").promptTokens > 0); +});