feat(delivery): add @agentworkforce/delivery — unified multi-target messaging

packages/delivery/package.json

@@ -0,0 +1,38 @@
+{
+  "name": "@agentworkforce/delivery",
+  "version": "0.1.0",
+  "private": false,
+  "type": "module",
+  "main": "dist/index.js",
+  "types": "dist/index.d.ts",
+  "exports": {
+    ".": {
+      "types": "./dist/index.d.ts",
+      "default": "./dist/index.js"
+    },
+    "./package.json": "./package.json"
+  },
+  "files": [
+    "dist",
+    "package.json"
+  ],
+  "repository": {
+    "type": "git",
+    "url": "https://github.com/AgentWorkforce/workforce",
+    "directory": "packages/delivery"
+  },
+  "publishConfig": {
+    "access": "public"
+  },
+  "scripts": {
+    "build": "tsc -p tsconfig.json",
+    "dev": "tsc -p tsconfig.json --watch --preserveWatchOutput",
+    "typecheck": "tsc -p tsconfig.json --noEmit",
+    "test": "tsc -p tsconfig.json && node --test dist/**/*.test.js dist/*.test.js",
+    "lint": "tsc -p tsconfig.json --noEmit"
+  },
+  "dependencies": {
+    "@agentworkforce/runtime": "workspace:*",
+    "@relayfile/relay-helpers": "^0.4.2"
+  }
+}

packages/delivery/src/delivery.ts

@@ -0,0 +1,244 @@
+import { slackClient, telegramClient } from '@relayfile/relay-helpers';
+import type { SlackClient, TelegramClient } from '@relayfile/relay-helpers';
+import type { WorkforceCtx } from '@agentworkforce/runtime';
+import {
+  resolveDeliveryTargets,
+  slackChannel,
+  telegramChat,
+  type DeliveryClient,
+  type DeliveryOptions,
+  type DeliveryResult,
+  type DeliveryTransports,
+  type SlackRef,
+  type TelegramRef
+} from './types.js';
+
+const WRITEBACK_TIMEOUT_MS = 45_000;
+
+/**
+ * Create a delivery client that auto-discovers configured transports from
+ * the persona context and sends to all of them.
+ *
+ * Blocking mode (the default):
+ *   const heads = await delivery.send(header);
+ *   await delivery.send(body, { replyTo: heads });
+ *
+ * Non-blocking parentRef mode (zero receipt round-trips):
+ *   const heads = await delivery.publish(header);
+ *   await delivery.send(body, { replyTo: heads, nonBlocking: true });
+ *
+ * Pass `transports` to inject mock clients for testing — the same injected
+ * client is used for both blocking and non-blocking paths (tests supply
+ * their own mock that short-circuits the writeback).
+ */
+export function createDelivery(
+  ctx: WorkforceCtx,
+  transports?: DeliveryTransports,
+  /** Override which transports to target (defaults to all configured). */
+  onlyTargets?: ReadonlyArray<'slack' | 'telegram'>
+): DeliveryClient {
+  const allTargets = resolveDeliveryTargets(ctx);
+  const targets = onlyTargets
+    ? allTargets.filter((t) => (onlyTargets as readonly string[]).includes(t))
+    : allTargets;
+
+  // Injected transports take priority. When not injected, construct real
+  // clients with appropriate timeouts.
+  const injectedSlack = transports?.slack;
+  const injectedTelegram = transports?.telegram;
+
+  const slackBlocking = injectedSlack ?? (targets.includes('slack')
+    ? slackClient({ writebackTimeoutMs: WRITEBACK_TIMEOUT_MS })
+    : undefined);
+  const slackNonBlocking = injectedSlack ?? (targets.includes('slack')
+    ? slackClient({ writebackTimeoutMs: 0 })
+    : undefined);
+  const telegramBlocking = injectedTelegram ?? (targets.includes('telegram')
+    ? telegramClient({ writebackTimeoutMs: WRITEBACK_TIMEOUT_MS })
+    : undefined);
+  const telegramNonBlocking = injectedTelegram ?? (targets.includes('telegram')
+    ? telegramClient({ writebackTimeoutMs: 0 })
+    : undefined);
+
+  return new DeliveryClientImpl(ctx, targets, {
+    slackBlocking,
+    slackNonBlocking,
+    telegramBlocking,
+    telegramNonBlocking
+  });
+}
+
+interface DeliveryTransportsInternal {
+  slackBlocking?: SlackClient;
+  slackNonBlocking?: SlackClient;
+  telegramBlocking?: TelegramClient;
+  telegramNonBlocking?: TelegramClient;
+}
+
+class DeliveryClientImpl implements DeliveryClient {
+  readonly targets: ReadonlyArray<'slack' | 'telegram'>;
+
+  private ctx: WorkforceCtx;
+  private t: DeliveryTransportsInternal;
+
+  constructor(
+    ctx: WorkforceCtx,
+    targets: Array<'slack' | 'telegram'>,
+    transports: DeliveryTransportsInternal
+  ) {
+    this.ctx = ctx;
+    this.targets = targets;
+    this.t = transports;
+  }
+
+  async send(text: string, opts?: DeliveryOptions): Promise<DeliveryResult> {
+    const nonBlocking = opts?.nonBlocking === true;
+    const refs: Array<SlackRef | TelegramRef> = [];
+    const errors: string[] = [];
+
+    const tasks: Promise<void>[] = [];
+
+    for (const target of this.targets) {
+      const parentRef = opts?.replyTo?.refs.find((r) => r.provider === target);
+      if (target === 'slack') {
+        tasks.push(
+          this.sendSlack(text, parentRef as SlackRef | undefined, nonBlocking)
+            .then((ref) => { if (ref) refs.push(ref); })
+            .catch((err) => { errors.push(`slack: ${String(err)}`); })
+        );
+      }
+      if (target === 'telegram') {
+        tasks.push(
+          this.sendTelegram(text, parentRef as TelegramRef | undefined, nonBlocking)
+            .then((ref) => { if (ref) refs.push(ref); })
+            .catch((err) => { errors.push(`telegram: ${String(err)}`); })
+        );
+      }
+    }
+
+    await Promise.all(tasks);
+
+    // In non-blocking mode, draft refs always succeed (no receipt wait to fail).
+    // Treat any ref as success. In blocking mode, require all targets to succeed.
+    const ok = nonBlocking
+      ? refs.length > 0
+      : errors.length === 0 && refs.length === this.targets.length;
+
+    if (!ok && errors.length > 0) {
+      this.ctx.log?.('warn', 'delivery.partial-failure', { errors, nonBlocking });
+    }
+    if (!ok && refs.length === 0) {
+      const detail = errors.length > 0 ? errors.join('; ') : 'all sends returned null (no configured targets)';
+      throw new Error(`Delivery failed to all targets: ${detail}`);
+    }
+
+    return { ok, refs };
+  }
+
+  async publish(text: string): Promise<DeliveryResult> {
+    return this.send(text, { nonBlocking: true });
+  }
+
+  // ── Slack ──────────────────────────────────────────────────────────────
+
+  private async sendSlack(
+    text: string,
+    parentRef: SlackRef | undefined,
+    nonBlocking: boolean
+  ): Promise<SlackRef | null> {
+    const channel = slackChannel(this.ctx);
+    if (!channel) return null;
+
+    if (nonBlocking) {
+      return this.sendSlackNonBlocking(channel, text, parentRef);
+    }
+    return this.sendSlackBlocking(channel, text, parentRef);
+  }
+
+  private async sendSlackBlocking(
+    channel: string,
+    text: string,
+    parentRef?: SlackRef
+  ): Promise<SlackRef | null> {
+    const client = this.t.slackBlocking;
+    if (!client) return null;
+
+    const result = parentRef?.draftRef
+      ? await client.post(channel, text, { replyTo: parentRef.draftRef })
+      : await client.post(channel, text);
+
+    if (!result.ts) {
+      this.ctx.log?.('warn', 'delivery.slack.no-receipt', { channel });
+      return null;
+    }
+
+    return {
+      provider: 'slack',
+      channel: result.channel,
+      ts: result.ts,
+      draftRef: result.ref
+    };
+  }
+
+  /**
+   * Non-blocking Slack: uses messages.write() directly with writebackTimeoutMs:0.
+   * The parentRef is embedded in the message body so the cloud orders the message
+   * under the parent server-side — zero receipt round-trips. The returned draftRef
+   * is the relay path, usable as a parent for subsequent threaded sends.
+   *
+   * Mirrors the x-reply-radar parentRef threading pattern (internal-agents).
+   */
+  private async sendSlackNonBlocking(
+    channel: string,
+    text: string,
+    parentRef?: SlackRef
+  ): Promise<SlackRef | null> {
+    const client = this.t.slackNonBlocking;
+    if (!client) return null;
+
+    const body: Record<string, unknown> = { text };
+    if (parentRef?.draftRef) {
+      // Embed parentRef in the body — the cloud lifts it from the streamed head
+      // and orders this message under the parent once the parent delivers.
+      body.parentRef = parentRef.draftRef;
+    }
+
+    const result = await client.messages.write({ channelId: channel }, body);
+
+    return {
+      provider: 'slack',
+      channel,
+      ts: '', // Not available yet (non-blocking)
+      draftRef: result.path
+    };
+  }
+
+  // ── Telegram ───────────────────────────────────────────────────────────
+
+  private async sendTelegram(
+    text: string,
+    parentRef: TelegramRef | undefined,
+    nonBlocking: boolean
+  ): Promise<TelegramRef | null> {
+    const chatId = telegramChat(this.ctx);
+    if (!chatId) return null;
+
+    const client = nonBlocking ? this.t.telegramNonBlocking : this.t.telegramBlocking;
+    if (!client) return null;
+
+    const result = parentRef?.messageId
+      ? await client.sendMessage(chatId, text, { replyToMessageId: Number(parentRef.messageId) || undefined })
+      : await client.sendMessage(chatId, text);
+
+    if (!nonBlocking && !result.ok) {
+      this.ctx.log?.('warn', 'delivery.telegram.no-receipt', { chatId });
+      return null;
+    }
+
+    return {
+      provider: 'telegram',
+      chatId: result.chatId != null ? String(result.chatId) : chatId,
+      messageId: result.ok ? result.messageId : ''
+    };
+  }
+}

packages/delivery/src/helpers.ts

@@ -0,0 +1,87 @@
+import type { WorkforceCtx } from '@agentworkforce/runtime';
+
+/**
+ * Resolve a persona input value from the runtime ctx.
+ *
+ * Resolution order (mirrors persona-kit): ctx.persona.inputs (already
+ * resolved from agent value → env → default by the runtime) → fall back
+ * to process.env directly when running outside the full runtime.
+ */
+export function input(ctx: WorkforceCtx, name: string): string | undefined {
+  const spec = ctx.persona.inputSpecs?.[name];
+  // ctx.persona.inputs is already resolved by the runtime (agent value →
+  // env → default). Check it first since it reflects the canonical value.
+  const fromCtx = ctx.persona.inputs?.[name];
+  if (fromCtx && String(fromCtx).trim()) return String(fromCtx).trim();
+  // Fall back to raw process.env for local dev outside the full runtime.
+  const fromEnv = process.env[spec?.env ?? name];
+  if (fromEnv && String(fromEnv).trim()) return String(fromEnv).trim();
+  // Last resort: the spec default.
+  const def = spec?.default;
+  if (def != null && String(def).trim()) return String(def).trim();
+  return undefined;
+}
+
+/**
+ * Split a comma-separated string into trimmed, non-empty entries.
+ */
+export function list(raw: string | undefined): string[] {
+  return (raw ?? '').split(',').map((s) => s.trim()).filter(Boolean);
+}
+
+/**
+ * Race a promise against a timeout. On timeout the timer rejects with an
+ * Error so the caller can catch and fall back. Always clears the timer.
+ */
+export async function withTimeout<T>(
+  p: Promise<T>,
+  ms: number,
+  label: string
+): Promise<T> {
+  let timer: ReturnType<typeof setTimeout>;
+  const timeout = new Promise<never>((_, rej) => {
+    timer = setTimeout(() => rej(new Error(`${label} timed out after ${ms}ms`)), ms);
+  });
+  try {
+    return await Promise.race([p, timeout]);
+  } finally {
+    clearTimeout(timer!);
+  }
+}
+
+/**
+ * Fetch with an AbortController timeout. Returns the response on success,
+ * or undefined on timeout/network error (never throws). Preserves caller-
+ * provided signal by racing our abort against it.
+ */
+export async function fetchWithTimeout(
+  url: string,
+  init: RequestInit = {},
+  timeoutMs: number = 8_000
+): Promise<Response | undefined> {
+  const controller = new AbortController();
+  const timer = setTimeout(() => controller.abort(), timeoutMs);
+  const signal = init.signal
+    ? anySignal([controller.signal, init.signal])
+    : controller.signal;
+  try {
+    return await fetch(url, { ...init, signal });
+  } catch {
+    return undefined;
+  } finally {
+    clearTimeout(timer);
+  }
+}
+
+/** Combine multiple AbortSignals — any one firing aborts the fetch. */
+function anySignal(signals: AbortSignal[]): AbortSignal {
+  const controller = new AbortController();
+  for (const sig of signals) {
+    if (sig.aborted) {
+      controller.abort(sig.reason);
+      return controller.signal;
+    }
+    sig.addEventListener('abort', () => controller.abort(sig.reason), { once: true });
+  }
+  return controller.signal;
+}

packages/delivery/src/index.ts

@@ -0,0 +1,15 @@
+export { createDelivery } from './delivery.js';
+export {
+  resolveDeliveryTargets,
+  slackChannel,
+  telegramChat,
+  type DeliveryClient,
+  type DeliveryOptions,
+  type DeliveryResult,
+  type DeliveryTransports,
+  type MessageRef,
+  type SlackRef,
+  type TelegramRef
+} from './types.js';
+
+export { input, list, withTimeout, fetchWithTimeout } from './helpers.js';