@netscript/plugin-streams-core

Schema, producer, config, telemetry, testing, and diagnostics primitives for NetScript streams. This page is written against the package's public surface reported by deno doc. For the full index of packages and plugins return to the reference overview.

Publishing change events sounds trivial until it has to be safe: typed payloads, idempotent appends, one producer per stream path, and a clean flush on shutdown. defineStreamSchema declares the collections a stream carries with standard-schema validation and a configured primary key; createDurableStream returns a path-singleton producer whose upsert/delete appends are idempotent and auto-claimed; and the diagnostics helpers inspect a schema or resolve the stream endpoint without opening a socket.

This is the layer the deployable @netscript/plugin-streams service builds on, and the layer the other NetScript plugins use when they project entities — executions, saga instances, sessions — into durable topics.

Exports

Export counts are the symbol counts deno doc reports for each entrypoint; a few domain types are re-exported through more than one subpath.

Root surface (@netscript/plugin-streams-core)

Schema definition

Producers

createDurableStream is a path singleton: two calls for the same stream path with compatible options return the same producer rather than opening a second writer.

Events

Endpoint resolution and diagnostics

inspectStreamTopic is a pure diagnostic: it reports on a schema and optional producer metadata without connecting to the streams server.

Producer port contract (v1)

The *V1 names are the versioned port contract a producer implementation satisfies and a supervisor consumes. They describe write acceptance, delivery outcome, reconnection, and buffering as data, which is what lets the retry policy live outside the transport.

Write outcomes

A write returns a receipt, not a boolean: acceptance is immediate, and the terminal outcome — delivered, rejected, cancelled, or unknown — settles later. unknown is a distinct outcome rather than a failure, because a producer that lost its connection mid-append cannot honestly report either success or rejection.

Bounded streams: retention and trim

StreamRetentionPolicyV1 is a create-time policy accepted by DurableStreamProducerOptions and StreamProducerConnectInputV1:

import type { StreamRetentionPolicyV1 } from '@netscript/plugin-streams-core';

const ttl: StreamRetentionPolicyV1 = { kind: 'ttl', ttlSeconds: 604800 };
const expiry: StreamRetentionPolicyV1 = { kind: 'expires-at', expiresAt: '2030-10-08T00:00:00Z' };

TTL must be a positive safe integer; expiry must be valid RFC3339 with a timezone. Construction validates and copies the policy before connecting. The adapter uses the upstream create options ttlSeconds / expiresAt, sending Stream-TTL / Stream-Expires-At only on PUT. On the shipped server, TTL is a sliding inactivity window renewed by reads and appends; HEAD and reopening with PUT do not renew it. TTL cannot bound a continuously active stream. Use absolute expiry or rotating day segments for a hard bound. Absolute expiry does not slide. Both expire the whole stream. Omitted retention keeps current behavior. Retention participates in singleton compatibility; rotate segment paths to change an existing server policy.

Helpers and StreamAdminError have one public import path, the package root. The adapter signature types are intentionally re-exported from ./admin as well, keeping that entrypoint self-contained for consumers and documentation tooling.

The port returns the existing StreamProducerTransportResultV1<T>. Authorization failures are unauthorized, request deadlines are timeout, caller cancellation is aborted, transient transport/server failures are retryable, and other client failures are non-retryable. Helpers throw StreamAdminError retaining that failure; absence is a successful null/false result. Requests default to 5,000 ms and perform one attempt, leaving retry policy to background workers. The shared failure union now includes unauthorized and timeout; exhaustive consumers must handle these new categories. Producer adapters still report retryable request failures as retryable; timeout alone does not opt a custom producer transport into retries. Both helpers emit client spans through StreamsInstrumentation (stream.head, stream.delete).

Use these helpers in background workers or triggers under a service identity resolved by buildStreamUrl and getStreamsAuth. The app is an interface and need not remain open. DurableStreamProducer.delete(collection, key) appends a tombstone; it does not reclaim stream storage. Stop a segment's producer before deleting it to prevent a reconnect from recreating it. Offset trim and server-side listing remain future protocol capabilities. The stream retention how-to pairs server retention with a scheduled trigger and background deletion worker, including day-baseline guidance.

The SSE contract (@netscript/plugin-streams-core/sse)

The ./sse subpath is the single runtime authority for NetScript's versioned stream SSE envelope. It distinguishes the wire event names the durable-stream server emits from the consumer event names a validated binding delivers, so a consumer never branches on an unvalidated frame.

Replay progress commits only on a control frame: reduceStreamSseReplayStateV1 applies a valid data frame without advancing the committed offset, so a consumer that reconnects mid-batch replays from the last acknowledged control point rather than skipping the tail of an interrupted batch. Errors are normalized into StreamSseErrorPayloadV1 and never advance replay state.

Telemetry and testing

@netscript/plugin-streams-core/telemetry exposes streamsInstrumentation — a registration object matching the minimal contract NetScript telemetry hosts understand — plus StreamSpanNames, StreamAttributes, and StreamProducerMetricNames. The meter, counter, gauge, tracer, and span ports are declared here as interfaces, so a producer emits reconnect metrics without importing an OpenTelemetry SDK.

@netscript/plugin-streams-core/testing provides MemoryStreamProducer (records MemoryStreamEvents instead of opening a socket) and createStreamTopicFixture (a small schema with one execution collection). Tests written against these run without network permissions.

Related pages

• @netscript/plugin-streams — the deployable streams plugin and service.
• @netscript/plugin-workers-core and @netscript/plugin-sagas-core — both project their records into durable topics through this producer.

Back to the reference overview.

Checked command sink

createStreamCommandOutboxSink uses the existing StreamProducerPort with an explicit upsert/delete payload envelope. Stable outbox message identity and correlation are passed through the supported context and serialized in State Protocol headers. Existing producer instrumentation owns W3C context and the producer owns all buffering/retries. The sink awaits delivered completion; FIFO acceptance, rejection, cancellation and delivery-unknown cannot settle. Native duplicate acknowledgements count as delivery. Relay redelivery starts a new producer operation with the same message identity, so downstream processing must be independently idempotent. Cancellation is cooperative around completion; the supplied producer retains lifecycle and permission ownership.

Exports: discovery metadata

The @netscript/plugin-streams-core/discovery entrypoint (./src/builders/discovery.ts) exports NETSCRIPT_CONTRIBUTION_BUILDERS. Pass these immutable patterns through the plugin SDK additionalBuilders option. The CLI composes first-party metadata before walking the project, so single-plugin updates succeed regardless of workers/streams order.