Gateway: per-watch notification switches, model list for new threads, and Ask before acting
This commit is contained in:
1 parent
a5222e14e8
commit
3cc7e40852
14 files changed
+472
-20
No files matched your search
+68
-9
@@ -6,6 +6,7 @@ import type {
|
||||
ErrorResponse,
|
||||
NewThreadRequest,
|
||||
NewThreadResponse,
|
||||
NotificationPrefs,
|
||||
PairRequest,
|
||||
PairResponse,
|
||||
ProjectsResponse,
|
||||
@@ -15,14 +16,17 @@ import type {
|
||||
SettleRecentResponse,
|
||||
SendMessageRequest,
|
||||
StatusResponse,
|
||||
ThreadModeRequest,
|
||||
} from "./contract.ts";
|
||||
import type { Db, DeviceRow } from "./db.ts";
|
||||
import type { Hub } from "./hub.ts";
|
||||
import type { Pusher } from "./push.ts";
|
||||
import { notificationPrefs, type Pusher } from "./push.ts";
|
||||
import { HostError, type T3Host } from "./t3/host.ts";
|
||||
import { RpcError } from "./t3/rpc.ts";
|
||||
import type { ApprovalDecision, ModelSelection, ThreadShell } from "./t3/types.ts";
|
||||
import {
|
||||
agentProviders,
|
||||
ASK_FIRST,
|
||||
buildFeed,
|
||||
buildRecent,
|
||||
buildSummary,
|
||||
@@ -110,6 +114,29 @@ export function createRoutes({ db, hub, cache, pusher, startedAt, log = () => {}
|
||||
const resolveModel = (host: T3Host, projectId: string): ModelSelection | null =>
|
||||
host.projects.get(projectId)?.defaultModelSelection ?? host.serverConfig?.settings?.defaultModelSelection ?? null;
|
||||
|
||||
/** Ask first, or else the host's default mode, or full access when that default also asks first. */
|
||||
const runtimeMode = (host: T3Host, askFirst: boolean): string => {
|
||||
const fallback = host.serverConfig?.settings?.defaultRuntimeMode ?? "full-access";
|
||||
return askFirst ? ASK_FIRST : fallback === ASK_FIRST ? "full-access" : fallback;
|
||||
};
|
||||
|
||||
/**
|
||||
* The model for a new thread: the one the watch picked, if the host offers it, or the project's
|
||||
* default. The default's options, such as reasoning effort, carry over only when it is kept.
|
||||
*/
|
||||
const pickModel = (host: T3Host, projectId: string, picked: NewThreadRequest["model"]): ModelSelection => {
|
||||
const fallback = resolveModel(host, projectId);
|
||||
if (picked && !(picked.instanceId === fallback?.instanceId && picked.slug === fallback.model)) {
|
||||
const offered = agentProviders(host)
|
||||
.find((provider) => provider.instanceId === picked.instanceId)
|
||||
?.models.some((model) => model.slug === picked.slug);
|
||||
if (!offered) throw new HttpError(400, "That model is not available on this host.");
|
||||
return { instanceId: picked.instanceId, model: picked.slug };
|
||||
}
|
||||
if (!fallback) throw new HttpError(400, "That project has no default model. Pick one, or set one in T3 Code.");
|
||||
return fallback;
|
||||
};
|
||||
|
||||
return {
|
||||
"/v1/health": {
|
||||
GET: () => json({ ok: true, hosts: hub.list().map(hostSummary) }),
|
||||
@@ -156,6 +183,18 @@ export function createRoutes({ db, hub, cache, pusher, startedAt, log = () => {}
|
||||
}),
|
||||
},
|
||||
|
||||
"/v1/device/notifications": {
|
||||
PUT: authed<"/v1/device/notifications">(async (req, device) => {
|
||||
const body = await readBody<Partial<NotificationPrefs>>(req);
|
||||
const prefs = notificationPrefs(device);
|
||||
for (const key of Object.keys(prefs) as Array<keyof NotificationPrefs>) {
|
||||
if (typeof body[key] === "boolean") prefs[key] = body[key];
|
||||
}
|
||||
db.devices.setNotifications(device.id, JSON.stringify(prefs));
|
||||
return json(prefs satisfies NotificationPrefs);
|
||||
}),
|
||||
},
|
||||
|
||||
"/v1/device/test-push": {
|
||||
POST: authed<"/v1/device/test-push">(async (_req, device) =>
|
||||
json(
|
||||
@@ -181,6 +220,7 @@ export function createRoutes({ db, hub, cache, pusher, startedAt, log = () => {}
|
||||
pairedAt: device.created_at,
|
||||
pushRegistered: device.apns_token !== null,
|
||||
apnsEnvironment: device.apns_env,
|
||||
notifications: notificationPrefs(device),
|
||||
},
|
||||
} satisfies StatusResponse);
|
||||
}),
|
||||
@@ -220,16 +260,21 @@ export function createRoutes({ db, hub, cache, pusher, startedAt, log = () => {}
|
||||
},
|
||||
|
||||
"/v1/projects": {
|
||||
GET: authed(() =>
|
||||
json({
|
||||
GET: authed(async () => {
|
||||
await Promise.all(hub.list().map((host) => host.refreshConfig()));
|
||||
return json({
|
||||
hosts: hub.list().map((host) => ({
|
||||
...hostSummary(host),
|
||||
projects: [...host.projects.values()]
|
||||
.sort((a, b) => a.title.localeCompare(b.title))
|
||||
.map((project) => ({ id: project.id, title: project.title, model: resolveModel(host, project.id)?.model ?? null })),
|
||||
.map((project) => {
|
||||
const model = resolveModel(host, project.id);
|
||||
return { id: project.id, title: project.title, model: model?.model ?? null, instanceId: model?.instanceId ?? null };
|
||||
}),
|
||||
providers: agentProviders(host),
|
||||
})),
|
||||
} satisfies ProjectsResponse),
|
||||
),
|
||||
} satisfies ProjectsResponse);
|
||||
}),
|
||||
},
|
||||
|
||||
"/v1/threads": {
|
||||
@@ -239,8 +284,7 @@ export function createRoutes({ db, hub, cache, pusher, startedAt, log = () => {}
|
||||
if (!text) throw new HttpError(400, "Say what the thread should do.");
|
||||
const host = findHost(String(body.hostId));
|
||||
if (!host.projects.has(String(body.projectId))) throw new HttpError(404, "Unknown project.");
|
||||
const modelSelection = resolveModel(host, body.projectId);
|
||||
if (!modelSelection) throw new HttpError(400, "That project has no default model. Set one in T3 Code.");
|
||||
const modelSelection = pickModel(host, body.projectId, body.model);
|
||||
const result = await host.launchThread({
|
||||
commandId: crypto.randomUUID(),
|
||||
creationSource: "mobile",
|
||||
@@ -248,7 +292,7 @@ export function createRoutes({ db, hub, cache, pusher, startedAt, log = () => {}
|
||||
title: titleFrom(text),
|
||||
generateTitle: true,
|
||||
modelSelection,
|
||||
runtimeMode: host.serverConfig?.settings?.defaultRuntimeMode ?? "full-access",
|
||||
runtimeMode: runtimeMode(host, body.askFirst === true),
|
||||
interactionMode: "default",
|
||||
workspaceStrategy: { type: "root" },
|
||||
initialMessage: { messageId: crypto.randomUUID(), text, attachments: [] },
|
||||
@@ -313,6 +357,21 @@ export function createRoutes({ db, hub, cache, pusher, startedAt, log = () => {}
|
||||
}),
|
||||
},
|
||||
|
||||
"/v1/threads/:hostId/:threadId/mode": {
|
||||
POST: authed<"/v1/threads/:hostId/:threadId/mode">(async (req) => {
|
||||
const { host, thread } = findThread(req.params.hostId, req.params.threadId);
|
||||
const { askFirst } = await readBody<ThreadModeRequest>(req);
|
||||
if (typeof askFirst !== "boolean") throw new HttpError(400, "Send askFirst as true or false.");
|
||||
await host.dispatch({
|
||||
type: "thread.runtime-mode.set",
|
||||
commandId: crypto.randomUUID(),
|
||||
threadId: thread.id,
|
||||
runtimeMode: runtimeMode(host, askFirst),
|
||||
});
|
||||
return accepted();
|
||||
}),
|
||||
},
|
||||
|
||||
"/v1/threads/:hostId/:threadId/stop": {
|
||||
POST: authed<"/v1/threads/:hostId/:threadId/stop">(async (req) => {
|
||||
const { host, thread } = findThread(req.params.hostId, req.params.threadId);
|
||||
|
||||
@@ -68,9 +68,21 @@ export interface StatusResponse {
|
||||
/** True once the watch has sent its APNs token. */
|
||||
pushRegistered: boolean;
|
||||
apnsEnvironment: "sandbox" | "production" | null;
|
||||
notifications: NotificationPrefs;
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Which pushes a watch wants. `needsMe` covers approvals and questions on your own threads, and
|
||||
* `subagents` covers the same from subagents. Host problems and test pushes always go out.
|
||||
*/
|
||||
export interface NotificationPrefs {
|
||||
needsMe: boolean;
|
||||
finished: boolean;
|
||||
failed: boolean;
|
||||
subagents: boolean;
|
||||
}
|
||||
|
||||
/** Something a thread is waiting on: an approval, or questions for the user. */
|
||||
export interface PendingRequest {
|
||||
id: string;
|
||||
@@ -157,6 +169,12 @@ export interface ThreadDetail {
|
||||
messages: Message[];
|
||||
/** True while a run is active, so Stop makes sense. */
|
||||
canStop: boolean;
|
||||
/** True when the thread runs in approval-required mode, asking before each action. */
|
||||
askFirst: boolean;
|
||||
}
|
||||
|
||||
export interface ThreadModeRequest {
|
||||
askFirst: boolean;
|
||||
}
|
||||
|
||||
export interface ProjectsResponse {
|
||||
@@ -167,11 +185,24 @@ export interface ProjectsResponse {
|
||||
title: string;
|
||||
/** The model a new thread will use, such as `claude-opus-5-5`. Null when none is configured. */
|
||||
model: string | null;
|
||||
/** The provider instance `model` belongs to, matching an entry in `providers`. */
|
||||
instanceId: string | null;
|
||||
}>;
|
||||
/** The providers on this host that can start a thread now, each with its models. */
|
||||
providers: AgentProvider[];
|
||||
}
|
||||
>;
|
||||
}
|
||||
|
||||
export interface AgentProvider {
|
||||
instanceId: string;
|
||||
/** Such as `Claude` or `Codex`. */
|
||||
name: string;
|
||||
/** False when the provider cannot run in approval-required mode. */
|
||||
canAskFirst: boolean;
|
||||
models: Array<{ slug: string; name: string }>;
|
||||
}
|
||||
|
||||
export interface PairRequest {
|
||||
code: string;
|
||||
deviceName: string;
|
||||
@@ -200,6 +231,10 @@ export interface NewThreadRequest {
|
||||
hostId: string;
|
||||
projectId: string;
|
||||
text: string;
|
||||
/** A model from the host's `providers`. Absent uses the project's default. */
|
||||
model?: { instanceId: string; slug: string };
|
||||
/** Starts the thread in approval-required mode. Absent or false uses the host's default mode. */
|
||||
askFirst?: boolean;
|
||||
}
|
||||
|
||||
export interface NewThreadResponse {
|
||||
|
||||
@@ -22,6 +22,8 @@ export interface DeviceRow {
|
||||
token_hash: string;
|
||||
apns_token: string | null;
|
||||
apns_env: "sandbox" | "production" | null;
|
||||
/** The watch's NotificationPrefs as JSON. Null until the watch changes one. */
|
||||
notifications: string | null;
|
||||
created_at: string;
|
||||
last_seen_at: string | null;
|
||||
revoked_at: string | null;
|
||||
@@ -65,6 +67,8 @@ export function openDb(path: string) {
|
||||
db.run(SCHEMA);
|
||||
const hostColumns = db.query<{ name: string }, []>("pragma table_info(hosts)").all().map((column) => column.name);
|
||||
if (!hostColumns.includes("alt_urls")) db.run("alter table hosts add column alt_urls text");
|
||||
const deviceColumns = db.query<{ name: string }, []>("pragma table_info(devices)").all().map((column) => column.name);
|
||||
if (!deviceColumns.includes("notifications")) db.run("alter table devices add column notifications text");
|
||||
|
||||
return {
|
||||
hosts: {
|
||||
@@ -107,6 +111,8 @@ export function openDb(path: string) {
|
||||
apnsEnv,
|
||||
});
|
||||
},
|
||||
setNotifications: (id: string, json: string) =>
|
||||
db.query("update devices set notifications = $json where id = $id").run({ id, json }),
|
||||
clearPush: (apnsToken: string) =>
|
||||
db.query("update devices set apns_token = null where apns_token = $apnsToken").run({ apnsToken }),
|
||||
revoke: (idOrName: string) =>
|
||||
|
||||
@@ -14,6 +14,8 @@ export interface Push {
|
||||
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;
|
||||
@@ -99,6 +101,7 @@ export class Notifier {
|
||||
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 } : {}) },
|
||||
collapseId: `${host.id}:${thread.id}`.slice(0, 64),
|
||||
});
|
||||
|
||||
+30
-4
@@ -1,11 +1,35 @@
|
||||
// Delivers pushes to paired watches through APNs, and remembers Apple's last answer so the watch's
|
||||
// status screen can show whether pushes are getting through. Without APNs settings it only logs.
|
||||
// status screen can show whether pushes are getting through. Each watch only gets the kinds of push
|
||||
// it asked for. Without APNs settings it only logs.
|
||||
|
||||
import type { Apns } from "./apns.ts";
|
||||
import type { PushResult } from "./contract.ts";
|
||||
import type { NotificationPrefs, PushResult } from "./contract.ts";
|
||||
import type { Db, DeviceRow } from "./db.ts";
|
||||
import type { Push } from "./notify.ts";
|
||||
|
||||
const ALL_ON: NotificationPrefs = { needsMe: true, finished: true, failed: true, subagents: true };
|
||||
|
||||
/** A watch's choices, with every kind it never set left on. */
|
||||
export function notificationPrefs(device: DeviceRow): NotificationPrefs {
|
||||
const saved: Partial<NotificationPrefs> = device.notifications ? JSON.parse(device.notifications) : {};
|
||||
return { ...ALL_ON, ...saved };
|
||||
}
|
||||
|
||||
/** Whether a watch with these choices wants this push. Host problems and tests always go out. */
|
||||
export function wants(prefs: NotificationPrefs, push: Push): boolean {
|
||||
switch (push.category) {
|
||||
case "approval":
|
||||
case "input":
|
||||
return push.fromSubagent ? prefs.subagents : prefs.needsMe;
|
||||
case "finished":
|
||||
return prefs.finished;
|
||||
case "failed":
|
||||
return prefs.failed;
|
||||
default:
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
export class Pusher {
|
||||
/** Apple's answer to the most recent push sent to any watch. */
|
||||
last: PushResult | null = null;
|
||||
@@ -20,9 +44,11 @@ export class Pusher {
|
||||
return this.apns !== null;
|
||||
}
|
||||
|
||||
/** Sends to every paired watch that registered for pushes. */
|
||||
/** Sends to every paired watch that registered for pushes and wants this kind. */
|
||||
sendAll = async (push: Push): Promise<void> => {
|
||||
const devices = this.db.devices.active().filter((device) => device.apns_token && device.apns_env);
|
||||
const devices = this.db.devices
|
||||
.active()
|
||||
.filter((device) => device.apns_token && device.apns_env && wants(notificationPrefs(device), push));
|
||||
if (!this.apns || devices.length === 0) {
|
||||
this.log(`push (not sent): [${push.category}] ${push.title}: ${push.body}`);
|
||||
return;
|
||||
|
||||
@@ -153,6 +153,19 @@ export class T3Host {
|
||||
return this.requireSocket().call("orchestration.dispatchCommand", command);
|
||||
}
|
||||
|
||||
/**
|
||||
* Rereads the server config, for current provider and model lists. Keeps the old one when the
|
||||
* server fails or takes over 3 s.
|
||||
*/
|
||||
async refreshConfig(): Promise<void> {
|
||||
if (!this.socket) return;
|
||||
const fresh = await Promise.race([
|
||||
this.socket.call<ServerConfig>("server.getConfig", {}).catch(() => null),
|
||||
Bun.sleep(3_000).then(() => null),
|
||||
]);
|
||||
if (fresh) this.serverConfig = fresh;
|
||||
}
|
||||
|
||||
async launchThread(input: LaunchThreadInput): Promise<{ threadId: string }> {
|
||||
return this.requireSocket().call("orchestration.launchThread", input);
|
||||
}
|
||||
|
||||
@@ -158,6 +158,20 @@ export interface EnvironmentDescriptor {
|
||||
export interface ServerConfig {
|
||||
environment?: { capabilities?: Record<string, unknown> };
|
||||
settings?: { defaultModelSelection?: ModelSelection | null; defaultRuntimeMode?: string };
|
||||
providers?: ServerProvider[];
|
||||
}
|
||||
|
||||
/** The parts of a provider in `server.getConfig` the gateway reads. */
|
||||
export interface ServerProvider {
|
||||
instanceId: string;
|
||||
driver: string;
|
||||
displayName?: string;
|
||||
enabled: boolean;
|
||||
installed: boolean;
|
||||
status: Open<"ready" | "warning" | "error" | "disabled">;
|
||||
availability?: Open<"available" | "unavailable">;
|
||||
supportedRuntimeModes?: string[];
|
||||
models: Array<{ slug: string; name: string; shortName?: string; isLegacy?: boolean }>;
|
||||
}
|
||||
|
||||
/** Commands sent through `orchestration.dispatchCommand`. */
|
||||
@@ -183,6 +197,7 @@ export type Command =
|
||||
}
|
||||
| { type: "thread.stop"; commandId: string; threadId: string }
|
||||
| { type: "thread.settle"; commandId: string; threadId: string }
|
||||
| { type: "thread.runtime-mode.set"; commandId: string; threadId: string; runtimeMode: string }
|
||||
| { type: "run.interrupt"; commandId: string; threadId: string; runId: string };
|
||||
|
||||
export interface LaunchThreadInput {
|
||||
|
||||
@@ -5,6 +5,7 @@
|
||||
|
||||
import { threadPhase, type Phase } from "./awareness.ts";
|
||||
import type {
|
||||
AgentProvider,
|
||||
Feed,
|
||||
FeedItem,
|
||||
HostStatusDetail,
|
||||
@@ -360,5 +361,28 @@ export async function buildThreadDetail(
|
||||
item: feedItem(host, thread, child?.phase ?? own, summary, via),
|
||||
messages: summary.messages,
|
||||
canStop: thread.activeRunId !== null,
|
||||
askFirst: thread.runtimeMode === ASK_FIRST,
|
||||
};
|
||||
}
|
||||
|
||||
/** T3's runtime mode that asks before each action. */
|
||||
export const ASK_FIRST = "approval-required";
|
||||
|
||||
/** The providers on a host that can start a thread now, leaving out legacy models. */
|
||||
export function agentProviders(host: T3Host): AgentProvider[] {
|
||||
return (host.serverConfig?.providers ?? [])
|
||||
.filter(
|
||||
(provider) =>
|
||||
provider.enabled &&
|
||||
provider.installed &&
|
||||
provider.availability !== "unavailable" &&
|
||||
(provider.status === "ready" || provider.status === "warning"),
|
||||
)
|
||||
.map((provider) => ({
|
||||
instanceId: provider.instanceId,
|
||||
name: provider.displayName ?? provider.driver,
|
||||
canAskFirst: provider.supportedRuntimeModes?.includes(ASK_FIRST) ?? true,
|
||||
models: provider.models.filter((model) => !model.isLegacy).map((model) => ({ slug: model.slug, name: model.name })),
|
||||
}))
|
||||
.filter((provider) => provider.models.length > 0);
|
||||
}
|
||||
Reference in new issue
Block a user