starbridge-mod

1import type { Register } from "claude-code";
2import {
3  AgentLoop,
4  HEADERS,
5  isPortFile,
6  type Line,
7  PROOF_HEADER,
8  pidRuns,
9  portTarget,
10  signCall,
11  socketPath,
12} from "./agent.ts";
13import { Handover } from "./handover.ts";
14import { configDir, Poller } from "./poller.ts";
15import { Sent, sessionElsewhere } from "./sent.ts";
16import { forParent, tagAsk, toSubagents, untagAsk } from "./subagent.ts";
17import { Switch, wantsLoop } from "./switch.ts";
18
19/**
20 * Starts the answer loop in each session a person keeps open (`wantsLoop`): through the machine's agent when it runs
21 * (agent.ts), else through the CLI (poller.ts); switch.ts picks. A hot reload loads this module
22 * afresh: the old loop dies with the old module, and `session.start` starts a new one. While the
23 * main agent's turn runs, an answer goes out with its next tool result instead (handover.ts).
24 * An answer submitted as a prompt stays on record until its turn starts, so a process that exits
25 * first does not lose it (sent.ts).
26 */
27/**
28 * Exits 0 when "$1" is a file owned by this user or root that neither its group nor others may
29 * write: what `ownedFile` (node.ts) checks, for a host that reads no owner or mode.
30 */
31const OWNED =
32  '[ -n "$(find -L "$1" -prune -type f \\( -user "$(id -u)" -o -user 0 \\) ! -perm -020 ! -perm -002 2>/dev/null)" ]';
33
34export const register: Register = (on) => {
35  let loop: Switch | undefined;
36  /** Prompts submitted whose turn has not started. */
37  let waiting = 0;
38  // `$.prompt.submit` resolves when the prompt's turn starts: mid-turn, once the running one ends.
39  const track = (submitted: Promise<unknown>) => {
40    waiting++;
41    const started = () => void waiting--;
42    submitted.then(started, started);
43    return submitted;
44  };
45  const handover = new Handover();
46  let sent: Sent | undefined;
47  /** Answers a subagent could not take: they go to the parent from then on (#1702). */
48  const failed = new Set<string>();
49  /** When each subagent's last tool call ended: a message sent before then, it has read. */
50  const roundEnded = new Map<string, number>();
51
52  on("session.start", async ($, e, next) => {
53    handover.ended();
54    const started = await next(e);
55    if (!wantsLoop(e.isInteractive, await $.env.get("CLAUDE_CODE_ENTRYPOINT"))) return started;
56    // The new loop takes over the lines the old one submitted and has not confirmed yet, so it
57    // confirms them rather than submitting them again (#1337).
58    const before = loop;
59    void before?.stop();
60    const env = {
61      STARBRIDGE_AGENT_SOCKET: await $.env.get("STARBRIDGE_AGENT_SOCKET"),
62      STARBRIDGE_CONFIG_DIR: await $.env.get("STARBRIDGE_CONFIG_DIR"),
63      STARBRIDGE_NO_AGENT: await $.env.get("STARBRIDGE_NO_AGENT"),
64      CLAUDE_CONFIG_DIR: await $.env.get("CLAUDE_CONFIG_DIR"),
65      XDG_CONFIG_HOME: await $.env.get("XDG_CONFIG_HOME"),
66      XDG_RUNTIME_DIR: await $.env.get("XDG_RUNTIME_DIR"),
67      HOME: await $.env.get("HOME"),
68      USERPROFILE: await $.env.get("USERPROFILE"),
69      OS: await $.env.get("OS"),
70    };
71    const dir = configDir(env);
72    const socket = socketPath(env);
73    // Kept across a hot reload, and new in each process (sent.ts); undefined on a build that
74    // lacks `$.state`.
75    const proc = { plugin: "starbridge-mod", key: "process" } as const;
76    let token: string | undefined;
77    try {
78      token = (await $.state.get(proc)).value;
79      if (!token) {
80        token = crypto.randomUUID();
81        await $.state.set(proc, token);
82      }
83    } catch {
84      token = undefined;
85    }
86    const fetch = async (method: string, path: string, body?: unknown) => {
87      const headers =
88        body === undefined ? HEADERS : { ...HEADERS, "content-type": "application/json" };
89      const init = { method, ...(body === undefined ? {} : { body: JSON.stringify(body) }) };
90      let r: Awaited<ReturnType<typeof $.http.fetch>>;
91      if (isPortFile(socket)) {
92        // Read each call: the agent writes a new port and token each time it starts.
93        const t = portTarget(await $.fs.read(socket));
94        const runs =
95          t &&
96          (await pidRuns(t.pid, env.OS === "Windows_NT", (argv) =>
97            $.process.run(argv, { timeoutMs: 10_000 }),
98          ));
99        if (!t || !runs) throw new Error(`no agent on ${socket}`);
100        const signed = await signCall(t.token);
101        r = await $.http.fetch(`http://127.0.0.1:${t.port}${path}`, {
102          ...init,
103          headers: { ...headers, ...signed.headers },
104        });
105        if (r.headers[PROOF_HEADER] !== signed.expect)
106          throw new Error(`no agent on ${socket}: the port answers without its proof`);
107      } else
108        r = await $.http.fetch(`http://agent${path}`, { ...init, socketPath: socket, headers });
109      return { status: r.status, text: r.text };
110    };
111    const log = (text: string) => $.ui.log(text, { to: "debug" });
112    const status = (text: string | undefined) => $.ui.status(text);
113    const sessionId = () => $.session.id();
114    const now = () => $.clock.now();
115    const sleep = (ms: number) => $.clock.sleep(ms);
116    const records = new Sent(
117      {
118        sessionId,
119        read: async (path) => ((await $.fs.exists(path)) ? $.fs.read(path) : undefined),
120        write: (path, text) => $.fs.write(path, text),
121        now,
122        submit: (text) => track($.prompt.submit({ text })),
123        elsewhere: async (session) => {
124          // Claude Code's own pid, as the parent of the shell it runs; Windows has no `sh`.
125          if (env.OS === "Windows_NT") return false;
126          const own = Number(
127            (await $.process.run(["sh", "-c", "echo $PPID"], { timeoutMs: 10_000 })).stdout.trim(),
128          );
129          if (!own) return false;
130          return sessionElsewhere(
131            session,
132            own,
133            { list: (path) => $.fs.list(path), read: (path) => $.fs.read(path) },
134            `${env.CLAUDE_CONFIG_DIR || `${env.HOME}/.claude`}/sessions`,
135            (pid) => pidRuns(pid, false, (argv) => $.process.run(argv, { timeoutMs: 10_000 })),
136          );
137        },
138        log,
139      },
140      dir,
141      token,
142    );
143    sent = records;
144    /** Whether subagent `agentId` runs now; when unsure, it does. */
145    const running = async (agentId: string) => {
146      try {
147        const status = (await $.agent.list()).find((a) => a.id === agentId)?.status;
148        return status === undefined || ["pending", "running", "waiting"].includes(status);
149      } catch {
150        return true;
151      }
152    };
153    // A subagent's answers go to it, whatever the main agent does (#1702).
154    const submit = async (lines: Line[]) => {
155      const me = await sessionId();
156      const { taken, rest } = await toSubagents(
157        lines,
158        async (agentId, group, text) => {
159          const send = () => $.session.send({ to: { agentId }, text });
160          // A subagent that ended is resumed with the message as its prompt, which its transcript
161          // keeps; one that runs holds it in memory until its next tool round.
162          if (!(await running(agentId))) return send();
163          return records.toAgent(
164            agentId,
165            group.map((l) => forParent(l, agentId)),
166            send,
167            () => running(agentId),
168          );
169        },
170        log,
171        failed,
172      );
173      // A /clear or /resume during the sends: the rest waits, unconfirmed, for session `me`.
174      if (rest.length > 0 && (await sessionId()) !== me) return taken;
175      const parent =
176        rest.length > 0 ? await handover.submit(rest, (left) => records.send(left)) : [];
177      return [...taken, ...parent];
178    };
179    const starting = () => waiting > 0;
180    loop = new Switch(
181      {
182        // The CLI honours STARBRIDGE_NO_AGENT too, so the CLI path then reaches the server itself.
183        agentUp: async () =>
184          !env.STARBRIDGE_NO_AGENT && (await fetch("GET", "/v1/status")).status < 300,
185        agent: (unconfirmed) =>
186          new AgentLoop(
187            {
188              sessionId,
189              cwd: () => $.session.cwd(),
190              fetch,
191              now,
192              sleep,
193              submit,
194              starting,
195              status,
196              log,
197            },
198            unconfirmed,
199          ),
200        poller: (unconfirmed) =>
201          new Poller(
202            {
203              sessionId,
204              run: (argv, timeoutMs) => $.process.run(argv, { timeoutMs }),
205              read: (path) => $.fs.read(path),
206              write: (path, text) => $.fs.write(path, text),
207              mtime: async (path) => (await $.fs.stat(path).catch(() => undefined))?.mtimeMs,
208              // The mod's host reads no owner or mode: find does, as plugin/hooks/cli.sh. Windows
209              // has no `sh`, and keeps owners in ACLs: there, as `ownedFile`, a file.
210              owned: async (path) =>
211                env.OS === "Windows_NT"
212                  ? (await $.fs.stat(path).catch(() => undefined))?.kind === "file"
213                  : (
214                      await $.process
215                        .run(["sh", "-c", OWNED, "sh", path], { timeoutMs: 10_000 })
216                        .catch(() => undefined)
217                    )?.exitCode === 0,
218              now,
219              sleep,
220              submit,
221              starting,
222              status,
223              log,
224            },
225            dir,
226            undefined,
227            undefined,
228            unconfirmed,
229          ),
230        sleep,
231        clearStatus: () => status(undefined),
232        log,
233      },
234      undefined,
235      before?.unconfirmed,
236    );
237    void records.recover().catch((e) => log(`starbridge: ${(e as Error).message}`));
238    return started;
239  });
240
241  // A subagent's run raises no `turn.start`; its `turn.complete` and tool calls carry its `agentId`.
242  on("turn.start", async (_$, e, next) => {
243    handover.started();
244    await sent?.started(e.text).catch(() => undefined);
245    return next(e);
246  });
247
248  on("turn.complete", async ($, e, next) => {
249    const r = await next(e);
250    if (e.agentId !== undefined) {
251      const agent = e.agentId;
252      const ended = roundEnded.get(agent);
253      if (ended !== undefined) void sent?.read(agent, ended).catch(() => undefined);
254      // A message that came after its last round waits for another one, which a subagent that
255      // finished never runs: the parent gets it instead (#1729).
256      void (async () => {
257        const list = () => $.agent.list();
258        if (!sent || !(await finished(list, (ms) => $.clock.sleep(ms), agent))) return;
259        roundEnded.delete(agent);
260        if (await sent.finished(agent, ended)) await sent.recover();
261      })().catch(() => undefined);
262    } else {
263      handover.ended();
264      // After a `/resume`, or a submit that failed.
265      void sent?.recover().catch(() => undefined);
266    }
267    return r;
268  });
269
270  on("tool.call", async ($, e, next) => {
271    if (e.agentId !== undefined) {
272      const agent = e.agentId;
273      // A new round: the subagent read every message sent before its last one ended.
274      const ended = roundEnded.get(agent);
275      if (ended !== undefined) void sent?.read(agent, ended).catch(() => undefined);
276      // The CLI keeps which subagent asked, so its answer goes back there (#1702).
277      const command =
278        e.tool === "Bash" && typeof e.command === "string" ? tagAsk(e.command, agent) : undefined;
279      const r = await next(command !== undefined && command !== e.command ? { ...e, command } : e);
280      roundEnded.set(agent, await $.clock.now());
281      return r;
282    }
283    const r = await next(e);
284    if (r.deny !== undefined) return r;
285    const lines = await handover.take((acks) => loop?.claim(acks) ?? Promise.resolve([]));
286    if (lines.length === 0) return r;
287    $.ui.log(`starbridge: handed ${lines.length} answer(s) with the ${e.tool} result`, {
288      to: "debug",
289    });
290    return { ...r, context: [...(r.context ?? []), ...lines] };
291  });
292
293  // The tag `tool.call` adds is not the subagent's, so the rules judge the command it wrote: where
294  // they would ask about the tagged one, the verdict on the untagged one stands (#1702).
295  on("tool.check", async ($, e, next) => {
296    const verdict = await next(e);
297    const input = e.input as { command?: unknown } | undefined;
298    if (verdict.decision !== "ask" || verdict.hook !== undefined) return verdict;
299    if (e.tool !== "Bash" || typeof input?.command !== "string") return verdict;
300    const command = untagAsk(input.command);
301    if (command === input.command) return verdict;
302    const plain = await $.tool.check({ tool: "Bash", input: { ...input, command } });
303    return plain.decision === "ask" ? verdict : plain;
304  });
305
306  // After a /clear or a /resume the process goes on under another session id, with no
307  // `session.start`, and the loop reads the id each step. Only the end of the process stops it.
308  on("session.end", async (_$, e, next) => {
309    handover.ended();
310    if (e.reason !== "clear" && e.reason !== "resume") {
311      await loop?.end();
312      loop = undefined;
313    }
314    return next(e);
315  });
316};
317
318/** A subagent's statuses once it runs no more rounds. */
319const DONE = ["completed", "failed", "killed"];
320
321/**
322 * Whether subagent `agentId`, whose turn just completed, runs no more rounds: done, or no longer
323 * listed. The engine may mark it a moment after its `turn.complete`, so it looks twice. An idle
324 * teammate wakes for a message, so it has not; nor, when the list cannot be read, has any.
325 */
326export async function finished(
327  list: () => Promise<{ id: string; status: string }[]>,
328  sleep: (ms: number) => Promise<void>,
329  agentId: string,
330): Promise<boolean> {
331  for (let i = 0; i < 2; i++) {
332    if (i > 0) await sleep(2000);
333    const agents = await list().catch(() => undefined);
334    if (!agents) return false;
335    const status = agents.find((a) => a.id === agentId)?.status;
336    if (status === undefined || DONE.includes(status)) return true;
337  }
338  return false;
339}
340

1/**
2 * The mod as a thin client of the machine's Starbridge agent (PROTOCOL.md, "Local agent API").
3 *
4 * Each session long-polls `GET /v1/sessions/<id>/events?wait=25` on the agent's unix socket,
5 * submits each event's line into the session and confirms it with `POST /v1/sessions/<id>/ack`.
6 * The agent holds the keys and the one connection to the server, so no session runs the CLI,
7 * takes a lease or watches files. When no agent answers, `Switch` falls back to `Poller`, the
8 * CLI-driven loop, until one does.
9 */
10
11import { configDir } from "./poller.ts";
12
13/** The agent API revision this mod speaks, sent in the `starbridge-api` header. */
14export const API = 1;
15export const VERSION = "0.1.8-rc.2";
16const CLIENT = `starbridge-mod/${VERSION}`;
17
18/** Event types this mod submits; others are skipped and left unconfirmed, as the API asks. */
19const KNOWN = new Set(["answer"]);
20
21/** An answer's line, and what confirms it. */
22export interface Line {
23  text: string;
24  ack: string;
25  /** The Claude Code subagent that asked, which gets the line instead of the session (#1702). */
26  subagent?: string;
27}
28
29/** Several answers as one message: one line each, in the order they came. */
30export const joinLines = (lines: Line[]) => lines.map((l) => l.text).join("\n");
31
32export interface Reply {
33  status: number;
34  text: string;
35}
36
37export interface AgentHost {
38  /** The session's id now; a `/clear` changes it. */
39  sessionId(): Promise<string>;
40  cwd(): Promise<string>;
41  /** One HTTP call on the agent's socket. Rejects when it cannot connect or the host aborts it. */
42  fetch(method: string, path: string, body?: unknown): Promise<Reply>;
43  now(): Promise<number>;
44  sleep(ms: number): Promise<void>;
45  /**
46   * Submits `lines`, every answer one poll brought, as one message (#1597), and returns the acks
47   * of those it submitted; a line it could not submit, or holds back for now, stays unconfirmed
48   * for a later try. Must not wait for the turn to start: a prompt submitted mid-turn waits for
49   * its own.
50   */
51  submit(lines: Line[]): string[] | Promise<string[]>;
52  /**
53   * True while a submitted prompt waits for its turn. Claude Code starts a plugin's queued prompt
54   * only once the mod's open `fetch` or `run` returns, so the loop then holds no long call (#1446).
55   */
56  starting(): boolean;
57  status(text: string | undefined): void;
58  log(text: string): void;
59}
60
61export interface AgentTiming {
62  /** Seconds the agent may hold one events request (the host aborts a call after 30 s). */
63  waitSeconds: number;
64  /**
65   * An events request that comes back sooner than this with nothing waits out the rest; while a
66   * prompt waits for its turn, requests don't wait and this is the time between them.
67   */
68  minCycleMs: number;
69  /** First retry delay after an error, doubled each time up to `maxBackoffMs`. */
70  backoffMs: number;
71  maxBackoffMs: number;
72}
73
74export const AGENT_TIMING: AgentTiming = {
75  waitSeconds: 25,
76  minCycleMs: 2_000,
77  backoffMs: 2_000,
78  maxBackoffMs: 60_000,
79};
80
81/**
82 * Where the agent listens, as the CLI's `socketPath` (cli/src/agent/api.ts) works it out:
83 * `$STARBRIDGE_AGENT_SOCKET`; on Windows (`OS=Windows_NT`) the port file `agent.port` in the
84 * config directory; for the default config directory `$XDG_RUNTIME_DIR/starbridge/agent.sock`
85 * when that is set; else `agent.sock` in the config directory.
86 */
87export function socketPath(env: {
88  STARBRIDGE_AGENT_SOCKET?: string;
89  STARBRIDGE_CONFIG_DIR?: string;
90  XDG_CONFIG_HOME?: string;
91  XDG_RUNTIME_DIR?: string;
92  HOME?: string;
93  USERPROFILE?: string;
94  OS?: string;
95}): string {
96  if (env.STARBRIDGE_AGENT_SOCKET) return env.STARBRIDGE_AGENT_SOCKET;
97  const dir = configDir(env);
98  if (env.OS === "Windows_NT") return `${dir}/agent.port`;
99  const standard = configDir({ ...env, STARBRIDGE_CONFIG_DIR: undefined });
100  if (env.XDG_RUNTIME_DIR && dir === standard)
101    return `${env.XDG_RUNTIME_DIR}/starbridge/agent.sock`;
102  return `${dir}/agent.sock`;
103}
104
105/**
106 * An address ending in `.port` names the file where an agent on loopback TCP wrote its port and
107 * its token, which calls prove they hold without sending it (the CLI's `PortFile` and `proof`).
108 */
109export const isPortFile = (address: string) => address.endsWith(".port");
110
111/**
112 * The port, the token and the agent's pid in port file `text`; undefined when it is not one, as
113 * the CLI's `readPortFile` reads it.
114 */
115export function portTarget(text: string): { port: number; token: string; pid: number } | undefined {
116  try {
117    const f = JSON.parse(text) as { port?: unknown; token?: unknown; pid?: unknown };
118    if (
119      Number.isInteger(f.port) &&
120      typeof f.token === "string" &&
121      f.token.length >= 32 &&
122      Number.isInteger(f.pid)
123    )
124      return { port: f.port as number, token: f.token, pid: f.pid as number };
125  } catch {}
126  return undefined;
127}
128
129/**
130 * Whether process `pid` runs, for a host that can run commands but not signal a process:
131 * `tasklist` on Windows, `ps` elsewhere. A call sends nothing to a port whose agent's pid runs no
132 * more, as the CLI does (#570): a dead agent's port shows in `netstat` to every local user, and
133 * whoever takes it would get the call's body before failing its proof. A check that cannot run
134 * counts as not running, so the mod falls back to the CLI.
135 */
136export async function pidRuns(
137  pid: number,
138  windows: boolean,
139  run: (argv: string[]) => Promise<{ exitCode: number | null; stdout: string }>,
140): Promise<boolean> {
141  try {
142    if (windows) {
143      // One CSV row per match, `"image","pid",…`; with none, a message in the system's language.
144      const r = await run(["tasklist", "/FI", `PID eq ${pid}`, "/FO", "CSV", "/NH"]);
145      return r.exitCode === 0 && new RegExp(`^"[^"]*","${pid}",`, "m").test(r.stdout);
146    }
147    const r = await run(["ps", "-p", String(pid), "-o", "pid="]);
148    return r.exitCode === 0 && r.stdout.trim() === String(pid);
149  } catch {
150    return false;
151  }
152}
153
154const hex = (bytes: Uint8Array) => [...bytes].map((b) => b.toString(16).padStart(2, "0")).join("");
155
156async function proof(token: string, role: "client" | "agent", nonce: string): Promise<string> {
157  const data = new TextEncoder().encode(`${token}:${role}:${nonce}`);
158  return hex(new Uint8Array(await crypto.subtle.digest("SHA-256", data)));
159}
160
161/**
162 * The headers that prove a call holds `token`, and the `starbridge-proof` header the agent's
163 * answer must carry: anything that answers without it took the port of an agent that stopped.
164 */
165export async function signCall(token: string): Promise<{
166  headers: Record<string, string>;
167  expect: string;
168}> {
169  const nonce = hex(crypto.getRandomValues(new Uint8Array(16)));
170  return {
171    headers: {
172      "starbridge-nonce": nonce,
173      authorization: `Starbridge ${await proof(token, "client", nonce)}`,
174    },
175    expect: await proof(token, "agent", nonce),
176  };
177}
178
179export const PROOF_HEADER = "starbridge-proof";
180
181/** The headers every call carries. */
182export const HEADERS: Record<string, string> = {
183  "starbridge-api": String(API),
184  "user-agent": CLIENT,
185};
186
187interface Event {
188  type?: unknown;
189  ack?: unknown;
190  line?: unknown;
191  decisionId?: unknown;
192  subagent?: unknown;
193}
194
195/** Why `AgentLoop.run` returned: `stop`, or the agent cannot serve this mod. */
196export type Ended = "stopped" | "unreachable";
197
198export class AgentLoop {
199  private stopped = false;
200  private failures = 0;
201  /** The session id `hello` last went out under. */
202  private greeted: string | undefined;
203
204  constructor(
205    private readonly host: AgentHost,
206    /** Lines submitted but not yet confirmed, shared with `Poller` so a switch submits nothing twice. */
207    private readonly unconfirmed: Set<string>,
208    private readonly t: AgentTiming = AGENT_TIMING,
209  ) {}
210
211  /** Ends the loop after the current step. */
212  stop() {
213    this.stopped = true;
214  }
215
216  /** Runs until `stop`, or until a call cannot reach the agent or gets a 426. */
217  async run(): Promise<Ended> {
218    while (!this.stopped) {
219      try {
220        await this.step();
221      } catch (e) {
222        // A call that never connected, one the host aborted (the agent is gone or hung), or a
223        // 426 (it speaks another API revision).
224        this.host.log(`starbridge: agent unreachable: ${(e as Error).message}`);
225        return "unreachable";
226      }
227    }
228    return "stopped";
229  }
230
231  /**
232   * Confirms the events among `acks` that the agent still hands this session, and returns those:
233   * a line a `wait` or `waiting` printed meanwhile is not handed over again (#1484).
234   */
235  async claim(acks: string[]): Promise<string[]> {
236    const me = await this.host.sessionId();
237    const r = await this.host.fetch("GET", `${path(me)}/events?wait=0`);
238    if (r.status < 200 || r.status >= 300) return [];
239    const pending = new Set((parse<{ events?: Event[] }>(r.text)?.events ?? []).map((e) => e.ack));
240    const mine = acks.filter((a) => pending.has(a));
241    if (mine.length === 0) return [];
242    const ack = await this.host.fetch("POST", `${path(me)}/ack`, { acks: mine });
243    if (ack.status < 200 || ack.status >= 300) return [];
244    // What it confirmed now: not a line a `wait` or `waiting` took since the read. An agent older
245    // than that names none.
246    return parse<{ acked?: string[] }>(ack.text)?.acked ?? mine;
247  }
248
249  /** Tells the agent the session ended, so its session-scoped state goes. */
250  async bye(): Promise<void> {
251    const me = this.greeted;
252    if (!me) return;
253    await this.host.fetch("POST", `${path(me)}/bye`).catch(() => undefined);
254  }
255
256  private async step() {
257    const me = await this.host.sessionId();
258    if (!me) {
259      await this.host.sleep(this.t.minCycleMs);
260      return;
261    }
262    if (this.greeted !== me) {
263      // After a /clear: the old id has no mod any more (#537).
264      const replaces = this.greeted ? { replaces: this.greeted } : {};
265      const r = await this.host.fetch("POST", `${path(me)}/hello`, {
266        cwd: await this.host.cwd(),
267        ...replaces,
268      });
269      if (!(await this.ok(r))) return;
270      this.greeted = me;
271    }
272    const started = await this.host.now();
273    const wait = this.host.starting() ? 0 : this.t.waitSeconds;
274    const r = await this.host.fetch("GET", `${path(me)}/events?wait=${wait}`);
275    if (!(await this.ok(r))) return;
276    const events = (parse<{ events?: Event[] }>(r.text)?.events ?? []).filter(
277      (e): e is Event & { ack: string; line: string } =>
278        typeof e.type === "string" &&
279        KNOWN.has(e.type) &&
280        typeof e.ack === "string" &&
281        typeof e.line === "string",
282    );
283    // A /clear during the call made another session current: the events wait, unconfirmed,
284    // until session `me` is resumed.
285    if (events.length > 0 && (await this.host.sessionId()) !== me) {
286      this.host.log(
287        `starbridge: held back the answers ${events.map((l) => l.ack).join(", ")}: their session ${me} ended`,
288      );
289      return;
290    }
291    const due = events.filter((e) => !this.unconfirmed.has(e.ack));
292    const taken =
293      due.length > 0
294        ? await this.host.submit(
295            due.map((e) => ({
296              text: e.line,
297              ack: e.ack,
298              ...(typeof e.subagent === "string" ? { subagent: e.subagent } : {}),
299            })),
300          )
301        : [];
302    for (const a of taken) this.unconfirmed.add(a);
303    const handed = taken.length;
304    const done = events.filter((e) => this.unconfirmed.has(e.ack)).map((e) => e.ack);
305    if (done.length > 0) {
306      const ack = await this.host.fetch("POST", `${path(me)}/ack`, { acks: done });
307      if (!(await this.ok(ack))) return;
308      for (const a of done) this.unconfirmed.delete(a);
309    }
310    const left = started + this.t.minCycleMs - (await this.host.now());
311    if (handed === 0 && left > 0) await this.host.sleep(left);
312  }
313
314  /** True on a 2xx; otherwise shows the agent's reason and waits before the next try. */
315  private async ok(r: Reply): Promise<boolean> {
316    if (r.status >= 200 && r.status < 300) {
317      if (this.failures > 0) this.host.status(undefined);
318      this.failures = 0;
319      return true;
320    }
321    const body = parse<{ error?: string; detail?: string }>(r.text);
322    const reason = body?.detail ?? body?.error ?? `status ${r.status}`;
323    // 426: the agent and this mod cannot work together; the CLI's own path still can.
324    if (r.status === 426) throw new Error(reason);
325    this.failures++;
326    const delay = Math.min(this.t.maxBackoffMs, this.t.backoffMs * 2 ** (this.failures - 1));
327    const text = `starbridge: ${reason}`;
328    this.host.status(text);
329    this.host.log(`${text}; retrying in ${Math.round(delay / 1000)} s`);
330    await this.host.sleep(delay);
331    return false;
332  }
333}
334
335function path(session: string): string {
336  return `/v1/sessions/${encodeURIComponent(session)}`;
337}
338
339function parse<T>(text: string): T | undefined {
340  try {
341    return JSON.parse(text) as T;
342  } catch {
343    return undefined;
344  }
345}
346

1import type { Line } from "./agent.ts";
2
3/**
4 * When an answer reaches the agent (#1484, #1584). Claude Code starts a prompt the mod submits only
5 * after the running turn, which can be minutes away, and the turn meanwhile works without the
6 * owner's word. So while the main agent's turn runs, its next tool result carries the answer as
7 * context, waiting or not, as Claude Code does with a message typed during a turn. An answer that
8 * comes after the turn's last tool call goes as a prompt once the turn ends. The mod's own
9 * `prompt.submit` hook never sees its own prompts, so it could not drop a second copy: a line goes
10 * out one way only.
11 */
12export class Handover {
13  private running = false;
14  /** Lines for the main agent's next tool call, by their line, with what confirms each. */
15  private readonly next = new Map<string, string>();
16  /**
17   * Lines a tool result carried, confirmed then: a loop that read one before that, and submits it
18   * now, finds it here and submits nothing.
19   */
20  private readonly handed = new Set<string>();
21
22  /**
23   * The loops' submit. Submits `lines` as one prompt with `enter` while no turn runs. During one,
24   * keeps them, with the acks that confirm them, for the next tool call, and takes none: they
25   * stay unconfirmed and come back in the next poll. A line a tool result already carried counts
26   * as taken without a second submit. Returns the acks taken, with those `enter` reports it took.
27   */
28  async submit(
29    lines: Line[],
30    enter: (lines: Line[]) => string[] | Promise<string[]>,
31  ): Promise<string[]> {
32    const carried = lines.filter((l) => this.handed.delete(l.text));
33    const rest = lines.filter((l) => !carried.includes(l));
34    if (this.running) {
35      for (const l of rest) this.next.set(l.text, l.ack);
36      return carried.map((l) => l.ack);
37    }
38    const entered = rest.length > 0 ? await enter(rest) : [];
39    return [...carried.map((l) => l.ack), ...entered];
40  }
41
42  /** The main agent's turn started. */
43  started() {
44    this.running = true;
45  }
46
47  /**
48   * The lines a main-agent tool result carries. `claim` confirms them first and returns the acks
49   * still due: a line a `wait` or `waiting` printed meanwhile goes no further, and one confirmed
50   * here is not handed over again after a reload or a restart.
51   */
52  async take(claim: (acks: string[]) => Promise<string[]>): Promise<string[]> {
53    if (this.next.size === 0) return [];
54    const lines = [...this.next];
55    this.next.clear();
56    const due = new Set(await claim(lines.map(([, ack]) => ack)));
57    const out = lines.filter(([, ack]) => due.has(ack)).map(([line]) => line);
58    for (const line of out) this.handed.add(line);
59    return out;
60  }
61
62  /**
63   * No turn runs: the main agent's turn ended, or the session started or ended. Lines kept for a
64   * tool call are still unconfirmed, so the next poll submits them, for the session that asked.
65   */
66  ended() {
67    this.running = false;
68    this.next.clear();
69  }
70}
71

1import type { Line } from "./agent.ts";
2
3/**
4 * The answer loop, apart from the engine's `$` so plain tests can drive it.
5 *
6 * Every session that loads the mod runs this loop, but only one per machine polls the server: the
7 * holder of a lease file in the CLI's config directory. The holder runs `starbridge answers
8 * --session <its id> --wait 25` back to back (the host aborts a call after 30 s); the others
9 * watch the CLI's state file and run `starbridge answers --session <id>` without `--wait` when
10 * it changes, which never touches the network. The CLI holds the keys, opens and verifies each
11 * answer, and hands a session only the answers to decisions it asked, until the poller confirms
12 * it submitted them with `starbridge answers --ack`.
13 */
14
15export interface Host {
16  /** The session's id now; a `/clear` changes it. */
17  sessionId(): Promise<string>;
18  run(
19    argv: string[],
20    timeoutMs: number,
21  ): Promise<{ exitCode: number | null; stdout: string; stderr: string }>;
22  /** Rejects when the file is missing. */
23  read(path: string): Promise<string>;
24  write(path: string, text: string): Promise<void>;
25  /** Undefined when the file is missing. */
26  mtime(path: string): Promise<number | undefined>;
27  /**
28   * Whether `path` is a file owned by this user or root that no one else may write (`ownedFile`,
29   * node.ts), so another user cannot pick the CLI that starts.
30   */
31  owned(path: string): Promise<boolean>;
32  now(): Promise<number>;
33  sleep(ms: number): Promise<void>;
34  /** `AgentHost.submit`: one message for the lines one read brought, returning the acks it took. */
35  submit(lines: Line[]): string[] | Promise<string[]>;
36  /** True while a submitted prompt waits for its turn: `AgentHost.starting`. */
37  starting(): boolean;
38  status(text: string | undefined): void;
39  log(text: string): void;
40}
41
42export interface Timing {
43  /** Seconds the server may hold one poll. */
44  waitSeconds: number;
45  /** How often a session that does not poll checks the state file. */
46  checkMs: number;
47  /** How often it reads its answers even when no change of the state file was seen. */
48  recheckMs: number;
49  /** How long a lease lasts past its last renewal. */
50  leaseMs: number;
51  /**
52   * A poll that comes back sooner than this with nothing waits out the rest; while a prompt waits
53   * for its turn, polls don't wait and this is the time between them.
54   */
55  minCycleMs: number;
56  /** First retry delay after an error, doubled each time up to `maxBackoffMs`. */
57  backoffMs: number;
58  maxBackoffMs: number;
59}
60
61export const TIMING: Timing = {
62  waitSeconds: 25,
63  checkMs: 2_000,
64  recheckMs: 30_000,
65  leaseMs: 60_000,
66  minCycleMs: 2_000,
67  backoffMs: 2_000,
68  maxBackoffMs: 60_000,
69};
70
71interface Lease {
72  session: string;
73  until: number;
74}
75
76/**
77 * `$STARBRIDGE_CONFIG_DIR`, else `$XDG_CONFIG_HOME/starbridge`, else `~/.config/starbridge`, as
78 * the CLI; on Windows (`OS=Windows_NT`) home is `USERPROFILE`, as the CLI reads it, whatever
79 * `HOME` a Git Bash set.
80 */
81/** Whether `path` is absolute, on Windows too: a relative one would depend on the agent's folder. */
82export const absolute = (path: string) => /^(\/|[A-Za-z]:[\\/])/.test(path);
83
84export function configDir(env: {
85  STARBRIDGE_CONFIG_DIR?: string;
86  XDG_CONFIG_HOME?: string;
87  HOME?: string;
88  USERPROFILE?: string;
89  OS?: string;
90}): string {
91  if (env.STARBRIDGE_CONFIG_DIR) return env.STARBRIDGE_CONFIG_DIR;
92  const home = env.OS === "Windows_NT" ? env.USERPROFILE || env.HOME : env.HOME || env.USERPROFILE;
93  return `${env.XDG_CONFIG_HOME || `${home}/.config`}/starbridge`;
94}
95
96/** The CLI's first error line, without the "starbridge: " the status line adds again. */
97function firstLine(stderr: string): string {
98  return (stderr.trim().split("\n")[0] ?? "").replace(/^starbridge:\s*/, "");
99}
100
101export class Poller {
102  private stopped = false;
103  private failures = 0;
104  /** Lines the host held back in the last `answers`. */
105  private held = 0;
106  /** The state file's time when this session last read it; a /clear makes it unread again. */
107  private seen: { session: string; mtime: number; at: number } | undefined;
108  /** The session id this poller last held the lease under; a `/clear` changes the id. */
109  private leasedAs: string | undefined;
110  private readonly leasePath: string;
111  private readonly cliPath: string;
112  private readonly statePath: string;
113  /** Resolves when the loop has ended. */
114  readonly done: Promise<void>;
115
116  constructor(
117    private readonly host: Host,
118    dir: string,
119    /** The CLI; by default the path setup recorded in `dir`, else `starbridge` on the PATH (#612). */
120    private readonly command?: string,
121    private readonly t: Timing = TIMING,
122    /**
123     * Lines submitted but not yet confirmed, so a retried confirm submits nothing twice; shared
124     * with `AgentLoop` (agent.ts) across a switch.
125     */
126    private readonly unconfirmed = new Set<string>(),
127  ) {
128    this.leasePath = `${dir}/mod-poller.json`;
129    this.cliPath = `${dir}/cli-path`;
130    this.statePath = `${dir}/state.json`;
131    // Nothing awaits a loop that ended before `stop`: a rejection would take the harness down (#1365).
132    this.done = this.loop().catch((e) => {
133      try {
134        this.host.log(`starbridge: the answer loop stopped: ${(e as Error).message}`);
135      } catch {}
136    });
137  }
138
139  /**
140   * Ends the loop after the current step and gives up the lease at once, so another session polls.
141   * Never rejects: it runs after the session ended, when the host may throw (#1365).
142   */
143  async stop(): Promise<void> {
144    this.stopped = true;
145    try {
146      const me = await this.host.sessionId();
147      const lease = await this.readLease();
148      if (lease && (lease.session === me || lease.session === this.leasedAs))
149        await this.host.write(this.leasePath, JSON.stringify({ ...lease, until: 0 }));
150    } catch {
151      // The lease runs out on its own.
152    }
153  }
154
155  private async cli(): Promise<string> {
156    if (this.command) return this.command;
157    try {
158      if (!(await this.host.owned(this.cliPath))) return "starbridge";
159      const recorded = (await this.host.read(this.cliPath)).trim();
160      // A binary removed since setup recorded it: the PATH may still hold another.
161      if (absolute(recorded) && (await this.host.owned(recorded))) return recorded;
162    } catch {}
163    return "starbridge";
164  }
165
166  private async loop() {
167    while (!this.stopped) {
168      try {
169        await this.step();
170      } catch (e) {
171        await this.fail((e as Error).message);
172      }
173    }
174  }
175
176  private async step() {
177    const me = await this.host.sessionId();
178    if (!me) {
179      await this.host.sleep(this.t.checkMs);
180      return;
181    }
182    if (await this.lease(me, this.t.leaseMs)) {
183      const started = await this.host.now();
184      const wait = this.host.starting() ? 0 : this.t.waitSeconds;
185      const handed = await this.answers(me, ["--wait", String(wait)]);
186      if (handed === undefined) return;
187      const left = started + this.t.minCycleMs - (await this.host.now());
188      if (handed === 0 && left > 0) await this.host.sleep(left);
189      return;
190    }
191    const mtime = await this.host.mtime(this.statePath);
192    const now = await this.host.now();
193    const unchanged = this.seen?.session === me && this.seen.mtime === mtime;
194    if (mtime === undefined || (unchanged && now - (this.seen?.at ?? 0) < this.t.recheckMs)) {
195      await this.host.sleep(this.t.checkMs);
196      return;
197    }
198    if ((await this.answers(me, [])) === undefined) return;
199    // A line the host held back for the turn's end is read again at the next check (#1484).
200    if (this.held === 0) this.seen = { session: me, mtime, at: now };
201    else await this.host.sleep(this.t.checkMs);
202  }
203
204  /**
205   * Confirms the lines among `acks` that the CLI still hands this session, and returns those: a
206   * line a `wait` or `waiting` printed meanwhile is not handed over again (#1484).
207   */
208  async claim(acks: string[]): Promise<string[]> {
209    const me = await this.host.sessionId();
210    const base = [await this.cli(), "answers", "--session", me];
211    const timeoutMs = (this.t.waitSeconds + 30) * 1000;
212    const r = await this.host.run(base, timeoutMs);
213    if (r.exitCode !== 0) return [];
214    const pending = new Set<string>();
215    for (const text of r.stdout.split("\n")) {
216      try {
217        const l = JSON.parse(text) as { decisionId?: unknown; ack?: unknown };
218        const ack = typeof l.ack === "string" ? l.ack : l.decisionId;
219        if (typeof ack === "string") pending.add(ack);
220      } catch {}
221    }
222    const mine = acks.filter((a) => pending.has(a));
223    if (mine.length === 0) return [];
224    const ack = await this.host.run([...base, ...mine.flatMap((a) => ["--ack", a])], timeoutMs);
225    if (ack.exitCode !== 0) return [];
226    // What it confirmed now: not a line a `wait` or `waiting` took since the read. A CLI older
227    // than that prints nothing.
228    try {
229      const acked = (JSON.parse(ack.stdout) as { acked?: unknown }).acked;
230      if (Array.isArray(acked)) return acked.filter((a): a is string => typeof a === "string");
231    } catch {}
232    return mine;
233  }
234
235  /**
236   * Runs one `starbridge answers`, submits each answer it hands over and confirms them. Returns
237   * how many, or undefined after an error, which it has already waited out.
238   */
239  private async answers(me: string, extra: string[]): Promise<number | undefined> {
240    const timeoutMs = (this.t.waitSeconds + 30) * 1000;
241    const base = [await this.cli(), "answers", "--session", me];
242    const r = await this.host.run([...base, ...extra], timeoutMs);
243    if (r.exitCode !== 0) {
244      await this.fail(firstLine(r.stderr) || `exit ${r.exitCode}`);
245      return undefined;
246    }
247    // Warnings on success, such as an answer that failed its checks.
248    for (const line of r.stderr.split("\n")) if (line.trim()) this.host.log(line);
249    if (this.failures > 0) this.host.status(undefined);
250    this.failures = 0;
251    const lines: Line[] = [];
252    for (const text of r.stdout.split("\n")) {
253      if (!text.trim()) continue;
254      let parsed: { decisionId?: unknown; ack?: unknown; line?: unknown; subagent?: unknown };
255      try {
256        parsed = JSON.parse(text);
257      } catch {
258        this.host.log(`starbridge: unreadable line from the CLI: ${text.slice(0, 200)}`);
259        continue;
260      }
261      const { decisionId: id, line } = parsed;
262      if (typeof id !== "string" || typeof line !== "string") continue;
263      lines.push({
264        text: line,
265        ack: typeof parsed.ack === "string" ? parsed.ack : id,
266        ...(typeof parsed.subagent === "string" ? { subagent: parsed.subagent } : {}),
267      });
268    }
269    // A /clear during the call made another session current: the answers wait, unconfirmed,
270    // until session `me` is resumed.
271    if (lines.length > 0 && (await this.host.sessionId()) !== me) {
272      this.host.log(
273        `starbridge: held back the answers ${lines.map((l) => l.ack).join(", ")}: their session ${me} ended`,
274      );
275      this.held = 0;
276      return 0;
277    }
278    const due = lines.filter((l) => !this.unconfirmed.has(l.ack));
279    const taken = due.length > 0 ? await this.host.submit(due) : [];
280    for (const a of taken) this.unconfirmed.add(a);
281    const handed = taken.length;
282    const held = due.length - handed;
283    const done = lines.filter((l) => this.unconfirmed.has(l.ack)).map((l) => l.ack);
284    if (done.length > 0) {
285      const ack = await this.host.run([...base, ...done.flatMap((id) => ["--ack", id])], timeoutMs);
286      if (ack.exitCode !== 0) {
287        await this.fail(firstLine(ack.stderr) || `exit ${ack.exitCode}`);
288        return undefined;
289      }
290      for (const id of done) this.unconfirmed.delete(id);
291    }
292    this.held = held;
293    return handed;
294  }
295
296  /** Waits before the next try, twice as long after each error in a row. */
297  private async fail(message: string) {
298    this.failures++;
299    const delay = Math.min(this.t.maxBackoffMs, this.t.backoffMs * 2 ** (this.failures - 1));
300    const text = `starbridge: ${message}`;
301    this.host.status(text);
302    this.host.log(`${text}; retrying in ${Math.round(delay / 1000)} s`);
303    // A poller that backs off keeps its lease, so the other sessions do not all start failing.
304    try {
305      const me = await this.host.sessionId();
306      const held = (await this.readLease())?.session;
307      if (me && held !== undefined && (held === me || held === this.leasedAs))
308        await this.lease(me, delay + this.t.leaseMs);
309    } catch {
310      // The lease is an optimisation; losing it costs one more poller.
311    }
312    await this.host.sleep(delay);
313  }
314
315  private async readLease(): Promise<Lease | undefined> {
316    try {
317      const lease = JSON.parse(await this.host.read(this.leasePath)) as Lease;
318      return typeof lease.session === "string" && typeof lease.until === "number"
319        ? lease
320        : undefined;
321    } catch {
322      return undefined;
323    }
324  }
325
326  /**
327   * Takes or renews the lease unless another live session holds it. Two sessions that write at
328   * once both read back the last write, so at most one goes on to poll; a double poll would
329   * still be harmless, since the CLI keeps each answer once and hands it over once.
330   */
331  private async lease(me: string, forMs: number): Promise<boolean> {
332    // A step still running when `stop` gave the lease up must not take it back.
333    if (this.stopped) return false;
334    const now = await this.host.now();
335    const held = await this.readLease();
336    const ours = held?.session === me || (held !== undefined && held.session === this.leasedAs);
337    if (held && !ours && held.until > now) return false;
338    const mine = ours && (held?.until ?? 0) > now;
339    await this.host.write(this.leasePath, JSON.stringify({ session: me, until: now + forMs }));
340    if (!mine) {
341      await this.host.sleep(100);
342      if ((await this.readLease())?.session !== me) return false;
343    }
344    this.leasedAs = me;
345    return true;
346  }
347}
348

1/**
2 * Answers the mod submitted as prompts, kept until their turn starts (#1517).
3 *
4 * The loops confirm a line to the CLI or the agent as soon as they submit it, so a `wait` or
5 * `waiting` never prints an answer already queued as a prompt. But Claude Code keeps a queued
6 * prompt in memory only: one still waiting for its turn when the process exits is gone, and
7 * `--resume` does not bring it back. So each line is written to a file of its session before it
8 * is submitted, and leaves it once its turn starts: `turn.start` carries its text, or
9 * `$.prompt.submit` resolves. A process that starts on a session whose file still holds lines
10 * from another process submits them again.
11 *
12 * Lines carry the token of the process that submitted them, which `$.state` keeps across a hot
13 * reload: a reload leaves the queued prompt in place, so a reloaded mod submits nothing again.
14 * Nor does a process that shares its session with another live one.
15 *
16 * Every operation reads the file afresh, since another process on the session writes it too.
17 */
18
19export interface SentHost {
20  /** The session's id now; a `/clear` or a `/resume` changes it. */
21  sessionId(): Promise<string>;
22  /** Undefined when the file is missing; rejects when it cannot be read. */
23  read(path: string): Promise<string | undefined>;
24  /** Creates the file's folder as needed. */
25  write(path: string, text: string): Promise<void>;
26  now(): Promise<number>;
27  /**
28   * `$.prompt.submit`: resolves when the prompt's turn starts, or to `{ drop }` when a hook
29   * refused it.
30   */
31  submit(text: string): Promise<unknown>;
32  /**
33   * Whether another live process runs `session`, such as a `claude --resume` in a second
34   * terminal: the lines it submitted may still be queued there.
35   */
36  elsewhere(session: string): Promise<boolean>;
37  log(text: string): void;
38}
39
40interface Line {
41  ack: string;
42  line: string;
43  /** The token of the process that submitted it last; empty when that submit failed. */
44  process: string;
45  at: number;
46  /** Submits a hook refused or the engine rejected; a process exiting is not one. */
47  refused: number;
48  /**
49   * Sent to this subagent, which has not read it yet (#1702); `line` is then what the parent gets
50   * if the process exits first.
51   */
52  agent?: string;
53}
54
55/** Refused submits after which a line is dropped, with a log line. */
56const MAX_REFUSED = 3;
57/** A line older than this is dropped: the server drops an unanswered decision after 30 days. */
58const MAX_AGE_MS = 30 * 24 * 3600_000;
59
60export class Sent {
61  /** Sessions whose file this process read. */
62  private readonly sessions = new Set<string>();
63  /** Sessions whose lines from other processes went out again. */
64  private readonly recovered = new Set<string>();
65  /** Every operation runs alone, from its read to its submit. */
66  private queue: Promise<unknown> = Promise.resolve();
67
68  constructor(
69    private readonly host: SentHost,
70    private readonly dir: string,
71    /**
72     * This process's token; undefined where `$.state` is missing, and then no line is submitted
73     * again, since a reload could not be told from a new process.
74     */
75    private readonly process: string | undefined,
76  ) {}
77
78  /**
79   * Records `batch`, then submits it as one prompt (#1597). Returns the acks of the lines that went
80   * out, now or before: not those it could not record, or that a `/clear` or `/resume` held back
81   * meanwhile, which the loop leaves unconfirmed for a later try.
82   */
83  send(batch: { text: string; ack: string }[]): Promise<string[]> {
84    return this.alone(async () => {
85      const me = await this.host.sessionId();
86      const at = await this.host.now();
87      let lines = await this.loaded(me);
88      if (!lines) return [];
89      const entries: Line[] = [];
90      /** Lines already out, from this process or another. */
91      const out: string[] = [];
92      for (const { text, ack } of batch) {
93        // A process that died between recording a line and confirming it leaves both: the loop
94        // hands the line over again, and the file holds it. It goes out once, as `recover` would.
95        let known: Line | undefined = lines.find((l) => l.ack === ack);
96        if (known && (this.process === undefined || known.process === this.process)) {
97          out.push(ack);
98          continue;
99        }
100        if (known && known.process !== "") {
101          if (await this.host.elsewhere(me).catch(() => false)) {
102            out.push(ack);
103            continue;
104          }
105          // The process may have run the line's turn before it exited: then the file lacks it. A
106          // process that started meanwhile may hold it now.
107          const before = known.process;
108          const fresh = await this.loaded(me);
109          if (!fresh) return out;
110          lines = [...fresh.filter((l) => !entries.some((e) => e.ack === l.ack)), ...entries];
111          known = lines.find((l) => l.ack === ack);
112          if (!known || (known.process !== "" && known.process !== before)) {
113            out.push(ack);
114            continue;
115          }
116        }
117        const entry: Line = known
118          ? { ...known, process: this.process ?? "" }
119          : { ack, line: text, process: this.process ?? "", at, refused: 0 };
120        lines = [...lines.filter((l) => l !== known), entry];
121        entries.push(entry);
122      }
123      if (entries.length === 0) return out;
124      try {
125        await this.save(me, lines);
126      } catch (e) {
127        const acks = entries.map((l) => l.ack).join(", ");
128        this.host.log(`starbridge: could not record the answers ${acks}: ${(e as Error).message}`);
129        return out;
130      }
131      if ((await this.host.sessionId()) !== me) {
132        // Left for session `me`: the loop holds the lines back until it is current again.
133        const left = lines.map((l) => (entries.includes(l) ? { ...l, process: "" } : l));
134        await this.save(me, left).catch(() => undefined);
135        return out;
136      }
137      this.enter(me, entries);
138      return [...out, ...entries.map((l) => l.ack)];
139    });
140  }
141
142  /**
143   * Records `batch` for subagent `agent`, as the prompt the parent gets should the process exit
144   * before the subagent reads it (#1702), then runs `deliver`, the send. A send that fails leaves
145   * no record. Claude Code keeps a message for a running subagent in memory until its next tool
146   * round, so `read` drops the record only once the subagent has moved on. `running` is asked
147   * again once the record is saved, right before the send: a subagent that finished meanwhile is
148   * resumed with the message, which needs no record, and `finished` must not hand it to the parent
149   * too (#1729).
150   */
151  toAgent(
152    agent: string,
153    batch: { text: string; ack: string }[],
154    deliver: () => Promise<{ isDelivered: boolean; reason?: string }>,
155    running: () => Promise<boolean> = async () => true,
156  ): Promise<{ isDelivered: boolean; reason?: string }> {
157    // Without a token a reload could not be told from a new process: nothing to recover by.
158    if (this.process === undefined) return deliver();
159    const process = this.process;
160    return this.alone(async () => {
161      const me = await this.host.sessionId();
162      const at = await this.host.now();
163      const acks = batch.map((l) => l.ack);
164      const lines = await this.loaded(me);
165      const entries = batch.map((l) => ({
166        ack: l.ack,
167        line: l.text,
168        process,
169        at,
170        refused: 0,
171        agent,
172      }));
173      const rest = lines?.filter((l) => !acks.includes(l.ack));
174      const saved =
175        rest !== undefined &&
176        (await this.save(me, [...rest, ...entries]).then(
177          () => true,
178          (e) => {
179            this.host.log(
180              `starbridge: could not record the answers ${acks.join(", ")}: ${(e as Error).message}`,
181            );
182            return false;
183          },
184        ));
185      // Sent without a record, an answer the subagent never read would be lost with the process.
186      if (!saved) return { isDelivered: false, reason: "could not record it" };
187      if (!(await running())) {
188        await this.save(me, rest).catch(() => undefined);
189        return deliver();
190      }
191      let r: { isDelivered: boolean; reason?: string };
192      try {
193        r = await deliver();
194      } catch (e) {
195        await this.save(me, rest).catch(() => undefined);
196        throw e;
197      }
198      if (!r.isDelivered) await this.save(me, rest).catch(() => undefined);
199      return r;
200    });
201  }
202
203  /**
204   * Subagent `agent` read what was sent to it up to `before`, the end of a tool round of its own:
205   * Claude Code hands a subagent its messages after each round. Those records go.
206   */
207  read(agent: string, before: number): Promise<void> {
208    return this.alone(async () => {
209      // After a hot reload the current session is the one its records are under.
210      await this.load(await this.host.sessionId()).catch(() => undefined);
211      for (const session of this.sessions) {
212        const lines = await this.load(session).catch(() => undefined);
213        if (!lines) continue;
214        const keep = lines.filter((l) => l.agent !== agent || l.at > before);
215        if (keep.length < lines.length) await this.save(session, keep).catch(() => undefined);
216      }
217    });
218  }
219
220  /**
221   * Subagent `agent` finished, its last tool round ending at `before` (undefined when this module
222   * saw none): what was sent to it after then, it never read (#1729). Those records become lines
223   * for the parent, which the next `recover` submits; the rest go. Returns whether any are left
224   * for the parent.
225   */
226  finished(agent: string, before: number | undefined): Promise<boolean> {
227    return this.alone(async () => {
228      await this.load(await this.host.sessionId()).catch(() => undefined);
229      let left = false;
230      for (const session of this.sessions) {
231        const lines = await this.load(session).catch(() => undefined);
232        if (!lines?.some((l) => l.agent === agent)) continue;
233        const keep: Line[] = [];
234        for (const l of lines) {
235          if (l.agent !== agent) keep.push(l);
236          else if (before === undefined || l.at > before) {
237            const { agent: _, ...line } = l;
238            keep.push({ ...line, process: "" });
239            left = true;
240          }
241        }
242        await this.save(session, keep).catch(() => undefined);
243      }
244      return left;
245    });
246  }
247
248  /**
249   * A turn started with `text`: each line it carries reached the model. Checks every session read
250   * so far, since a `/clear` may come between the submit and the turn.
251   */
252  started(text: string): Promise<void> {
253    return this.alone(async () => {
254      await this.load(await this.host.sessionId());
255      for (const session of this.sessions) {
256        const lines = await this.load(session);
257        const left = lines.filter((l) => !text.includes(l.line));
258        if (left.length < lines.length) await this.save(session, left).catch(() => undefined);
259      }
260    });
261  }
262
263  /**
264   * Submits again the current session's lines that another process submitted, or whose submit
265   * failed, once per session id, and not while another live process runs the session. Then these come from this process, and only a turn start or a
266   * failed submit changes them.
267   */
268  recover(): Promise<void> {
269    return this.alone(async () => {
270      if (this.process === undefined) return;
271      const me = await this.host.sessionId();
272      if (!me) return;
273      let lines = await this.load(me);
274      // Other processes' lines go out once another live process no longer runs this session:
275      // checked again at each turn's end until then.
276      const theirs = (l: Line) => l.process !== "" && l.process !== this.process;
277      const key = (l: Line) => `${l.ack}\n${l.process}`;
278      const found = new Set(lines.filter(theirs).map(key));
279      let take = false;
280      if (!this.recovered.has(me)) {
281        take = !(found.size > 0 && (await this.host.elsewhere(me).catch(() => false)));
282        // A process that exited may have run some of its lines' turns since the first read, and
283        // one that started meanwhile may have recorded its own, which stay with it.
284        if (take && found.size > 0) lines = await this.load(me);
285      }
286      const due = lines.filter((l) => l.process === "" || (take && found.has(key(l))));
287      if (due.length === 0) {
288        if (take) this.recovered.add(me);
289        return;
290      }
291      const keep = lines.filter((l) => !due.includes(l));
292      const again: Line[] = [];
293      for (const l of due)
294        if (l.refused >= MAX_REFUSED)
295          this.host.log(`starbridge: gave up on the answer ${l.ack}: refused ${l.refused} times`);
296        else again.push({ ...l, process: this.process });
297      // Marked once saved: a failed write leaves the lines to the next try.
298      await this.save(me, [...keep, ...again]);
299      if (take) this.recovered.add(me);
300      if ((await this.host.sessionId()) !== me) {
301        // Left for session `me`, as `send` leaves a line.
302        this.recovered.delete(me);
303        const free = again.map((l) => ({ ...l, process: "" }));
304        await this.save(me, [...keep, ...free]).catch(() => undefined);
305        return;
306      }
307      for (const l of again)
308        this.host.log(`starbridge: submitting the answer ${l.ack} again: its prompt never ran`);
309      if (again.length > 0) this.enter(me, again);
310    });
311  }
312
313  /** Submits `entries` as one prompt, one line each. */
314  private enter(session: string, entries: Line[]) {
315    const acks = entries.map((l) => l.ack);
316    this.host.submit(entries.map((l) => l.line).join("\n")).then(
317      (r) => {
318        const drop = (r as { drop?: unknown } | undefined)?.drop;
319        if (typeof drop === "string") void this.failed(session, acks, `refused: ${drop}`);
320        else void this.done(session, acks);
321      },
322      (e) => void this.failed(session, acks, (e as Error)?.message ?? String(e)),
323    );
324  }
325
326  /** `$.prompt.submit` resolved: the lines' turn started. */
327  private done(session: string, acks: string[]) {
328    return this.alone(async () => {
329      const lines = await this.load(session);
330      if (lines.some((l) => acks.includes(l.ack)))
331        await this.save(
332          session,
333          lines.filter((l) => !acks.includes(l.ack)),
334        ).catch(() => undefined);
335    });
336  }
337
338  /** The submit failed: the next `recover` submits the lines again. */
339  private failed(session: string, acks: string[], why: string) {
340    this.host.log(`starbridge: the answers ${acks.join(", ")} did not go in: ${why}`);
341    return this.alone(async () => {
342      const lines = await this.load(session);
343      if (!lines.some((l) => acks.includes(l.ack))) return;
344      await this.save(
345        session,
346        lines.map((l) =>
347          acks.includes(l.ack) ? { ...l, process: "", refused: l.refused + 1 } : l,
348        ),
349      ).catch(() => undefined);
350    });
351  }
352
353  private alone<T>(op: () => Promise<T>): Promise<T> {
354    const run = this.queue.then(op);
355    this.queue = run.catch(() => undefined);
356    return run;
357  }
358
359  private path(session: string) {
360    return `${this.dir}/mod-sent/${encodeURIComponent(session)}.json`;
361  }
362
363  /**
364   * Rejects when the file cannot be read, so that no write replaces lines it could not see. One
365   * that does not parse is replaced: holding every answer back on it would be worse.
366   */
367  private async load(session: string): Promise<Line[]> {
368    this.sessions.add(session);
369    const text = await this.host.read(this.path(session));
370    let lines: Line[] = [];
371    try {
372      const parsed = JSON.parse(text ?? "[]");
373      if (Array.isArray(parsed)) lines = parsed.filter(isLine);
374    } catch {}
375    const now = await this.host.now();
376    return lines.filter((l) => now - l.at < MAX_AGE_MS);
377  }
378
379  /** `load`, or undefined with a log line when the file cannot be read. */
380  private async loaded(session: string): Promise<Line[] | undefined> {
381    try {
382      return await this.load(session);
383    } catch (e) {
384      this.host.log(`starbridge: could not read the answers sent: ${(e as Error).message}`);
385      return undefined;
386    }
387  }
388
389  /** Writes `session`'s lines; rejects when the write fails. */
390  private async save(session: string, lines: Line[]): Promise<void> {
391    await this.host.write(this.path(session), JSON.stringify(lines));
392  }
393}
394
395function isLine(l: unknown): l is Line {
396  const x = l as Partial<Line> | null;
397  return (
398    typeof x?.ack === "string" &&
399    typeof x.line === "string" &&
400    typeof x.process === "string" &&
401    typeof x.at === "number" &&
402    typeof x.refused === "number" &&
403    (x.agent === undefined || typeof x.agent === "string")
404  );
405}
406
407/**
408 * Whether a live Claude Code process other than `own` runs session `session`, from the
409 * `<pid>.json` files Claude Code keeps in `<its config folder>/sessions`; files of processes that
410 * ended stay behind, so `runs` checks each pid.
411 */
412export async function sessionElsewhere(
413  session: string,
414  own: number,
415  fs: { list(dir: string): Promise<{ name: string }[]>; read(path: string): Promise<string> },
416  dir: string,
417  runs: (pid: number) => Promise<boolean>,
418): Promise<boolean> {
419  for (const f of await fs.list(dir).catch(() => [])) {
420    if (!/^\d+\.json$/.test(f.name) || f.name === `${own}.json`) continue;
421    try {
422      const s = firstObject(await fs.read(`${dir}/${f.name}`));
423      if (
424        s?.sessionId === session &&
425        typeof s.pid === "number" &&
426        s.pid !== own &&
427        (await runs(s.pid))
428      )
429        return true;
430    } catch {}
431  }
432  return false;
433}
434
435/**
436 * The object a session record starts with. Claude Code rewrites the file without truncating it,
437 * so a shorter record leaves the end of the previous one after it, as cli/src/claude.ts reads it.
438 */
439function firstObject(text: string): { pid?: unknown; sessionId?: unknown } | undefined {
440  for (let end = text.indexOf("}"); end !== -1; end = text.indexOf("}", end + 1)) {
441    try {
442      const value = JSON.parse(text.slice(0, end + 1));
443      return value && typeof value === "object" ? value : undefined;
444    } catch {}
445  }
446  return undefined;
447}
448

1import { joinLines, type Line } from "./agent.ts";
2
3/**
4 * Answers to the questions a Claude Code subagent asked (#1702). A subagent's Bash runs with its
5 * parent's session id and no id of its own, so the mod tags its `starbridge ask` with the
6 * subagent's id, which the CLI keeps with the decision, and sends the answer to that subagent
7 * rather than submitting it as the parent's prompt. A subagent that runs gets it at its next tool
8 * call; one that finished is resumed with it. One that cannot take it leaves it to the parent.
9 */
10
11/** The variable the CLI reads the asking subagent's id from. */
12export const AGENT_VAR = "STARBRIDGE_CLAUDE_AGENT";
13
14/** Characters a Claude Code agent id or teammate address takes, as the CLI checks them. */
15const AGENT_ID = /^[A-Za-z0-9_.@-]{1,200}$/;
16
17/** A command that runs `starbridge ask`, by name or path, anywhere in it. */
18const ASKS = /(?:^|[\s;&|(`'"/])starbridge(?:\.exe|\.cmd|\.js)?['"]?\s+ask(?=$|\s)/;
19
20const TAG = new RegExp(`^export ${AGENT_VAR}=([A-Za-z0-9_.@-]{1,200})\\n`);
21
22/**
23 * The Bash command subagent `agentId` runs, with the subagent's id exported first when it runs
24 * `starbridge ask`; any other command, or an id the CLI would refuse, unchanged.
25 */
26export function tagAsk(command: string, agentId: string): string {
27  if (!AGENT_ID.test(agentId) || !ASKS.test(command) || TAG.test(command)) return command;
28  return `export ${AGENT_VAR}=${agentId}\n${command}`;
29}
30
31/**
32 * The command as the subagent wrote it: the permission check judges that, so the tag never makes
33 * a command the rules allow ask anyone.
34 */
35export function untagAsk(command: string): string {
36  return command.replace(TAG, "");
37}
38
39/** One message for subagent `agentId`: the lines, and the text `Line.text` joins them into. */
40export type Send = (
41  agentId: string,
42  lines: Line[],
43  text: string,
44) => Promise<{ isDelivered: boolean; reason?: string }>;
45
46/** What the parent gets for a line its subagent `agent` could not take. */
47export function forParent(l: Line, agent: string): Line {
48  return {
49    text: `${l.text}\nYour subagent ${agent} asked this and could not take the answer: act on it, or pass it on.`,
50    ack: l.ack,
51  };
52}
53
54/**
55 * Sends each subagent its answers, one message per subagent in the order they came. Returns the
56 * acks it delivered, and the lines left for the parent: those no subagent asked, then those a
57 * subagent could not take, which say so. A line in `failed`, which a send already failed, goes to
58 * the parent without another try: retried while the parent's turn holds it, it could reach both.
59 */
60export async function toSubagents(
61  lines: Line[],
62  send: Send,
63  log: (text: string) => void,
64  failed: Set<string> = new Set(),
65): Promise<{ taken: string[]; rest: Line[] }> {
66  const byAgent = new Map<string, Line[]>();
67  const rest: Line[] = [];
68  for (const l of lines) {
69    if (l.subagent === undefined) rest.push(l);
70    else if (failed.has(l.ack)) rest.push(forParent(l, l.subagent));
71    else byAgent.set(l.subagent, [...(byAgent.get(l.subagent) ?? []), l]);
72  }
73  const taken: string[] = [];
74  for (const [agent, group] of byAgent) {
75    let why: string | undefined;
76    try {
77      const r = await send(agent, group, joinLines(group));
78      if (r.isDelivered) {
79        taken.push(...group.map((l) => l.ack));
80        continue;
81      }
82      why = r.reason ?? "not delivered";
83    } catch (e) {
84      why = (e as Error).message;
85    }
86    log(
87      `starbridge: subagent ${agent} could not take ${group.map((l) => l.ack).join(", ")}: ${why}`,
88    );
89    for (const l of group) {
90      failed.add(l.ack);
91      rest.push(forParent(l, agent));
92    }
93  }
94  return { taken, rest };
95}
96

1/**
2 * Picks the session's answer loop: `AgentLoop` (agent.ts) while the machine's agent answers on
3 * its socket, else `Poller` (poller.ts), which runs the CLI as before the agent existed. It
4 * checks for the agent every `probeMs` while the CLI path runs, and goes back to the CLI path
5 * when the agent stops answering.
6 */
7
8import type { AgentLoop } from "./agent.ts";
9import type { Poller } from "./poller.ts";
10
11/** Entry points of a run that ends with its last turn: `claude -p` and Agent SDK scripts. */
12const ONE_SHOT = new Set(["sdk-cli", "sdk-ts", "sdk-py"]);
13
14/**
15 * Whether a session gets an answer loop: one under the REPL, or one a host such as Claude
16 * desktop's Code tab or an IDE keeps open for a person (#863). The engine calls those not
17 * interactive, as it does a `claude -p` run, so the host's `CLAUDE_CODE_ENTRYPOINT` tells them
18 * apart; a run with none set is `claude -p`'s own.
19 */
20export function wantsLoop(isInteractive: boolean, entrypoint: string | undefined): boolean {
21  return isInteractive || (!!entrypoint && !ONE_SHOT.has(entrypoint));
22}
23
24export interface Paths {
25  /** True when an agent answers `GET /v1/status` with a 2xx. */
26  agentUp(): Promise<boolean>;
27  agent(unconfirmed: Set<string>): AgentLoop;
28  poller(unconfirmed: Set<string>): Poller;
29  sleep(ms: number): Promise<void>;
30  /** Clears the status line: each loop clears only the errors it showed itself. */
31  clearStatus(): void;
32  log(text: string): void;
33}
34
35export class Switch {
36  private stopped = false;
37  private agent: AgentLoop | undefined;
38  private poller: Poller | undefined;
39  /** The last agent loop, for its `bye` at the end of the session. */
40  private lastAgent: AgentLoop | undefined;
41  /** Resolves when the loop has ended. */
42  readonly done: Promise<void>;
43
44  constructor(
45    private readonly paths: Paths,
46    private readonly probeMs = 30_000,
47    /**
48     * Shared by both loops, so a line submitted before a switch is not submitted again after; and
49     * with the loop that replaces this one, which only confirms it then (#1337).
50     */
51    readonly unconfirmed = new Set<string>(),
52  ) {
53    // Nothing awaits a loop that ended before `stop`: a rejection would take the harness down (#1365).
54    this.done = this.loop().catch((e) => {
55      try {
56        this.paths.log(`starbridge: the answer loop stopped: ${(e as Error).message}`);
57      } catch {}
58    });
59  }
60
61  /** Which loop runs now, for tests and the debug log. */
62  get mode(): "agent" | "cli" | undefined {
63    return this.agent ? "agent" : this.poller ? "cli" : undefined;
64  }
65
66  /**
67   * Confirms through the loop that runs now the lines among `acks` still due to this session, and
68   * returns those; none when no loop runs or the call fails.
69   */
70  async claim(acks: string[]): Promise<string[]> {
71    try {
72      return (await (this.agent ?? this.poller)?.claim(acks)) ?? [];
73    } catch {
74      return [];
75    }
76  }
77
78  /** Ends the loop; the Poller gives up its lease at once. */
79  async stop(): Promise<void> {
80    this.stopped = true;
81    this.agent?.stop();
82    await this.poller?.stop();
83  }
84
85  /**
86   * Stops, and tells the agent this session ended. Waits at most 2 s for that: a hung agent
87   * must not hold up the session's exit, and a missed `bye` only leaves a stale entry in `status`.
88   */
89  async end(): Promise<void> {
90    await this.stop();
91    const bye = this.lastAgent?.bye();
92    if (bye) await Promise.race([bye, this.paths.sleep(2_000)]);
93  }
94
95  private async loop() {
96    while (!this.stopped) {
97      if (await this.up()) {
98        this.paths.log("starbridge: answers through the agent");
99        this.paths.clearStatus();
100        const agent = this.paths.agent(this.unconfirmed);
101        this.agent = agent;
102        this.lastAgent = agent;
103        // `stop` may have come while the probe ran, before `this.agent` was set.
104        if (this.stopped) agent.stop();
105        await agent.run();
106        this.agent = undefined;
107        continue;
108      }
109      if (this.stopped) return;
110      this.paths.log("starbridge: no agent; answers through the CLI");
111      this.paths.clearStatus();
112      const poller = this.paths.poller(this.unconfirmed);
113      this.poller = poller;
114      while (!this.stopped) {
115        await this.paths.sleep(this.probeMs);
116        if (!this.stopped && (await this.up())) break;
117      }
118      // Lets the step in progress finish and confirm what it submitted.
119      await poller.stop();
120      await poller.done;
121      this.poller = undefined;
122    }
123  }
124
125  private async up(): Promise<boolean> {
126    try {
127      return await this.paths.agentUp();
128    } catch {
129      return false;
130    }
131  }
132}
133