Contrib Kit Protocols

@primafuture/contrib-kit-protocols adds framework-neutral typed commands, notification delivery, and ordered transformation pipelines to the lifecycle and contribution catalog owned by Contrib Kit Core. Protocols 1.1 additionally introduces typed collect-all requests over dynamically registered contribution sources.

The complete 1.0 command runtime provides parallel and generation-owned serial commands, exact handler registrations, dynamic contributor-first providers, caller and lifecycle cancellation, deadlines, managed middleware, stateless execution, runtime validation, safe diagnostics, and deterministic cleanup.

The notification runtime provides typed payload, context, and listener-data axes, Core-backed filtering and ordering, immutable recipient snapshots, parallel or strict serial delivery, explicit continue/stop failure policy, cancellation, deadlines, stateless publishing, and cleanup-safe late settlement.

The pipeline runtime provides typed Input → State → Output transformations, asynchronous initialization and finalization, deterministic Core-backed stage selection, strict sequential stage execution, explicit short-circuiting, validation, cancellation, deadlines, contributor-first stage providers, and cleanup-safe late settlement.

Installation

pnpm add @primafuture/contrib-kit-core @primafuture/contrib-kit-protocols

Both packages are ESM-only. Protocols has no runtime dependency beyond its compatible Core peer and does not require DOM, Node, or framework ambient types.

Define and collect typed values

interface MetricsRequest {
	readonly window: "minute" | "hour";
}

interface MetricsContext {
	readonly tenantId: string;
}

interface MetricsSourceData {
	readonly priority: number;
}

const metrics = contribKitProtocols.defineCollector<
	MetricsRequest,
	number,
	MetricsContext,
	MetricsSourceData
>()({
	pointId: "metrics:collect",
	contractVersion: 1,
	concurrency: { mode: "parallel" },
	orderingPolicy: {
		order(sources) {
			return [...sources].sort((left, right) => right.data.priority - left.data.priority);
		},
	},
});

const registry = contribKitCore.createExtensionRegistry();
const collectorOwner = registry.rootScope.registerPoint(metrics);
const protocols = contribKitProtocols.createContributionProtocols({ registry });
const source = protocols.collectors.registerSource(metrics, {
	contributionId: "metrics:requests",
	data: { priority: 100 },
	source(request, execution) {
		return readRequestCount(execution.context.tenantId, request.window);
	},
});

await source.whenActive();
const report = await protocols.collectors.collect(
	metrics,
	{ window: "minute" },
	{
		context: { tenantId: "tenant-1" },
		timeoutMs: 2_000,
		concurrency: { mode: "serial" },
	},
);

for (const result of report.results) {
	if (result.status === "collected") {
		console.log(result.contributionId, result.value);
	} else {
		console.error(result.contributionId, result.error);
	}
}

await source.dispose();
await collectorOwner.close().whenCleanupFinished();
await protocols.dispose();
await registry.dispose();

collect() resolves one immutable Core snapshot and reports sources in that same order. Parallel is the default; serial mode starts at most one source and advances in a new microtask only after its actual settlement. A call-level policy overrides the definition default. Source failure, invalid returned value, or removal affects only that source record; caller cancellation, a deadline, point close, or runtime/registry disposal rejects the whole invocation. Source cleanup still joins the callback's actual late settlement.

For a one-off collection without persistent Protocols runtime ownership, use the stateless helper:

const oneOffReport = await contribKitProtocols.collectContributions(
	registry,
	metrics,
	{ window: "hour" },
	{ context: { tenantId: "tenant-1" }, concurrency: { mode: "parallel" } },
);

collectContributions() shares the same validation, Core snapshot, scheduling, cancellation and reporting engine. Its transient runtime owns no source, provider, point, or registry and detaches only after actual finality of already leased callbacks.

Contributor-first plugins can provide a source before the host registers the point:

const provider = protocols.collectors.provideSource(metrics, {
	contributionId: "metrics:dynamic",
	data: { priority: 50 },
	source(request, execution) {
		return readDynamicCount(execution.context.tenantId, request.window);
	},
});

const dynamicOwner = registry.rootScope.registerPoint(metrics);
await provider.whenFirstActive();

await dynamicOwner.close().whenClosed();
// provider.state === "waiting"

registry.rootScope.registerPoint(metrics);
await provider.whenFirstActive();
await provider.dispose();

Each exact point generation receives a new Core registration identity. Recoverable activation collisions leave the provider blocked; after removing the competing source, call provider.retry(). Disposing the provider removes only its own current and historical source registrations and leaves the point and registry active.

Define and transform a pipeline

interface NormalizeContext {
	readonly locale: string;
}

interface StageData {
	readonly priority: number;
}

const normalizeDocument = contribKitProtocols.definePipeline<
	string,
	readonly string[],
	string,
	NormalizeContext,
	StageData
>()({
	pointId: "documents:normalize",
	contractVersion: 1,
	initialize(input) {
		return [input.trim()];
	},
	finalize(state, execution) {
		return `${state.join(" ")} (${execution.context.locale})`;
	},
	orderingPolicy: {
		order(stages) {
			return [...stages].sort((left, right) => right.data.priority - left.data.priority);
		},
	},
});

const registry = contribKitCore.createExtensionRegistry();
const pipelineOwner = registry.rootScope.registerPoint(normalizeDocument);
const protocols = contribKitProtocols.createContributionProtocols({ registry });
const stage = protocols.pipelines.registerStage(normalizeDocument, {
	contributionId: "documents:add-title",
	data: { priority: 100 },
	stage(state, execution) {
		const nextState = [...state, "title"];

		return nextState.length >= 2
			? execution.stop(nextState)
			: nextState;
	},
});

await stage.whenActive();
const output = await protocols.pipelines.transform(
	normalizeDocument,
	"Document",
	{ context: { locale: "en" }, timeoutMs: 2_000 },
);

await stage.dispose();
await pipelineOwner.close().whenCleanupFinished();
await protocols.dispose();
await registry.dispose();

Each transformation resolves one immutable ordered stage snapshot from the initial validated state. Stages within that invocation never overlap; a following stage or the finalizer starts only after the preceding callback's actual settlement. Separate transformations may execute concurrently. An authentic invocation-bound execution.stop(nextState) skips later stages but still validates nextState and runs the finalizer. Registration changes affect only later snapshots.

Caller cancellation, timeout, point close, runtime disposal, or registry disposal rejects the transformation according to the first irreversible reason. Removing a running stage also rejects the transformation immediately, while stage cleanup waits for its actual late settlement. The runtime owns exact registrations created through registerStage; it never owns the pipeline point or Core registry.

Transform without a persistent runtime

For a one-off transformation without persistent stage or provider ownership, use the stateless helper:

const output = await contribKitProtocols.transformPipeline(
	registry,
	normalizeDocument,
	"One-off document",
	{ context: { locale: "en" }, timeoutMs: 2_000 },
);

transformPipeline() uses the same validators, initializer, ordered stage snapshot, short-circuit, finalizer, cancellation, deadline, and error mapping as the bound runtime. Its transient registry observation remains alive only through the exact invocation's actual callback and lease finality. It owns no stage, provider, point, or Core registry and never extends the caller-visible business Promise with cleanup.

Provide a stage before the pipeline exists

An independently loaded contributor can observe the exact canonical pipeline point without knowing when its owner will load:

const provider = protocols.pipelines.provideStage(normalizeDocument, {
	contributionId: "documents:add-title",
	data: { priority: 100 },
	stage(state) {
		return [...state, "title"];
	},
});

// The provider starts as waiting when the point is absent. Each point generation
// creates a fresh exact Core registration over the same captured stage callbacks.
await provider.whenFirstActive();

// Replacement or explicit removal blocks automatic recovery in the same generation.
if (provider.state === "blocked") provider.retry();

await provider.dispose();

The frozen provider handle owns one point observer and every stage registration it creates. Point close returns it to waiting; re-registering the same canonical point creates a new registration identity. Provider disposal joins current and historical stage cleanup but never disposes the pipeline point or Core registry.

Define and publish a notification

interface DocumentChanged {
	readonly documentId: string;
}

interface DeliveryContext {
	readonly tenantId: string;
}

interface ListenerData {
	readonly priority: number;
}

const documentChanged = contribKitProtocols.defineNotification<
	DocumentChanged,
	DeliveryContext,
	ListenerData
>()({
	pointId: "documents:changed",
	contractVersion: 1,
	orderingPolicy: {
		order(listeners) {
			return [...listeners].sort((left, right) => right.data.priority - left.data.priority);
		},
	},
});

const registry = contribKitCore.createExtensionRegistry();
const protocols = contribKitProtocols.createContributionProtocols({ registry });
const notificationOwner = registry.rootScope.registerPoint(documentChanged);
const listener = protocols.events.registerListener(documentChanged, {
	contributionId: "search:document-indexer",
	data: { priority: 100 },
	async listener(payload, delivery): Promise<void> {
		if (delivery.signal.aborted) return;
		await updateSearchIndex(delivery.context.tenantId, payload.documentId);
	},
});

await listener.whenActive();
const report = await protocols.events.publish(
	documentChanged,
	{ documentId: "document-1" },
	{ context: { tenantId: "tenant-1" }, timeoutMs: 2_000 },
);

console.log(report.deliveredCount, report.failedCount);
await listener.dispose();
await notificationOwner.close().whenCleanupFinished();
await protocols.dispose();
await registry.dispose();

publish() resolves listeners once, starts the frozen snapshot in deterministic Core order, and reports every recipient in that same order even when parallel callbacks settle differently. A listener failure creates its own failed report item. Caller cancellation, timeout, point close, runtime disposal, or registry disposal rejects the whole publish operation. Removing one listener affects only that recipient; its cleanup still waits for the callback's actual late settlement.

Use { mode: "serial", onListenerFailure: "continue" } when listeners must never overlap but every recipient should be attempted. Use "stop" when the first failed recipient must prevent all later callbacks; those recipients are reported as skipped/priorListenerFailed. A serial successor starts only after the preceding listener's actual settlement, even when that listener ignored cancellation.

For a one-off publish without a persistent Protocols runtime, call the stateless helper:

const report = await contribKitProtocols.publishNotification(
	registry,
	documentChanged,
	{ documentId: "document-2" },
	{ context: { tenantId: "tenant-1" }, timeoutMs: 2_000 },
);

publishNotification() shares the bound runtime's validation, resolution, cancellation, deadline and reporting semantics. It owns no listener or provider and releases its transient registry observation without disposing the Core registry.

Provide a listener before the notification point exists

Dynamic plugins can declare notification dependencies without a central dependency graph. provideListener() observes the exact notification point reference and creates one listener registration for every later point generation:

const provider = protocols.events.provideListener(documentChanged, {
	contributionId: "search:document-indexer",
	data: { priority: 100 },
	async listener(payload, delivery): Promise<void> {
		await updateSearchIndex(delivery.context.tenantId, payload.documentId);
	},
});

// A separately loaded plugin may register the point later.
const notificationOwner = registry.rootScope.registerPoint(documentChanged);
await provider.whenFirstActive();

await notificationOwner.close().whenClosed();
// The provider is now waiting. Re-registering the same canonical point creates a
// fresh listener registration with a new Core registration identity.
registry.rootScope.registerPoint(documentChanged);

// Collision or explicit removal leaves recovery under explicit host control.
if (provider.state === "blocked") provider.retry();
await provider.dispose();

The frozen provider handle exposes future-only lifecycle subscriptions, the current Core registration, and an attempt-local error. It owns only its availability observer and exact listener registrations. It never owns the notification point or Core registry. A new point generation recovers automatically; replacement or collision in the current generation requires explicit retry() and therefore cannot create a registration ping-pong.

Define and execute a command

import * as contribKitCore from "@primafuture/contrib-kit-core";
import * as contribKitProtocols from "@primafuture/contrib-kit-protocols";

interface RenameInput {
	readonly documentId: string;
	readonly title: string;
}

interface RenameContext {
	readonly actorId: string;
}

const renameDocument = contribKitProtocols.defineCommand<
	RenameInput,
	void,
	RenameContext
>()({
	pointId: "documents:rename",
	contractVersion: 1,
	concurrency: { mode: "serial", maxPending: 32 },
});

const registry = contribKitCore.createExtensionRegistry();
const pointOwner = registry.rootScope.registerPoint(renameDocument);
const protocols = contribKitProtocols.createContributionProtocols({
	registry,
	defaultTimeoutMs: 5_000,
	cleanupTimeoutMs: 10_000,
});

const handler = protocols.commands.registerHandler(renameDocument, {
	contributionId: "documents:rename-handler",
	async handler(input, execution): Promise<void> {
		if (execution.signal.aborted) return;
		await saveDocumentTitle(input.documentId, input.title, execution.context.actorId);
	},
});

await handler.whenActive();
const callerController = new AbortController();
await protocols.commands.execute(
	renameDocument,
	{ documentId: "document-1", title: "New title" },
	{
		context: { actorId: "user-1" },
		signal: callerController.signal,
		timeoutMs: 2_000,
	},
);

await handler.dispose();
await protocols.dispose();
await pointOwner.close().whenClosed();
await registry.dispose();

parallel is the default command policy. A serial command owns one FIFO for each exact handler generation, shared by every Protocols runtime that resolves that handler. maxPending bounds only waiting calls; 0 allows no queue. A per-call timeoutMs overrides the runtime default, while null disables it for that call. Caller cancellation and deadlines reject the business promise immediately and abort the handler signal. If a running handler ignores abort, the next serial call and handler cleanup still wait for its actual settlement, preserving non-overlap.

Add deterministic middleware

Middleware is local to one Protocols runtime. Global and exact-command registrations are merged for each invocation when it actually starts, ordered by descending priority and then by middlewareId:

const telemetry = protocols.commands.registerMiddleware({
	middlewareId: "telemetry",
	priority: 100,
	async intercept(input, context, next): Promise<unknown> {
		recordCommandStarted(context.command.pointId);
		try {
			return await next(input);
		} finally {
			recordCommandFinished(context.command.pointId);
		}
	},
});

const authorization = protocols.commands.registerMiddleware({
	middlewareId: "rename-authorization",
	command: renameDocument,
	priority: 200,
	async intercept(input, context, next): Promise<void> {
		await requireRenamePermission(context.context.actorId, input.documentId);
		return await next();
	},
});

await protocols.commands.execute(
	renameDocument,
	{ documentId: "document-1", title: "Approved title" },
	{ context: { actorId: "user-1" } },
);

await authorization.dispose();
await telemetry.dispose();

An interceptor can transform input with next(nextInput), transform the downstream output, or short-circuit without calling next(). Each next capability is usable at most once. Removing middleware first excludes it from new snapshots, aborts only invocations that already captured it, waits for their actual finality, and then runs its cleanup callback.

Execute without a persistent runtime

For a one-off call with no persistent middleware or handler ownership, use the free function:

const result = await contribKitProtocols.executeCommand(
	registry,
	renameDocument,
	{ documentId: "document-1", title: "One-off title" },
	{ context: { actorId: "user-1" }, timeoutMs: 2_000 },
);

executeCommand() uses the same validators, handler resolution, cancellation, deadline, error mapping, and handler-generation serial queue as a bound runtime. Its transient registry subscription remains alive only until the invocation's actual resource finality and never extends the caller-visible business promise.

Provide a handler before the command point exists

Dynamic plugins can declare their own dependency without a central dependency graph. provideHandler() observes the exact command point reference and creates one handler registration for every later point generation:

const provider = protocols.commands.provideHandler(renameDocument, {
	contributionId: "documents:rename-handler",
	async handler(input, execution): Promise<void> {
		await saveDocumentTitle(input.documentId, input.title, execution.context.actorId);
	},
});

// A separately loaded plugin may register the point later.
const pointOwner = registry.rootScope.registerPoint(renameDocument);
await provider.whenFirstActive();

await pointOwner.close().whenClosed();
// The provider is now waiting. Re-registering the same canonical point creates a
// fresh handler registration; replacement recovery remains an explicit retry.
registry.rootScope.registerPoint(renameDocument);
await provider.whenFirstActive();

await provider.dispose();

The provider owns only its observer and exact handler registrations. It never owns the point or registry. After a competing replacement, it stays blocked until the host calls retry() or a new exact point generation becomes available.

The host owns the Core registry and its point owners. A protocols runtime owns only the executions, deliveries, transformations, middleware, providers, and exact handler, listener, or stage registrations created through that runtime; protocols.dispose() never disposes the registry.

cleanupTimeoutMs bounds the public disposal promise of Protocols-owned runtimes, providers, and middleware handles. A timeout marks the public handle disposed and aborts its cleanup signal, while actual resource finality continues safely in the background. User cleanup still never begins before every leased invocation has settled. Direct Core registration handles retain Core's own cleanup contract.

Handler removal, point close, registry disposal, or protocols runtime disposal immediately removes the affected invocation's business authority. A handler that ignores its abort signal may settle later, but cleanup waits for that actual finality and no late result can overwrite the caller-visible outcome.

Only package-root imports are public. src/*, internal/*, and package metadata subpaths are deliberately closed.