Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 29 additions & 3 deletions apps/pwa/src/routes/sessions.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,26 @@ const LONG_POLL_MS = readLongPollMs();
// enough that a mis-paste of a whole file can't queue thousands of lines
// against a prompt that runs them one at a time.
const MAX_PASTED_LINES = 50;
// The most output one scrollback row may hold. The CLI flushes at ~16k, so a
// normal chunk is one row; a burst bigger than this becomes several rows in
// order rather than one truncated one.
export const MAX_CHUNK = 20000;

/**
* One posted chunk as the rows it will be stored in, in order.
*
* Every byte survives: nothing is dropped for being past the cap, which is
* what the old `slice(0, MAX_CHUNK)` did — quietly, and only to the copy the
* scrollback kept, so it showed up as a session that read differently after a
* reload than it had live.
*/
export function splitChunk(text, max = MAX_CHUNK) {
const s = String(text ?? "");
if (!s) return [];
const parts = [];
for (let i = 0; i < s.length; i += max) parts.push(s.slice(i, i + max));
return parts;
}

// Arrow keys pressed on the session page. The command queue carries opaque
// strings, so a key rides it as a sentinel the CLI decodes — ESC leads it
Expand Down Expand Up @@ -172,7 +192,13 @@ sessionsRouter.post("/api/sessions/:id/output", cliAuth, async (req, res) => {
// at the old column and stays wrong in the scrollback for good.
if (resized) publish(session.id, { type: "size", cols, rows });

if (chunk) {
// Split rather than truncate. A row is capped so one flush can't put an
// unbounded string in the table, but a chunk that overran the cap used to be
// stored short while the live watchers got all of it — so the same session
// read two different ways (watching, or reloading the page) disagreed, and
// the bytes past the cap were gone for good. A build that prints a wall of
// output in one 150ms window is exactly when that happens.
for (const part of splitChunk(chunk)) {
// seq is per-session and monotonic so a reconnecting browser can ask for
// "everything after N" instead of replaying the whole scrollback.
//
Expand All @@ -187,14 +213,14 @@ sessionsRouter.post("/api/sessions/:id/output", cliAuth, async (req, res) => {
`INSERT INTO session_output (session_id,seq,chunk,created_at)
SELECT ?, COALESCE(MAX(seq), 0) + 1, ?, ? FROM session_output WHERE session_id = ?
RETURNING seq`,
[session.id, chunk.slice(0, 20000), now, session.id]
[session.id, part, now, session.id]
);
const seq = Number(inserted.seq);
await run(
`DELETE FROM session_output WHERE session_id = ? AND seq <= ?`,
[session.id, seq - SCROLLBACK]
);
publish(session.id, { type: "out", seq, chunk });
publish(session.id, { type: "out", seq, chunk: part });
}
if (engine !== session.engine) publish(session.id, { type: "engine", engine });
res.json({ ok: true });
Expand Down
29 changes: 25 additions & 4 deletions apps/pwa/test/sessions-output-seq.test.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -216,20 +216,41 @@ test("sessions: an empty chunk writes no output row", skip, async () => {
assert.deepEqual(await rowsFor(sid), [], "no row for an empty chunk");
});

test("sessions: a long chunk is still truncated to the stored limit", skip, async () => {
test("sessions: an ordinary chunk is one row, at the stored limit", skip, async () => {
const { one, rowsFor } = await app();

const reg = await one("/api/sessions", { name: "truncate" });
const reg = await one("/api/sessions", { name: "at-cap" });
const sid = reg.body.id;

await one(`/api/sessions/${sid}/output`, { chunk: "x".repeat(20050) });
await one(`/api/sessions/${sid}/output`, { chunk: "x".repeat(20000) });

const rows = await rowsFor(sid);
assert.equal(rows.length, 1);
assert.equal(rows[0].chunk.length, 20000, "chunk is capped at 20000 chars");
assert.equal(rows[0].chunk.length, 20000, "a row is still capped at 20000 chars");
assert.equal(Number(rows[0].seq), 1);
});

test("sessions: a chunk past the row limit is split, not truncated", skip, async () => {
const { one, rowsFor } = await app();

// The row cap keeps one flush from writing an unbounded string. It used to be
// applied with slice(), so the overflow was published live to every watcher
// and then thrown away — the same session read one way while you watched it
// and another way after a reload, with no sign anything was missing. A build
// that dumps a wall of output inside one 150ms flush is exactly that case.
const chunk = `${"x".repeat(20000)}TAIL-MUST-SURVIVE\n`;
const reg = await one("/api/sessions", { name: "oversized" });
const sid = reg.body.id;

await one(`/api/sessions/${sid}/output`, { chunk });

const rows = await rowsFor(sid);
assert.equal(rows.length, 2, "the overflow becomes another row");
assert.ok(rows.every((r) => r.chunk.length <= 20000), "each row still respects the cap");
assert.equal(rows.map((r) => r.chunk).join(""), chunk, "every byte posted must be stored");
assert.deepEqual(rows.map((r) => Number(r.seq)), [1, 2], "and in order, so a replay reassembles it");
});

test("sessions: a foreign key still cannot append output", skip, async () => {
const { one, two, rowsFor } = await app();

Expand Down
8 changes: 7 additions & 1 deletion src/billing.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ import { spawnSync } from "node:child_process";

import { loadBusiness, loadTimers, newId, updateBusiness, updateTimers } from "./business-store.mjs";
import { clientLabel, parseFields, resolveClient } from "./clients.mjs";
import { captureSpec } from "./pty.mjs";
import { GATEWAYS, defaultGateway, gatewayState } from "./payments.mjs";
import { chargeFor, describeRate, formatMoney, isDollarPegged, isFiat, rateFor } from "./rates.mjs";
import { humanDuration, selectEntries, windowFrom } from "./timer.mjs";
Expand Down Expand Up @@ -289,7 +290,12 @@ function handOff(record, invoice, business, fields, write, run) {
return 0;
}

const result = run("coinpay", args, { stdio: "inherit" });
// Mirrored like every other hand-off: sending an invoice is exactly the kind
// of thing you want to read back from the session page afterwards.
const launch = captureSpec({ cmd: "coinpay", args });
let result;
try { result = run(launch.cmd, launch.args, { stdio: "inherit" }); }
finally { launch.stop(); }
if (result?.error) { write(err(String(result.error.message || result.error))); return 1; }
if (result?.status) { write(err(`coinpay exited ${result.status} — invoice ${record.id} is still a local draft`)); return result.status; }
updateBusiness((data) => {
Expand Down
11 changes: 10 additions & 1 deletion src/commands.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import { capture, killSession, remoteStatus, sendPrompt } from "./herd.mjs";
import { herdStart, isRemoteMember, roster, waitForMany, waitMember } from "./herd-cli.mjs";
import { endTask, findTask, readTasks, startTask } from "./herd-tasks.mjs";
import { shellInvocation } from "./shell.mjs";
import { captureSpec } from "./pty.mjs";
import { identity, loginAuto, logout as forgetCreds } from "./auth.mjs";
import { expandAlias, getAlias, loadAliases, removeAlias, setAlias } from "./aliases.mjs";
import { CORE_CLI_COMMAND_NAMES, PIT_COMMANDS } from "./cli-schema.mjs";
Expand Down Expand Up @@ -116,7 +117,15 @@ const SHELL = {
// has the reasoning, including why a headless run stays non-interactive.
const { shell: sh, args: shArgs } = shellInvocation(cmd);
ctx.out(` ▶ shell: ${cmd}`);
const res = spawnSync(sh, shArgs, { stdio: "inherit" });
// Captured for the session mirror like the pit's own `!cmd`. A blocking
// spawn holds the event loop, so the follower's poll never runs and the
// whole command arrives in the drain stop() does — batched rather than
// live, which is still the difference between reading it from a phone and
// not.
const launch = captureSpec({ cmd: sh, args: shArgs });
let res;
try { res = spawnSync(launch.cmd, launch.args, { stdio: "inherit" }); }
finally { launch.stop(); }
if (res.error) throw res.error;
const code = res.status ?? 1;
if (code !== 0) {
Expand Down
53 changes: 17 additions & 36 deletions src/engines.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -32,11 +32,11 @@
// a session that starts fresh is a small disappointment, and one that starts
// with a flag the engine does not have is a crash.
import { spawn } from "node:child_process";
import { existsSync, mkdtempSync, readFileSync, rmSync, statSync, writeFileSync } from "node:fs";
import { homedir, tmpdir } from "node:os";
import { existsSync, readFileSync, statSync } from "node:fs";
import { homedir } from "node:os";
import path from "node:path";

import { followFile, ptyEnabled, ptySpec, scriptFlavor, stripScriptBanner } from "./pty.mjs";
import { captureSpec } from "./pty.mjs";

export const ENGINES = {
opencode: {
Expand Down Expand Up @@ -439,21 +439,29 @@ export function runCmd(cmd, args = [], { capture = false } = {}) {
let child;
const spec = spawnSpec(cmd, args);
const stdio = capture ? ["inherit", "pipe", "pipe"] : "inherit";
try { child = spawn(spec.cmd, spec.args, { stdio }); }
catch (e) { resolve({ ok: false, error: e }); return; }
// The `capture` branch already reaches a watching browser: it re-writes
// every byte through this process's own stdout/stderr, which the mirror
// tees. The inherited branch does not — those bytes go to the tty and
// nowhere else — so it goes under a pty when a mirror is live. This is what
// an upgrade, a plugin install and an `mcp add` all run through, and all
// three used to be a rule, a blank stretch, and a result line.
const launch = capture ? { ...spec, stop: () => {} } : captureSpec(spec);
const finish = (result) => { try { launch.stop(); } catch { /* already drained */ } resolve(result); };
try { child = spawn(launch.cmd, launch.args, { stdio }); }
catch (e) { finish({ ok: false, error: e }); return; }
let output = "";
if (capture) {
for (const [stream, sink] of [[child.stdout, process.stdout], [child.stderr, process.stderr]]) {
stream?.on("data", (chunk) => { output += chunk.toString(); sink.write(chunk); });
}
}
child.on("error", (e) => resolve({ ok: false, error: e, output }));
child.on("error", (e) => finish({ ok: false, error: e, output }));
// "exit" fires as soon as the process is gone, which with pipes can leave
// the last chunk still queued — the one line we are trying to read. "close"
// waits for the streams too. With stdio inherited there are no streams, so
// the two are the same moment and existing callers are unaffected; the
// distinction is kept explicit so neither branch changes by accident.
child.on(capture ? "close" : "exit", (code, signal) => resolve({ ok: true, code, signal, output }));
child.on(capture ? "close" : "exit", (code, signal) => finish({ ok: true, code, signal, output }));
});
}

Expand Down Expand Up @@ -501,35 +509,8 @@ export function openPassthrough(target, args = [], { onOutput } = {}) {
// the child the tty's own file descriptors, so none of its bytes ever pass
// through this process. See src/pty.mjs for why this is script(1) and not
// a pipe or node-pty.
let transcript = null;
let workDir = null;
let stopFollow = null;
let launch = { ...spec, stdio: "inherit" };
if (ptyEnabled(onOutput)) {
try {
workDir = mkdtempSync(path.join(tmpdir(), "moshcode-pty-"));
transcript = path.join(workDir, "transcript");
writeFileSync(transcript, "");
const wrapped = ptySpec(spec.cmd, spec.args, transcript, scriptFlavor());
if (wrapped) {
launch = { ...wrapped, stdio: "inherit" };
let first = true;
stopFollow = followFile(transcript, (chunk) => {
const clean = stripScriptBanner(chunk, first);
first = false;
if (clean) onOutput(clean);
});
}
} catch {
// Capture is a nicety; never let it stop the session from opening.
transcript = null;
}
}

const cleanup = () => {
try { stopFollow?.(); } catch { /* nothing left to drain */ }
if (workDir) { try { rmSync(workDir, { recursive: true, force: true }); } catch { /* temp dir */ } }
};
const launch = captureSpec(spec, onOutput);
const cleanup = () => launch.stop();

let child;
try { child = spawn(launch.cmd, launch.args, { stdio: "inherit", env }); }
Expand Down
22 changes: 22 additions & 0 deletions src/mirror.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,28 @@ export function pressKey(name, rl = null, stdin = process.stdin) {
const FLUSH_MS = 150; // batch writes so a busy render is one request, not fifty
const MAX_BUFFER = 16000; // flush early once a batch gets big

// Where a child process's output should be copied while a mirror is watching.
//
// Module-level rather than threaded through every call, because "is anyone
// watching this pit" is one fact about the process and the launchers that need
// it are scattered: the shell, the installers, the upgrader, the plugin/skill/
// MCP hand-offs. Passing it down by hand is what left most of them writing
// straight to the tty with the session page showing nothing — each new launcher
// had to remember, and none of them did. src/pty.mjs reads this as its default,
// so capture is what a launcher gets for free and opting out is the deliberate
// act.
let activeSink = null;

/** Point child capture at this mirror (or null when the pit stops mirroring). */
export function setActiveSink(sink) {
activeSink = typeof sink === "function" ? sink : null;
}

/** The sink a child's output should be copied to, or null when unmirrored. */
export function activeChildSink() {
return activeSink;
}

export function createMirror({
version = "",
cwd = process.cwd(),
Expand Down
6 changes: 5 additions & 1 deletion src/payments.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
// `/payments connect stripe` records a *reference* — vault and key name — and
// says out loud where the secret should go.
import { spawnSync } from "node:child_process";
import { captureSpec } from "./pty.mjs";

import { loadBusiness, updateBusiness } from "./business-store.mjs";
import { parseFields } from "./clients.mjs";
Expand Down Expand Up @@ -178,7 +179,10 @@ function connectGateway(args, write, run) {
return 1;
}
write(info(`handing you to ${bone(gateway.bin)} — it owns its own session`));
const result = run(gateway.bin, gateway.connect, { stdio: "inherit" });
const launch = captureSpec({ cmd: gateway.bin, args: gateway.connect });
let result;
try { result = run(launch.cmd, launch.args, { stdio: "inherit" }); }
finally { launch.stop(); }
if (result?.error) { write(err(String(result.error.message || result.error))); return 1; }
if (result?.status) {
write(err(`${gateway.bin} ${gateway.connect.join(" ")} exited ${result.status} — nothing recorded`));
Expand Down
59 changes: 58 additions & 1 deletion src/pty.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,11 @@
// `script` disagree on both flag names and argument order, and anything we
// cannot positively identify falls back to today's plain `inherit`.
import { spawnSync } from "node:child_process";
import { closeSync, existsSync, openSync, readSync, statSync } from "node:fs";
import { closeSync, existsSync, mkdtempSync, openSync, readSync, rmSync, statSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import { StringDecoder } from "node:string_decoder";
import { activeChildSink } from "./mirror.mjs";

/**
* POSIX single-quote escaping, for argv that has to survive being flattened
Expand Down Expand Up @@ -178,3 +181,57 @@ export function ptyEnabled(sink, flavor = scriptFlavor()) {
if (process.env.MOSHCODE_MIRROR_PTY === "0") return false;
return Boolean(flavor);
}

/**
* Wrap a spawn spec so a copy of everything the child prints reaches `onOutput`
* while the child still owns the real terminal.
*
* The whole capture dance in one place — temp transcript, the flavour-specific
* `script` argv, the follower, the banner strip, the cleanup — because every
* launcher in the pit needs it, and each one growing its own copy is how a
* shell command ended up invisible in the mirror while `/agents claude` was
* captured: both spawn `inherit`, and only one of them had been taught this.
*
* `onOutput` defaults to whatever the live mirror is (src/mirror.mjs), so a
* launcher gets capture without having to know the mirror exists — the reverse
* of how this started, where each launcher had to be taught separately and only
* two ever were. Pass `null` to opt a launch out.
*
* Returns `{ cmd, args, stop }`. With nothing watching, or on a box with no
* `script(1)` we can drive, `cmd`/`args` come back exactly as passed in and
* `stop` is a no-op — the caller spawns what it always spawned. `stop()` must
* be called once the child exits: it drains the tail of the transcript (the
* last lines of a command are usually the ones you were waiting for) and
* removes the temp dir.
*/
export function captureSpec({ cmd, args = [] }, onOutput = activeChildSink(), { flavor = scriptFlavor() } = {}) {
const plain = { cmd, args, stop: () => {} };
if (!ptyEnabled(onOutput, flavor)) return plain;
let workDir = null;
try {
workDir = mkdtempSync(path.join(tmpdir(), "moshcode-pty-"));
const transcript = path.join(workDir, "transcript");
writeFileSync(transcript, "");
const wrapped = ptySpec(cmd, args, transcript, flavor);
if (!wrapped) throw new Error("no script(1) spec for this flavour");
let first = true;
const stopFollow = followFile(transcript, (chunk) => {
const clean = stripScriptBanner(chunk, first);
first = false;
if (clean) onOutput(clean);
});
const dir = workDir;
return {
cmd: wrapped.cmd,
args: wrapped.args,
stop() {
try { stopFollow(); } catch { /* nothing left to drain */ }
try { rmSync(dir, { recursive: true, force: true }); } catch { /* temp dir */ }
},
};
} catch {
// Capture is a nicety; never let it stop a command from running.
if (workDir) { try { rmSync(workDir, { recursive: true, force: true }); } catch { /* temp dir */ } }
return plain;
}
}
Loading
Loading