@salesforce/agentic-common

Shared primitives and common utilities for the Salesforce agentic DX packages. Provides a typed event bus, clock abstraction, ID generation, error utilities, log record shape, thin log-emit helpers, a Salesforce org connection interface, and the JSONWebToken family used by the agent-SDK's connectivity resolvers.

Quick Start

Closed source. This package is published to npm under the Salesforce Public Code License and is for use by Salesforce only.

import { EventBus, LogBus } from '@salesforce/agentic-common';

type MyEvent = { type: 'hello'; name: string };

const bus = new EventBus<MyEvent>();
const unsubscribe = bus.on((event) => {
  console.log(`hello, ${event.name}`);
});

bus.emit({ type: 'hello', name: 'world' });

unsubscribe();
bus.dispose();

const logs = new LogBus();
logs.on((record) => console.log(`[${record.level}] ${record.message}`));
logs.info('ready', { agentId: 'a1' });

API Reference

OrgConnection / RealOrgConnection

Interface for making authenticated HTTP requests to a Salesforce org, carrying identity metadata alongside transport capabilities. RealOrgConnection is the production implementation backed by @salesforce/core.

interface OrgConnection {
  request<T>(opts: {
    method: 'GET' | 'POST' | 'PUT' | 'PATCH' | 'DELETE';
    url: string;
    body?: string;
    headers?: { [name: string]: string };
  }): Promise<T>;

  getAccessToken(): string;
  getInstanceUrl(): string;
  getUsername(): string;
  getOrgId(): string;
  getOrgAliasOrUsername(): string;
  getInferredSfApiEnv(): SfApiEnv;
}

OrgConnectionFactory / RealOrgConnectionFactory

Factory for creating authenticated OrgConnection instances. RealOrgConnectionFactory is the production implementation backed by @salesforce/core.

interface OrgConnectionFactory {
  createFromOrgAliasOrUsername(orgAliasOrUsername: string): Promise<OrgConnection>;
  createFromTargetOrg(options?: { projectRoot?: string }): Promise<OrgConnection>;
  createFromCredentials(accessToken: string, instanceUrl: string): Promise<OrgConnection>;
}

SfApiEnv

Enum of Salesforce API environments used for gateway routing. Consumers access the inferred value via OrgConnection.getInferredSfApiEnv() which uses the instance URL and the SF_API_ENV environment variable override.

enum SfApiEnv {
  Dev = 'dev',
  Perf = 'perf',
  Prod = 'prod',
  Stage = 'stage',
  Test = 'test',
}

inferSfApiEnv(instanceUrl, options?): SfApiEnv

Maps a Salesforce instance URL to a SfApiEnv. OrgConnection.getInferredSfApiEnv() is the typical entry point; direct callers use this when they have an instanceUrl but not a connection (e.g. building a Salesforce platform MCP URL from a token + instance URL).

Resolution order:

1. If SF_API_ENV is set to a valid value, return it.
2. STM patterns → Stage.
3. OrgFarm dev / perf / test sub-segments under .pc-rnd. → Dev / Perf / Test.
4. .pc-rnd. host with no recognized sub-segment → options.pcRndFallback ?? Test. .pc-rnd. is an internal-only OrgFarm domain (by definition not Prod), so the default degrades a future unrecognized sub-segment to a sibling non-prod host rather than leaking traffic from a non-prod org to the Prod servlet.
5. .crm.dev workspaces, localhost.sfdcdev., .internal. → Dev.
6. Anything else → Prod.

function inferSfApiEnv(instanceUrl: string, options?: { pcRndFallback?: SfApiEnv }): SfApiEnv;

import { SfApiEnv, inferSfApiEnv } from '@salesforce/agentic-common';

inferSfApiEnv('https://myorg.test1.pc-rnd.salesforce.com'); // SfApiEnv.Test
inferSfApiEnv('https://myorg.qa1.pc-rnd.salesforce.com'); // SfApiEnv.Test (default fallback)
inferSfApiEnv('https://myorg.qa1.pc-rnd.salesforce.com', { pcRndFallback: SfApiEnv.Prod }); // SfApiEnv.Prod

Clock / RealClock

Abstract time source for dependency injection. Use RealClock in production; extend Clock in tests for deterministic time control.

abstract class Clock {
  abstract now(): Date;
  // Default: returns ts + 1ms. Override for test scenarios needing different stepping.
  nextAfter(ts: Date): Date;
}

class RealClock extends Clock {
  now(): Date; // returns new Date()
}

getErrorMessage(err: unknown): string

Safely extracts a message string from any thrown value — Error instances, error-like objects with a message property, or arbitrary values (coerced via String()).

import { getErrorMessage } from '@salesforce/agentic-common';

try { ... } catch (err) {
  console.error(getErrorMessage(err));
}

isAbortError(err: unknown): boolean

Returns true for either platform shape of an abort signal error: the DOM AbortError (err.name === 'AbortError', e.g. from AbortController.abort() against fetch / undici.request) or the Node ABORT_ERR (err.code === 'ABORT_ERR'). Useful in retry loops and error handlers that need to distinguish caller-initiated cancellation from transport failures.

import { isAbortError } from '@salesforce/agentic-common';

try {
  await someAsyncWork(signal);
} catch (err) {
  if (isAbortError(err)) return; // caller cancelled; not an error to surface
  throw err;
}

resolveProxyDispatcher() / createProxyAwareFetch(dispatcher?)

Proxy-routing helpers for Node's fetch. resolveProxyDispatcher() returns an undici.EnvHttpProxyAgent built from HTTPS_PROXY / HTTP_PROXY / NO_PROXY env vars (or their lowercase forms — undici honors both casings), or undefined when none is set. createProxyAwareFetch(dispatcher?) returns a fetch-compatible function that routes outbound calls through the supplied dispatcher via undici.fetch; when dispatcher is undefined, it returns globalThis.fetch unchanged so the no-proxy path has zero overhead.

Designed to be called from harness factories' create(); consumers normally don't call them directly. Both production harness factories (MastraHarnessFactory, ClaudeHarnessFactory) build a proxy-aware fetch via these helpers and thread it into every in-process HTTP call site (LLM gateway language-model builders + MCP remote transports). No globalThis mutation.

EnvHttpProxyAgent captures HTTPS_PROXY and HTTP_PROXY at construction; NO_PROXY is re-evaluated per dispatch. Set HTTPS_PROXY / HTTP_PROXY BEFORE the first create() call so the dispatcher snapshot reflects the right values.

import { createProxyAwareFetch, resolveProxyDispatcher } from '@salesforce/agentic-common';

const dispatcher = resolveProxyDispatcher();
const fetchFn = createProxyAwareFetch(dispatcher);
await fetchFn('https://example.com/v1/things');

UniqueIDGenerator / UUIDGenerator

Interface + default implementation for generating unique identifiers. Tests can inject deterministic implementations.

interface UniqueIDGenerator {
  getUniqueId(): string;
}

class UUIDGenerator implements UniqueIDGenerator {
  getUniqueId(): string; // wraps crypto.randomUUID()
}

EventBus<T>

A minimal, typed event bus. One type parameter, one emission channel. Error-isolated (a throwing listener never breaks emit or other listeners).

class EventBus<T> {
  readonly listenerCount: number;

  on(callback: EventListener<T>): Unsubscribe;
  emit(event: T): void;
  forwardTo(target: EventBus<T>, enrich?: (event: T) => T): Unsubscribe;
  forwardWhileSubscribed(target: EventBus<T>, enrich?: (event: T) => T): Unsubscribe;
  onSubscriberPresenceChange(callback: SubscriberPresenceListener): Unsubscribe;
  dispose(): void;
}

type EventListener<T> = (event: T) => void;
type SubscriberPresenceListener = (hasSubscribers: boolean) => void;
type Unsubscribe = () => void;

| Method | Description | | ----------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | on(callback) | Subscribe. Returns an Unsubscribe function — no need to hold the callback reference for removal. | | emit(event) | Deliver to all listeners. Listener errors are caught and ignored so one bad listener can't cascade. | | forwardTo(target, ?) | Subscribe to this bus and re-emit every event onto target. Optional enrich transforms events. Eager — the upstream subscription stays attached for the full lifetime of the link, even when target has zero listeners. | | forwardWhileSubscribed(target, ?) | Lazy peer of forwardTo. Only attaches the upstream subscription while target.listenerCount > 0; detaches when target loses its last listener. Composes naturally — chaining across multiple buses propagates "no listener anywhere downstream" all the way up the chain so an expensive producer can short-circuit on this.listenerCount > 0 reads. | | onSubscriberPresenceChange(cb) | Subscribe to listener-presence transitions. The callback fires true on 0 → 1 transitions and false on N → 0. Does NOT fire on intermediate listener add/remove. Returns an idempotent Unsubscribe. Used by forwardWhileSubscribed; also useful when an expensive producer wants to start/stop work based on whether anyone is listening at all. | | dispose() | Remove all listeners. Fires a final false presence notification if there were active listeners. Safe to call multiple times. | | listenerCount | Current listener count (useful for leak-check assertions in tests, and for forwardWhileSubscribed-style "is anyone listening?" checks). |

LogRecord / LogBus

LogRecord is the structured log shape used across packages. LogBus extends EventBus<LogRecord> with level-named convenience methods so emit sites read as bus.warn('msg', { ctx }) instead of repeating the { level, message, timestamp, ... } literal.

type LogLevel = 'debug' | 'info' | 'warn' | 'error';

type LogRecord = {
  level: LogLevel;
  message: string;
  timestamp: Date;
  context?: Record<string, unknown>;
  error?: Error;
};

class LogBus extends EventBus<LogRecord> {
  debug(message: string, context?: Record<string, unknown>): void;
  info(message: string, context?: Record<string, unknown>): void;
  warn(message: string, context?: Record<string, unknown>, error?: Error): void;
  error(message: string, error?: Error, context?: Record<string, unknown>): void;
}

JSONWebToken / FixedJSONWebToken / DynamicJSONWebToken

Auth primitive for Salesforce-fronted services (LLM gateway, MCP servers, future Apex). FixedJSONWebToken parses a literal JWT string and reports expiration with a 30-second buffer; DynamicJSONWebToken wraps a FixedJSONWebToken and auto-refreshes via the org's minting endpoint (default /ide/auth) when expired. Both implement the same JSONWebToken interface so consumers don't need to know which kind they hold.

interface JSONWebToken<JWTHeaders, JWTPayload> {
  getValue(): Promise<string>; // raw serialized JWT
  getHeaders(): Promise<JWTHeaders>;
  getPayload(): Promise<JWTPayload>;
  getTenantKey(): Promise<string>; // from header `tnk`
  getFeatureId(): string;
  isExpired(): boolean;
  onLog(callback: (record: LogRecord) => void): Unsubscribe;
}

// Resolves a fresh org connection from access token + instance URL, mints + validates the first JWT.
function createJWT(options: {
  accessToken: string;
  instanceUrl: string;
  mintingPath?: string; // default '/ide/auth'
  featureId?: string; // default 'VibesService' (or LLMG_FEATURE_ID env var)
}): Promise<JSONWebToken>;

// Same fail-fast first-mint behavior, but uses an existing OrgConnection.
function createJWTFromConnection(
  orgConnection: OrgConnection,
  options?: CreateJWTFromConnectionOptions, // { mintingPath?: string; featureId?: string }
): Promise<JSONWebToken>;

JWTOptions, CreateJWTFromConnectionOptions, RequiredJWTHeaders, and RequiredJWTPayload are all exported types so consumers can name the parameter / generic-instantiation shapes without redeclaring them.

JWT lifecycle log records flow through onLog. The token's auto-refresh emits 'Refreshing expired JWT' (debug), 'JWT refreshed' (info, with durationMs), and 'JWT refresh failed' (error, with the wrapped exception) so operators have visibility without subscribing to a separate telemetry channel.

Retryer / BackoffRetryer / NoOpRetryer

Generic retry orchestration. Consumers depend on the Retryer interface and supply per-call decisions (which errors / results are retryable, how to extract a server-driven Retry-After, what to log) via callbacks. Backoff timing, jittering, deadline enforcement, and abort handling are construction details of BackoffRetryer — invisible to the caller.

interface Retryer {
  execute<T>(attemptFn: () => Promise<T>, callbacks?: RetryCallbacks<T>): Promise<T>;
}

type RetryOptions = {
  maxAttempts?: number; // Default: 3 (>= 1)
  initialDelayMs?: number; // Default: 100
  maxDelayMs?: number; // Default: 2000 (caps *computed* backoff only)
  maxRetryAfterMs?: number; // Default: 60_000 (caps *server-driven* hints only)
  backoffFactor?: number; // Default: 2
  maxTotalElapsedMs?: number; // Default: Infinity
};

type RetryCallbacks<T> = {
  signal?: AbortSignal;
  isRetryableError?: (err: unknown) => boolean;
  isRetryableResult?: (result: T) => boolean;
  getRetryAfterMs?: (result: T) => number | undefined;
  onRetry?: (info: RetryAttemptInfo<T>) => void;
  onExhausted?: (info: RetryExhaustedInfo<T>) => void;
  drainResult?: (result: T) => Promise<void>;
};

type RetryAttemptInfo<T> = {
  attempt: number; // 1-indexed attempt number that just failed
  delayMs: number; // jittered or server-driven delay about to be waited
  error?: unknown; // set if the attempt threw a retryable error
  result?: T; // set if the attempt returned a retryable result
};

type RetryExhaustedInfo<T> = {
  attempts: number; // total attempts made
  error?: unknown;
  result?: T;
  reason: 'attempts' | 'deadline';
};

The fully-resolved defaults are exported as DEFAULT_RETRY_OPTIONS ({ maxAttempts: 3, initialDelayMs: 100, maxDelayMs: 2000, maxRetryAfterMs: 60_000, backoffFactor: 2, maxTotalElapsedMs: Infinity }) so consumers and tests can share one source of truth.

Use BackoffRetryer in production:

import { BackoffRetryer } from '@salesforce/agentic-common';

const retryer = new BackoffRetryer({ maxAttempts: 3, initialDelayMs: 100 });
const response = await retryer.execute(() => fetch(url), {
  isRetryableError: (err) => (err as { code?: string }).code === 'ECONNRESET',
  isRetryableResult: (res) => res.status >= 500,
  onRetry: ({ attempt, delayMs }) => log.warn(`retry ${attempt} in ${delayMs}ms`),
});

Use NoOpRetryer in tests where retry behavior is irrelevant:

import { NoOpRetryer } from '@salesforce/agentic-common';

const client = new SomeClient(opts, { retryer: new NoOpRetryer() });

getErrorMessageWithStack(err: unknown): string

Like getErrorMessage, but returns the full stack trace for Error instances.

wrapError(err: unknown, message: string): Error

Creates a new Error with a prefixed message and the original error as cause.

splitFrontmatterAndBody(content: string): FrontmatterSplit

Splits a markdown document into its YAML frontmatter block and body content. Recognizes the standard --- delimited frontmatter block at the start of a document and normalizes CRLF line endings to LF. When no frontmatter block is present, frontmatter is null and body is the full input.

The frontmatter is returned unparsed so callers can choose whether to parse it as YAML, JSON, or treat it as opaque text — keeping a YAML runtime out of @salesforce/agentic-common.

import { splitFrontmatterAndBody } from '@salesforce/agentic-common';

const { frontmatter, body } = splitFrontmatterAndBody(fileContents);
// frontmatter: "name: terse-replies\ndescription: Keep replies short."
// body:        "Keep replies under three sentences.\n"

type FrontmatterSplit = {
  frontmatter: string | null;
  body: string;
};

backfillCreatedAt<T extends { createdAt?: Date }>(messages: T[], clock: Clock): T[]

Backfills missing createdAt timestamps on a batch of message-shaped records, stepping per-position via clock.nextAfter so a bulk insert produces strictly-ascending timestamps rather than ms-precision ties tie-broken by stable sort. Records that already carry a createdAt pass through unchanged (same reference, no clone). Records missing one are cloned with a freshly-stepped value: the first missing position seeds from clock.now(), each subsequent missing position uses clock.nextAfter(<prior>).

Used by sfdx-agent-sdk's ChatSession.addContext() and both harnesses' addContext boundaries so the read-side "every Message has populated createdAt; sort ascending" contract is upheld regardless of consumer-construction style. Generic over the record shape, so any consumer with a createdAt?: Date field can reuse the same policy.

import { backfillCreatedAt, RealClock } from '@salesforce/agentic-common';

const filled = backfillCreatedAt(messages, new RealClock());

buildSummaryPrompt(transcript: string): string

Returns a third-person summarization-prompt string asking the model to compress the supplied transcript into a context summary. Used by both production harnesses (@salesforce/sfdx-agent-harness-mastra / @salesforce/sfdx-agent-harness-claude) inside their compactThread flows so the prompt wording stays uniform across implementations — a freshly-summarized thread reads the same regardless of which harness produced it.

import { buildSummaryPrompt } from '@salesforce/agentic-common';

const prompt = buildSummaryPrompt(transcriptText);
// → "Summarize the following conversation into a concise context summary. ..."

Development

See DEVELOPING.md for build-from-source setup, scripts, and monorepo commands.

See ARCHITECTURE.md for implementation notes.