privateer-mail

1// Loaded per invocation by privateer start. No model, account or tool policies.
2const PREFIX = "privateer-mail:";
3// A hosted session has nobody to type /privateer-mail resume after an error.
4const HOSTED_RETRY_MS = 60000;
5let config, session, key, timer, flight, pending, prepared, nudge, offer, typed, resumeAt;
6let active = false, busy = false, paused = false, seen = 0, generation = 0, statusText, nudged = "";
7
8// The captain's fleet commands, as a front door and the Pi session offer them.
9// Only the person at the keyboard runs them, as the captain, not as the ship.
10const CAPTAIN = {
11  stow: { description: "Stow this machine before an update: /stow, /stow status, /stow resume, /stow resume all",
12    argv: rest => ({ "": ["stow"], status: ["stow", "status"], resume: ["stow", "resume"],
13      "resume all": ["stow", "resume", "--all"] })[rest] },
14  yolo: { description: "Fleet tool approvals for every first mate: /yolo on, off or status",
15    argv: rest => ({ "": ["permissions", "show"], status: ["permissions", "show"],
16      on: ["permissions", "yolo"], off: ["permissions", "standard"] })[rest] },
17  afk: { description: "Away: hold the fleet's mail except emergencies, /afk [until 8am] [instructions]",
18    argv: rest => (rest === "" ? ["afk"] : ["afk", rest]) },
19  back: { description: "Back from away: a digest now, the liaison's summary next",
20    argv: rest => (rest === "" ? ["back"] : undefined) },
21  ships: { description: "The fleet's ships and what each handles",
22    argv: rest => (rest === "" ? ["ships"] : undefined) },
23};
24
25async function status($, text) {
26  if (text === statusText) return;
27  statusText = text;
28  await $.ui.status(text);
29}
30
31async function call($, command, args) {
32  const result = await $.process.run([config.cli, "--home", config.home, command, "--json", JSON.stringify(args)], {
33    timeoutMs: 3000, env: { PRIVATEER_ROLE: "agent", PRIVATEER_ACTOR: config.ship },
34  });
35  if (result.exitCode !== 0 || result.stdout.length > 65536) throw Error("mail probe failed");
36  return JSON.parse(result.stdout);
37}
38
39async function owned($, gen) {
40  const value = await call($, "ship.presence", { ship: config.ship });
41  return active && generation === gen && value.attended === true && value.lease?.holder === config.holder
42    && await $.session.id() === session;
43}
44
45async function healthy($) {
46  // Classic settings hooks stand down only while a loaded mod is healthy.
47  // No marker is set by the launcher: disabled/refused mods retain the old hooks.
48  await $.env.set("PRIVATEER_CLAUDE_MAIL_UNTIL", String(await $.clock.now() + 12000));
49}
50
51async function pause($) {
52  paused = true;
53  resumeAt = config?.hosted ? await $.clock.now() + HOSTED_RETRY_MS : undefined;
54}
55
56async function save($) {
57  await $.store.set(key, { through: seen, paused, retry: resumeAt !== undefined, busy, holder: config.holder,
58    pending: pending ? { text: pending.text, after: pending.after, through: pending.through } : undefined });
59}
60
61function receipt(snapshot) {
62  return { home: config.home, ship: config.ship, session, through: snapshot.through_seq };
63}
64
65function marker(snapshot) { return PREFIX + JSON.stringify(receipt(snapshot)); }
66
67function message(snapshot) {
68  const inbox = [config.cli, "--home", config.home, "mail.inbox", "--json",
69    JSON.stringify({ address: config.ship, after_seq: snapshot.after_seq, limit: 40 })];
70  return `${marker(snapshot)}\n${snapshot.count} new mailbox arrivals for ${config.ship}. `
71    + `Read bodies with this argv: ${JSON.stringify(inbox)}. Page as needed; reply with correlation IDs, `
72    + `and acknowledge only handled messages. This notification is not an acknowledgement. `
73    + `Existing charter and production deployment approval rules apply. Headers: ${JSON.stringify(snapshot.headers)}`
74    + (snapshot.attention ? ` ${snapshot.attention}` : "");
75}
76
77async function snapshot($, gen) {
78  if (!await owned($, gen)) {
79    if (active && gen === generation) {
80      await $.env.set("PRIVATEER_CLAUDE_MAIL_UNTIL", undefined);
81      await status($, "⚓ mail held by another session");
82    }
83    return;
84  }
85  const value = await call($, "mail.pending", { address: config.ship, after_seq: seen });
86  if (!active || gen !== generation) return;
87  if (value.address !== config.ship || ![value.count, value.actionable, value.through_seq,
88      value.after_seq, value.acknowledged_through].every(n => Number.isSafeInteger(n) && n >= 0)
89      || value.through_seq < value.after_seq || value.actionable > value.count
90      || !Array.isArray(value.headers) || value.headers.length > Math.min(6, value.count)) throw Error("invalid mailbox snapshot");
91  seen = Math.max(seen, value.acknowledged_through);
92  if (paused && resumeAt !== undefined && await $.clock.now() >= resumeAt) {
93    paused = false; resumeAt = undefined; await save($);
94  }
95  await healthy($);
96  await status($, paused ? `⚓ ${value.count} new mail • paused; /privateer-mail resume` : `⚓ ${value.count} new mail`);
97  return value;
98}
99
100async function deliver($, gen) {
101  const value = await snapshot($, gen);
102  if (!value || !value.actionable || busy || paused || pending || prepared || nudge || offer) return;
103  if ((await $.prompt.read()).text.length || !await owned($, gen) || busy || paused || pending || nudge || offer) return;
104  const item = { gen, text: message(value), after: value.after_seq, through: value.through_seq };
105  pending = item;
106  try {
107    await save($);
108    const result = await $.prompt.submit({ text: item.text });
109    if (result.drop && pending === item) {
110      pending = undefined; await pause($); await save($);
111      await status($, "⚓ native delivery declined; /privateer-mail resume");
112    }
113    // A queued prompt is not delivery evidence. turn.start records the receipt.
114  } catch (error) {
115    if (pending === item) pending = undefined;
116    await pause($);
117    try { await save($); } catch { /* No delivery watermark advanced. */ }
118    throw error;
119  }
120}
121
122// Runtime-owned scheduled work for a hosted session. Eligibility, budgets and
123// leases live in Rust; the mod reports idleness and starts one offered turn.
124async function work($, gen) {
125  if (!config.hosted || !active || gen !== generation) return;
126  const idle = !busy && !paused && !pending && !prepared && !nudge && !offer
127    && !(await $.prompt.read()).text.length;
128  let sample = null;
129  try {
130    const { context } = await $.session.usage();
131    if (Number.isFinite(context?.tokens) && context?.window > 0) {
132      sample = { tokens: Math.round(context.tokens), window: context.window };
133    }
134  } catch { /* the context reading is optional */ }
135  if (!await owned($, gen)) return;
136  const response = await call($, "work.poll", { ship: config.ship, holder: config.holder, idle, attention: sample });
137  const next = response?.offer;
138  if (!next) return;
139  const release = () => call($, "work.release", { ship: config.ship, holder: config.holder, offer_id: next.id });
140  if (gen !== generation || offer || !idle || busy || pending || nudge) { await release(); return; }
141  if (typeof next.id !== "string" || typeof next.task_id !== "string"
142      || typeof next.prompt !== "string" || next.prompt.length > 24000) throw Error("invalid work offer");
143  const item = { gen, id: next.id, task: next.task_id, text: next.prompt, started: false };
144  offer = item;
145  try {
146    const result = await $.prompt.submit({ text: item.text });
147    if (result.drop && offer === item) { offer = undefined; await release(); }
148  } catch (error) {
149    if (offer === item) { offer = undefined; await release(); }
150    throw error;
151  }
152}
153
154async function poll($) {
155  if (!active || flight) return;
156  const gen = generation;
157  const operation = deliver($, gen).then(() => work($, gen).catch(async () => {
158    if (active && gen === generation) await status($, "⚓ work delivery unavailable; durable offer retained");
159  })).catch(async () => {
160    if (active && gen === generation) await status($, "⚓ mail unavailable; inbox preserved; /privateer-mail resume");
161  });
162  flight = operation;
163  try { await operation; }
164  finally { if (flight === operation) flight = undefined; }
165}
166
167// Why this hosted first mate's turn began, for the ship's activity. Best
168// effort and never awaited: a refusal or a slow CLI changes no delivery.
169function woke($, trigger, item) {
170  if (!config?.hosted) return;
171  const args = { ship: config.ship, holder: config.holder, trigger };
172  if (Number.isSafeInteger(item?.after) && Number.isSafeInteger(item?.through) && item.after <= item.through) {
173    args.after_seq = item.after; args.through_seq = item.through;
174  }
175  void Promise.resolve().then(() => call($, "wake.hosted", args)).catch(() => {});
176}
177
178// The ship's attention: its live context after each turn and each compaction,
179// reported by the session that holds the lease. Returns the runtime's reply
180// (the summary, with compact: true when this turn crossed the compaction line).
181// Best effort.
182async function attention($, kind, tokens, extra = {}) {
183  try {
184    const gen = generation;
185    const { context } = await $.session.usage();
186    tokens = tokens ?? context?.tokens;
187    if (!active || !Number.isFinite(tokens) || !(context?.window > 0) || !await owned($, gen)) return;
188    return await call($, "attention.report", { ship: config.ship, kind, tokens: Math.round(tokens),
189      window: context.window, holder: config.holder, ...extra });
190  } catch { /* attention reports are best effort */ }
191}
192
193// A resumable hosted first mate compacts its own conversation when the runtime
194// says this settled turn crossed the line. The session API runs the same call
195// /compact makes, between turns: it starts no model turn, and it is not a prompt,
196// so this mod's prompt.submit hook never sees it. It also never reads or changes
197// the pane's input box, so a draft there does not block it. A declined compaction
198// (a {skip} result, or a refusal) leaves the crossing armed, so the runtime asks
199// again at the next settled turn; the compaction report disarms it.
200async function compact($, gen, reply) {
201  if (reply?.compact !== true || gen !== generation || !active || busy || paused
202      || pending || prepared || nudge || offer) return;
203  try {
204    if (!await owned($, gen)) return;
205    const result = await $.session.compact();
206    if (result?.skip) return;
207  } catch { /* best effort: a refused compaction is asked again at the next settled turn */ }
208}
209
210function followText(idle) {
211  const list = idle.map(t => `- ${t.id}: ${t.title}`).join("\n");
212  return `Before you stop: these tasks are marked running, but no job, crew or open request is moving them:\n${list}\n`
213    + "Continue each one now (start its crew, its check or its next step), or record why it waits: "
214    + "set it blocked with task.update, or pin what wakes it with note.pin "
215    + `{"ship":"${config.ship}","focus":"task:<id>","key":"waiting","value":"..."}. `
216    + "Do not end on a promise to do it later.";
217}
218
219// A turn can end on "I'll carry on with X", leaving X running with nothing on
220// it. After a completed turn, give the first mate one more prompt to continue
221// such tasks or record why they wait, once per distinct set, so it cannot loop.
222async function follow($) {
223  const gen = generation;
224  try {
225    if (!active || busy || paused || pending || prepared || nudge || offer || !await owned($, gen)) return;
226    const idle = (await call($, "ship.bearings", { ship: config.ship })).idle_tasks;
227    if (!Array.isArray(idle) || !idle.length) { nudged = ""; return; }
228    const set = idle.map(t => t.id).sort().join(" ");
229    if (set === nudged || gen !== generation || busy || pending || nudge
230        || (await $.prompt.read()).text.length) return;
231    nudged = set;
232    const item = { gen, text: followText(idle) };
233    nudge = item;
234    const result = await $.prompt.submit({ text: item.text });
235    if (result.drop && nudge === item) nudge = undefined;
236  } catch { nudge = undefined; }
237}
238
239async function captain($, name, args) {
240  const command = CAPTAIN[name];
241  const argv = command.argv(args.trim());
242  if (!argv) return { text: `Usage: ${command.description}` };
243  // An operator role, not the ship's agent identity: the captain is typing.
244  let result;
245  try {
246    result = await $.process.run([config.cli, "--home", config.home, ...argv], {
247      cwd: config.home.replace(/\/[^/]+$/, ""), timeoutMs: 30000, env: { PRIVATEER_ROLE: "operator" },
248    });
249  } catch (error) {
250    return { text: `/${name} failed: ${String(error?.message ?? error).slice(0, 2000)}` };
251  }
252  const text = String(result.exitCode === 0 ? result.stdout : (result.stderr || result.stdout)).trim();
253  return { text: text.slice(0, 256 * 1024) || `/${name} done.` };
254}
255
256async function stop($) {
257  active = false; generation++;
258  timer?.cancel(); timer = undefined;
259  pending = undefined; prepared = undefined; nudge = undefined; offer = undefined; typed = undefined;
260  await $.env.set("PRIVATEER_CLAUDE_MAIL_UNTIL", undefined);
261}
262
263async function recover($, e, next) {
264  await stop($);
265  await status($, "⚓ attended mail inactive; native hooks retained");
266  return next(e);
267}
268
269async function bind($, interactive) {
270  if (interactive === false) return;
271  if (!config) {
272    const role = await $.env.get("PRIVATEER_ROLE");
273    const harness = await $.env.get("PRIVATEER_ATTENDED_HARNESS");
274    const job = await $.env.get("PRIVATEER_JOB");
275    config = { home: await $.env.get("PRIVATEER_HOME"), cli: await $.env.get("PRIVATEER_CONTROL_CLI"),
276      ship: await $.env.get("PRIVATEER_ACTOR"), holder: await $.env.get("PRIVATEER_PRESENCE_HOLDER"),
277      hosted: await $.env.get("PRIVATEER_HOSTED") === "1" };
278    if (job || role !== "agent" || harness !== "claude" || !config.home?.startsWith("/")
279        || !config.cli?.startsWith("/") || !config.ship || !config.holder) { config = undefined; return; }
280  }
281  if (interactive === undefined && !(await $.session.surfaces()).length) return;
282  const id = await $.session.id();
283  if (active && session === id) { await healthy($); return; }
284  await stop($);
285  session = id;
286  key = PREFIX + JSON.stringify([config.home, config.ship, session]);
287  const stored = await $.store.get(key);
288  seen = Number.isSafeInteger(stored?.through) && stored.through >= 0 ? stored.through : 0;
289  paused = stored?.paused === true;
290  // A hosted pane restarted after an error pause (its Mac woke mid-reply) still
291  // has nobody to resume it: it retries a minute later, as after the error itself.
292  // A typed pause stays paused. Records from before `retry` count as error pauses.
293  resumeAt = paused && config.hosted && stored?.retry !== false ? await $.clock.now() + HOSTED_RETRY_MS : undefined;
294  busy = stored?.holder === config.holder && stored?.busy === true;
295  if (stored?.holder === config.holder && typeof stored?.pending?.text === "string"
296      && stored.pending.text.length <= 65536 && Number.isSafeInteger(stored.pending.through)
297      && stored.pending.through > seen) {
298    const after = Number.isSafeInteger(stored.pending.after) ? stored.pending.after : seen;
299    pending = { ...stored.pending, after, gen: generation };
300  }
301  // Recover accepted wake receipts even if the process died before store.set.
302  for (const row of (await $.session.messages()).slice(-128)) {
303    if (row.role !== "user") continue;
304    for (const line of row.text.slice(0, 8192).split("\n")) {
305      const at = line.indexOf(PREFIX);
306      if (at < 0) continue;
307      try {
308        const data = JSON.parse(line.slice(at + PREFIX.length));
309        if (data.home === config.home && data.ship === config.ship && data.session === session
310            && Number.isSafeInteger(data.through) && data.through >= 0) seen = Math.max(seen, data.through);
311      } catch { /* Not a receipt. Never interpret arbitrary conversation text. */ }
312    }
313  }
314  active = true;
315  if (!await owned($, generation)) { await stop($); return; }
316  await healthy($);
317  // Bound this plugin's persistent receipt index; mail remains in SQLite.
318  const keys = (await $.store.keys()).filter(k => k.startsWith(PREFIX) && k !== key);
319  for (const old of keys.slice(0, Math.max(0, keys.length - 127))) await $.store.delete(old);
320  await $.command.register({ name: "privateer-mail", description: "Pause, resume or inspect ship mail", argumentHint: "pause|resume|status", immediate: true });
321  for (const [name, command] of Object.entries(CAPTAIN)) await $.command.register({ name, description: command.description });
322  timer = $.clock.every(5000, async () => { await poll($); });
323}
324
325export function register(on) {
326  on("session.start", async ($, e, next) => {
327    await bind($, e.isInteractive);
328    const result = await next(e);
329    if (active) void poll($);
330    return result;
331  }).catch(($, e, next) => recover($, e, next));
332  on("classic.SessionStart", async ($, e, next) => {
333    await bind($);
334    return next(e);
335  }).catch(($, e, next) => recover($, e, next));
336  on("session.end", async ($, e, next) => { await stop($); return next(e); });
337  on("prompt.submit", async ($, e, next) => {
338    if (!config) return next(e);
339    await bind($);
340    if (e.origin?.kind === "plugin") {
341      const ours = [pending, nudge, offer].find(item => item?.gen === generation && e.text.includes(item.text));
342      if (e.origin.name === "privateer-mail" && (!active || paused || !ours
343          || !await owned($, generation))) return { drop: "Privateer mail wake superseded or paused; inbox preserved." };
344      return next(e);
345    }
346    if (!active) return next(e);
347    if (!["composer", "bridge"].includes(e.origin?.kind)) return next(e);
348    // An actual captain prompt permits recovery; never add mail to a queued
349    // mid-turn submission or change the person's text or draft.
350    paused = false; resumeAt = undefined;
351    await save($);
352    if (busy || e.turnId || flight || pending) return next(e);
353    const gen = generation;
354    // The captain's own prompt starts a turn even with no mail beside it.
355    typed = { gen, text: e.text };
356    const value = await snapshot($, gen);
357    if (!value?.count || !await owned($, gen)) return next(e);
358    const item = { gen, text: e.text, after: value.after_seq, through: value.through_seq };
359    prepared = item;
360    const result = await next({ ...e, context: [...(e.context ?? []), message(value)] });
361    if (result.drop && prepared === item) { prepared = undefined; paused = true; await save($); }
362    return result;
363  }).catch(($, e, next) => recover($, e, next));
364  on("turn.start", async ($, e, next) => {
365    if (active) {
366      busy = true;
367      const item = pending && e.text.includes(pending.text) ? pending : prepared?.text === e.text ? prepared : undefined;
368      if (item && item.gen === generation) {
369        seen = Math.max(seen, item.through);
370        woke($, item === prepared ? "captain" : "mail", item);
371        if (item === pending) pending = undefined;
372        if (item === prepared) prepared = undefined;
373      } else if (typed?.gen === generation && typed.text === e.text) woke($, "captain");
374      typed = undefined;
375      if (nudge && e.text.includes(nudge.text)) {
376        nudge = undefined;
377        woke($, "follow");
378      }
379      if (offer && !offer.started && e.text.includes(offer.text)) {
380        offer.started = true;
381        // Do not invent acceptance on failure; the runtime keeps the claim.
382        void call($, "work.started", { ship: config.ship, holder: config.holder, offer_id: offer.id }).catch(() => {});
383      }
384      await save($);
385    }
386    return next(e);
387  }).catch(($, e, next) => recover($, e, next));
388  on("turn.complete", async ($, e, next) => {
389    const main = active && !e.agentId;
390    const completed = main && !e.isAborted && e.reason === "answer";
391    if (main) {
392      busy = false;
393      if (!completed) await pause($);
394      await save($);
395      if (offer?.started) {
396        const done = offer;
397        offer = undefined;
398        // An ambiguous native stop is reconciled by the durable deadline.
399        void call($, "work.settled", { ship: config.ship, holder: config.holder, offer_id: done.id }).catch(() => {});
400      }
401    }
402    const result = await next(e);
403    // settled: the turn ended its answer with no tool pending, so the line may act.
404    if (main) {
405      const gen = generation;
406      void attention($, "turn", undefined, { settled: completed }).then(reply => compact($, gen, reply));
407    }
408    if (completed) void follow($);
409    return result;
410  }).catch(($, e, next) => recover($, e, next));
411  on("session.compact", async ($, e, next) => {
412    const result = await next(e);
413    if (active && !e.agentId && e.trigger !== "precompute" && !result?.skip && Number.isFinite(result?.tokensBefore)) {
414      void attention($, "compaction", result.tokensBefore);
415    }
416    return result;
417  });
418  on("command.run", async ($, e, next) => {
419    if (CAPTAIN[e.command]) {
420      if (!config) return next(e);
421      if (!["composer", "bridge"].includes(e.origin?.kind)) return { text: `/${e.command} runs only from the captain's keyboard.` };
422      return captain($, e.command, e.args);
423    }
424    if (e.command !== "privateer-mail") return next(e);
425    if (!active) return { text: "Privateer attended mail is inactive in this session." };
426    const arg = e.args.trim();
427    if (arg === "pause") { paused = true; resumeAt = undefined; }
428    else if (arg === "resume") { paused = false; resumeAt = undefined; }
429    await save($);
430    if (arg === "resume") void poll($);
431    await status($, paused ? "⚓ mail paused; /privateer-mail resume" : "⚓ mail enabled");
432    return { text: paused ? "Ship mail paused; messages remain in the inbox." : "Ship mail enabled." };
433  });
434}
435