postmaster

1import type { EngineInterface, Register, SessionSendArgs } from "claude-code";
2import { startOutbox, reportOutboxError } from "./outbox";
3import { startReplies } from "./replies";
4import type { Outbox, PostmasterCalls, Replies } from "./protocol";
5
6type ModuleTimers = {
7  setInterval(fn: () => void, ms: number): unknown;
8  clearInterval(handle: unknown): void;
9};
10type WorkerServices = { outbox: Outbox; replies: Replies };
11type Worker = { status: "waiting" } | { status: "started"; calls: PostmasterCalls; services: Promise<WorkerServices> };
12
13function isModuleTimers(value: unknown): value is ModuleTimers {
14  return typeof value === "object" && value !== null
15    && "setInterval" in value && typeof value.setInterval === "function"
16    && "clearInterval" in value && typeof value.clearInterval === "function";
17}
18
19function callsFor($: EngineInterface): PostmasterCalls {
20  return {
21    env: { get: (name: string) => {
22      if (name === "HOME") return $.env.get("HOME");
23      if (name === "XDG_STATE_HOME") return $.env.get("XDG_STATE_HOME");
24      return Promise.reject(new Error(`Unsupported environment name: ${name}`));
25    } },
26    fs: {
27      read: (path: string) => $.fs.read(path),
28      write: (path: string, text: string) => $.fs.write(path, text),
29      exists: (path: string) => $.fs.exists(path),
30      list: (path?: string) => $.fs.list(path),
31    },
32    session: {
33      id: () => $.session.id(),
34      send: (message: SessionSendArgs) => $.session.send(message),
35    },
36    now: Date.now,
37    every: (ms, fn) => {
38      const timers: unknown = globalThis;
39      if (!isModuleTimers(timers)) throw new Error("Module interval timers are unavailable");
40      const interval = timers.setInterval(fn, ms);
41      return { cancel: () => timers.clearInterval(interval) };
42    },
43  };
44}
45
46export const register: Register = (on) => {
47  let worker: Worker = { status: "waiting" };
48  const startWorker = (calls: PostmasterCalls): Promise<WorkerServices> => {
49    if (worker.status === "started") return worker.services;
50    const services = startReplies(calls).then(replies => startOutbox(calls, replies).then(outbox => ({ outbox, replies })));
51    const active: Extract<Worker, { status: "started" }> = {
52      status: "started", calls,
53      services: services.catch(async error => {
54        await reportOutboxError(calls, error);
55        if (worker === active) worker = { status: "waiting" };
56        throw error;
57      }),
58    };
59    worker = active;
60    return active.services;
61  };
62
63  on("session.start", async ($, e, next) => {
64    await startWorker(callsFor($)).catch(() => {});
65    return next(e);
66  }).catch(async ($, e, next) => {
67    await reportOutboxError(callsFor($), e);
68    return next(e);
69  });
70
71  on("session.receive", async ($, e, next) => {
72    try {
73      const services = await startWorker(callsFor($));
74      const reply = await services.replies.receive(e);
75      if (reply.kind === "consumed") return { consumed: reply.reason };
76    } catch { /* A failed start or reply handler leaves the message for the session. */ }
77    return next(e);
78  }).catch(async ($, e, next) => {
79    await reportOutboxError(callsFor($), e);
80    return next(e);
81  });
82};
83

1import { BATCH, PATHS, POLL_MS, VERSION, isEntryId, isOutboxClaim, isOutboxEntry } from "./protocol";
2import type { Heartbeat, Outbox, OutboxClaim, OutboxEntry, OutboxResult, PostmasterCalls, Replies, RouteToken } from "./protocol";
3
4type EntryRead = { status: "valid"; entry: OutboxEntry } | { status: "unreadable" };
5type PendingResult = { path: string; result: OutboxResult };
6type RouteWithdrawal = { kind: "awaiting durable write"; token: RouteToken } | { kind: "durable" };
7
8let errorCount = 0;
9
10function errorText(error: unknown): string {
11  if (typeof error === "object" && error !== null && "message" in error && typeof error.message === "string") return error.message;
12  return error instanceof Error ? error.message : String(error);
13}
14
15export async function stateDirectory(calls: PostmasterCalls): Promise<string> {
16  const xdg = await calls.env.get("XDG_STATE_HOME");
17  const root = typeof xdg === "string" && xdg !== "" ? xdg : `${await calls.env.get("HOME")}/.local/state`;
18  if (!root.startsWith("/")) throw new Error("No absolute home or XDG_STATE_HOME is available");
19  return `${root.replace(/\/$/, "")}/message/postmaster`;
20}
21
22async function writeError(calls: PostmasterCalls, directory: string, error: unknown): Promise<void> {
23  const at_ms = calls.now();
24  const n = ++errorCount;
25  await calls.fs.write(`${directory}/${PATHS.errors}/${at_ms}-${n}.json`, JSON.stringify({ at_ms, error: errorText(error) })).catch(() => {});
26}
27
28export async function reportOutboxError(calls: PostmasterCalls, error: unknown): Promise<void> {
29  try { await writeError(calls, await stateDirectory(calls), error); }
30  catch { /* Without a state directory there is no permitted error path. */ }
31}
32
33async function textHash(text: string): Promise<string> {
34  const bytes = new TextEncoder().encode(text);
35  const digest = await crypto.subtle.digest("SHA-256", bytes);
36  return Array.from(new Uint8Array(digest), byte => byte.toString(16).padStart(2, "0")).join("");
37}
38
39export async function startOutbox(calls: PostmasterCalls, replies: Replies): Promise<Outbox> {
40  const directory = await stateDirectory(calls);
41  const session_id = await calls.session.id();
42  if (session_id === "") throw new Error("No session id is available");
43  const run = `${calls.now()}-${crypto.randomUUID()}`;
44  const pending = new Map<string, PendingResult>();
45  const pendingWithdrawals = new Map<string, RouteToken>();
46  let polling = false;
47  let cancelled = false;
48
49  const path = (folder: string, id: string) => `${directory}/${folder}/${id}.json`;
50  const heartbeat = async (): Promise<void> => {
51    const value: Heartbeat = { session_id, at_ms: calls.now(), version: VERSION, run };
52    await calls.fs.write(`${directory}/${PATHS.heartbeat}`, JSON.stringify(value));
53  };
54  const readEntry = async (id: string): Promise<EntryRead> => {
55    try {
56      const value: unknown = JSON.parse(await calls.fs.read(path(PATHS.outbox, id)));
57      return isOutboxEntry(value) && value.id === id ? { status: "valid", entry: value } : { status: "unreadable" };
58    } catch { return { status: "unreadable" }; }
59  };
60  const flush = async (): Promise<void> => {
61    for (const [id, item] of pending) {
62      try {
63        await calls.fs.write(item.path, JSON.stringify(item.result));
64        pending.delete(id);
65      } catch (error) { await writeError(calls, directory, error); }
66    }
67  };
68  const withdraw = async (token: RouteToken): Promise<RouteWithdrawal> => {
69    try {
70      await replies.notDelivered(token);
71      return { kind: "durable" };
72    } catch (error) {
73      await writeError(calls, directory, error);
74      return { kind: "awaiting durable write", token };
75    }
76  };
77  const flushWithdrawals = async (): Promise<void> => {
78    for (const [key, token] of pendingWithdrawals) {
79      const withdrawal = await withdraw(token);
80      if (withdrawal.kind === "durable") pendingWithdrawals.delete(key);
81    }
82  };
83  const result = async (id: string, claimed_ms: number, sha256: string, outcome: OutboxResult["outcome"], detail: string): Promise<void> => {
84    pending.set(id, { path: path(PATHS.results, id), result: { id, outcome, detail, sha256, claimed_ms, finished_ms: calls.now() } });
85    await flush();
86  };
87  const deliver = async (id: string): Promise<void> => {
88    if (pending.has(id) || await calls.fs.exists(path(PATHS.results, id)) || await calls.fs.exists(path(PATHS.claimed, id))) return;
89    const read = await readEntry(id);
90    const claimed_ms = calls.now();
91    const claim: OutboxClaim = { id, claimed_ms, run };
92    let sha256 = read.status === "valid" ? read.entry.sha256 : "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
93    let sendAttempted = false;
94    await calls.fs.write(path(PATHS.claimed, id), JSON.stringify(claim));
95    try {
96      const present = await calls.fs.exists(path(PATHS.outbox, id));
97      sha256 = await textHash(read.status === "valid" ? read.entry.text : "");
98      if (!present) return await result(id, claimed_ms, sha256, "withdrawn", "sender withdrew entry");
99      if (read.status === "unreadable") return await result(id, claimed_ms, sha256, "failed", "unreadable entry");
100      const entry = read.entry;
101      if (entry.deadline_ms < calls.now()) return await result(id, claimed_ms, sha256, "expired", "entry deadline passed");
102      const token = await replies.sending(entry, calls.now());
103      sendAttempted = true;
104      const answer = await calls.session.send({ to: { sessionId: entry.session_id }, text: entry.text });
105      if (!answer.isDelivered) {
106        const withdrawal = await withdraw(token);
107        if (withdrawal.kind === "awaiting durable write") pendingWithdrawals.set(token.key, withdrawal.token);
108      }
109      await result(id, claimed_ms, sha256, answer.isDelivered ? "delivered" : "failed", answer.isDelivered ? "" : answer.reason);
110    } catch (error) {
111      await writeError(calls, directory, error);
112      await result(id, claimed_ms, sha256, sendAttempted ? "abandoned" : "failed", errorText(error));
113    }
114  };
115  const oldClaim = async (id: string): Promise<void> => {
116    if (pending.has(id) || await calls.fs.exists(path(PATHS.results, id))) return;
117    const value: unknown = JSON.parse(await calls.fs.read(path(PATHS.claimed, id)));
118    if (!isOutboxClaim(value) || value.id !== id) throw new Error(`Unreadable claim: ${id}`);
119    if (value.run === run) return;
120    const read = await readEntry(id);
121    const sha256 = await textHash(read.status === "valid" ? read.entry.text : "");
122    await result(id, value.claimed_ms, sha256, "abandoned", "earlier worker left a claim without a result");
123  };
124  const listIds = async (folder: string): Promise<string[]> => {
125    if (!await calls.fs.exists(`${directory}/${folder}`)) {
126      try { return await listedIds(folder); } catch { return []; }
127    }
128    return listedIds(folder);
129  };
130  const listedIds = async (folder: string): Promise<string[]> => {
131    const files = await calls.fs.list(`${directory}/${folder}`);
132    return files.filter(file => file.kind === "file" && !file.isLink && file.name.endsWith(".json") && isEntryId(file.name.slice(0, -5)))
133      .sort((a, b) => a.mtimeMs - b.mtimeMs).map(file => file.name.slice(0, -5));
134  };
135  const pollBody = async (): Promise<void> => {
136    if (polling || cancelled) return;
137    polling = true;
138    try {
139      await heartbeat();
140      await flushWithdrawals();
141      await flush();
142      for (const id of await listIds(PATHS.claimed)) {
143        await oldClaim(id).catch(error => writeError(calls, directory, error));
144      }
145      let handled = 0;
146      for (const id of await listIds(PATHS.outbox)) {
147        if (handled === BATCH) break;
148        await (async () => {
149          if (pending.has(id) || await calls.fs.exists(path(PATHS.claimed, id)) || await calls.fs.exists(path(PATHS.results, id))) return;
150          handled++;
151          await heartbeat();
152          await deliver(id);
153        })().catch(error => writeError(calls, directory, error));
154      }
155    } finally { polling = false; }
156  };
157  const poll = (): Promise<void> => pollBody().catch(error => writeError(calls, directory, error));
158  const timer = calls.every(POLL_MS, () => { void poll(); });
159  return { poll, cancel: () => { cancelled = true; timer.cancel(); } };
160}
161

1import type { SessionReceiveInput } from "claude-code";
2import { stateDirectory } from "./outbox";
3import { PATHS, REPLY_WINDOW_MS, isDeliveryRouteFile } from "./protocol";
4import type { DeliveryRoute, DeliveryRouteFile, PostmasterCalls, Replies, ReplyDestination, ReplyLog } from "./protocol";
5
6type ReplyWrapper = { kind: "reply"; socket: string; name: string; body: string } | { kind: "not a reply" };
7
8function wrapper(text: string): ReplyWrapper {
9  const lines = text.split("\n");
10  const first = lines[0];
11  const last = lines[lines.length - 1];
12  if (first === undefined || !/^<cross-session-message\s.*>\r?$/.test(first)
13    || last?.replace(/\r$/, "") !== "</cross-session-message>") return { kind: "not a reply" };
14  const socket = /\sfrom="([^"]+)"/.exec(first)?.[1];
15  const name = /\sfrom-name="([^"]+)"/.exec(first)?.[1];
16  if (socket === undefined || !socket.startsWith("uds:/") || name === undefined) return { kind: "not a reply" };
17  return { kind: "reply", socket, name, body: lines.slice(1, -1).join("\n") };
18}
19
20function destination(sessionId: string): ReplyDestination {
21  return sessionId === "" ? { kind: "unroutable" } : { kind: "session", sessionId };
22}
23
24function routeFile(route: DeliveryRoute, status: DeliveryRouteFile["route"]): DeliveryRouteFile {
25  return {
26    route: status, socket: route.socket, sender: route.sender,
27    reply_to_session_id: route.destination.kind === "session" ? route.destination.sessionId : "",
28    session_id: route.sessionId, at_ms: route.atMs,
29  };
30}
31
32export async function startReplies(calls: PostmasterCalls): Promise<Replies> {
33  const directory = await stateDirectory(calls);
34  const routes = new Map<string, Map<string, DeliveryRoute>>();
35  const pendingWithdrawals = new Map<string, DeliveryRoute>();
36  let routeCount = 0;
37  let replyCount = 0;
38  let forwardCount = 0;
39  const path = (folder: string, key: string) => `${directory}/${folder}/${key}.json`;
40  const inWindow = (atMs: number, now: number) => atMs <= now && atMs >= now - REPLY_WINDOW_MS;
41  const remember = (route: DeliveryRoute): void => {
42    let socketRoutes = routes.get(route.socket);
43    if (socketRoutes === undefined) {
44      socketRoutes = new Map();
45      routes.set(route.socket, socketRoutes);
46    }
47    socketRoutes.set(route.key, route);
48  };
49  const expire = (now: number): void => {
50    for (const [socket, socketRoutes] of routes) {
51      for (const [key, route] of socketRoutes) if (!inWindow(route.atMs, now)) socketRoutes.delete(key);
52      if (socketRoutes.size === 0) routes.delete(socket);
53    }
54  };
55  const delivered = `${directory}/${PATHS.delivered}`;
56  const present = await calls.fs.exists(delivered);
57  const files = await calls.fs.list(delivered).catch(error => {
58    if (present) throw error;
59    return [];
60  });
61  const now = calls.now();
62  for (const file of files) {
63    if (file.kind !== "file" || file.isLink || !/^\d+-\d+\.json$/.test(file.name)) continue;
64    const key = file.name.slice(0, -5);
65    const count = Number(key.slice(key.lastIndexOf("-") + 1));
66    if (Number.isSafeInteger(count)) routeCount = Math.max(routeCount, count);
67    try {
68      const value: unknown = JSON.parse(await calls.fs.read(path(PATHS.delivered, key)));
69      if (!isDeliveryRouteFile(value) || value.route !== "open" || !inWindow(value.at_ms, now)) continue;
70      remember({ key, socket: value.socket, sender: value.sender, destination: destination(value.reply_to_session_id), sessionId: value.session_id, atMs: value.at_ms });
71    } catch { /* An unreadable route cannot identify a reply. */ }
72  }
73
74  return {
75    sending: async (entry, atMs) => {
76      expire(atMs);
77      const route: DeliveryRoute = {
78        key: `${atMs}-${++routeCount}`, socket: entry.socket, sender: entry.sender,
79        destination: destination(entry.reply_to_session_id), sessionId: entry.session_id, atMs,
80      };
81      await calls.fs.write(path(PATHS.delivered, route.key), JSON.stringify(routeFile(route, "open")));
82      remember(route);
83      return { key: route.key };
84    },
85    notDelivered: async token => {
86      let route = pendingWithdrawals.get(token.key);
87      if (route === undefined) {
88        for (const [socket, socketRoutes] of routes) {
89          route = socketRoutes.get(token.key);
90          if (route === undefined) continue;
91          socketRoutes.delete(token.key);
92          if (socketRoutes.size === 0) routes.delete(socket);
93          pendingWithdrawals.set(token.key, route);
94          break;
95        }
96      }
97      if (route === undefined) return;
98      await calls.fs.write(path(PATHS.delivered, token.key), JSON.stringify(routeFile(route, "withdrawn")));
99      pendingWithdrawals.delete(token.key);
100    },
101    receive: async (e: SessionReceiveInput) => {
102      if (e.origin.kind !== "peer") return { kind: "pass" };
103      const reply = wrapper(e.text);
104      if (reply.kind === "not a reply") return { kind: "pass" };
105      const at_ms = calls.now();
106      expire(at_ms);
107      const socketRoutes = routes.get(reply.socket);
108      if (socketRoutes === undefined) return { kind: "pass" };
109      const matching = [...socketRoutes.values()];
110      const senders = [...new Set(matching.map(route => route.sender))];
111      const targets = new Set<string>();
112      for (const route of matching) if (route.destination.kind === "session") targets.add(route.destination.sessionId);
113      const answerTo = senders.length === 1 ? senders.join(", ") : `one of ${senders.join(", ")}`;
114      const key = `${at_ms}-${++replyCount}`;
115      const forwarded: string[] = [];
116      for (const sessionId of targets) {
117        const text = `Message from ${reply.name} (their answer to ${answerTo}, passed on by postmaster) [${at_ms}-${++forwardCount}]:\n\n${reply.body}`;
118        try {
119          const answer = await calls.session.send({ to: { sessionId }, text });
120          if (answer.isDelivered) forwarded.push(sessionId);
121        } catch { /* The answer is still logged and consumed when forwarding fails. */ }
122      }
123      const replying = matching[matching.length - 1];
124      if (replying === undefined) return { kind: "pass" };
125      const log: ReplyLog = { at_ms, socket: reply.socket, name: reply.name, session_id: replying.sessionId, senders, body: reply.body, forwarded };
126      await calls.fs.write(path(PATHS.replies, key), JSON.stringify(log));
127      return { kind: "consumed", reason: "postmaster reply" };
128    },
129  };
130}
131

1import type { EngineInterface, SessionReceiveInput } from "claude-code";
2
3export const POLL_MS = 500;
4export const BATCH = 20;
5export const VERSION = "0.1.0";
6export const REPLY_WINDOW_MS = 30 * 60 * 1000;
7export const PATHS = { outbox: "outbox", claimed: "claimed", results: "results", heartbeat: "heartbeat.json", errors: "errors", delivered: "delivered", replies: "replies" } as const;
8
9export type OutboxEntry = {
10  id: string;
11  session_id: string;
12  socket: string;
13  text: string;
14  sha256: string;
15  deadline_ms: number;
16  sender: string;
17  reply_to_session_id: string;
18};
19export type OutboxClaim = { id: string; claimed_ms: number; run: string };
20export type OutboxResult = {
21  id: string;
22  outcome: "delivered" | "failed" | "expired" | "withdrawn" | "abandoned";
23  detail: string;
24  sha256: string;
25  claimed_ms: number;
26  finished_ms: number;
27};
28export type Heartbeat = { session_id: string; at_ms: number; version: string; run: string };
29export type PostmasterCalls = {
30  env: Pick<EngineInterface["env"], "get">;
31  fs: Pick<EngineInterface["fs"], "write" | "exists" | "list"> & {
32    read(path: string): Promise<string>;
33  };
34  session: Pick<EngineInterface["session"], "id" | "send">;
35  now(): number;
36  every(ms: number, fn: () => void): { cancel(): void };
37};
38export type Outbox = { poll(): Promise<void>; cancel(): void };
39export type ReplyDestination = { kind: "session"; sessionId: string } | { kind: "unroutable" };
40export type DeliveryRoute = {
41  key: string;
42  socket: string;
43  sender: string;
44  destination: ReplyDestination;
45  sessionId: string;
46  atMs: number;
47};
48export type RouteToken = { key: string };
49export type ReplyDisposition = { kind: "consumed"; reason: "postmaster reply" } | { kind: "pass" };
50export type DeliveryRouteFile = {
51  route: "open" | "withdrawn";
52  socket: string;
53  sender: string;
54  reply_to_session_id: string;
55  session_id: string;
56  at_ms: number;
57};
58export type ReplyLog = {
59  at_ms: number;
60  socket: string;
61  name: string;
62  session_id: string;
63  senders: string[];
64  body: string;
65  forwarded: string[];
66};
67export type Replies = {
68  sending(entry: OutboxEntry, atMs: number): Promise<RouteToken>;
69  notDelivered(token: RouteToken): Promise<void>;
70  receive(e: SessionReceiveInput): Promise<ReplyDisposition>;
71};
72
73function record(value: unknown): value is Record<string, unknown> {
74  return typeof value === "object" && value !== null && !Array.isArray(value);
75}
76function timestamp(value: unknown): value is number {
77  return typeof value === "number" && Number.isSafeInteger(value) && value >= 0;
78}
79export function isEntryId(value: unknown): value is string {
80  return typeof value === "string" && /^[a-f0-9]{32}$/i.test(value);
81}
82function hash(value: unknown): value is string {
83  return typeof value === "string" && /^[a-f0-9]{64}$/.test(value);
84}
85export function isOutboxEntry(value: unknown): value is OutboxEntry {
86  return record(value) && isEntryId(value.id) && typeof value.session_id === "string" && value.session_id !== ""
87    && typeof value.socket === "string" && typeof value.text === "string" && hash(value.sha256) && timestamp(value.deadline_ms)
88    && typeof value.sender === "string" && value.sender !== "" && typeof value.reply_to_session_id === "string";
89}
90export function isOutboxClaim(value: unknown): value is OutboxClaim {
91  return record(value) && isEntryId(value.id) && timestamp(value.claimed_ms) && typeof value.run === "string" && value.run !== "";
92}
93export function isOutboxResult(value: unknown): value is OutboxResult {
94  return record(value) && isEntryId(value.id) && typeof value.outcome === "string"
95    && ["delivered", "failed", "expired", "withdrawn", "abandoned"].includes(value.outcome)
96    && typeof value.detail === "string" && hash(value.sha256) && timestamp(value.claimed_ms) && timestamp(value.finished_ms);
97}
98export function isHeartbeat(value: unknown): value is Heartbeat {
99  return record(value) && typeof value.session_id === "string" && value.session_id !== ""
100    && timestamp(value.at_ms) && typeof value.version === "string" && value.version !== ""
101    && typeof value.run === "string" && value.run !== "";
102}
103export function isDeliveryRouteFile(value: unknown): value is DeliveryRouteFile {
104  return record(value) && (value.route === "open" || value.route === "withdrawn")
105    && typeof value.socket === "string" && value.socket.startsWith("uds:/")
106    && typeof value.sender === "string" && value.sender !== ""
107    && typeof value.reply_to_session_id === "string"
108    && typeof value.session_id === "string" && value.session_id !== "" && timestamp(value.at_ms);
109}
110export function isReplyLog(value: unknown): value is ReplyLog {
111  return record(value) && timestamp(value.at_ms)
112    && typeof value.socket === "string" && value.socket.startsWith("uds:/")
113    && typeof value.name === "string" && value.name !== ""
114    && typeof value.session_id === "string" && value.session_id !== ""
115    && Array.isArray(value.senders) && value.senders.length > 0
116    && value.senders.every((sender: unknown) => typeof sender === "string" && sender !== "") && typeof value.body === "string"
117    && Array.isArray(value.forwarded) && value.forwarded.every((id: unknown) => typeof id === "string" && id !== "");
118}
119