Durable streams

A NetScript stream is a typed, durable change-log: producers write entity state into a durable-stream server, and any number of HTTP/SSE consumers materialize the latest value per key. alpha

NetScript's streams capability is the typed, change-data backbone the other plugins lean on — workers, sagas, and the auth service all publish their live state through it. The producer half is implemented: you define a typed stream schema with defineStreamSchema, open a producer with createDurableStream (or the Service-facing createServiceStreamProducer), and upsert/delete/flush entity state over a durable-stream server that runs as an Aspire resource on port :4437.

The producer runtime in @netscript/plugin-streams-core writes through @durable-streams/client with idempotent delivery. The topic-centric manifest sugar in @netscript/plugin-streams (defineStreamProducer / defineStreamConsumer) is not wired to a transport: a producer's publish() returns a rejected promise and a consumer's subscribe() throws synchronously — both with StreamUnsupportedOperationError, pointing you at the core package. There is also no in-process consumer subscribe() yet — consumption is over the durable-stream server's HTTP/SSE protocol, which Fresh clients read. Build against the core producer package, not the manifest helpers.

What it is

A NetScript stream is an entity-oriented change log. You describe a set of collections — each a named entity type with a primary key — and the producer publishes upsert and delete operations keyed by that primary key. Downstream readers materialize the latest value per key and observe a live, replayable view of your domain state. This is the same contracts-first instinct as oRPC services, but applied to state replication instead of request/response: the schema is the type contract that both producer and any HTTP/SSE consumer are locked to.

Streams sit alongside the other long-running capabilities rather than replacing them. Reach for a durable saga

when you need message-driven orchestration with compensation; reach for a

trigger when inbound HTTP or a file-watch should kick off work. A stream is the read-model fan-out: each of those plugins runs a thin streams/producer.ts that mirrors its execution state through createDurableStream, so a Fresh dashboard can watch saga, worker, and trigger progress live without polling a request/response API.

The workflow this replaces is familiar: a dashboard polls GET /executions on an interval, the interval is always wrong — too fast for the server, too slow for the operator watching a stuck import — and every new consumer re-implements the same polling loop against a slightly different endpoint. With a stream, the contract moves to the data itself. The live-dashboard tutorial builds exactly this: worker execution state published once through a producer, read by a Fresh page that updates without a polling loop anywhere in it. And because the schema is a frozen, typed collection map, what a stream carries is enumerable from its definition — inspectStreamTopic renders it as a JSON-stable report, which is as useful to a CLI doctor or a coding agent as it is to a test.

Learn → / Do →

Minimal example — produce, then consume

The producer side is two calls: freeze a typed schema, open a stream, write entity state. The consumer side is an HTTP/SSE read of the same :4437 stream path — there is no in-process subscribe() handle, so a Fresh island (or any SSE client) reads the durable log directly and materializes the latest value per key. This is deliberate: exposing one HTTP/SSE surface instead of an in-process subscribe() means a browser island and a server-side consumer read the durable log through the exact same path and wire format, with no separate in-VM subscription API to keep in sync.

// streams/executions-schema.ts
// Author against the core package — this is the live surface.
import { defineStreamSchema } from '@netscript/plugin-streams-core';
import { z } from 'zod';

// Each collection is an entity type with a primary key. The schema is the
// type contract producers and HTTP/SSE consumers are locked to.
export const executionsSchema = defineStreamSchema({
  execution: {
    schema: z.object({
      id: z.string().min(1),
      status: z.enum(['queued', 'running', 'succeeded', 'failed']),
      updatedAt: z.string().datetime(),
    }),
    type: 'execution',
    primaryKey: 'id',
  },
});

// streams/producer.ts
import { createDurableStream } from '@netscript/plugin-streams-core';
import { executionsSchema } from './executions-schema.ts';

// createDurableStream returns a singleton producer per streamPath and begins
// connecting to the :4437 durable-stream server immediately.
const producer = createDurableStream({
  streamPath: '/workers/executions',
  schema: executionsSchema,
  producerId: 'workers-service',
});

// upsert/delete are synchronous enqueues keyed by the collection primary key.
producer.upsert('execution', {
  id: 'exec-1',
  status: 'running',
  updatedAt: new Date().toISOString(),
});
producer.delete('execution', 'exec-0');

// flush before graceful shutdown; it rethrows the connect error if the
// producer never connected (see known limitations).
await producer.flush();

// islands/ExecutionsView.tsx — read the durable log directly
import { getStreamsUrl } from '@netscript/plugin-streams-core';

// There is no in-process subscribe(); consumption is an HTTP/SSE read of the
// same stream path the producer writes to. getStreamsUrl resolves the :4437
// base from Aspire discovery / VITE env (see runtime resolvers below).
const base = getStreamsUrl();
const source = new EventSource(`${base}/workers/executions`);

const latest = new Map<string, unknown>(); // materialize latest value per key
source.onmessage = (ev) => {
  const change = JSON.parse(ev.data) as { key: string; value?: unknown };
  if (change.value === undefined) latest.delete(change.key);
  else latest.set(change.key, change.value);
};

// @netscript/plugin-streams (the manifest root) re-exports topic-centric
// helpers that are NOT implemented. They fail loud, by design.
import {
  defineStreamProducer,
  defineStreamConsumer,
} from '@netscript/plugin-streams';

// The producer handle is returned, but publish() rejects.
const producer = defineStreamProducer(/* topic */);
// await producer.publish(event) // -> rejects with StreamUnsupportedOperationError

// The consumer's subscribe() throws synchronously the moment you call it.
const consumer = defineStreamConsumer(/* topic */);
// consumer.subscribe(handler) // -> throws StreamUnsupportedOperationError

// Correct path: import createDurableStream / defineStreamSchema from
// '@netscript/plugin-streams-core' instead (see the other tabs).

Key types first — the stream definition API

A stream schema is a map of collections. Each collection is a CollectionDefinition — a Standard-Schema validator, a State-Protocol type discriminator, and the primaryKey property the producer keys writes by. defineStreamSchema(collections) freezes that map into a StateSchema that both the producer and any HTTP/SSE consumer are locked to.

defineStreamSchema returns a frozen StateSchema<TDef> — the durable-streams runtime attaches per-collection event helpers (insert/update/upsert/delete, the CollectionEventHelpers) so the schema can both validate and emit State-Protocol ChangeEvents. The supported Operation set is insert | update | delete | upsert.

Producer options — createDurableStream

createDurableStream(options) takes a DurableStreamProducerOptions and returns a DurableStreamProducer. It is a singleton factory keyed by streamPath: calling it twice with the same path returns the same live producer (a closed one is replaced). Writes are idempotent via @durable-streams/client's IdempotentProducer (stable producerId + auto-claim), so duplicate enqueues do not double-apply downstream.

The DurableStreamProducer it returns exposes a small, synchronous-write surface with an async flush/close for shutdown. (StreamProducerPort is the implemented-by interface — the same four members, with entityType widened to string.)

Service-side producers — createServiceStreamProducer

When the writer is a backend Service — for example an ingestion worker that emits a doc.ready completion event from the callback where the work finishes — reach for createServiceStreamProducer instead of wiring createDurableStream and the URL/auth resolvers by hand. It is a thin wrapper over createDurableStream that reuses the exact same singleton producer and Aspire discovery (getStreamsUrl / getStreamsAuth), and it adds one guard: assertResolvable (default true) eagerly resolves the streams URL and auth at construction, so a Service that forgot to declare the streams reference throws immediately rather than silently dropping every write.

// services/ingestion/emit-completion.ts
import {
  createServiceStreamProducer,
  defineStreamSchema,
} from "@netscript/plugin-streams-core";
import { z } from "zod";

const completions = defineStreamSchema({
  completion: {
    schema: z.object({ id: z.string().min(1), status: z.string() }),
    type: "completion",
    primaryKey: "id",
  },
});

// assertResolvable defaults to true: this throws now if neither
// DURABLE_STREAMS_URL nor services__streams__http__0 is wired.
const producer = createServiceStreamProducer({
  streamPath: "/support-chat/completions",
  schema: completions,
  producerId: "support-chat-ingestion",
});

producer.upsert("completion", { id: "run-1", status: "done" });
await producer.flush();

A Service that declares ServiceReferences: ["streams"] always has the streams endpoint wired, so the guard resolves without throwing. Set assertResolvable: false to keep the tolerant, lazy-connect behavior of createDurableStream when a Service may legitimately start before the streams service is reachable. ServiceStreamProducerOptions is DurableStreamProducerOptions plus this optional assertResolvable flag; the returned DurableStreamProducer is identical, so the write/flush/close surface below applies unchanged.

Runtime & transport — URL and auth resolution

The producer never hardcodes a host. createDurableStream resolves the durable-stream base URL through getStreamsUrl() and the auth header through getStreamsAuth(), both of which read the environment so the same code works under Aspire, in a browser build, or against an explicit override. buildStreamUrl joins a stream path onto that base, and inspectStreamTopic returns a JSON-stable diagnostic report for a schema (handy in tests and CLI doctors).

Known limitations

Be deliberate about what the alpha producer does and does not guarantee.

Endpoints & manifest

The streams plugin is registered as a utility/infra plugin — note it requires neither a database nor KV (requiresDb=false, requiresKv=false), unlike workers, sagas, and triggers. Its durable-stream service listens on :4437 and is wired into the Aspire resource graph so workers, sagas, and auth can publish through it. The port is overridable via STREAMS_PORT or PORT.

The plugin is referenced from netscript.config.ts as ./plugins/streams/mod.ts. Because workers, sagas, and triggers each list streams in their dependencies, it is installed first in the dependency graph and its :4437 service comes up so dependent producers have somewhere to write.

Production notes

Kept in sync, compared

Convex is the reference point for live client state: sync there is a property of its hosted database — client queries are reactive subscriptions, and the platform re-runs them when the underlying data changes. A NetScript stream scopes the same promise differently: sync is a property of a declared contract, not of the database. You freeze a schema of collections, a producer writes latest-state per key to the :4437 durable-stream service in your own Aspire graph, and any HTTP/SSE consumer materializes that view — independent of which store actually holds your data (the plugin itself requires neither a database nor KV). The practical consequence: you choose per-collection what becomes live state, and everything outside that schema stays on the request/response and persistence paths you already have.

Reference

This hub is intentionally thin — the full generated API lives in the reference.

streams