feat(scheduler): pipeline notify — ctx.notify for proactive channel delivery

.ai/context.md

@@ -349,8 +349,19 @@ canvas-QR card and `vesper connections pair`. **WhatsApp-Web (personal account)*
 `@vesper/channel-whatsapp-web` package — the SOLE runtime dependency (Baileys; see the opt-in carve-out
 under Stack), lazy-registered by the daemon, with rotating-QR pairing into a vault-backed session.
 
+**Pipeline notify (`ctx.notify`) SHIPPED.** The outbound, pipeline-initiated complement to the inbound
+chatbot flow (`specs/pipeline-notify.md`): a running pipeline can push a notification to the user out a
+connected channel. `PipelineContext.notify(text, opts?)` is gated by `NETWORK_FETCH` and backed by a
+`NotifyFn` injected through `BuildContextDeps` + `SchedulerOptions` exactly where `complete` is threaded
+(top-level + sub-agent). Core stays decoupled (`channel?: string`, not the connections `ChannelId`); the
+host `makeNotifyFn` (CLI) resolves the channel + the pairing-persisted owner `defaultChatId`, sends
+through the daemon's already-authenticated running handler, and audits each send on the `events` table
+(`notification_sent`/`notification_failed`, body/chat id never logged) — no migration, no new capability,
+no new dependency. A missing channel/destination/resolver is graceful (`{delivered:false, reason}`); only
+a capability violation throws. Issue-capped: the record is the spec + `cycle-log.md` + the commit (Rule 11).
+
 **Agent docs** — single-source `.ai/` drives Claude Code, opencode, Codex, Gemini, and Cursor via
-`bun run sync:ai` (`scripts/sync-ai-docs.ts`). Suite: **870 tests / 0 fail**; Biome clean; no
+`bun run sync:ai` (`scripts/sync-ai-docs.ts`). Suite: **890 tests / 0 fail**; Biome clean; no
 provider SDKs (the lone runtime dep is the isolated, opt-in Baileys in `@vesper/channel-whatsapp-web`).
 
 **Next:** the Vesper World UI redesign (Omar dislikes the current look — a design prompt is in hand);

.ai/generated/rules.mdc

@@ -357,8 +357,19 @@ canvas-QR card and `vesper connections pair`. **WhatsApp-Web (personal account)*
 `@vesper/channel-whatsapp-web` package — the SOLE runtime dependency (Baileys; see the opt-in carve-out
 under Stack), lazy-registered by the daemon, with rotating-QR pairing into a vault-backed session.
 
+**Pipeline notify (`ctx.notify`) SHIPPED.** The outbound, pipeline-initiated complement to the inbound
+chatbot flow (`specs/pipeline-notify.md`): a running pipeline can push a notification to the user out a
+connected channel. `PipelineContext.notify(text, opts?)` is gated by `NETWORK_FETCH` and backed by a
+`NotifyFn` injected through `BuildContextDeps` + `SchedulerOptions` exactly where `complete` is threaded
+(top-level + sub-agent). Core stays decoupled (`channel?: string`, not the connections `ChannelId`); the
+host `makeNotifyFn` (CLI) resolves the channel + the pairing-persisted owner `defaultChatId`, sends
+through the daemon's already-authenticated running handler, and audits each send on the `events` table
+(`notification_sent`/`notification_failed`, body/chat id never logged) — no migration, no new capability,
+no new dependency. A missing channel/destination/resolver is graceful (`{delivered:false, reason}`); only
+a capability violation throws. Issue-capped: the record is the spec + `cycle-log.md` + the commit (Rule 11).
+
 **Agent docs** — single-source `.ai/` drives Claude Code, opencode, Codex, Gemini, and Cursor via
-`bun run sync:ai` (`scripts/sync-ai-docs.ts`). Suite: **870 tests / 0 fail**; Biome clean; no
+`bun run sync:ai` (`scripts/sync-ai-docs.ts`). Suite: **890 tests / 0 fail**; Biome clean; no
 provider SDKs (the lone runtime dep is the isolated, opt-in Baileys in `@vesper/channel-whatsapp-web`).
 
 **Next:** the Vesper World UI redesign (Omar dislikes the current look — a design prompt is in hand);

AGENTS.md

@@ -351,8 +351,19 @@ canvas-QR card and `vesper connections pair`. **WhatsApp-Web (personal account)*
 `@vesper/channel-whatsapp-web` package — the SOLE runtime dependency (Baileys; see the opt-in carve-out
 under Stack), lazy-registered by the daemon, with rotating-QR pairing into a vault-backed session.
 
+**Pipeline notify (`ctx.notify`) SHIPPED.** The outbound, pipeline-initiated complement to the inbound
+chatbot flow (`specs/pipeline-notify.md`): a running pipeline can push a notification to the user out a
+connected channel. `PipelineContext.notify(text, opts?)` is gated by `NETWORK_FETCH` and backed by a
+`NotifyFn` injected through `BuildContextDeps` + `SchedulerOptions` exactly where `complete` is threaded
+(top-level + sub-agent). Core stays decoupled (`channel?: string`, not the connections `ChannelId`); the
+host `makeNotifyFn` (CLI) resolves the channel + the pairing-persisted owner `defaultChatId`, sends
+through the daemon's already-authenticated running handler, and audits each send on the `events` table
+(`notification_sent`/`notification_failed`, body/chat id never logged) — no migration, no new capability,
+no new dependency. A missing channel/destination/resolver is graceful (`{delivered:false, reason}`); only
+a capability violation throws. Issue-capped: the record is the spec + `cycle-log.md` + the commit (Rule 11).
+
 **Agent docs** — single-source `.ai/` drives Claude Code, opencode, Codex, Gemini, and Cursor via
-`bun run sync:ai` (`scripts/sync-ai-docs.ts`). Suite: **870 tests / 0 fail**; Biome clean; no
+`bun run sync:ai` (`scripts/sync-ai-docs.ts`). Suite: **890 tests / 0 fail**; Biome clean; no
 provider SDKs (the lone runtime dep is the isolated, opt-in Baileys in `@vesper/channel-whatsapp-web`).
 
 **Next:** the Vesper World UI redesign (Omar dislikes the current look — a design prompt is in hand);

cycle-log.md

@@ -969,3 +969,46 @@ Backend->Client->Review workflow; the review's 2 real HIGH gaps were then fixed
   (the plugin is registered only in the daemon — the UI/daemon is the source of truth; the CLI doesn't load Baileys
   for a list); the compiled `vesper-desktop` binary omits whatsapp-web (dynamic import not bundled) until Launch wires
   it; re-pairing an already-live whatsapp-web opens a second socket (rare edge). Signal (signal-cli) still open.
+
+## Pipeline notify (`ctx.notify` — proactive channel delivery) — SHIPPED
+- The outbound, pipeline-initiated complement to the shipped inbound `ChatSink` flow. A running pipeline can now
+  push a notification to the user out a connected channel. `OutboundIntent.kind:"notify"` + `ChannelHandler.send`
+  already existed (used only by the operator `vesper connections send`); the gap was a pipeline-facing seam.
+  Spec: `specs/pipeline-notify.md`. Issue-capped: this entry + the commit are the record (Rule 11). Omar approved
+  SPEC + PLAN at the advancement gates; chose graceful-degradation + reuse-`NETWORK_FETCH` over throw + new cap.
+- DESIGN (mirror `complete`, stay decoupled): `ctx.notify(text, opts?)` on `PipelineContext`, gated by
+  `NETWORK_FETCH` (the egress cap `send` already needs). A `NotifyFn` is injected through `BuildContextDeps` +
+  `SchedulerOptions` exactly where `complete` is threaded (top-level run AND the `subagent.ts` child context).
+  KEY DECISION: the core `NotifyIntent`/`NotifyOutcome` use `channel?: string`, NOT the connections `ChannelId`
+  union — so `vesper-core/scheduler` keeps ZERO dependency on the connections feature layer (the import is
+  cycle-safe either way; decoupling is the better architecture). The host (`makeNotifyFn`, CLI) owns channel
+  identity. DIVERGENCE from `complete`: a missing resolver is GRACEFUL (`{delivered:false, reason:"unavailable"}`),
+  never throws — a side-channel must not crash a pipeline; only a capability violation throws.
+- HOST RESOLUTION (`packages/vesper-cli/src/make-notify.ts`): channel = explicit `intent.channel` (must be running)
+  -> `config.notify.defaultChannel` (if running) -> first running channel with a paired owner. chatId = explicit
+  -> `config.connections.<id>.params.defaultChatId` (the destination scan-to-connect ALREADY persists at pairing,
+  `pairing-coordinator.ts#persistLinked`) — so a pipeline never handles a chat id. Sends through the daemon's
+  ALREADY-AUTHENTICATED running handler (`registry.list().find`), never a fresh handler (that stays the operator
+  `sendVia` path). Audits every actual send attempt on the `events` table (`notification_sent`/`notification_failed`,
+  reusing `recordConnectionEvent`, which strips `text`/body) — NO migration, payload is `{channel}` only (never the
+  body or chat id; a test asserts neither serializes).
+- DAEMON WIRING: the Scheduler is constructed BEFORE `buildChannelRegistry`, so `makeNotifyFn` late-binds the
+  registry through a `getRegistry: () => channelRegistry` getter read only at notify time (`channelRegistry` is a
+  `let` assigned right after the registry builds). Avoided reordering the whole startup; `uiStore` was moved a few
+  lines up so it can be the notify-audit sink passed into the constructor.
+- SPEC DELTA (the one deviation): the spec's acceptance said `normalizeNotify` "SHALL surface a dropped-record
+  warning". The codebase has NO warnings channel in `config.ts` — `normalizePresence`/`normalizeConnection` all
+  SILENTLY drop malformed input. Matched that precedent (drop, never throw) rather than invent a one-off warning
+  path; behavior is otherwise identical (unknown/non-string `defaultChannel` dropped). Reconcile the contract
+  wording if a warnings channel is ever added.
+- GOTCHA: adding `notify` to the `PipelineContext` interface broke 5 hand-rolled context mocks in pipeline +
+  subagent tests (tsc: "Property 'notify' is missing") — they had no notify stub. Fixed with a one-line
+  `notify: async () => ({ delivered:false })` per mock. A reminder that widening a core interface ripples into
+  every hand-rolled test double; a shared `fakeContext` factory would localize this (follow-up).
+- Verified: 890 tests / 0 fail (+20: 5 context + 2 scheduler-context + 4 config + 9 make-notify); 100% line+func
+  coverage on the two new units; biome clean (exit 0); tsc adds 0 NEW errors (the 5 mock errors fixed; pre-existing
+  exactOptional/`as`-cast errors in unchanged code remain, CI skips tsc); NO new dependency; NO migration; NO new
+  capability; transport mocked end-to-end (suite sends to nothing). NOT exercised against a live channel.
+- FOLLOW-UPS: rate-limiting/anti-spam on notifications (declared out-of-scope; every send is audited so abuse is
+  visible); rich/structured messages (plain text only in v1); a shared `fakeContext` test factory; downstream
+  consumers can now wire delivery (`pipeline-career.md`, `pipeline-secretary.md`) onto `ctx.notify`.

packages/pipelines/orchestrator-demo/handler.test.ts

@@ -130,6 +130,7 @@ describe("orchestrator-demo pipeline", () => {
       readSignals: () => {
         throw new Error("unused");
       },
+      notify: async () => ({ delivered: false }),
     });
 
     expect(recorded).toEqual([{ status: "ok", summary: "research complete" }]);

packages/pipelines/router/handler.test.ts

@@ -101,6 +101,9 @@ function makeFakeContext(options: {
     readSignals() {
       throw new Error("readSignals is not supported in this fake context");
     },
+    async notify() {
+      return { delivered: false };
+    },
   };
 
   return { ctx, completePrompts, spawned, recordedRuns, progress };

packages/pipelines/selftest/handler.test.ts

@@ -84,6 +84,9 @@ function makeFakeContext(options: {
     readSignals() {
       throw new Error("readSignals is not supported in this fake context");
     },
+    async notify() {
+      return { delivered: false };
+    },
   };
 
   return { ctx, completeCalls, recordedRuns };

packages/pipelines/skill-train/handler.test.ts

@@ -47,6 +47,9 @@ function makeCtx(params: Record<string, unknown>): {
     readSignals() {
       throw new Error("readSignals is not supported in this fake context");
     },
+    async notify() {
+      return { delivered: false };
+    },
   };
   return { ctx, recorded };
 }

packages/vesper-cli/src/commands/daemon-run.ts

@@ -2,6 +2,7 @@ import { Database } from "bun:sqlite";
 import { mkdir } from "node:fs/promises";
 import {
   ApprovalTokenStore,
+  type ChannelRegistry,
   channelStates,
   DEFAULT_AGENT_MATCHERS,
   detectAvailableCLIs,
@@ -19,6 +20,7 @@ import { loadConfig, saveConfig } from "../config.ts";
 import { buildChannelRegistry, makeChannelSink } from "../connections-wiring.ts";
 import { removePidFile, resolveDaemonState, writePidFile } from "../daemon-lifecycle.ts";
 import type { Command } from "../dispatch.ts";
+import { makeNotifyFn } from "../make-notify.ts";
 import { loadOptionalChannels } from "../optional-channels.ts";
 import { PairingCoordinator } from "../pairing-coordinator.ts";
 import { dbPath, pidPath, runDir, socketPath, uiPort } from "../paths.ts";
@@ -62,20 +64,29 @@ export const daemonRunCommand: Command = {
     const installed = await detectAvailableCLIs();
     const complete = makeCompleteFn(config, installed);
 
+    // The UI store also serves the router's editable template default_params (#4)
+    // and is the audit sink for `ctx.notify`; opened before the scheduler so the
+    // notify resolver can be wired into the constructor.
+    const uiStore = openStore(dbPath());
+
+    // `ctx.notify` resolver: delivers a pipeline notification out a connected
+    // channel. The channel registry is built further below (after the scheduler), so
+    // the resolver late-binds it through a getter read only at notify time.
+    let channelRegistry: ChannelRegistry | undefined;
+    const notify = makeNotifyFn({ getRegistry: () => channelRegistry, config, store: uiStore });
+
     // Construct the Scheduler granting only the capabilities the built-in
-    // pipelines actually declare (deny-by-default), with the CLI resolver, then
-    // register the pipelines so their handlers + tasks are available to the tick loop.
+    // pipelines actually declare (deny-by-default), with the CLI + notify resolvers,
+    // then register the pipelines so their handlers + tasks are available to the tick loop.
     const registry = new HandlerRegistry();
     const scheduler = new Scheduler({
       db,
       registry,
       grants: grantedCapabilities(),
       complete,
+      notify,
       redactSummaries: config.storage?.redactRunSummaries === true,
     });
-    // The UI store is opened first so the router can read editable template
-    // default_params through it (#4) — an edited template then affects its runs.
-    const uiStore = openStore(dbPath());
     registerPipelines(scheduler, registry, {
       getDefaultParams: (handlerId) => uiStore.getTemplate(handlerId)?.defaultParams ?? {},
     });
@@ -104,6 +115,8 @@ export const daemonRunCommand: Command = {
       vault,
       store: uiStore,
     });
+    // Late-bind the live registry into the notify resolver (declared before the scheduler).
+    channelRegistry = channels.registry;
     // Pairing (scan-to-connect): the coordinator multiplexes the daemon's single
     // inbound stream into active QR/link pairing sessions and persists the captured
     // chat id on link. Exposed to the UI's POST /api/connections/:id/pair route and

packages/vesper-cli/src/config.test.ts

@@ -155,6 +155,28 @@ describe("normalizeConfig — connections", () => {
   });
 });
 
+describe("normalizeConfig — notify", () => {
+  test("keeps a defaultChannel that names a catalog channel", () => {
+    expect(normalizeConfig({ notify: { defaultChannel: "telegram" } }).notify).toEqual({
+      defaultChannel: "telegram",
+    });
+  });
+
+  test("drops a defaultChannel that is not a catalog channel", () => {
+    expect(normalizeConfig({ notify: { defaultChannel: "slack" } }).notify).toBeUndefined();
+  });
+
+  test("drops a non-string defaultChannel", () => {
+    expect(normalizeConfig({ notify: { defaultChannel: 5 } }).notify).toBeUndefined();
+  });
+
+  test("omits notify entirely when absent or malformed", () => {
+    expect(normalizeConfig({ cli: { adapters: {} } }).notify).toBeUndefined();
+    expect(normalizeConfig({ notify: "nope" }).notify).toBeUndefined();
+    expect(normalizeConfig({ notify: {} }).notify).toBeUndefined();
+  });
+});
+
 describe("loadConfig / saveConfig", () => {
   test("missing file yields the default", async () => {
     expect(await loadConfig(tempConfigPath())).toEqual(DEFAULT_CONFIG);

packages/vesper-cli/src/config.ts

@@ -47,6 +47,15 @@ export interface VesperConfig {
   };
   /** Per-channel messaging wiring (Connections). Secrets stay in the vault. */
   readonly connections?: Readonly<Record<string, ConnectionConfig>>;
+  /** Proactive-notification routing for `ctx.notify` (pipeline -> connected channel). */
+  readonly notify?: {
+    /**
+     * Catalog channel id `ctx.notify` delivers through when a pipeline names none.
+     * When unset the host resolves the first enabled+running channel with a paired
+     * destination. An unknown id is dropped during normalization.
+     */
+    readonly defaultChannel?: string;
+  };
 }
 
 /** A fresh config with no default and no overrides. */
@@ -161,6 +170,19 @@ function normalizeConnections(raw: unknown): VesperConfig["connections"] | undef
   return Object.keys(result).length > 0 ? result : undefined;
 }
 
+/**
+ * Coerce untrusted `notify` config. Keeps `defaultChannel` only when it names a
+ * known catalog channel (mirrors `normalizeConnection`'s catalog gate); an unknown
+ * or non-string id is dropped, never thrown. Returns undefined when nothing valid
+ * remains so the host falls back to first-eligible resolution.
+ */
+function normalizeNotify(raw: unknown): VesperConfig["notify"] | undefined {
+  if (!isObject(raw)) return undefined;
+  const defaultChannel = asString(raw.defaultChannel);
+  if (defaultChannel === undefined || channelById(defaultChannel) === undefined) return undefined;
+  return { defaultChannel };
+}
+
 /** Coerce untrusted `ui` config; keeps only a string `theme`. */
 function normalizeUi(raw: unknown): VesperConfig["ui"] | undefined {
   if (!isObject(raw)) return undefined;
@@ -195,6 +217,8 @@ export function normalizeConfig(raw: unknown): VesperConfig {
   if (ui !== undefined) result = { ...result, ui };
   const connections = normalizeConnections(raw.connections);
   if (connections !== undefined) result = { ...result, connections };
+  const notify = normalizeNotify(raw.notify);
+  if (notify !== undefined) result = { ...result, notify };
   return result;
 }