# 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

![A producer defines a typed stream schema and writes upsert/delete operations into the durable-stream server on port 4437; the durable log fans out over HTTP/SSE to Fresh consumers that materialize the latest value per key.](https://rickylabs.github.io/netscript/netscript/assets/diagrams/streams-pipeline.svg)

*Streams pipeline: producer (defineStreamSchema + createDurableStream) → durable log on :4437 → HTTP/SSE → Fresh consumers (latest value per key).*

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`](https://rickylabs.github.io/netscript/reference/streams/), 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.

> Status — producers write via the core package; manifest helpers fail loud
>
> The producer is implemented:
>
> createDurableStream(...)
>
> from
>
> @netscript/plugin-streams-core
>
> writes
>
> upsert
>
> /
>
> delete
>
> /
>
> flush
>
> through
>
> @durable-streams/client
>
> to the
>
> :4437
>
> Aspire service, and workers, sagas, and auth already mirror their state through it. What is
>
> not
>
> supported: the manifest helpers
>
> defineStreamProducer
>
> /
>
> defineStreamConsumer
>
> in
>
> @netscript/plugin-streams
>
> fail loud — a producer's
>
> publish()
>
> returns a
>
> rejected
>
> promise and a consumer's
>
> subscribe()
>
> throws
>
> synchronously, both with
>
> StreamUnsupportedOperationError
>
> — use
>
> @netscript/plugin-streams-core
>
> instead. There is also no in-process consumer
>
> subscribe()
>
> ; consumption is via the durable-stream HTTP/SSE server (read by Fresh clients).

## 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](https://rickylabs.github.io/netscript/netscript/durable-workflows/sagas/)

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

[trigger](https://rickylabs.github.io/netscript/netscript/durable-workflows/triggers/) 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](https://rickylabs.github.io/netscript/tutorials/live-dashboard/05-live-stream/) 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 →

[Learn — stream a live dashboard  Track D 05: publish execution state from a producer and read it over HTTP/SSE into a live Fresh dashboard.](https://rickylabs.github.io/netscript/netscript/tutorials/live-dashboard/05-live-stream/)

[Do — add the streams plugin  Public package install adds the stream plugin dependency and user-owned glue; local netscript-dev scaffolding supports contributor-source samples.](https://rickylabs.github.io/netscript/netscript/orchestration-runtime/how-to/add-a-plugin/)

[Understand — the plugin system  How thread-isolated plugins like streams register contributions and wire into the Aspire resource graph.](https://rickylabs.github.io/netscript/netscript/explanation/plugin-system/)

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

```ts
// 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',
  },
});
```

```ts
// 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();
```

```ts
// 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);
};
```

```ts
// @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).
```

> Do not call the manifest topic helpers
>
> defineStreamProducer
>
> and
>
> defineStreamConsumer
>
> exported from
>
> @netscript/plugin-streams
>
> are
>
> not
>
> implemented: a producer's
>
> publish()
>
> returns a
>
> rejected
>
> promise and a consumer's
>
> subscribe()
>
> throws
>
> synchronously, both with
>
> StreamUnsupportedOperationError
>
> — they are not silent no-ops and they will not publish anything. Always reach for
>
> createDurableStream
>
> /
>
> defineStreamSchema
>
> from
>
> @netscript/plugin-streams-core
>
> for real producer work. See the
>
> streams reference
>
> for the full export map.

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

**CollectionDefinition — one entry per collection in defineStreamSchema**

| Name | Type | Description |
| --- | --- | --- |
| `schema` | `unknown (Standard Schema validator)` | Standard-Schema-compatible validator (e.g. a zod object) used by durable-streams to validate the collection payload. |
| `type` | `string (required)` | State Protocol type discriminator emitted for every event in this collection (e.g. 'execution'). |
| `primaryKey` | `string (required)` | Property name on the value used as the entity primary key; upsert/delete are keyed by this property. |

`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 `ChangeEvent`s. 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.

**DurableStreamProducerOptions (createDurableStream argument)**

| Name | Type | Description |
| --- | --- | --- |
| `streamPath` | `string (required)` | Stream path relative to the base URL, e.g. '/workers/executions'. This is the singleton key and the path consumers read over HTTP/SSE. |
| `schema` | `StateSchema (required)` | The frozen schema returned by defineStreamSchema; binds the producer to its collection map. |
| `producerId` | `string (required)` | Stable producer identity used for idempotent delivery (IdempotentProducer auto-claim). Keep it stable across restarts for duplicate-safe writes. |
| `signal` | `AbortSignal?` | Optional abort signal consulted while opening the stream connection; aborting cancels the connect (an AbortError is treated as expected, not a failure). |

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

**DurableStreamProducer — methods**

| Member | Shape | Behavior |
| --- | --- | --- |
| `upsert(entityType, value)` | `(K, Record) => void` | Enqueue an upsert keyed by the collection `primaryKey`; skipped (warns) if the key is missing/empty or the collection is unknown. |
| `delete(entityType, key)` | `(K, string) => void` | Enqueue a delete by primary key; skipped (warns) on an empty key. |
| `flush()` | `() => Promise<void>` | Await pending writes before shutdown; **rethrows** the connect error if the producer never connected. |
| `close()` | `() => Promise<void>` | Flush, close the underlying handle, and release the singleton for this `streamPath`. |
| `streamPath` | `string (readonly)` | The stream path this producer owns. |
| `closed` | `boolean (get)` | Whether shutdown has begun; further `upsert`/`delete` calls are ignored. |

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

```ts
// 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).

**Runtime resolvers (@netscript/plugin-streams-core)**

| Name | Type | Description |
| --- | --- | --- |
| `getStreamsUrl()` | `() => string` | Resolves the durable-stream base URL. Server: DURABLE_STREAMS_URL override, else Aspire's services__streams__http__0 discovery var. Browser: VITE_services__streams__http__0 (or the VITE_STREAMS_URL shorthand). Throws a descriptive error if none resolve. |
| `getStreamsAuth()` | `() => Record` | Builds the auth header from STREAMS_SECRET (or DURABLE_STREAMS_SECRET) as { Authorization: 'Bearer ' }; returns {} when no secret is set. |
| `buildStreamUrl(path, baseUrl?)` | `(string, string?) => string` | Joins a stream path onto the resolved base (or an explicit baseUrl), trimming a trailing slash on the base. |
| `inspectStreamTopic(input)` | `(input) => StreamTopicInspectionReport` | Diagnostic: returns a JSON-stable report (package, target, summary, details with collections/streamPath/producerId) for a schema + optional producer metadata. |

**Environment variables read by the resolvers**

| Name | Type | Description |
| --- | --- | --- |
| `DURABLE_STREAMS_URL` | `server override` | Explicit base URL; takes precedence over Aspire discovery (e.g. http://localhost:4437). |
| `services__streams__http__0` | `Aspire (server)` | Injected by the Aspire resource graph; the default server-side discovery path. |
| `VITE_services__streams__http__0` | `browser` | Vite-injected reference for browser/Fresh consumers; VITE_STREAMS_URL is the convenience shorthand. |
| `STREAMS_SECRET / DURABLE_STREAMS_SECRET` | `auth` | Bearer secret for getStreamsAuth(); when set, every connect sends Authorization: Bearer . |

## Known limitations

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

> Writes are dropped after a connect failure (no reconnect)
>
> If the producer cannot reach the
>
> :4437
>
> durable-stream server at startup, it logs a
>
> console.warn
>
> and then
>
> silently skips every subsequent `upsert`/`delete`
>
> — there is no reconnect loop in the current alpha.
>
> flush()
>
> rethrows that connect error so a graceful shutdown surfaces the failure. (Writes issued
>
> before
>
> the connection completes are buffered and drained once it opens; the drop only applies after a connect
>
> error
>
> .) Treat a healthy
>
> :4437
>
> service as a hard precondition for durable delivery; do not assume buffered writes will be replayed once the server returns.

> No in-process consumer — read over HTTP/SSE
>
> There is no in-process
>
> subscribe()
>
> handle. Consumption happens over the durable-stream server's HTTP/SSE protocol, which Fresh clients read to materialize the latest value per key. Model your read side as an HTTP/SSE consumer of the
>
> :4437
>
> stream, not as an in-process callback.

> Local HTTP can limit concurrent stream consumers
>
> The local AppHost serves the generated app over
>
> http://
>
> . Under HTTP/1.1, browsers typically allow about six concurrent connections per origin, so several long-lived HTTP/SSE streams from one Fresh page can starve later requests or stream subscriptions. Use HTTPS for HTTP/2 when a local dashboard needs many simultaneous stream consumers.

> Operator visibility is console.warn (alpha)
>
> The producer reports skips and connect failures through
>
> console.warn
>
> today — tracked as
>
> AP-13
>
> architecture debt until the telemetry-integration wave supplies a structured reporter. Until then, scrape the
>
> [DurableStreamProducer]
>
> warn lines if you need to alert on dropped writes.

## 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`.

**Streams plugin — runtime facts**

| Property | Value |
| --- | --- |
| Plugin location | `plugins/streams/` |
| Producer package | `@netscript/plugin-streams-core` (`createDurableStream`, `createServiceStreamProducer`, `defineStreamSchema`) |
| Manifest import | `@netscript/plugin-streams` — topic helpers **throw** `StreamUnsupportedOperationError` |
| Transport client | `@durable-streams/client` (`IdempotentProducer`) |
| Dev service port | `:4437` (durable-stream Aspire service; override with `STREAMS_PORT`/`PORT`) |
| provider.kind | `stream` · category `plugin` · pluginType `utility` |
| Requires DB / KV | `false` / `false` |
| First-party producers | workers, sagas, triggers, auth (each `streams/producer.ts` → `createDurableStream`) |
| Consumer surface | HTTP/SSE from the `:4437` server (Fresh clients) — **no** in-process `subscribe()` |

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

> Footguns before you ship
>
> - **Build on the core package, not the manifest helpers.** `createDurableStream` / `defineStreamSchema` from `@netscript/plugin-streams-core` are real; the `defineStreamProducer`/`defineStreamConsumer` helpers in `@netscript/plugin-streams` fail loud with `StreamUnsupportedOperationError`.
> - **A healthy `:4437` server is a hard precondition.** A startup connect failure drops every later write with no reconnect — bring Aspire (or an explicit `DURABLE_STREAMS_URL`) up first, and treat a `flush()` rejection on shutdown as a real delivery failure, not noise.
> - **Keep `producerId` stable across restarts.** Idempotent delivery is keyed by it; a churning id defeats the duplicate-safety guarantee.
> - **Model the read side as HTTP/SSE.** There is no in-process `subscribe()`; resolve the base with `getStreamsUrl()` and consume the `:4437` stream path directly.
> - **Set the auth secret on both ends.** When the server expects a bearer token, export `STREAMS_SECRET` (or `DURABLE_STREAMS_SECRET`) wherever the producer runs so `getStreamsAuth()` can attach it.

## 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](https://rickylabs.github.io/netscript/netscript/reference/streams/)

> Where the real surface comes from
>
> The producer you build on lives in
>
> @netscript/plugin-streams-core
>
> :
>
> createDurableStream
>
> , the Service-facing
>
> createServiceStreamProducer
>
> ,
>
> DurableStreamProducer
>
> ,
>
> defineStreamSchema
>
> , the
>
> buildStreamUrl
>
> /
>
> getStreamsUrl
>
> /
>
> getStreamsAuth
>
> resolvers,
>
> inspectStreamTopic
>
> , and the
>
> StreamProducerPort
>
> /
>
> DurableStreamProducerOptions
>
> types. The
>
> @netscript/plugin-streams
>
> manifest re-exports
>
> streamsPlugin
>
> plus the fail-loud topic helpers. See the
>
> full reference
>
> for every exported symbol and the
>
> cli
>
> /
>
> scaffolding
>
> /
>
> e2e
>
> /
>
> aspire
>
> sub-paths.

[Look up — @netscript/plugin-streams-core  The generated API for createDurableStream, defineStreamSchema, the URL/auth resolvers, inspectStreamTopic, and the StreamProducerPort type.](https://rickylabs.github.io/netscript/netscript/reference/streams/)

[Learn — stream a live dashboard  Track D 05: publish execution state and read it over HTTP/SSE into a live Fresh dashboard.](https://rickylabs.github.io/netscript/netscript/tutorials/live-dashboard/05-live-stream/)

[Understand — the plugin system  How thread-isolated plugins like streams register contributions and wire into the Aspire resource graph.](https://rickylabs.github.io/netscript/netscript/explanation/plugin-system/)

[Related — durable sagas  Message-driven orchestration with compensation; sagas mirror their state through a streams producer.](https://rickylabs.github.io/netscript/netscript/durable-workflows/sagas/)

[Triggers & ingress](https://rickylabs.github.io/netscript/netscript/durable-workflows/triggers/) [Database & Prisma](https://rickylabs.github.io/netscript/netscript/data-persistence/database/)