Gateway: review fixes for stale summaries, history cursors that survive a moving window, timeouts on T3 calls, push size, T3's snooze rules, and input checks
Gateway / check (push) Successful in 9s
Gateway / image (push) Successful in 21s

This commit is contained in:
gib committed 2026-11-02 14:59:32 -05:00
1 parent edc4b88990
commit 99b300855e
13 files changed
+208 -58

No files matched your search

+4 -2
View File
@@ -103,7 +103,7 @@ directory), the search, the subagent's title or the tool's name, trimmed to abou
Monitors are left out, since they run for hours. It is null when no step is in progress. The gateway Monitors are left out, since they run for hours. It is null when no step is in progress. The gateway
rereads a running thread at most every 5 seconds, so it can lag that much. rereads a running thread at most every 5 seconds, so it can lag that much.
`changes` totals the files the thread's latest run changed. It is null when the run changed no `changes` totals the files the thread's latest run changed, counting only edits that ran. It is null when the run changed no
files, and, like `lastMessage`, until the gateway has read the thread. files, and, like `lastMessage`, until the gateway has read the thread.
``` ```
@@ -234,7 +234,9 @@ POST /v1/threads/:hostId/:threadId/unsnooze
``` ```
Snoozing hides a finished thread from Recent until `until`, which must be in the next 30 days. Then Snoozing hides a finished thread from Recent until `until`, which must be in the next 30 days. Then
it comes back on its own. A snoozed thread that starts working or waiting on you shows as usual. it comes back on its own. A snoozed thread that starts working or waiting on you shows as usual,
and one that fails after it was snoozed comes back to Recent, the same rules T3 uses. A snoozed
thread that finishes again does not push.
Thread detail fills `request` and `via` the same way as the feed when a subagent waits on you. Thread detail fills `request` and `via` the same way as the feed when a subagent waits on you.
+6 -2
View File
@@ -340,7 +340,7 @@ export function createRoutes({ db, hub, cache, pusher, startedAt, log = () => {}
GET: authed<"/v1/threads/:hostId/:threadId/history">(async (req) => { GET: authed<"/v1/threads/:hostId/:threadId/history">(async (req) => {
const { host, thread } = findThread(req.params.hostId, req.params.threadId); const { host, thread } = findThread(req.params.hostId, req.params.threadId);
const cursor = new URL(req.url).searchParams.get("cursor") ?? ""; const cursor = new URL(req.url).searchParams.get("cursor") ?? "";
if (!/^[wh]./.test(cursor)) throw new HttpError(400, "Send the earlierCursor from the thread."); if (!/^(w.{1,256}|h.{1,4096})$/.test(cursor)) throw new HttpError(400, "Send the earlierCursor from the thread.");
return json(await loadHistory(host, thread, cursor)); return json(await loadHistory(host, thread, cursor));
}), }),
}, },
@@ -462,12 +462,16 @@ const isAnswers = (value: unknown): value is Record<string, string | string[]> =
(answer) => typeof answer === "string" || (Array.isArray(answer) && answer.every((item) => typeof item === "string")), (answer) => typeof answer === "string" || (Array.isArray(answer) && answer.every((item) => typeof item === "string")),
); );
/** The request's JSON object. Anything else, such as `null` or an array, is a 400. */
async function readBody<T>(req: Request): Promise<Partial<T> & T> { async function readBody<T>(req: Request): Promise<Partial<T> & T> {
let body: unknown;
try { try {
return (await req.json()) as Partial<T> & T; body = await req.json();
} catch { } catch {
throw new HttpError(400, "The body must be JSON."); throw new HttpError(400, "The body must be JSON.");
} }
if (typeof body !== "object" || body === null || Array.isArray(body)) throw new HttpError(400, "The body must be a JSON object.");
return body as Partial<T> & T;
} }
/** A thread title from the first line of the prompt. T3 replaces it with a generated one. */ /** A thread title from the first line of the prompt. T3 replaces it with a generated one. */
+17
View File
@@ -75,6 +75,23 @@ export function resolvePhase(thread: AwarenessFields): Phase | null {
// Commands and monitors are as often a dev server or its watchdog as a wait, so they hold // Commands and monitors are as often a dev server or its watchdog as a wait, so they hold
// completion only when the agent marks them awaited. // completion only when the agent marks them awaited.
/**
* Whether a snooze still hides the thread, as T3's own `isSnoozed` decides: until its time, unless
* the thread waits on someone, or failed after it was snoozed.
*/
export function isSnoozed(
thread: Pick<ThreadShell, "snoozedUntil" | "snoozedAt" | "latestRunCompletedAt" | "status" | "pendingRuntimeRequest" | "pendingGuestMessageCount">,
now = Date.now(),
): boolean {
const until = thread.snoozedUntil ? Date.parse(thread.snoozedUntil) : NaN;
if (!(until > now)) return false;
if (thread.pendingRuntimeRequest !== null || (thread.pendingGuestMessageCount ?? 0) > 0) return false;
const snoozedAt = thread.snoozedAt ? Date.parse(thread.snoozedAt) : null;
const completedAt = thread.latestRunCompletedAt ? Date.parse(thread.latestRunCompletedAt) : null;
const wokeOnError = thread.status === "failed" && (snoozedAt === null || (completedAt !== null && completedAt > snoozedAt));
return !wokeOnError;
}
function backgroundWorkHoldsCompletion(tasks: NonNullable<ThreadShell["pendingBackgroundTasks"]>): boolean { function backgroundWorkHoldsCompletion(tasks: NonNullable<ThreadShell["pendingBackgroundTasks"]>): boolean {
return tasks.some((task) => task.awaited === true || task.kind === "subagent" || task.kind === "background_task"); return tasks.some((task) => task.awaited === true || task.kind === "subagent" || task.kind === "background_task");
} }
+9 -1
View File
@@ -29,7 +29,15 @@ if (!apns) log("APNs is not configured. Pushes will only be logged.");
const pusher = new Pusher(db, apns, log); const pusher = new Pusher(db, apns, log);
const notifier = new Notifier(pusher.sendAll, cache, log); const notifier = new Notifier(pusher.sendAll, cache, log);
const hub = new Hub(db, { onThreadChange: notifier.onThreadChange, onStatus: notifier.onHostStatus, log }); const hub = new Hub(db, {
onThreadChange: (change) => {
// A removed thread's summary would otherwise stay cached for the gateway's lifetime.
if (!change.next && change.previous) cache.forget(change.host, change.previous.id);
notifier.onThreadChange(change);
},
onStatus: notifier.onHostStatus,
log,
});
hub.sync(); hub.sync();
setInterval(() => hub.sync(), HOST_SYNC_MS); setInterval(() => hub.sync(), HOST_SYNC_MS);
notifier.checkTokenExpiry(hub.rows()); notifier.checkTokenExpiry(hub.rows());
+11 -4
View File
@@ -3,7 +3,7 @@
// thread already waiting counts as a change, so each request notifies once. Nothing notifies while // 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. // a host loads its first snapshot, so a restart never replays old news.
import { threadPhase, type Phase } from "./awareness.ts"; import { isSnoozed, threadPhase, type Phase } from "./awareness.ts";
import type { PendingRequest, PushCategory, PushPayload } from "./contract.ts"; import type { PendingRequest, PushCategory, PushPayload } from "./contract.ts";
import type { HostRow } from "./db.ts"; import type { HostRow } from "./db.ts";
import type { T3Host, ThreadChange } from "./t3/host.ts"; import type { T3Host, ThreadChange } from "./t3/host.ts";
@@ -25,6 +25,9 @@ export type SendPush = (push: Push) => Promise<void>;
const NOTIFYING: ReadonlySet<Phase> = new Set(["waiting_for_approval", "waiting_for_input", "completed", "failed"]); const NOTIFYING: ReadonlySet<Phase> = new Set(["waiting_for_approval", "waiting_for_input", "completed", "failed"]);
const BODY_CHARS = 180; const BODY_CHARS = 180;
const TITLE_CHARS = 80;
/** An option value longer than this can't be sent whole, so the question goes without buttons. */
const OPTION_VALUE_CHARS = 200;
const SUMMARY_WAIT_MS = 3_000; const SUMMARY_WAIT_MS = 3_000;
const TOKEN_WARNING_MS = 7 * 24 * 60 * 60_000; const TOKEN_WARNING_MS = 7 * 24 * 60 * 60_000;
@@ -51,6 +54,8 @@ export class Notifier {
const before = this.keys.get(id); const before = this.keys.get(id);
this.keys.set(id, key); this.keys.set(id, key);
if (initial || phase === null || !NOTIFYING.has(phase) || key === before) return; if (initial || phase === null || !NOTIFYING.has(phase) || key === before) return;
// T3 keeps a snoozed thread asleep through background wakes, so its finishing again is not news.
if (phase === "completed" && isSnoozed(next)) return;
void this.notify(host, next, phase).catch((error: unknown) => void this.notify(host, next, phase).catch((error: unknown) =>
this.log(`push for ${id} failed: ${error instanceof Error ? error.message : String(error)}`), this.log(`push for ${id} failed: ${error instanceof Error ? error.message : String(error)}`),
); );
@@ -97,9 +102,9 @@ export class Notifier {
const parent = thread.delegatedFrom?.title; const parent = thread.delegatedFrom?.title;
const text = body(phase, thread.lastError ?? null, summary); const text = body(phase, thread.lastError ?? null, summary);
await this.send({ await this.send({
title: parent ?? thread.title, title: trimText(parent ?? thread.title, TITLE_CHARS),
subtitle: project ? `${host.label} · ${project}` : host.label, subtitle: project ? `${host.label} · ${project}` : host.label,
body: parent ? `Subagent ${thread.title}: ${text}` : text, body: parent ? `Subagent ${trimText(thread.title, TITLE_CHARS)}: ${text}` : text,
category: category(phase), category: category(phase),
fromSubagent: parent !== undefined, fromSubagent: parent !== undefined,
data: { hostId: host.id, threadId: thread.id, ...(requestId ? { requestId } : {}), ...pushQuestion(phase, summary) }, data: { hostId: host.id, threadId: thread.id, ...(requestId ? { requestId } : {}), ...pushQuestion(phase, summary) },
@@ -117,10 +122,12 @@ function pushQuestion(phase: Phase, summary: ThreadSummary | undefined): Pick<Pu
const question = questions[0]; const question = questions[0];
if (phase !== "waiting_for_input" || questions.length !== 1 || !question || question.multiSelect) return {}; if (phase !== "waiting_for_input" || questions.length !== 1 || !question || question.multiSelect) return {};
if (question.options.length === 0 && !question.allowCustomAnswer) return {}; if (question.options.length === 0 && !question.allowCustomAnswer) return {};
// A cut value would answer with a string that matches no option.
if (question.options.some(({ value }) => value.length > OPTION_VALUE_CHARS)) return {};
return { return {
question: { question: {
id: question.id, id: question.id,
options: question.options.slice(0, 4).map(({ label, value }) => ({ label: trimText(label, 40), value: value.slice(0, 200) })), options: question.options.slice(0, 4).map(({ label, value }) => ({ label: trimText(label, 40), value })),
allowCustomAnswer: question.allowCustomAnswer, allowCustomAnswer: question.allowCustomAnswer,
}, },
}; };
+28 -9
View File
@@ -6,6 +6,7 @@ import type { Apns } from "./apns.ts";
import type { NotificationPrefs, PushResult } from "./contract.ts"; import type { NotificationPrefs, PushResult } from "./contract.ts";
import type { Db, DeviceRow } from "./db.ts"; import type { Db, DeviceRow } from "./db.ts";
import type { Push } from "./notify.ts"; import type { Push } from "./notify.ts";
import { trimText } from "./views.ts";
const ALL_ON: NotificationPrefs = { needsMe: true, finished: true, failed: true, subagents: true }; const ALL_ON: NotificationPrefs = { needsMe: true, finished: true, failed: true, subagents: true };
@@ -30,6 +31,32 @@ export function wants(prefs: NotificationPrefs, push: Push): boolean {
} }
} }
/** APNs refuses a payload over 4096 bytes. This leaves room for the JSON APNs adds. */
const PAYLOAD_LIMIT = 3900;
/**
* The APNs payload for a push, kept under APNs' size limit. Long titles and bodies in multi-byte
* scripts can overflow it, so the question's buttons go first, then the body is cut.
*/
export function apnsPayload(push: Push) {
const build = (data: Push["data"], body: string) => ({
aps: {
alert: { title: push.title, ...(push.subtitle ? { subtitle: push.subtitle } : {}), body },
sound: "default",
category: push.category,
"thread-id": data.threadId ?? data.hostId ?? push.category,
},
...data,
});
const fits = (payload: object) => Buffer.byteLength(JSON.stringify(payload)) <= PAYLOAD_LIMIT;
let payload = build(push.data, push.body);
if (fits(payload)) return payload;
const { question: _, ...plain } = push.data;
payload = build(plain, push.body);
for (let chars = 120; !fits(payload) && chars >= 20; chars -= 50) payload = build(plain, trimText(push.body, chars));
return payload;
}
export class Pusher { export class Pusher {
/** Apple's answer to the most recent push sent to any watch. */ /** Apple's answer to the most recent push sent to any watch. */
last: PushResult | null = null; last: PushResult | null = null;
@@ -63,15 +90,7 @@ export class Pusher {
if (!device.apns_token || !device.apns_env) { if (!device.apns_token || !device.apns_env) {
return { ok: false, at, status: 0, reason: "This watch has not registered for notifications." }; return { ok: false, at, status: 0, reason: "This watch has not registered for notifications." };
} }
const payload = { const payload = apnsPayload(push);
aps: {
alert: { title: push.title, ...(push.subtitle ? { subtitle: push.subtitle } : {}), body: push.body },
sound: "default",
category: push.category,
"thread-id": push.data.threadId ?? push.data.hostId ?? push.category,
},
...push.data,
};
const answer = await this.apns.send({ const answer = await this.apns.send({
deviceToken: device.apns_token, deviceToken: device.apns_token,
environment: device.apns_env, environment: device.apns_env,
+4
View File
@@ -74,6 +74,9 @@ export class HostError extends Error {
} }
} }
/** How long a read from a T3 server may take before the gateway gives up on it. */
const HTTP_TIMEOUT_MS = 15_000;
export class T3Host { export class T3Host {
status: HostStatus = "connecting"; status: HostStatus = "connecting";
lastError: string | null = null; lastError: string | null = null;
@@ -213,6 +216,7 @@ export class T3Host {
private async http<T>(path: string, init: RequestInit = {}): Promise<T> { private async http<T>(path: string, init: RequestInit = {}): Promise<T> {
const response = await fetch(new URL(path, this.activeUrl), { const response = await fetch(new URL(path, this.activeUrl), {
signal: AbortSignal.timeout(HTTP_TIMEOUT_MS),
...init, ...init,
headers: { headers: {
authorization: `Bearer ${this.config.token}`, authorization: `Bearer ${this.config.token}`,
+24 -4
View File
@@ -26,6 +26,9 @@ export class RpcError extends Error {
} }
} }
/** How long a unary call waits for its answer. */
const CALL_TIMEOUT_MS = 20_000;
interface Pending { interface Pending {
onChunk?: (value: unknown) => void; onChunk?: (value: unknown) => void;
resolve: (value: unknown) => void; resolve: (value: unknown) => void;
@@ -83,11 +86,28 @@ export class RpcSocket {
}); });
} }
/** Calls a unary method and resolves with its success value. */ /** Calls a unary method and resolves with its success value. Fails after `timeoutMs`. */
call<T = unknown>(tag: string, payload: unknown): Promise<T> { call<T = unknown>(tag: string, payload: unknown, timeoutMs = CALL_TIMEOUT_MS): Promise<T> {
return new Promise<T>((resolve, reject) => { return new Promise<T>((resolve, reject) => {
const id = this.request(tag, payload, { resolve: resolve as (v: unknown) => void, reject }); let timer: ReturnType<typeof setTimeout> | undefined;
if (id === null) reject(new RpcError("socket is not open")); const id = this.request(tag, payload, {
resolve: (value) => {
clearTimeout(timer);
resolve(value as T);
},
reject: (error) => {
clearTimeout(timer);
reject(error);
},
});
if (id === null) {
reject(new RpcError("socket is not open"));
return;
}
timer = setTimeout(() => {
this.pending.delete(id);
reject(new RpcError(`${tag} got no answer in ${timeoutMs / 1000} s`));
}, timeoutMs);
}); });
} }
+4
View File
@@ -74,6 +74,10 @@ export interface ThreadShell {
settledOverride?: Open<"settled"> | null; settledOverride?: Open<"settled"> | null;
/** Set while the user has snoozed the thread. It stays set after the time passes. */ /** Set while the user has snoozed the thread. It stays set after the time passes. */
snoozedUntil?: string | null; snoozedUntil?: string | null;
snoozedAt?: string | null;
latestRunCompletedAt?: string | null;
/** Messages from guests waiting for the thread. One wakes a snoozed thread. */
pendingGuestMessageCount?: number;
} }
export interface ShellSnapshot { export interface ShellSnapshot {
+43 -28
View File
@@ -3,7 +3,7 @@
// are large (hundreds of KB), so they are reduced to a small summary and cached per thread until // are large (hundreds of KB), so they are reduced to a small summary and cached per thread until
// the thread's `updatedAt` moves. // the thread's `updatedAt` moves.
import { resolvePhase, threadPhase, type Phase } from "./awareness.ts"; import { isSnoozed, resolvePhase, threadPhase, type Phase } from "./awareness.ts";
import type { import type {
Activity, Activity,
AgentProvider, AgentProvider,
@@ -76,7 +76,7 @@ export function summarize(bounded: BoundedThread, thread: ThreadShell): ThreadSu
request: pendingRequest(thread, turnItems), request: pendingRequest(thread, turnItems),
activity: currentActivity(turnItems), activity: currentActivity(turnItems),
changedFiles: latestChanges(turnItems), changedFiles: latestChanges(turnItems),
earlierCursor: windowCursor(start, bounded), earlierCursor: windowCursor(start, visible, bounded),
}; };
} }
@@ -90,12 +90,13 @@ const toMessage = (message: Pick<ConversationMessage, "id" | "role" | "text" | "
createdAt: message.createdAt, createdAt: message.createdAt,
}); });
// History cursors are the gateway's own. `w<n>` means the bounded window's messages before index n, // History cursors are the gateway's own. `w<messageId>` means the bounded window's messages before
// which the thread detail leaves out past its 20. `h<cursor>` is T3's cursor for older pages. // that message, which the thread detail leaves out past its 20. The window moves as the thread
// grows, so the cursor names a message, not a position. `h<cursor>` is T3's cursor for older pages.
/** The cursor for what comes before the window's message `start`. */ /** The cursor for what comes before `visible[start]`. */
function windowCursor(start: number, bounded: BoundedThread): string | null { function windowCursor(start: number, visible: ConversationMessage[], bounded: BoundedThread): string | null {
if (start > 0) return `w${start}`; if (start > 0) return `w${visible[start]!.id}`;
return bounded.hasMoreHistory && bounded.historyCursor ? `h${bounded.historyCursor}` : null; return bounded.hasMoreHistory && bounded.historyCursor ? `h${bounded.historyCursor}` : null;
} }
@@ -105,11 +106,17 @@ function windowCursor(start: number, bounded: BoundedThread): string | null {
*/ */
export async function loadHistory(host: T3Host, thread: ThreadShell, cursor: string): Promise<HistoryResponse> { export async function loadHistory(host: T3Host, thread: ThreadShell, cursor: string): Promise<HistoryResponse> {
if (cursor.startsWith("w")) { if (cursor.startsWith("w")) {
const end = Number(cursor.slice(1));
const bounded = await host.bounded(thread.id); const bounded = await host.bounded(thread.id);
const visible = visibleMessages(bounded.projection.messages); const visible = visibleMessages(bounded.projection.messages);
const start = Math.max(0, Math.min(end, visible.length) - MESSAGES_PER_THREAD); const end = visible.findIndex((message) => message.id === cursor.slice(1));
return { messages: visible.slice(start, end).map(toMessage), earlierCursor: windowCursor(start, bounded) }; if (end < 0) {
// The message left the window. T3's pages pick up before the window, and the watch drops
// any message it already has.
const before = windowCursor(0, visible, bounded);
return before ? loadHistory(host, thread, before) : { messages: [], earlierCursor: null };
}
const start = Math.max(0, end - MESSAGES_PER_THREAD);
return { messages: visible.slice(start, end).map(toMessage), earlierCursor: windowCursor(start, visible, bounded) };
} }
let next: string | null = cursor.slice(1); let next: string | null = cursor.slice(1);
for (let pages = 0; next && pages < 3; pages++) { for (let pages = 0; next && pages < 3; pages++) {
@@ -134,13 +141,13 @@ export async function loadHistory(host: T3Host, thread: ThreadShell, cursor: str
return { messages: [], earlierCursor: next && `h${next}` }; return { messages: [], earlierCursor: next && `h${next}` };
} }
/** The files the latest run changed, summed per file, most lines changed first. */ /** The files the latest run changed, summed per file, most lines changed first. Only edits that ran count. */
export function latestChanges(turnItems: TurnItem[]): ChangedFile[] { export function latestChanges(turnItems: TurnItem[]): ChangedFile[] {
const runId = turnItems.findLast((item) => item.runId)?.runId; const runId = turnItems.findLast((item) => item.runId)?.runId;
if (!runId) return []; if (!runId) return [];
const files = new Map<string, ChangedFile>(); const files = new Map<string, ChangedFile>();
for (const item of turnItems) { for (const item of turnItems) {
if (item.type !== "file_change" || item.runId !== runId || !item.fileName) continue; if (item.type !== "file_change" || item.runId !== runId || item.status !== "completed" || !item.fileName) continue;
const file = files.get(item.fileName) ?? { path: item.fileName, name: item.fileName.split("/").pop()!, additions: 0, deletions: 0 }; const file = files.get(item.fileName) ?? { path: item.fileName, name: item.fileName.split("/").pop()!, additions: 0, deletions: 0 };
file.additions += item.additions ?? 0; file.additions += item.additions ?? 0;
file.deletions += item.deletions ?? 0; file.deletions += item.deletions ?? 0;
@@ -271,26 +278,34 @@ export class SummaryCache {
constructor(private readonly log: (message: string) => void = () => {}) {} constructor(private readonly log: (message: string) => void = () => {}) {}
/** The cached summary, if it matches the thread's current version and, while running, is recent. */ /**
* The summary to show: one matching the thread's current version, or for a running thread the
* last one read, which is at most a few seconds behind. Its request can be stale, so callers
* match it against the thread's pending request.
*/
peek(host: T3Host, thread: ThreadShell): ThreadSummary | undefined { peek(host: T3Host, thread: ThreadShell): ThreadSummary | undefined {
const entry = this.entries.get(key(host, thread.id)); const entry = this.entries.get(key(host, thread.id));
if (entry?.updatedAt !== thread.updatedAt) return undefined; if (!entry) return undefined;
if (thread.activeRunId !== null && Date.now() - entry.fetchedAt > RUNNING_SUMMARY_MS) return undefined; return entry.updatedAt === thread.updatedAt || thread.activeRunId !== null ? entry.summary : undefined;
return entry.summary; }
/** Whether the cached summary is current: the same version and, while running, under 5 s old. */
fresh(host: T3Host, thread: ThreadShell): boolean {
const entry = this.entries.get(key(host, thread.id));
if (entry?.updatedAt !== thread.updatedAt) return false;
return thread.activeRunId === null || Date.now() - entry.fetchedAt <= RUNNING_SUMMARY_MS;
} }
get(host: T3Host, thread: ThreadShell): Promise<ThreadSummary> { get(host: T3Host, thread: ThreadShell): Promise<ThreadSummary> {
const cached = this.peek(host, thread); if (this.fresh(host, thread)) return Promise.resolve(this.entries.get(key(host, thread.id))!.summary);
if (cached) return Promise.resolve(cached);
const k = `${key(host, thread.id)}@${thread.updatedAt}`; const k = `${key(host, thread.id)}@${thread.updatedAt}`;
const fetchedAt = Date.now();
const running = this.inflight.get(k); const running = this.inflight.get(k);
if (running) return running; if (running) return running;
const promise = host const promise = host
.bounded(thread.id) .bounded(thread.id)
.then((bounded) => { .then((bounded) => {
const summary = summarize(bounded, thread); const summary = summarize(bounded, thread);
this.entries.set(key(host, thread.id), { updatedAt: thread.updatedAt, fetchedAt, summary }); this.entries.set(key(host, thread.id), { updatedAt: thread.updatedAt, fetchedAt: Date.now(), summary });
return summary; return summary;
}) })
.finally(() => this.inflight.delete(k)); .finally(() => this.inflight.delete(k));
@@ -384,6 +399,10 @@ function attachDelegates(candidates: Candidate[], hosts: T3Host[]): Candidate[]
const requestAge = (thread: ThreadShell) => thread.pendingRuntimeRequest?.createdAt ?? ""; const requestAge = (thread: ThreadShell) => thread.pendingRuntimeRequest?.createdAt ?? "";
/** The summary's request, if it is still the one the thread waits on. */
const currentRequest = (thread: ThreadShell, summary: ThreadSummary | undefined): PendingRequest | null =>
summary?.request && summary.request.id === thread.pendingRuntimeRequest?.id ? summary.request : null;
function feedItem( function feedItem(
host: T3Host, host: T3Host,
thread: ThreadShell, thread: ThreadShell,
@@ -410,7 +429,7 @@ function feedItem(
deletions: summary.changedFiles.reduce((sum, file) => sum + file.deletions, 0), deletions: summary.changedFiles.reduce((sum, file) => sum + file.deletions, 0),
} }
: null, : null,
request: (via ? via.summary : summary)?.request ?? null, request: via ? currentRequest(via.thread, via.summary) : currentRequest(thread, summary),
via: via ? { hostId: via.host.id, threadId: via.thread.id, title: via.thread.title } : null, via: via ? { hostId: via.host.id, threadId: via.thread.id, title: via.thread.title } : null,
}; };
} }
@@ -455,14 +474,14 @@ export async function buildSummary(hosts: T3Host[], cache: SummaryCache): Promis
} }
} }
const read = [next, current].filter((pick) => pick !== null); const read = [next, current].filter((pick) => pick !== null);
const missing = read.filter(({ host, thread }) => !cache.peek(host, thread)); const missing = read.filter(({ host, thread }) => !cache.fresh(host, thread));
if (missing.length > 0) { if (missing.length > 0) {
await Promise.race([ await Promise.race([
Promise.all(missing.map(({ host, thread }) => cache.get(host, thread).catch(() => undefined))), Promise.all(missing.map(({ host, thread }) => cache.get(host, thread).catch(() => undefined))),
Bun.sleep(SUMMARY_WAIT_MS), Bun.sleep(SUMMARY_WAIT_MS),
]); ]);
} }
const request = next && cache.peek(next.host, next.thread)?.request; const request = next && currentRequest(next.thread, cache.peek(next.host, next.thread));
const asked = request?.prompt ?? request?.questions[0]?.question; const asked = request?.prompt ?? request?.questions[0]?.question;
return { return {
needsYou, needsYou,
@@ -484,10 +503,6 @@ export async function buildSummary(hosts: T3Host[], cache: SummaryCache): Promis
}; };
} }
/** True while a snooze is set and its time has not come. */
const isSnoozed = (thread: ThreadShell, now: number) =>
!!thread.snoozedUntil && Date.parse(thread.snoozedUntil) > now;
/** Every thread the feed shows, in feed order, with subagent requests on their parents' cards. */ /** Every thread the feed shows, in feed order, with subagent requests on their parents' cards. */
export function feedCandidates(hosts: T3Host[], now = Date.now()) { export function feedCandidates(hosts: T3Host[], now = Date.now()) {
const found: Candidate[] = []; const found: Candidate[] = [];
@@ -530,7 +545,7 @@ export async function buildFeed(hosts: T3Host[], cache: SummaryCache): Promise<F
const shown = [...needsYou, ...working, ...recent].flatMap((candidate) => (candidate.via ? [candidate, candidate.via] : [candidate])); const shown = [...needsYou, ...working, ...recent].flatMap((candidate) => (candidate.via ? [candidate, candidate.via] : [candidate]));
const missing = shown const missing = shown
.filter(({ host, thread }) => !cache.peek(host, thread)) .filter(({ host, thread }) => !cache.fresh(host, thread))
.map(({ host, thread }) => cache.get(host, thread).catch(() => undefined)); .map(({ host, thread }) => cache.get(host, thread).catch(() => undefined));
if (missing.length > 0) await Promise.race([Promise.all(missing), Bun.sleep(SUMMARY_WAIT_MS)]); if (missing.length > 0) await Promise.race([Promise.all(missing), Bun.sleep(SUMMARY_WAIT_MS)]);
+23 -1
View File
@@ -1,7 +1,7 @@
// Cases ported from T3 Code's agentAwareness.test.ts, so the gateway notifies when T3's own app would. // Cases ported from T3 Code's agentAwareness.test.ts, so the gateway notifies when T3's own app would.
import { describe, expect, it } from "bun:test"; import { describe, expect, it } from "bun:test";
import { threadPhase } from "../src/awareness.ts"; import { isSnoozed, threadPhase } from "../src/awareness.ts";
import type { ThreadShell } from "../src/t3/types.ts"; import type { ThreadShell } from "../src/t3/types.ts";
const thread = (overrides: Partial<ThreadShell> = {}): ThreadShell => ({ const thread = (overrides: Partial<ThreadShell> = {}): ThreadShell => ({
@@ -104,3 +104,25 @@ describe("threadPhase", () => {
expect(threadPhase(thread({ delegatedFrom, pendingRuntimeRequest: request("command") }))).toBe("waiting_for_approval"); expect(threadPhase(thread({ delegatedFrom, pendingRuntimeRequest: request("command") }))).toBe("waiting_for_approval");
}); });
}); });
describe("isSnoozed", () => {
const now = Date.parse("2026-11-01T12:00:00.000Z");
const snoozed = {
snoozedUntil: "2026-11-01T18:00:00.000Z",
snoozedAt: "2026-11-01T10:00:00.000Z",
latestRunCompletedAt: "2026-11-01T09:00:00.000Z",
status: "completed",
pendingRuntimeRequest: null,
} as const;
it("hides a thread until its time, as T3 does", () => {
expect(isSnoozed(snoozed, now)).toBe(true);
expect(isSnoozed({ ...snoozed, snoozedUntil: "2026-11-01T11:00:00.000Z" }, now)).toBe(false);
});
it("wakes a thread that waits on someone or failed after the snooze", () => {
expect(isSnoozed({ ...snoozed, pendingRuntimeRequest: { id: "r", kind: "command", createdAt: "" } }, now)).toBe(false);
expect(isSnoozed({ ...snoozed, status: "failed", latestRunCompletedAt: "2026-11-01T11:00:00.000Z" }, now)).toBe(false);
expect(isSnoozed({ ...snoozed, status: "failed" }, now)).toBe(true);
});
});
+21 -1
View File
@@ -2,7 +2,7 @@ import { describe, expect, it } from "bun:test";
import type { DeviceRow } from "../src/db.ts"; import type { DeviceRow } from "../src/db.ts";
import type { Push } from "../src/notify.ts"; import type { Push } from "../src/notify.ts";
import { notificationPrefs, wants } from "../src/push.ts"; import { apnsPayload, notificationPrefs, wants } from "../src/push.ts";
const push = (category: Push["category"], fromSubagent = false): Push => ({ const push = (category: Push["category"], fromSubagent = false): Push => ({
title: "Receipt OCR", title: "Receipt OCR",
@@ -35,3 +35,23 @@ describe("wants", () => {
expect(wants({ needsMe: false, finished: false, failed: false, subagents: false }, push("host"))).toBe(true); expect(wants({ needsMe: false, finished: false, failed: false, subagents: false }, push("host"))).toBe(true);
}); });
}); });
describe("apnsPayload", () => {
it("drops the question's buttons, then cuts the body, to stay under APNs' 4 KB", () => {
const wide = "界".repeat(400);
const big: Push = {
...push("input"),
title: "界".repeat(80),
body: wide,
data: {
hostId: "host-1",
threadId: "thread-1",
question: { id: "q", options: [1, 2, 3, 4].map((n) => ({ label: `${n}${"界".repeat(39)}`, value: "界".repeat(200) })), allowCustomAnswer: true },
},
};
const payload = apnsPayload(big);
expect(Buffer.byteLength(JSON.stringify(payload))).toBeLessThanOrEqual(3900);
expect("question" in payload).toBe(false);
expect(apnsPayload(push("finished"))).toMatchObject({ aps: { alert: { body: "Finished." } }, threadId: "thread-1" });
});
});
+14 -6
View File
@@ -266,12 +266,19 @@ describe("thread history", () => {
const host = { bounded: () => Promise.resolve(window) } as unknown as T3Host; const host = { bounded: () => Promise.resolve(window) } as unknown as T3Host;
it("pages back through the bounded window, then hands over to T3's cursor", async () => { it("pages back through the bounded window, then hands over to T3's cursor", async () => {
expect(summarize(window as BoundedThread, thread()).earlierCursor).toBe("w25"); expect(summarize(window as BoundedThread, thread()).earlierCursor).toBe("wm25");
const first = await loadHistory(host, thread(), "w25"); const first = await loadHistory(host, thread(), "wm25");
expect([first.messages[0]?.text, first.messages.at(-1)?.text, first.earlierCursor]).toEqual(["m5", "m24", "w5"]); expect([first.messages[0]?.text, first.messages.at(-1)?.text, first.earlierCursor]).toEqual(["m5", "m24", "wm5"]);
const second = await loadHistory(host, thread(), "w5"); const second = await loadHistory(host, thread(), "wm5");
expect([second.messages.length, second.earlierCursor]).toEqual([5, "ht3-cursor"]); expect([second.messages.length, second.earlierCursor]).toEqual([5, "ht3-cursor"]);
}); });
it("names a message, so new messages moving the window do not repeat or skip any", async () => {
const grown = { ...window, projection: { ...window.projection, messages: [...messages.slice(3), message("user", "m45"), message("assistant", "m46"), message("user", "m47")] } };
const moved = { bounded: () => Promise.resolve(grown) } as unknown as T3Host;
const page = await loadHistory(moved, thread(), "wm25");
expect([page.messages[0]?.text, page.messages.at(-1)?.text]).toEqual(["m5", "m24"]);
});
}); });
describe("projectThreads", () => { describe("projectThreads", () => {
@@ -297,14 +304,15 @@ describe("projectThreads", () => {
describe("latestChanges", () => { describe("latestChanges", () => {
it("sums each file the latest run changed, most lines first", () => { it("sums each file the latest run changed, most lines first", () => {
const change = (runId: string, fileName: string, additions: number, deletions: number) => const change = (runId: string, fileName: string, additions: number, deletions: number, status = "completed") =>
({ type: "file_change", runId, fileName, additions, deletions }); ({ type: "file_change", runId, status, fileName, additions, deletions });
expect( expect(
latestChanges([ latestChanges([
change("run-1", "old.ts", 50, 0), change("run-1", "old.ts", 50, 0),
change("run-2", "gateway/compose.yml", 3, 2), change("run-2", "gateway/compose.yml", 3, 2),
change("run-2", "gateway/Containerfile", 31, 0), change("run-2", "gateway/Containerfile", 31, 0),
change("run-2", "gateway/compose.yml", 6, 4), change("run-2", "gateway/compose.yml", 6, 4),
change("run-2", "declined.ts", 80, 0, "cancelled"),
{ type: "assistant_message", runId: "run-2" }, { type: "assistant_message", runId: "run-2" },
]), ]),
).toEqual([ ).toEqual([