aa: stack prompts while thinking + git-pull each turn + live telemetry

+223 -14

2 changed files

expand all

help

bridge

aa-bridge.mjs

system

public

aesthetic.computer

disks

aa.mjs

+57 -1

help/bridge/aa-bridge.mjs

··· 13 13 // GET /api/history — return prior user/assistant events for this user's session 14 14 15 15 import http from "http"; 16 - import { spawn } from "child_process"; 16 + import { spawn, exec } from "child_process"; 17 + import { promisify } from "util"; 17 18 import { readFile, writeFile, mkdir } from "fs/promises"; 18 19 import { existsSync } from "fs"; 19 20 import { homedir } from "os"; 20 21 import { dirname, join } from "path"; 21 22 import { randomUUID } from "crypto"; 23 + 24 + const execP = promisify(exec); 22 25 23 26 const PORT = parseInt(process.env.AA_PORT || "3004", 10); 24 27 const ADMIN_SUB = process.env.ADMIN_SUB; ··· 236 239 return events; 237 240 } 238 241 242 + // ───────── git sync ───────── 243 + // Pull remote changes into WORK_DIR before each turn so claude always starts 244 + // from a fresh tree. --autostash tucks in-flight edits; --rebase keeps linear 245 + // history. We never abort the turn on pull failure — claude gets to see the 246 + // repo state and can reconcile. 247 + async function gitPull(cwd = WORK_DIR) { 248 + const started = Date.now(); 249 + try { 250 + const { stdout, stderr } = await execP("git pull --rebase --autostash", { 251 + cwd, 252 + timeout: 30_000, 253 + maxBuffer: 1024 * 1024, 254 + }); 255 + const out = ((stdout || "") + (stderr || "")).trim(); 256 + let summary = "updated"; 257 + if (/Already up to date/i.test(out)) summary = "up to date"; 258 + else if (/Fast-forward/.test(out)) summary = "fast-forwarded"; 259 + else if (/Successfully rebased/.test(out)) summary = "rebased"; 260 + else if (/CONFLICT/.test(out)) summary = "conflicts"; 261 + return { 262 + ok: true, 263 + summary, 264 + output: out.split("\n").slice(-20).join("\n"), 265 + durationMs: Date.now() - started, 266 + }; 267 + } catch (err) { 268 + const out = ((err.stdout || "") + (err.stderr || "")).trim(); 269 + return { 270 + ok: false, 271 + summary: "failed", 272 + output: (out || err.message || "").split("\n").slice(-20).join("\n"), 273 + durationMs: Date.now() - started, 274 + }; 275 + } 276 + } 277 + 239 278 // ───────── claude spawn ───────── 240 279 // 241 280 // Git attribution: commits made through this bridge keep the *author* as ··· 315 354 316 355 const sessionId = await getSessionId(sub); 317 356 sse(res, "start", { sessionId, cwd: WORK_DIR }); 357 + 358 + // Pre-spawn: pull any remote changes so claude works from fresh state. 359 + const pull = await gitPull(WORK_DIR); 360 + sse(res, "git-pull", pull); 318 361 319 362 const child = spawnClaude(message, sessionId); 320 363 let buffer = ""; ··· 586 629 res.writeHead(500, { "Content-Type": "application/json", ...corsHeaders(origin) }); 587 630 res.end(JSON.stringify({ error: err.message })); 588 631 } 632 + return; 633 + } 634 + 635 + if (req.url === "/api/pull" && req.method === "POST") { 636 + const sub = await validateBearer(req.headers.authorization); 637 + if (!sub || (!DEV_BYPASS && sub !== ADMIN_SUB)) { 638 + res.writeHead(sub ? 403 : 401, corsHeaders(origin)); 639 + res.end(); 640 + return; 641 + } 642 + const pull = await gitPull(WORK_DIR); 643 + res.writeHead(200, { "Content-Type": "application/json", ...corsHeaders(origin) }); 644 + res.end(JSON.stringify(pull)); 589 645 return; 590 646 } 591 647

+166 -13

system/public/aesthetic.computer/disks/aa.mjs

··· 17 17 let abortCtrl = null; 18 18 let userHandleRef = null; 19 19 let hudRef = null; 20 + let verbose = false; // /verbose toggles rich telemetry 21 + const queue = []; // prompts stacked while pending 20 22 21 23 let msgCounter = 0; 22 24 const nextId = () => `aa-${Date.now()}-${msgCounter++}`; ··· 125 127 if (text === "/reset") return reset(); 126 128 if (text === "/clear") { 127 129 client.system.messages.length = 0; 130 + queue.length = 0; 128 131 invalidate(); 129 132 return; 130 133 } 131 134 if (text === "/cancel" || text === "/stop") { 132 135 if (abortCtrl) abortCtrl.abort(); 136 + return; 137 + } 138 + if (text === "/verbose") { 139 + verbose = !verbose; 140 + pushSystem(`verbose ${verbose ? "on" : "off"}`); 141 + return; 142 + } 143 + if (text === "/queue") { 144 + pushSystem(queue.length ? `${queue.length} queued` : "queue empty"); 145 + return; 146 + } 147 + if (text === "/pull") return manualPull(); 148 + 149 + // Echo the prompt immediately so the transcript reflects submission order. 150 + const me = userHandleRef?.() || "@jeffrey"; 151 + pushMessage(me, text); 152 + 153 + if (pending) { 154 + queue.push(text); 155 + pushSystem(`→ queued (${queue.length})`); 133 156 return; 134 157 } 135 158 await sendToBridge(text); ··· 211 234 method: "POST", 212 235 headers: { Authorization: `Bearer ${token}` }, 213 236 }); 237 + queue.length = 0; 214 238 pushSystem("session reset."); 215 239 } catch (err) { 216 240 pushSystem(`reset failed: ${err.message}`); 217 241 } 218 242 } 219 243 244 + async function manualPull() { 245 + try { 246 + const res = await fetch(`${ENDPOINT}/api/pull`, { 247 + method: "POST", 248 + headers: { Authorization: `Bearer ${token}` }, 249 + }); 250 + if (!res.ok) { 251 + pushSystem(`pull failed: ${res.status}`); 252 + return; 253 + } 254 + const data = await res.json(); 255 + renderGitPull(data); 256 + } catch (err) { 257 + pushSystem(`pull failed: ${err.message}`); 258 + } 259 + } 260 + 261 + function renderGitPull(p) { 262 + if (!p) return; 263 + const ms = p.durationMs != null ? ` ${Math.max(1, Math.round(p.durationMs / 100) / 10)}s` : ""; 264 + // "up to date" is the common case — keep it out of the transcript unless verbose. 265 + if (p.ok && p.summary === "up to date" && !verbose) return; 266 + const glyph = p.ok ? "⟲" : "⟲!"; 267 + const head = `${glyph} pull ${p.summary}${ms}`; 268 + if (verbose && p.output) pushSystem(`${head}\n${p.output}`); 269 + else pushSystem(head); 270 + } 271 + 272 + function shorten(s, max = 80) { 273 + if (!s) return ""; 274 + const t = String(s); 275 + return t.length > max ? t.slice(0, max - 1) + "…" : t; 276 + } 277 + 278 + function toolLabel(block) { 279 + const name = block.name || "?"; 280 + const input = block.input || {}; 281 + let arg = ""; 282 + switch (name) { 283 + case "Read": 284 + case "Write": 285 + case "Edit": 286 + case "MultiEdit": 287 + case "NotebookEdit": 288 + arg = input.file_path || input.notebook_path || ""; 289 + break; 290 + case "Bash": 291 + arg = input.command || ""; 292 + break; 293 + case "Grep": 294 + case "Glob": 295 + arg = input.pattern || ""; 296 + break; 297 + case "WebFetch": 298 + case "WebSearch": 299 + arg = input.url || input.query || ""; 300 + break; 301 + case "TodoWrite": 302 + arg = `${input.todos?.length ?? 0} items`; 303 + break; 304 + case "Task": 305 + arg = input.description || input.subagent_type || ""; 306 + break; 307 + default: 308 + // Generic fallback: first string-ish value in input 309 + for (const v of Object.values(input)) { 310 + if (typeof v === "string") { arg = v; break; } 311 + } 312 + } 313 + // Strip the common home prefix so paths stay readable on a phone. 314 + if (arg.startsWith("/Users/aesthetic/")) arg = arg.slice("/Users/aesthetic/".length); 315 + arg = shorten(arg.replace(/\s+/g, " ").trim(), 100); 316 + return arg ? `⚙ ${name} ${arg}` : `⚙ ${name}`; 317 + } 318 + 220 319 async function loadHistory() { 221 320 try { 222 321 const res = await fetch(`${ENDPOINT}/api/history`, { ··· 266 365 } 267 366 268 367 async function sendToBridge(text) { 269 - if (pending) { pushSystem("still thinking — type /cancel to stop."); return; } 270 - 271 - const me = userHandleRef?.() || "@jeffrey"; 272 - pushMessage(me, text); 273 368 pending = true; 274 369 abortCtrl = new AbortController(); 275 370 276 371 // Streamed reply: the aa bubble is created lazily on the first text chunk, 277 - // then mutated in place so the reply appears progressively. 372 + // then mutated in place so the reply appears progressively. Each assistant 373 + // turn may emit multiple text blocks interleaved with tool_use blocks, so 374 + // we close the current bubble when a tool_use lands and start a fresh one 375 + // on the next text chunk. 278 376 let streamMsg = null; 279 377 let pendingText = ""; 280 378 const applyStream = () => { ··· 283 381 streamMsg.text = pendingText; 284 382 invalidate(); 285 383 } 384 + }; 385 + const flushBubble = () => { 386 + if (streamMsg && !pendingText) removeMessage(streamMsg); 387 + streamMsg = null; 388 + pendingText = ""; 286 389 }; 287 390 288 391 try { ··· 320 423 const evt = parseSSE(block); 321 424 if (!evt) continue; 322 425 323 - if (evt.event === "claude") { 426 + if (evt.event === "git-pull") { 427 + renderGitPull(evt.data); 428 + } else if (evt.event === "claude") { 324 429 const ev = evt.data; 325 - if (ev.type === "assistant" && ev.message?.content) { 430 + if (ev.type === "system" && ev.subtype === "init") { 431 + const parts = []; 432 + if (ev.model) parts.push(ev.model); 433 + if (ev.session_id) parts.push(ev.session_id.slice(0, 8)); 434 + if (parts.length) pushSystem(`◌ ${parts.join(" · ")}`); 435 + } else if (ev.type === "assistant" && ev.message?.content) { 326 436 for (const b of ev.message.content) { 327 437 if (b.type === "text" && b.text) { 328 438 pendingText += b.text; 329 439 applyStream(); 440 + } else if (b.type === "tool_use") { 441 + // Seal the current assistant bubble so the tool line lands 442 + // between text blocks rather than replacing them. 443 + if (streamMsg && pendingText) { 444 + streamMsg.text = pendingText; 445 + streamMsg = null; 446 + pendingText = ""; 447 + } 448 + pushSystem(toolLabel(b)); 449 + } else if (b.type === "thinking" && verbose && b.thinking) { 450 + pushSystem(`… ${shorten(b.thinking.replace(/\s+/g, " "), 200)}`); 330 451 } 331 - // tool_use / thinking / other blocks deliberately suppressed 332 452 } 333 - } else if (ev.type === "result" && ev.result) { 334 - pendingText = ev.result; 335 - applyStream(); 453 + } else if (ev.type === "user" && verbose && ev.message?.content) { 454 + const c = ev.message.content; 455 + const blocks = Array.isArray(c) ? c : []; 456 + for (const b of blocks) { 457 + if (b.type === "tool_result") { 458 + const body = typeof b.content === "string" 459 + ? b.content 460 + : Array.isArray(b.content) 461 + ? b.content.filter((x) => x.type === "text").map((x) => x.text).join("\n") 462 + : ""; 463 + if (body) { 464 + const preview = body.split("\n").slice(0, 3).join(" "); 465 + pushSystem(`← ${shorten(preview, 180)}`); 466 + } 467 + } 468 + } 469 + } else if (ev.type === "result") { 470 + if (ev.result && !pendingText) { 471 + // Rare: result arrived with no prior streamed text — render it. 472 + pendingText = ev.result; 473 + applyStream(); 474 + } 475 + const parts = []; 476 + if (ev.duration_ms != null) parts.push(`${(ev.duration_ms / 1000).toFixed(1)}s`); 477 + if (ev.total_cost_usd != null) parts.push(`$${Number(ev.total_cost_usd).toFixed(4)}`); 478 + const u = ev.usage || {}; 479 + if (u.input_tokens) parts.push(`${u.input_tokens}in`); 480 + if (u.output_tokens) parts.push(`${u.output_tokens}out`); 481 + if (u.cache_read_input_tokens) parts.push(`${u.cache_read_input_tokens}cache`); 482 + if (ev.num_turns) parts.push(`${ev.num_turns}t`); 483 + if (parts.length) pushSystem(`✓ ${parts.join(" · ")}`); 336 484 } 337 - // tool_result (in user events) also suppressed 338 485 } else if (evt.event === "error") { 339 486 pushMessage("aa", `! ${evt.data.message || "error"}`); 340 487 } else if (evt.event === "done") { ··· 355 502 } 356 503 } 357 504 } catch (err) { 358 - if (streamMsg && !pendingText) removeMessage(streamMsg); 505 + flushBubble(); 359 506 if (err.name === "AbortError") pushSystem("cancelled."); 360 507 else pushSystem(`error: ${err.message}`); 361 508 } finally { 362 509 pending = false; 363 510 abortCtrl = null; 511 + // Drain the next queued prompt if any. Defer a tick so the current turn's 512 + // "done" state renders before the next one starts streaming. 513 + if (queue.length > 0) { 514 + const next = queue.shift(); 515 + setTimeout(() => { sendToBridge(next); }, 40); 516 + } 364 517 } 365 518 } 366 519

Configure Feed

Configure Feed