From db031dd447fa0296eb0bc6e860f8a14a41c6c49c Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Sun, 26 Jul 2026 10:41:58 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20content-blind=20push=20notifications=20?= =?UTF-8?q?=E2=80=94=20daemon=20backbone=20(P2=20push,=20slice=201)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a session blocks on a tool approval, the daemon now sends a CONTENT-BLIND wake-up to the session owner's registered devices off-LAN, behind a swappable PushTransport seam. Expo Push is the v1 transport; a self-hosted content-blind relay swaps in behind the same interface later (the migration is a transport change, not a redesign). codeoid-mobile consumes this after the next codeoid release — this slice is the headless, device-free backbone. Protocol (@codeoid/protocol): - push.register / push.unregister client messages + a PUSH capability + PUSH_TOKEN_MAX limit; zod schemas + the compile-time coverage samples. Daemon: - push_registrations SQLite table + owner-scoped CRUD, keyed on the ZeroID identity (owner_sub == sessions.created_by), tenant-scoped by account/project — so only the session's creator is alerted, never across tenants. - PushTransport seam (src/daemon/push): ExpoPushTransport (batched at 100/req, bare fetch, best-effort/never-throws) + a noop transport, selected by config.push.transport ("expo" | "none", default "none"). - PushService.notifyApproval resolves a blocked session's owner to their devices and delivers a content-blind { sessionId, kind } note — never a tool name, args, or description. - Emit at SessionManager.#statusObserver — the single daemon-wide chokepoint that fires regardless of client attachment — gated on waiting_approval; fire-and-forget so a push hiccup can't touch the status path. - push.register / push.unregister handlers, self-scoped to the caller's auth (no bespoke scope, mirroring session.ui_response). - The daemon advertises the PUSH capability on auth.ok only when a transport is configured, so clients feature-detect push support. Config: a push block + CODEOID_PUSH_TRANSPORT / CODEOID_EXPO_ACCESS_TOKEN env overrides. Tests (bun): push_registrations store (owner/tenant isolation, refresh, owner-scoped delete); Expo transport content-blindness + batching + auth header + error-swallow; the createPushTransport factory; PushService routing; register/unregister handlers over the wire; and an end-to-end emit test that drives a real session to waiting_approval and asserts the outbound Expo payload is content-blind. Signed-off-by: Yash Datta Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/protocol/src/schemas.test.ts | 2 + packages/protocol/src/schemas.ts | 17 +++ packages/protocol/src/types.ts | 35 ++++- src/config.ts | 32 +++++ src/daemon/push/expo.ts | 65 +++++++++ src/daemon/push/index.ts | 4 + src/daemon/push/push.test.ts | 184 ++++++++++++++++++++++++ src/daemon/push/service.ts | 68 +++++++++ src/daemon/push/types.ts | 42 ++++++ src/daemon/server.ts | 9 +- src/daemon/session-manager.ts | 46 ++++++ src/daemon/store.ts | 74 +++++++++- src/tests/push-emit.test.ts | 192 ++++++++++++++++++++++++++ src/tests/push-handlers.test.ts | 77 +++++++++++ src/tests/push-store.test.ts | 76 ++++++++++ 15 files changed, 920 insertions(+), 3 deletions(-) create mode 100644 src/daemon/push/expo.ts create mode 100644 src/daemon/push/index.ts create mode 100644 src/daemon/push/push.test.ts create mode 100644 src/daemon/push/service.ts create mode 100644 src/daemon/push/types.ts create mode 100644 src/tests/push-emit.test.ts create mode 100644 src/tests/push-handlers.test.ts create mode 100644 src/tests/push-store.test.ts diff --git a/packages/protocol/src/schemas.test.ts b/packages/protocol/src/schemas.test.ts index 38e9b055..f81a3510 100644 --- a/packages/protocol/src/schemas.test.ts +++ b/packages/protocol/src/schemas.test.ts @@ -160,6 +160,8 @@ const samples: { [T in ClientTypes]: Extract } = { "pipeline.pack.remove": { type: "pipeline.pack.remove", id: "r41", packId: "aif-sdlc" }, "pipeline.pack.trust": { type: "pipeline.pack.trust", id: "r42", packId: "aif-sdlc", trusted: true }, "pipeline.pack.select": { type: "pipeline.pack.select", id: "r43", packId: "aif-sdlc" }, + "push.register": { type: "push.register", id: "r44", token: "ExponentPushToken[abc]", platform: "ios" }, + "push.unregister": { type: "push.unregister", id: "r45", token: "ExponentPushToken[abc]" }, }; describe("fidelity — valid samples round-trip unchanged", () => { diff --git a/packages/protocol/src/schemas.ts b/packages/protocol/src/schemas.ts index a7f22cda..86c7ae69 100644 --- a/packages/protocol/src/schemas.ts +++ b/packages/protocol/src/schemas.ts @@ -170,6 +170,21 @@ export const sessionApproveSchema = z.object({ updatedInput: z.record(z.string(), z.unknown()).optional(), }); +const pushTokenField = z.string().min(1).max(LIMITS.PUSH_TOKEN_MAX); + +export const pushRegisterSchema = z.object({ + ...base, + type: z.literal("push.register"), + token: pushTokenField, + platform: z.enum(["ios", "android"]), +}); + +export const pushUnregisterSchema = z.object({ + ...base, + type: z.literal("push.unregister"), + token: pushTokenField, +}); + export const sessionUiResponseSchema = z .object({ ...base, @@ -584,6 +599,8 @@ export const clientMessageSchema = z.discriminatedUnion("type", [ pipelinePackRemoveSchema, pipelinePackTrustSchema, pipelinePackSelectSchema, + pushRegisterSchema, + pushUnregisterSchema, ]); /** diff --git a/packages/protocol/src/types.ts b/packages/protocol/src/types.ts index 2b938b44..e19e434b 100644 --- a/packages/protocol/src/types.ts +++ b/packages/protocol/src/types.ts @@ -71,6 +71,14 @@ export const CAPABILITIES = { * full-buffer replay. */ SCROLLBACK_PAGING: "scrollback.paging", + /** + * Push notifications. Declared by the DAEMON when a push transport is + * configured (so clients can feature-detect before registering) and by + * CLIENTS that can receive them. A client registers a device token via + * `push.register`; the daemon then sends content-blind wake-ups (only opaque + * ids, never tool args) when one of that user's sessions blocks on approval. + */ + PUSH: "push", } as const; export type Capability = (typeof CAPABILITIES)[keyof typeof CAPABILITIES]; @@ -126,6 +134,8 @@ export const LIMITS = { * governs concurrency at run time. */ COLLABORATION_ROLE_COUNT_MAX: 8, + /** Max device push-token length (`push.register`). Expo tokens are ~40 chars. */ + PUSH_TOKEN_MAX: 512, } as const; // ============================================================================= @@ -804,7 +814,9 @@ export type ClientMessage = | PipelinePackInstallMsg | PipelinePackRemoveMsg | PipelinePackTrustMsg - | PipelinePackSelectMsg; + | PipelinePackSelectMsg + | PushRegisterMsg + | PushUnregisterMsg; interface BaseClientMsg { /** Request ID for correlating responses */ @@ -1085,6 +1097,27 @@ export interface SessionApproveMsg extends BaseClientMsg { updatedInput?: Record; } +/** Device platform for a push registration. */ +export type PushPlatform = "ios" | "android"; + +/** + * Register a device to receive push notifications for the authenticated + * user's sessions. Scoped to the caller's identity (ZeroID `sub`) and tenant + * (account/project) server-side — a client can only register its own devices. + * `token` is the opaque transport token (an Expo push token today). + */ +export interface PushRegisterMsg extends BaseClientMsg { + type: "push.register"; + token: string; + platform: PushPlatform; +} + +/** Remove a previously-registered device token (e.g. on sign-out). */ +export interface PushUnregisterMsg extends BaseClientMsg { + type: "push.unregister"; + token: string; +} + /** * Answer a provider-initiated dialog (`session.ui_request`). Exactly one of * the payload fields applies per method: diff --git a/src/config.ts b/src/config.ts index 60c88429..3ce0e825 100644 --- a/src/config.ts +++ b/src/config.ts @@ -566,6 +566,24 @@ const PipelineSchema = z }) .default({ enabled: true, defaultPack: null, packs: [], registries: [] }); +/** + * Push notifications (docs/push.md). When a session blocks on a tool approval, + * the daemon sends a CONTENT-BLIND wake-up (an opaque session id only, never + * tool args) to the configured transport so the session owner's registered + * devices are alerted off-LAN. `transport: "none"` (default) disables push. + */ +const PushSchema = z + .object({ + /** Delivery transport. "expo" routes through Expo's push service (which + * relays to APNs/FCM); "none" disables push. A self-hosted content-blind + * relay transport swaps in behind this seam later. */ + transport: z.enum(["expo", "none"]).default("none"), + /** Expo access token (Bearer) for the push API. Optional — Expo accepts + * unauthenticated sends, but a token enables receipts + higher limits. */ + expoAccessToken: z.string().optional(), + }) + .default({ transport: "none" }); + const RootSchema = z.object({ daemonUrl: z.string().default("ws://127.0.0.1:7400"), dbPath: z.string().default("codeoid.db"), @@ -592,6 +610,7 @@ const RootSchema = z.object({ mcpServers: McpServersSchema, hooks: HooksSchema, embed: EmbedSchema, + push: PushSchema, fork: z .object({ /** Shell command run once in a freshly-created fork worktree to make it @@ -773,6 +792,14 @@ export interface CodeoidConfig { enabled: boolean; entries: HookEntryConfig[]; }; + /** + * Push notifications. Optional in the type so hand-built test configs stay + * minimal; loadConfig always populates it (schema default: transport "none"). + */ + push?: { + transport: "expo" | "none"; + expoAccessToken?: string; + }; /** * Embed trust — origins permitted to frame the web UI and pre-authenticate * it via the URL-hash credential handoff. The web UI's trusted-framing-origin @@ -846,6 +873,10 @@ const ENV_OVERRIDES: readonly EnvOverride[] = [ // Hooks kill switch — disable every configured hook per-invocation without // touching config.json. Entries themselves are file-config only. { env: "CODEOID_HOOKS_ENABLED", path: "hooks.enabled", kind: "boolean" }, + // Push transport switch + Expo token — per-invocation without touching + // config.json (e.g. CODEOID_PUSH_TRANSPORT=expo). + { env: "CODEOID_PUSH_TRANSPORT", path: "push.transport", kind: "string" }, + { env: "CODEOID_EXPO_ACCESS_TOKEN", path: "push.expoAccessToken", kind: "string" }, { env: "CODEOID_TURN_STALL_TIMEOUT_MS", path: "session.turnStallTimeoutMs", kind: "int" }, { env: "CODEOID_MCP_TOOL_TIMEOUT_MS", path: "session.mcpToolTimeoutMs", kind: "int" }, // Embed-SSO trusted framing origins (comma-separated). Each is an exact @@ -1048,6 +1079,7 @@ export function loadConfig(opts: LoadOptions = {}): CodeoidConfig { mcpServers: parsed.mcpServers, hooks: parsed.hooks, embed: parsed.embed, + push: parsed.push, fork: parsed.fork, }; } diff --git a/src/daemon/push/expo.ts b/src/daemon/push/expo.ts new file mode 100644 index 00000000..45856e2c --- /dev/null +++ b/src/daemon/push/expo.ts @@ -0,0 +1,65 @@ +/** + * Expo Push transport. Delivers via Expo's push service, which holds the + * APNs/FCM credentials and relays to the OS — so a self-hosted daemon never + * needs its own Apple/Google credentials for v1. + * + * CONTENT-BLIND: the alert copy is generic and `data` carries only the opaque + * sessionId + kind, so Expo (a third party in the delivery path) never sees a + * tool name, args, or description. The tradeoff Expo introduces — a third party + * seeing device tokens + timing — is why the self-hosted relay is the eventual + * end state; this transport keeps the payload content-blind so that migration + * changes nothing about what's exposed. + * + * Best-effort: logs and swallows failures so a push outage can never wedge the + * status-change path that fires it. + */ +import type { PushNotification, PushTarget, PushTransport } from "./types.js"; + +const EXPO_PUSH_URL = "https://exp.host/--/api/v2/push/send"; +/** Expo accepts at most 100 messages per request. */ +const BATCH_SIZE = 100; +const TIMEOUT_MS = 10_000; + +export class ExpoPushTransport implements PushTransport { + readonly name = "expo"; + + constructor(private readonly accessToken?: string) {} + + async send(targets: PushTarget[], note: PushNotification): Promise { + if (targets.length === 0) return; + const messages = targets.map((t) => ({ + to: t.token, + title: "codeoid", + body: "A session needs your approval", + // Content-blind: opaque ids only. The app fetches real context over its + // authenticated socket after the user taps. + data: { sessionId: note.sessionId, kind: note.kind }, + priority: "high", + // Collapse repeated pings for the same session on the device. + threadId: note.sessionId, + })); + for (let i = 0; i < messages.length; i += BATCH_SIZE) { + await this.#post(messages.slice(i, i + BATCH_SIZE)); + } + } + + async #post(batch: unknown[]): Promise { + try { + const res = await fetch(EXPO_PUSH_URL, { + method: "POST", + headers: { + "content-type": "application/json", + accept: "application/json", + ...(this.accessToken ? { authorization: `Bearer ${this.accessToken}` } : {}), + }, + body: JSON.stringify(batch), + signal: AbortSignal.timeout(TIMEOUT_MS), + }); + if (!res.ok) { + console.error(`[codeoid/push] Expo push HTTP ${res.status} ${res.statusText}`); + } + } catch (err) { + console.error("[codeoid/push] Expo push error:", err); + } + } +} diff --git a/src/daemon/push/index.ts b/src/daemon/push/index.ts new file mode 100644 index 00000000..fc70381d --- /dev/null +++ b/src/daemon/push/index.ts @@ -0,0 +1,4 @@ +export { PushService, createPushTransport } from "./service.js"; +export type { PushConfig, SessionOwner } from "./service.js"; +export { ExpoPushTransport } from "./expo.js"; +export type { PushNotification, PushTarget, PushTransport } from "./types.js"; diff --git a/src/daemon/push/push.test.ts b/src/daemon/push/push.test.ts new file mode 100644 index 00000000..58c83d69 --- /dev/null +++ b/src/daemon/push/push.test.ts @@ -0,0 +1,184 @@ +/** + * Push transport + service. The load-bearing property under test is + * CONTENT-BLINDNESS: the Expo payload must carry only an opaque session id + + * kind and generic copy — never a tool name, args, or description — so no + * session content leaves the daemon even through a third-party transport. + */ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { Store } from "../store.js"; +import { ExpoPushTransport } from "./expo.js"; +import { createPushTransport, PushService } from "./service.js"; +import type { PushNotification, PushTarget, PushTransport } from "./types.js"; + +describe("ExpoPushTransport", () => { + let calls: Array<{ url: string; init: RequestInit }>; + let origFetch: typeof fetch; + + beforeEach(() => { + calls = []; + origFetch = globalThis.fetch; + globalThis.fetch = (async (url: unknown, init: unknown) => { + calls.push({ url: String(url), init: init as RequestInit }); + return new Response(JSON.stringify({ data: [] }), { status: 200 }); + }) as typeof fetch; + }); + + afterEach(() => { + globalThis.fetch = origFetch; + }); + + test("payload is CONTENT-BLIND — opaque id + kind only, no tool details", async () => { + await new ExpoPushTransport().send( + [{ token: "ExponentPushToken[x]", platform: "ios" }], + { sessionId: "s-123", kind: "approval" }, + ); + expect(calls).toHaveLength(1); + expect(calls[0].url).toBe("https://exp.host/--/api/v2/push/send"); + const body = JSON.parse(calls[0].init.body as string) as Array>; + expect(body).toHaveLength(1); + const msg = body[0]; + expect(msg.to).toBe("ExponentPushToken[x]"); + expect(msg.title).toBe("codeoid"); // generic + expect(msg.data).toEqual({ sessionId: "s-123", kind: "approval" }); + // The ENTIRE data object must contain nothing but the two opaque fields. + expect(Object.keys(msg.data as object).sort()).toEqual(["kind", "sessionId"]); + }); + + test("Bearer auth is sent only when an access token is configured", async () => { + await new ExpoPushTransport("tok-abc").send( + [{ token: "t", platform: "android" }], + { sessionId: "s", kind: "approval" }, + ); + expect((calls[0].init.headers as Record).authorization).toBe("Bearer tok-abc"); + + calls.length = 0; + await new ExpoPushTransport().send( + [{ token: "t", platform: "android" }], + { sessionId: "s", kind: "approval" }, + ); + expect((calls[0].init.headers as Record).authorization).toBeUndefined(); + }); + + test("batches at 100 messages per request", async () => { + const targets: PushTarget[] = Array.from({ length: 250 }, (_, i) => ({ + token: `t${i}`, + platform: "ios" as const, + })); + await new ExpoPushTransport().send(targets, { sessionId: "s", kind: "approval" }); + expect(calls).toHaveLength(3); // 100 + 100 + 50 + }); + + test("no targets → no HTTP call", async () => { + await new ExpoPushTransport().send([], { sessionId: "s", kind: "approval" }); + expect(calls).toHaveLength(0); + }); + + test("swallows a non-2xx response (best-effort delivery)", async () => { + globalThis.fetch = (async () => new Response("bad", { status: 400 })) as unknown as typeof fetch; + await expect( + new ExpoPushTransport().send([{ token: "t", platform: "ios" }], { + sessionId: "s", + kind: "approval", + }), + ).resolves.toBeUndefined(); + }); + + test("swallows a network error", async () => { + globalThis.fetch = (async () => { + throw new Error("network down"); + }) as unknown as typeof fetch; + await expect( + new ExpoPushTransport().send([{ token: "t", platform: "ios" }], { + sessionId: "s", + kind: "approval", + }), + ).resolves.toBeUndefined(); + }); +}); + +describe("createPushTransport", () => { + test("undefined / none → noop transport", () => { + expect(createPushTransport(undefined).name).toBe("none"); + expect(createPushTransport({ transport: "none" }).name).toBe("none"); + }); + + test("expo → Expo transport", () => { + expect(createPushTransport({ transport: "expo" }).name).toBe("expo"); + }); +}); + +describe("PushService.notifyApproval", () => { + let tmp: string; + let store: Store; + + beforeEach(() => { + tmp = mkdtempSync(join(tmpdir(), "codeoid-push-svc-")); + store = new Store(join(tmp, "codeoid.db")); + }); + + afterEach(() => { + store.close(); + rmSync(tmp, { recursive: true, force: true }); + }); + + function fakeTransport() { + const sent: Array<{ targets: PushTarget[]; note: PushNotification }> = []; + const transport: PushTransport = { + name: "expo", + async send(targets, note) { + sent.push({ targets, note }); + }, + }; + return { transport, sent }; + } + + test("noop transport → disabled, delivers nothing", async () => { + const svc = new PushService(store, { name: "none", async send() {} }); + expect(svc.enabled).toBe(false); + store.registerPush("tok", "ios", "user:a", "acc", "proj"); + await svc.notifyApproval("s-1", { sub: "user:a", accountId: "acc", projectId: "proj" }); + // enabled=false short-circuits before any transport work — nothing to assert + // beyond not throwing. + }); + + test("delivers a content-blind note to the owner's devices only", async () => { + store.registerPush("tok-a", "ios", "user:a", "acc", "proj"); + store.registerPush("tok-b", "ios", "user:b", "acc", "proj"); // other owner + const { transport, sent } = fakeTransport(); + const svc = new PushService(store, transport); + expect(svc.enabled).toBe(true); + + await svc.notifyApproval("s-1", { sub: "user:a", accountId: "acc", projectId: "proj" }); + expect(sent).toHaveLength(1); + expect(sent[0].targets.map((t) => t.token)).toEqual(["tok-a"]); + expect(sent[0].note).toEqual({ sessionId: "s-1", kind: "approval" }); + }); + + test("no registered devices → no send", async () => { + const { transport, sent } = fakeTransport(); + await new PushService(store, transport).notifyApproval("s-1", { + sub: "nobody", + accountId: "acc", + projectId: "proj", + }); + expect(sent).toHaveLength(0); + }); + + test("swallows a store lookup failure", async () => { + const throwingStore = { + listPushForOwner() { + throw new Error("db gone"); + }, + } as unknown as Store; + const { transport, sent } = fakeTransport(); + await new PushService(throwingStore, transport).notifyApproval("s-1", { + sub: "u", + accountId: "a", + projectId: "p", + }); + expect(sent).toHaveLength(0); + }); +}); diff --git a/src/daemon/push/service.ts b/src/daemon/push/service.ts new file mode 100644 index 00000000..4fcc59c4 --- /dev/null +++ b/src/daemon/push/service.ts @@ -0,0 +1,68 @@ +/** + * Push service: resolves a blocked session's owner to their registered devices + * and delivers a content-blind wake-up through the configured transport. + * + * Routing key is the owner's ZeroID identity (`sub` == `sessions.created_by`), + * tenant-scoped by account/project — so only the human who owns the session is + * alerted, and never across tenants. + */ +import type { Store } from "../store.js"; +import { ExpoPushTransport } from "./expo.js"; +import type { PushNotification, PushTransport } from "./types.js"; + +/** Config shape this module needs (a subset of CodeoidConfig["push"]). */ +export interface PushConfig { + transport: "expo" | "none"; + expoAccessToken?: string; +} + +const noopTransport: PushTransport = { + name: "none", + async send() {}, +}; + +/** Build the transport for the daemon's push config. */ +export function createPushTransport(config: PushConfig | undefined): PushTransport { + if (!config || config.transport === "none") return noopTransport; + if (config.transport === "expo") return new ExpoPushTransport(config.expoAccessToken); + return noopTransport; +} + +/** Owner identity + tenant of a session — the push routing key. */ +export interface SessionOwner { + sub: string; + accountId: string; + projectId: string; +} + +export class PushService { + /** False for the noop transport, so callers can skip the store lookup entirely. */ + readonly enabled: boolean; + + constructor( + private readonly store: Store, + private readonly transport: PushTransport, + ) { + this.enabled = transport.name !== "none"; + } + + /** + * Deliver a content-blind "a session needs your approval" wake-up to the + * owner's devices. Best-effort — the transport swallows delivery failures; + * this wraps the store lookup so a DB hiccup can't escape onto the caller's + * status-change path either. + */ + async notifyApproval(sessionId: string, owner: SessionOwner): Promise { + if (!this.enabled) return; + let targets: Array<{ token: string; platform: "ios" | "android" }>; + try { + targets = this.store.listPushForOwner(owner.sub, owner.accountId, owner.projectId); + } catch (err) { + console.error("[codeoid/push] registration lookup failed:", err); + return; + } + if (targets.length === 0) return; + const note: PushNotification = { sessionId, kind: "approval" }; + await this.transport.send(targets, note); + } +} diff --git a/src/daemon/push/types.ts b/src/daemon/push/types.ts new file mode 100644 index 00000000..d19341be --- /dev/null +++ b/src/daemon/push/types.ts @@ -0,0 +1,42 @@ +/** + * Push notification transport seam. + * + * The whole point of this abstraction is content-blindness + swappability: the + * daemon hands a transport only opaque routing ids, so no session content ever + * leaves the box regardless of which transport is configured. Expo Push is the + * v1 transport (routes through Expo's service to APNs/FCM); a self-hosted + * content-blind relay (APNs .p8 + FCM + iOS Notification Service Extension + * poll-back) swaps in behind this same interface later — the migration is a + * transport change, not a redesign. + */ +import type { PushPlatform } from "../../protocol/types.js"; + +/** A device to deliver to. */ +export interface PushTarget { + token: string; + platform: PushPlatform; +} + +/** + * The CONTENT-BLIND payload a transport delivers. Deliberately carries only an + * opaque session id + a kind — never a tool name, args, or description. The app + * resolves human-readable context over its authenticated socket after the user + * taps. Even when the transport is a third party (Expo), no session content + * leaves the daemon. + */ +export interface PushNotification { + /** Opaque session id the user should open. */ + sessionId: string; + /** What happened. Only "approval" today (a session is blocked awaiting approval). */ + kind: "approval"; +} + +/** + * Delivery seam. Implementations map the content-blind PushNotification to + * their wire format and fan it out to the targets. Never throws — delivery is + * best-effort and runs off the daemon's status-change hot path. + */ +export interface PushTransport { + readonly name: string; + send(targets: PushTarget[], note: PushNotification): Promise; +} diff --git a/src/daemon/server.ts b/src/daemon/server.ts index db3d5ac6..f3dda0ff 100644 --- a/src/daemon/server.ts +++ b/src/daemon/server.ts @@ -157,6 +157,9 @@ export class DaemonServer { #mcpRegistry: McpRegistry | null = null; #mcpHub: McpHub | null = null; #bunServer: ReturnType | null = null; + /** Capabilities advertised on auth.ok — SERVER_CAPABILITIES plus PUSH when a + * push transport is configured, so clients can feature-detect push support. */ + #advertisedCapabilities: string[]; #sockets = new Map(); #frontends: Frontend[] = []; #httpHandlers: Array<(req: IncomingMessage, res: ServerResponse) => boolean> = []; @@ -164,6 +167,10 @@ export class DaemonServer { constructor(config: DaemonConfig) { this.#config = config; + const pushOn = (config.fullConfig?.push?.transport ?? "none") !== "none"; + this.#advertisedCapabilities = pushOn + ? [...SERVER_CAPABILITIES, CAPABILITIES.PUSH] + : SERVER_CAPABILITIES; this.#store = new Store(config.dbPath); this.#transcriptStore = new TranscriptStore(config.transcriptDir); this.#shutdown = new ShutdownManager(); @@ -565,7 +572,7 @@ export class DaemonServer { }, scopes: data.auth.scopes, protocolVersion: PROTOCOL_VERSION, - capabilities: SERVER_CAPABILITIES, + capabilities: self.#advertisedCapabilities, // Registered backends, default first — feeds the new-session // provider picker (see AuthOkMsg.providers). providers: self.#manager.providerIds(), diff --git a/src/daemon/session-manager.ts b/src/daemon/session-manager.ts index 6c314297..e7a60ba0 100644 --- a/src/daemon/session-manager.ts +++ b/src/daemon/session-manager.ts @@ -19,6 +19,7 @@ import { } from "./providers/registry.js"; import type { HookBus } from "./hooks/bus.js"; import type { Store } from "./store.js"; +import { createPushTransport, PushService } from "./push/index.js"; import { hasScope, SCOPES } from "../protocol/scopes.js"; import { applyPatches, getManifest, getSnapshot } from "./settings/store.js"; import { RateLimiter } from "./rate-limit.js"; @@ -204,6 +205,9 @@ export class SessionManager { #config?: CodeoidConfig; #compressionRegistry?: CompressionRegistry; #dispatcher: Dispatcher; + /** Content-blind push notifications — resolves a blocked session's owner to + * their registered devices. Noop transport when push is disabled (default). */ + #pushService: PushService; /** SDLC pipeline manager — undefined when the pipeline is disabled (default). */ #pipelines?: PipelineManager; /** Pack curation surface (registries + install/trust/select) — always present, @@ -221,6 +225,22 @@ export class SessionManager { /** Stable observer identity — every Session reports status transitions here. */ #statusObserver = (sessionId: string, status: SessionInfo["status"]): void => { this.#dispatcher.onSessionStatus(sessionId, status); + // Content-blind push: a session that just blocked on approval alerts its + // owner's registered devices off-LAN. Fire-and-forget — a push hiccup must + // never touch the status path (the emit itself only ever sees an opaque + // session id, never the tool's args/description). + if (status === "waiting_approval" && this.#pushService.enabled) { + const session = this.#sessions.get(sessionId); + if (session) { + void this.#pushService + .notifyApproval(sessionId, { + sub: session.createdBy, + accountId: session.accountId, + projectId: session.projectId, + }) + .catch((err) => console.error("[codeoid/push] notify failed:", err)); + } + } // A pipeline phase driving this session awaits its next rest; resolve it. // (Non-run sessions never have a waiter, so this is a no-op for them.) // @@ -281,6 +301,7 @@ export class SessionManager { this.#makeDispatcherHost(), opts?.config?.dispatch, ); + this.#pushService = new PushService(store, createPushTransport(opts?.config?.push)); // SDLC pipeline (docs/sdlc-pipeline.md) — off by default; when enabled, the // manager shares the daemon DB (one connection) and rehydrates non-terminal // pipelines on construction (resume). The runner drives prompt/slash phases @@ -655,6 +676,10 @@ mcpHub: this.#mcpHub, return this.#packTrust(msg, auth); case "pipeline.pack.select": return this.#packSelect(msg, auth); + case "push.register": + return this.#pushRegister(msg, auth); + case "push.unregister": + return this.#pushUnregister(msg, auth); default: { // Inbound messages are cast from raw JSON at the transport, so an // unknown/malformed `type` reaches here. Without this the function @@ -2853,6 +2878,27 @@ mcpHub: this.#mcpHub, return { type: "response.ok", requestId: msg.id }; } + #pushRegister( + msg: Extract, + auth: AuthContext, + ): DaemonMessage { + // No scope gate: registration is inherently self-scoped — the token is + // bound to the caller's own identity (auth.sub) + tenant, so a client can + // only ever register its own device. Gated by authentication + the PUSH + // capability, mirroring how session.ui_response avoids a bespoke scope. + this.#store.registerPush(msg.token, msg.platform, auth.sub, auth.accountId, auth.projectId); + return { type: "response.ok", requestId: msg.id }; + } + + #pushUnregister( + msg: Extract, + auth: AuthContext, + ): DaemonMessage { + // Owner-scoped delete — a client can only remove its own device token. + this.#store.unregisterPush(msg.token, auth.sub); + return { type: "response.ok", requestId: msg.id }; + } + /** * Cross-session search — fans out to the memory engine, groups by * session, and returns a ranked list with evidence snippets. Requires diff --git a/src/daemon/store.ts b/src/daemon/store.ts index 4fb12c7f..466d2b78 100644 --- a/src/daemon/store.ts +++ b/src/daemon/store.ts @@ -5,7 +5,7 @@ */ import { Database } from "bun:sqlite"; -import type { ModelInfo, SessionInfo, SessionStatus } from "../protocol/types.js"; +import type { ModelInfo, PushPlatform, SessionInfo, SessionStatus } from "../protocol/types.js"; // ── Dispatch queue types (P4) ───────────────────────────────────────────── @@ -197,6 +197,23 @@ export class Store { ); CREATE INDEX IF NOT EXISTS idx_session_pins_session ON session_pins(session_id); + -- Device push registrations, keyed by the owner's ZeroID identity + -- (owner_sub == sessions.created_by == AuthContext.sub), tenant-scoped by + -- account/project. NOT session-scoped: a device registers once and is + -- notified for any of that owner's sessions that block on approval. The + -- token is the opaque transport token (an Expo push token today). + CREATE TABLE IF NOT EXISTS push_registrations ( + token TEXT PRIMARY KEY, + platform TEXT NOT NULL, + owner_sub TEXT NOT NULL, + account_id TEXT NOT NULL, + project_id TEXT NOT NULL, + created_at INTEGER NOT NULL, + last_seen INTEGER + ); + CREATE INDEX IF NOT EXISTS idx_push_reg_owner + ON push_registrations(owner_sub, account_id, project_id); + -- Last live model catalog per provider (claude, gemini, openai, ...), -- as reported by that provider's backend. Served as the models.list -- fallback on boots where no session has run a turn yet, so the picker @@ -346,6 +363,61 @@ export class Store { return rows.map((r) => r.file_path); } + // ── Push registrations ──────────────────────────────────────────────── + + /** + * Register (or refresh) a device token for an owner. Re-registering the same + * token updates its owner/tenant/last_seen — a physical device maps to one + * token, and a token can only belong to whoever last registered it. + */ + registerPush( + token: string, + platform: PushPlatform, + ownerSub: string, + accountId: string, + projectId: string, + ): void { + const now = Date.now(); + this.#db + .prepare( + `INSERT INTO push_registrations + (token, platform, owner_sub, account_id, project_id, created_at, last_seen) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(token) DO UPDATE SET + platform = excluded.platform, + owner_sub = excluded.owner_sub, + account_id = excluded.account_id, + project_id = excluded.project_id, + last_seen = excluded.last_seen`, + ) + .run(token, platform, ownerSub, accountId, projectId, now, now); + } + + /** + * Remove a device token — owner-scoped so a client can only unregister its + * own device, never another user's token. + */ + unregisterPush(token: string, ownerSub: string): void { + this.#db + .prepare("DELETE FROM push_registrations WHERE token = ? AND owner_sub = ?") + .run(token, ownerSub); + } + + /** Device tokens to notify for a given owner within a tenant. */ + listPushForOwner( + ownerSub: string, + accountId: string, + projectId: string, + ): Array<{ token: string; platform: PushPlatform }> { + const rows = this.#db + .prepare( + `SELECT token, platform FROM push_registrations + WHERE owner_sub = ? AND account_id = ? AND project_id = ?`, + ) + .all(ownerSub, accountId, projectId) as Array<{ token: string; platform: PushPlatform }>; + return rows; + } + // ── Sessions ────────────────────────────────────────────────────────── createSession(session: SessionInfo & { accountId: string; projectId: string }): void { diff --git a/src/tests/push-emit.test.ts b/src/tests/push-emit.test.ts new file mode 100644 index 00000000..1d3b4a74 --- /dev/null +++ b/src/tests/push-emit.test.ts @@ -0,0 +1,192 @@ +/** + * End-to-end: a manager-owned session that blocks on a tool approval fires a + * CONTENT-BLIND push to the owner's registered device. Drives a real Session + * (MockSessionProvider) through SessionManager so the daemon-wide status + * observer runs, and captures the outbound Expo POST to assert content-blindness. + */ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdirSync, mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { CodeoidConfig } from "../config.js"; +import { MockSessionProvider } from "../daemon/providers/mock/session-provider.js"; +import { SessionManager } from "../daemon/session-manager.js"; +import { Store } from "../daemon/store.js"; +import { TranscriptStore } from "../daemon/transcript.js"; +import type { ProviderEvent } from "../daemon/providers/interface.js"; +import { ALL_SCOPES } from "../protocol/scopes.js"; +import type { AuthContext, SessionInfo } from "../protocol/types.js"; + +const AUTH: AuthContext = { + sub: "user:push", + scopes: [...ALL_SCOPES] as AuthContext["scopes"], + delegationDepth: 0, + accountId: "acc", + projectId: "proj", +}; +const CLIENT = { id: "c", auth: AUTH, send: () => {} }; + +/** One scripted turn: call a hard-gated fleet tool (requires approval in ANY + * mode), then finish — so the session parks at waiting_approval. */ +function fleetSendTurn(): ProviderEvent[] { + return [ + { + type: "tool_start", + toolId: "t1", + sdkToolUseId: "sdk-t1", + name: "mcp__codeoid_fleet__fleet_send", + input: { session: "x", message: "go" }, + approvalId: "approval-1", + } as ProviderEvent, + { + type: "turn_done", + result: { + providerId: "mock", + model: "mock-model", + inputTokens: 1, + outputTokens: 1, + cacheReadTokens: 0, + cacheCreationTokens: 0, + totalCostUsd: 0, + durationMs: 1, + }, + } as ProviderEvent, + ]; +} + +function mkConfig(dbPath: string, transcriptDir: string, transport: "expo" | "none"): CodeoidConfig { + return { + daemonUrl: "ws://127.0.0.1:7400", + dbPath, + transcriptDir, + auth: { baseUrl: "http://localhost:8899" }, + zeroidUrl: "http://localhost:8899", + workspaceIndex: { enabled: false, episodeThreshold: 5, timeThresholdMs: 60_000, debounceMs: 15_000 }, + compress: { enabled: false, excludeCommands: [], excludePatterns: [], compressPipes: false, minBytes: 1024 }, + labeling: {}, + telemetry: { osc8: "auto" }, + autoRotate: { + enabled: false, + warnPct: 0.6, + rotatePct: 0.8, + hardRotatePct: 0.9, + minTurnsBeforeRotate: 3, + strategy: "task-anchor", + }, + session: {}, + conductor: { enabled: false, name: "conductor", provider: "claude" }, + dispatch: { + enabled: false, + tickMs: 999_999, + leaseMs: 60_000, + failureLimit: 2, + maxConcurrentWorkers: 2, + workerToolBudget: 7, + retryBaseMs: 0, + }, + pipeline: { enabled: false, defaultPack: null, packs: [] }, + push: { transport }, + }; +} + +let tmp: string; +let workdir: string; +let store: Store; +let transcript: TranscriptStore; +let manager: SessionManager; +let pushCalls: Array[]>; +let origFetch: typeof fetch; + +function setup(transport: "expo" | "none") { + tmp = mkdtempSync(join(tmpdir(), "codeoid-push-emit-")); + workdir = join(tmp, "repo"); + mkdirSync(workdir, { recursive: true }); + store = new Store(join(tmp, "codeoid.db")); + transcript = new TranscriptStore(join(tmp, "transcripts")); + manager = new SessionManager(store, transcript, undefined, undefined, undefined, { + config: mkConfig(join(tmp, "codeoid.db"), join(tmp, "transcripts"), transport), + _testProviderFactory: () => new MockSessionProvider("mock", [fleetSendTurn()]), + }); +} + +beforeEach(() => { + pushCalls = []; + origFetch = globalThis.fetch; + globalThis.fetch = (async (url: unknown, init: unknown) => { + if (String(url).includes("exp.host")) { + pushCalls.push(JSON.parse((init as RequestInit).body as string)); + } + return new Response(JSON.stringify({ data: [] }), { status: 200 }); + }) as typeof fetch; +}); + +afterEach(async () => { + globalThis.fetch = origFetch; + try { + await manager.drain(3_000); + } catch { + // best-effort + } + try { + await transcript.flush(); + } catch { + // best-effort + } + store.close(); + rmSync(tmp, { recursive: true, force: true }); +}); + +async function until(cond: () => boolean, ms = 3_000): Promise { + const deadline = Date.now() + ms; + while (!cond()) { + if (Date.now() > deadline) throw new Error("condition not reached"); + await new Promise((r) => setTimeout(r, 5)); + } +} + +async function createAndSend(): Promise { + const created = await manager.handle( + { type: "session.create", id: "req", name: "blocker", workdir }, + AUTH, + CLIENT, + ); + const sessionId = (created as { data: SessionInfo }).data.id; + await manager.handle({ type: "session.send", id: "s1", sessionId, text: "go" }, AUTH, CLIENT); + return sessionId; +} + +describe("push emit on waiting_approval", () => { + test("a blocked session pushes a content-blind wake-up to the owner's device", async () => { + setup("expo"); + store.registerPush("ExponentPushToken[dev]", "ios", AUTH.sub, AUTH.accountId, AUTH.projectId); + + const sessionId = await createAndSend(); + await until(() => pushCalls.length > 0); + + const batch = pushCalls[0]; + expect(batch).toHaveLength(1); + const msg = batch[0]; + expect(msg.to).toBe("ExponentPushToken[dev]"); + // Content-blind: the opaque session id + kind, nothing else. + expect(msg.data).toEqual({ sessionId, kind: "approval" }); + // The tool name that triggered the block must NOT appear anywhere. + expect(JSON.stringify(msg)).not.toContain("fleet_send"); + }); + + test("no push when the owner has no registered devices", async () => { + setup("expo"); + // No registerPush — a block should look up zero targets and send nothing. + await createAndSend(); + // Give the turn time to reach waiting_approval, then confirm no push fired. + await new Promise((r) => setTimeout(r, 150)); + expect(pushCalls).toHaveLength(0); + }); + + test("no push when the transport is disabled (transport: none)", async () => { + setup("none"); + store.registerPush("ExponentPushToken[dev]", "ios", AUTH.sub, AUTH.accountId, AUTH.projectId); + await createAndSend(); + await new Promise((r) => setTimeout(r, 150)); + expect(pushCalls).toHaveLength(0); + }); +}); diff --git a/src/tests/push-handlers.test.ts b/src/tests/push-handlers.test.ts new file mode 100644 index 00000000..87dd5531 --- /dev/null +++ b/src/tests/push-handlers.test.ts @@ -0,0 +1,77 @@ +/** + * push.register / push.unregister routed through the real SessionManager.handle() + * — persistence keyed to the CALLER's identity (never a body-supplied owner) and + * owner-scoped delete, over the wire. + */ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { SessionManager } from "../daemon/session-manager.js"; +import { Store } from "../daemon/store.js"; +import { TranscriptStore } from "../daemon/transcript.js"; +import { ALL_SCOPES } from "../protocol/scopes.js"; +import type { AuthContext } from "../protocol/types.js"; + +const AUTH: AuthContext = { + sub: "user:t", + scopes: [...ALL_SCOPES] as AuthContext["scopes"], + delegationDepth: 0, + accountId: "acc", + projectId: "proj", +}; +const CLIENT = { id: "c", auth: AUTH, send: () => {} }; + +let tmp: string; +let store: Store; +let transcript: TranscriptStore; +let manager: SessionManager; + +beforeEach(() => { + tmp = mkdtempSync(join(tmpdir(), "codeoid-push-h-")); + store = new Store(join(tmp, "codeoid.db")); + transcript = new TranscriptStore(join(tmp, "transcripts")); + manager = new SessionManager(store, transcript, undefined, undefined, undefined, {}); +}); + +afterEach(async () => { + try { + await manager.drain(2_000); + } catch { + // best-effort + } + store.close(); + rmSync(tmp, { recursive: true, force: true }); +}); + +describe("push.register / push.unregister handlers", () => { + test("push.register persists the token under the CALLER's identity + tenant", async () => { + const r = await manager.handle( + { type: "push.register", id: "1", token: "ExponentPushToken[z]", platform: "ios" }, + AUTH, + CLIENT, + ); + expect(r.type).toBe("response.ok"); + expect(store.listPushForOwner("user:t", "acc", "proj")).toEqual([ + { token: "ExponentPushToken[z]", platform: "ios" }, + ]); + // Not visible to another tenant. + expect(store.listPushForOwner("user:t", "other", "proj")).toEqual([]); + }); + + test("push.unregister removes only the caller's own token", async () => { + await manager.handle( + { type: "push.register", id: "1", token: "tok", platform: "android" }, + AUTH, + CLIENT, + ); + // A different user owns a different token in the same tenant. + store.registerPush("tok-other", "ios", "user:other", "acc", "proj"); + + const r = await manager.handle({ type: "push.unregister", id: "2", token: "tok" }, AUTH, CLIENT); + expect(r.type).toBe("response.ok"); + expect(store.listPushForOwner("user:t", "acc", "proj")).toEqual([]); + // The other user's token is untouched. + expect(store.listPushForOwner("user:other", "acc", "proj")).toHaveLength(1); + }); +}); diff --git a/src/tests/push-store.test.ts b/src/tests/push-store.test.ts new file mode 100644 index 00000000..7fd70067 --- /dev/null +++ b/src/tests/push-store.test.ts @@ -0,0 +1,76 @@ +/** + * push_registrations persistence — owner-scoped device tokens keyed by the + * ZeroID identity (sub == sessions.created_by), tenant-scoped by account/project. + * The guarantees the push routing relies on: register/refresh, tenant + owner + * isolation on lookup, and owner-scoped delete. + */ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { Store } from "../daemon/store.js"; + +let tmp: string; +let store: Store; + +beforeEach(() => { + tmp = mkdtempSync(join(tmpdir(), "codeoid-push-store-")); + store = new Store(join(tmp, "codeoid.db")); +}); + +afterEach(() => { + store.close(); + rmSync(tmp, { recursive: true, force: true }); +}); + +describe("push_registrations store", () => { + test("register then list, scoped to owner + tenant", () => { + store.registerPush("tok-a", "ios", "user:a", "acc", "proj"); + expect(store.listPushForOwner("user:a", "acc", "proj")).toEqual([ + { token: "tok-a", platform: "ios" }, + ]); + }); + + test("lookup is isolated by owner AND by tenant", () => { + store.registerPush("tok-a", "ios", "user:a", "acc", "proj"); + store.registerPush("tok-b", "android", "user:b", "acc", "proj"); // other owner + store.registerPush("tok-a2", "ios", "user:a", "acc2", "proj"); // other account + store.registerPush("tok-a3", "ios", "user:a", "acc", "proj2"); // other project + + expect(store.listPushForOwner("user:a", "acc", "proj")).toEqual([ + { token: "tok-a", platform: "ios" }, + ]); + expect(store.listPushForOwner("user:b", "acc", "proj")).toEqual([ + { token: "tok-b", platform: "android" }, + ]); + expect(store.listPushForOwner("user:c", "acc", "proj")).toEqual([]); + }); + + test("multiple devices for one owner all come back", () => { + store.registerPush("tok-1", "ios", "user:a", "acc", "proj"); + store.registerPush("tok-2", "android", "user:a", "acc", "proj"); + const tokens = store.listPushForOwner("user:a", "acc", "proj").map((r) => r.token).sort(); + expect(tokens).toEqual(["tok-1", "tok-2"]); + }); + + test("re-registering the same token updates owner/platform (no duplicate row)", () => { + store.registerPush("tok-x", "ios", "user:a", "acc", "proj"); + // Same physical device, now owned by user:b (e.g. a shared device re-signed-in). + store.registerPush("tok-x", "android", "user:b", "acc", "proj"); + expect(store.listPushForOwner("user:a", "acc", "proj")).toEqual([]); + expect(store.listPushForOwner("user:b", "acc", "proj")).toEqual([ + { token: "tok-x", platform: "android" }, + ]); + }); + + test("unregister is owner-scoped — cannot delete another user's token", () => { + store.registerPush("tok-a", "ios", "user:a", "acc", "proj"); + store.registerPush("tok-b", "ios", "user:b", "acc", "proj"); + // user:a attempts to unregister user:b's token — must be a no-op. + store.unregisterPush("tok-b", "user:a"); + expect(store.listPushForOwner("user:b", "acc", "proj")).toHaveLength(1); + // user:a unregisters its own token — removed. + store.unregisterPush("tok-a", "user:a"); + expect(store.listPushForOwner("user:a", "acc", "proj")).toEqual([]); + }); +});