wristcode/gateway/src/notify.ts

// Decides when to push. A thread notifies when its phase becomes one the user cares about
// (waiting for them, finished, failed), and again only after it moves on. A new approval on a
// thread already waiting counts as a change, so each request notifies once. Nothing notifies while
// a host loads its first snapshot, so a restart never replays old news.

import { threadPhase, type Phase } from "./awareness.ts";
import type { PendingRequest, PushCategory, PushPayload } from "./contract.ts";
import type { HostRow } from "./db.ts";
import type { T3Host, ThreadChange } from "./t3/host.ts";
import { trimText, type SummaryCache, type ThreadSummary } from "./views.ts";

export interface Push {
  title: string;
  subtitle?: string;
  body: string;
  category: PushCategory;
  /** True for a subagent's request, which watches can mute apart from their own threads'. */
  fromSubagent?: boolean;
  data: PushPayload;
  /** Replaces an earlier push about the same thing on the watch. */
  collapseId: string;
}

export type SendPush = (push: Push) => Promise<void>;

const NOTIFYING: ReadonlySet<Phase> = new Set(["waiting_for_approval", "waiting_for_input", "completed", "failed"]);
const BODY_CHARS = 180;
const SUMMARY_WAIT_MS = 3_000;
const TOKEN_WARNING_MS = 7 * 24 * 60 * 60_000;

export class Notifier {
  /** The last phase key seen per thread, `phase:requestId`. */
  private readonly keys = new Map<string, string>();
  /** Hosts already warned about, so each warning goes out once a day at most. */
  private readonly warned = new Map<string, number>();

  constructor(
    private readonly send: SendPush,
    private readonly cache: SummaryCache,
    private readonly log: (message: string) => void = () => {},
  ) {}

  onThreadChange = ({ host, previous, next, initial }: ThreadChange) => {
    const id = `${host.id}:${(next ?? previous)?.id}`;
    if (!next) {
      this.keys.delete(id);
      return;
    }
    const phase = threadPhase(next);
    const key = `${phase}:${next.pendingRuntimeRequest?.id ?? ""}`;
    const before = this.keys.get(id);
    this.keys.set(id, key);
    if (initial || phase === null || !NOTIFYING.has(phase) || key === before) return;
    void this.notify(host, next, phase).catch((error: unknown) =>
      this.log(`push for ${id} failed: ${error instanceof Error ? error.message : String(error)}`),
    );
  };

  /** Warns once when a host starts rejecting the gateway's token. */
  onHostStatus = (host: T3Host) => {
    if (host.status !== "auth_invalid") return;
    this.warnHost(host.id, host.label, `T3 rejected the gateway's token. Pair ${host.label} again.`);
  };

  /** Warns about tokens that expire within a week. Run it daily. */
  checkTokenExpiry(rows: HostRow[], now = Date.now()) {
    for (const row of rows) {
      if (!row.token_expires_at) continue;
      const left = Date.parse(row.token_expires_at) - now;
      if (left > TOKEN_WARNING_MS) continue;
      const days = Math.max(0, Math.ceil(left / (24 * 60 * 60_000)));
      this.warnHost(row.id, row.label, `The gateway's T3 token expires in ${days} day${days === 1 ? "" : "s"}. Pair ${row.label} again.`);
    }
  }

  private warnHost(hostId: string, label: string, body: string) {
    const last = this.warned.get(hostId) ?? 0;
    if (Date.now() - last < 24 * 60 * 60_000) return;
    this.warned.set(hostId, Date.now());
    void this.send({
      title: label,
      body,
      category: "host",
      data: { hostId },
      collapseId: `host:${hostId}`,
    }).catch(() => {});
  }

  private async notify(host: T3Host, thread: NonNullable<ThreadChange["next"]>, phase: Phase) {
    const summary = await Promise.race([
      this.cache.get(host, thread).catch(() => undefined),
      Bun.sleep(SUMMARY_WAIT_MS).then(() => undefined),
    ]);
    const project = host.projects.get(thread.projectId)?.title;
    const requestId = thread.pendingRuntimeRequest?.id;
    // A subagent's push is about the thread that delegated to it, which is the one the user knows.
    const parent = thread.delegatedFrom?.title;
    const text = body(phase, thread.lastError ?? null, summary);
    await this.send({
      title: parent ?? thread.title,
      subtitle: project ? `${host.label} · ${project}` : host.label,
      body: parent ? `Subagent ${thread.title}: ${text}` : text,
      category: category(phase),
      fromSubagent: parent !== undefined,
      data: { hostId: host.id, threadId: thread.id, ...(requestId ? { requestId } : {}), ...pushQuestion(phase, summary) },
      collapseId: `${host.id}:${thread.id}`.slice(0, 64),
    });
  }
}

/**
 * A lone single-choice question's options, so the notification can offer them as buttons. Labels
 * are trimmed and the list capped to keep the payload under APNs' 4 KB.
 */
function pushQuestion(phase: Phase, summary: ThreadSummary | undefined): Pick<PushPayload, "question"> {
  const questions = summary?.request?.questions ?? [];
  const question = questions[0];
  if (phase !== "waiting_for_input" || questions.length !== 1 || !question || question.multiSelect) return {};
  if (question.options.length === 0 && !question.allowCustomAnswer) return {};
  return {
    question: {
      id: question.id,
      options: question.options.slice(0, 4).map(({ label, value }) => ({ label: trimText(label, 40), value: value.slice(0, 200) })),
      allowCustomAnswer: question.allowCustomAnswer,
    },
  };
}

function category(phase: Phase): PushCategory {
  switch (phase) {
    case "waiting_for_approval":
      return "approval";
    case "waiting_for_input":
      return "input";
    case "failed":
      return "failed";
    default:
      return "finished";
  }
}

/**
 * An approval push's body: the files a file-change approval would edit, as "Wants to edit
 * compose.yml +9 −6" or "Wants to edit 3 files: a.ts, b.ts, c.ts", or else the command it wants to
 * run. The watch's cards word it the same way.
 */
export function approvalBody(request: PendingRequest | null): string {
  const files = request?.kind === "file-change" ? request.files : [];
  const [file] = files;
  if (files.length === 1 && file) {
    const counts = [
      file.additions === null ? null : `+${file.additions}`,
      file.deletions === null ? null : `\u2212${file.deletions}`,
    ].filter((count) => count !== null);
    return trimText(["Wants to edit", file.name, ...counts].join(" "), BODY_CHARS);
  }
  if (files.length > 1) {
    return trimText(`Wants to edit ${files.length} files: ${files.map((file) => file.name).join(", ")}`, BODY_CHARS);
  }
  const prompt = request?.prompt;
  return prompt ? `Wants to run: ${trimText(prompt, BODY_CHARS)}` : "Needs your approval.";
}

function body(phase: Phase, lastError: string | null, summary: ThreadSummary | undefined): string {
  switch (phase) {
    case "waiting_for_approval":
      return approvalBody(summary?.request ?? null);
    case "waiting_for_input": {
      const question = summary?.request?.questions[0]?.question;
      return question ? trimText(question, BODY_CHARS) : "Has a question for you.";
    }
    case "failed":
      return lastError ? trimText(lastError, BODY_CHARS) : "The run failed.";
    default:
      return summary?.lastAssistantMessage ? trimText(summary.lastAssistantMessage, BODY_CHARS) : "Finished.";
  }
}