@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.

Optional getApiVersion() returns the effective Salesforce API version in bare NN.0 form. Consumers should feature-detect it so structural implementations written before this addition remain compatible.

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;
  getApiVersion?(): string;
}

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>;
}

The disk-backed resolve paths invalidate the relevant @salesforce/core cache before reading, so a connection built after another process rewrites ~/.sfdx / .sf (e.g. sf org login, a default-org switch, or an alias reassociation) reflects the current on-disk auth rather than a stale cached value: createFromTargetOrg clears the ConfigAggregator entry for the exact key it reads (scoped to projectPath / process.cwd(), never a process-wide wipe), and the alias path clears StateAggregator (the alias→username store, keyed on Global.DIR). createFromCredentials reads no disk state, so it clears nothing. Resolution is also serialized process-wide so concurrent first-time resolves can't corrupt each other's cache reads.

OrgConnectionProvider / connectionProviderFor / reuseThenReresolve

An OrgConnectionProvider is a () => Promise<OrgConnection> closure. DynamicJSONWebToken (and createJWTFromConnection) invoke it on every mint, so a re-mint after JWT expiry re-resolves the connection — which re-reads the current on-disk auth. This closes the stale-access-token window: when the auth for an alias/default org changes mid-session, the next mint uses the new credentials.

type OrgConnectionProvider = () => Promise<OrgConnection>;

// Builds a provider that re-resolves a disk-backed org on every call (alias wins, else target org).
function connectionProviderFor(
  factory: OrgConnectionFactory,
  opts: { orgAlias?: string; projectRoot?: string },
): OrgConnectionProvider;

// Returns `initial` on the first call, then delegates to `reresolve` — so a resolver that already
// resolved a connection eagerly reuses it for the initial mint (no extra refreshAuth round-trip)
// while re-mints still re-read disk.
function reuseThenReresolve(initial: OrgConnection, reresolve: OrgConnectionProvider): OrgConnectionProvider;

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). |

Event-bus hierarchy (TimestampedEventBus / EnvironmentAwareEventBus / BaseTelemetryEventBus)

A small hierarchy of buses that stamp base fields onto every event they originate, so those fields live close to the event's origin rather than being appended by each downstream subscriber. The events are a plain extends chain; each bus layer contributes only its OWN base fields via a super-chained create*Event builder, and a concrete bus's emit method folds its domain fields onto that base. Because the built event is a concrete type, there is no type assertion anywhere in the hierarchy.

type TimestampedEvent = { timestamp: Date };
// EnvironmentFields = the flat identity fields (see below)
type EnvironmentAwareEvent = TimestampedEvent & EnvironmentFields;
type BaseTelemetryEvent = EnvironmentAwareEvent & { type: string };

// Adds `timestamp` (from an injected Clock). Generic over the full event E it carries.
class TimestampedEventBus<E extends TimestampedEvent> extends EventBus<E> {
  constructor(clock?: Clock);
  protected createTimestampedEvent(timestamp?: Date): TimestampedEvent; // this layer's base fields
}

// Additionally adds the flat environment fields (read once at construction).
class EnvironmentAwareEventBus<E extends EnvironmentAwareEvent> extends TimestampedEventBus<E> {
  constructor(clock?: Clock, environment?: EnvironmentFields);
  protected createEnvironmentAwareEvent(timestamp?: Date): EnvironmentAwareEvent; // super() + env fields
}

// Reusable telemetry bus — any service subclasses it over its own telemetry-event union.
class BaseTelemetryEventBus<E extends BaseTelemetryEvent = BaseTelemetryEvent> extends EnvironmentAwareEventBus<E> {
  protected createBaseTelemetryEvent(timestamp?: Date): EnvironmentAwareEvent; // super() (type is per-variant)
  emitTelemetry(event: E, timestamp?: Date): void; // stamps base fields onto an already-built event
}

Two emit roles: emit(fullEvent) (inherited) is a raw passthrough used by forwardTo (so a forwarded event is never re-stamped); a concrete bus's public method builds the event via the create*Event chain. The optional timestamp arg pins the value so a telemetry event and a colocated log record can share a tick.

A downstream service subclasses BaseTelemetryEventBus over its own union and adds concrete emit<Variant> methods that build each event through the inherited createBaseTelemetryEvent() — env-stamped, flat, log-pipeline-friendly telemetry with no per-service bus plumbing:

type MyEvent = BaseTelemetryEvent & ({ type: 'request'; route: string } | { type: 'query'; ms: number });

class MyTelemetryBus extends BaseTelemetryEventBus<MyEvent> {
  emitRequest(route: string, timestamp?: Date): void {
    this.emit({ ...this.createBaseTelemetryEvent(timestamp), type: 'request', route }); // subscribers get it + timestamp + env
  }
}

LogRecord / LogBus

LogRecord is the structured log shape used across packages. LogBus extends EnvironmentAwareEventBus<LogRecord> with level-named convenience methods so emit sites read as bus.warn('msg', { ctx }) instead of repeating the { level, message, timestamp, ... } literal. Every record is built through the super-chained builders, carrying timestamp and the flat environment identity fields, stamped by the origin bus.

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

type LogPayload = {
  // what a caller supplies
  level: LogLevel;
  message: string;
  context?: Record<string, unknown>;
  error?: Error;
};

// Delivered to onLog() subscribers: payload + timestamp + flat env identity fields.
type LogRecord = EnvironmentAwareEvent & LogPayload;

class LogBus extends EnvironmentAwareEventBus<LogRecord> {
  emitLog(payload: LogPayload, timestamp?: Date): void;
  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;
}

EnvironmentFields

Optional, flat identity fields stamped onto log/telemetry events by an EnvironmentAwareEventBus. Read once from generic environment variables via readEnvironmentContext(). Every field is optional — present only when its variable holds a non-empty value.

type EnvironmentFields = {
  instanceType?: string; // INSTANCE_TYPE — what kind of running instance this is
  instanceId?: string; // INSTANCE_ID   — this specific running instance
  orgId?: string; // ORG_ID
  userId?: string; // USER_ID
  featureId?: string; // FEATURE_ID, falling back to legacy LLMG_FEATURE_ID
};

function readEnvironmentContext(env?: NodeJS.ProcessEnv): EnvironmentFields;
function resolveFeatureId(env?: NodeJS.ProcessEnv): string | undefined; // FEATURE_ID → LLMG_FEATURE_ID → undefined

The library reads only generic variable names; a deployment maps its own concepts onto them at the environment boundary (e.g. a DX Workspace container exports INSTANCE_TYPE=$DXW_TEMPLATE_NAME / INSTANCE_ID=$SFDX_INSTANCE_ID), so this package never learns deployment-specific names. These fields are read once at bus construction and are a convenience for single-tenant deployments (one process = one org/user); they are subject to change. Multi-tenant consumers should pass an explicit EnvironmentFields bag (or per-event identity) instead of relying on the env read. Consumers that set none of the vars get an empty bag and nothing is stamped.

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.

DynamicJSONWebToken mints against an OrgConnectionProvider (not a fixed connection): it calls the provider on the initial mint and on every re-mint, so a disk-backed provider makes each expiry-driven re-mint re-read the current on-disk auth (see connectionProviderFor / reuseThenReresolve above).

Its constructor takes an optional trailing Retryer (default BackoffRetryer): the mint round-trip (only the POST — the connection is resolved once, outside the retry) is retried on transient transport failures (connection reset / timeout / DNS hiccup / socket hang-up / HTTP 502-504), so a network blip during the eager mint doesn't terminalize the caller. INVALID_SESSION_ID is not retried — it surfaces immediately for service-side self-healing. Inject a NoOpRetryer in tests.

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 mints against a caller-supplied provider.
// The provider is invoked on the initial mint AND every re-mint — pass connectionProviderFor(...)
// so re-mints re-read the current on-disk auth. (Breaking change, W-23721341: the first parameter
// was previously a fixed `OrgConnection`.)
function createJWTFromConnection(
  orgConnectionProvider: OrgConnectionProvider,
  options?: CreateJWTFromConnectionOptions,
): 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.