feat(agents c0mpute): live job status in the spinner (v0.122.0)

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",

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; }

src/c0agent.mjs

@@ -697,15 +697,26 @@ 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 }) },
     );
     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) {

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);
+});