StreamsForge Documentation

Overview

StreamsForge is a streaming platform built on Microsoft Orleans 10 (.NET 10). A source puts events onto a stream — a synthetic generator for the seeded demo world, or a database, a change log, a FIX session, a NATS subject, a file, an HTTP endpoint, a peer instance. You declare what to compute over those streams in a SQL-like dialect, StreamsForge runs it continuously, and the answer leaves through a sink:

Where StreamsForge fits

StreamsForge is a streaming SQL engine with its own connectors, state, governance and console, running as one .NET process. Rows arrive from a source — a database's change log, a FIX session, a NATS subject, a file, an HTTP endpoint, a peer instance — you declare what to compute over them in SQL, the engine maintains the answer incrementally as a materialized table or a windowed pipeline, and the result leaves through a sink. Nothing in that sentence requires a broker, a JVM, a scheduler, a separate state store or a second team.

The comparison that fits best is Flink SQL / ksqlDB / Materialize — genuine incremental view maintenance over changing inputs, not a batch job on a timer — and, inside a Microsoft shop, Azure Stream Analytics, which is what a streaming query usually means there already. The difference is packaging and reach: those systems are clusters you operate next to your application, in a language your .NET team probably does not write, fed by a broker you also operate. StreamsForge is the same class of semantics reachable as dotnet run, as one container, or as the same registration calls inside your own ASP.NET Core host.

Compared with the market

Read this as "what you have to stand up and keep alive", not as a benchmark. Every row is a property of the deployment, and the honest limits are in the next section.

Where Azure Stream Analytics wins, stated plainly, because it is the one on this list a Microsoft-shop buyer is most likely to already own: it is fully managed with nothing to operate, it scales by streaming units with no cluster to provision, and it guarantees exactly-once processing — which StreamsForge does not. Reach for StreamsForge instead when the workload has to run on-premises or inside your own application, when the inputs are a database's change log or a FIX session rather than Event Hubs, when you need a maintained materialized table rather than a windowed query, or when the catalog itself needs per-entity access control and an audit trail.

What StreamsForge is not

Each of these is a deliberate boundary with the reasoning written down somewhere in this document, not an unshipped feature.

Typical deployments

Five shapes this platform was actually built and verified against. Each names the concrete wiring, so the sections below can be read in the order a given shape needs.

Quick start

# prerequisites: .NET 10 SDK (~/.dotnet), Bun / Node 22+
cd ~/work/crates-foundation/orleans
cd web && bun install && bun run build && cd ..
~/.dotnet/dotnet run --project src/StreamsForge.Host     # http://localhost:5199

Open http://localhost:5199 and sign in:

Frontend development with hot reload: cd web && bun run dev → http://localhost:5173 (proxies API and websockets to :5199). Engine tests: ~/.dotnet/dotnet test tests/StreamsForge.Engine.Tests — 196 tests covering the parser, validator, window/join semantics, Z-set algebra, and search index.

Core concepts

Sources

A source is a named stream with a declared schema (String, Double, Long, Bool, Timestamp, Json) and a synthetic generator profile. Json fields can declare their nested shape — a drill-down schema of sub-fields — which documents the payload, powers editor autocomplete, and drives generation for custom sources. Every event carries reserved fields _ts (epoch ms) and _source.

Pipelines

A pipeline is a continuous streaming query over sources. It runs as one Orleans grain: events flow in, the query's operator chain (join → filter → window/aggregate → project) runs incrementally, and result rows stream out to the console. Pipelines understand event time: a watermark trails the newest event by 1 second; windows close when the watermark passes them, and later ("late") events are dropped and counted.

Tables

A table is a persistent materialized view — the current answer to a query, maintained incrementally as data arrives. Unlike pipelines, tables have no windows: aggregates are running totals that update in place. Tables can read other tables, and their state survives restarts. See Materialized tables.

Everything is an entity

Sources, pipelines, tables, and users are all managed the same way: create, edit, delete, start/stop — from the console or the REST API, gated by roles.

The seeded demo world

Every fresh instance (empty data dir on Orleans, empty Redis on Dapr, every containerized cold start) seeds the same self-consistent market-data world — six generator sources, seven pipelines, five tables. Everything below is live out of the box or one click away from Running.

Sources (synthetic generators)

Complex nested financial instruments (plan 002)

The structures + order_events pair is the platform's deep-dive example — nested payloads and entity lifecycles, not just flat ticks:

• Multileg structures, flattened with UNNEST. Each structures event carries a typed legs array (an IR swap's pay/receive legs with ccy/notional/rate fields; an option strategy's 2–4 legs with strike/expiry fields). The Running leg_exposure table flattens every instrument into per-leg rows and aggregates per-currency notional across all legs:

  SELECT l ->> 'ccy' AS ccy, SUM(l -> 'notional') AS notional, COUNT(*) AS legs
  FROM structures s, UNNEST(s.legs) AS l GROUP BY l ->> 'ccy'

  Note the fine print this example exists to teach: -> keeps the raw numeric JSON node so SUM accumulates; ->> would extract text and sum to zero.
• Multistage lifecycle, materialized with LATEST BY. The Running order_states table keeps exactly one current row per live order — retract/assert pairs stream out as each order advances:

  SELECT order_id, symbol, side, stage, stage_rank, stage_ts, qty, filled_qty, px
  FROM order_events LATEST BY (order_id)

  Row history is enabled in LastN(8) mode, so clicking a row in the console shows the order's stage trail (NEW, ACK, the PART_FILLs, then FILLED/CANCELED) — the recent-versions walk, not just the endpoints.
• Windowed fill activity. The Running fill-rate-5s pipeline counts PART_FILL/FILLED events per symbol in 5-second tumbling windows over the same feed.

Pipelines (windowed SQL)

Tables (materialized views)

The Stopped entries are left that way on purpose — starting one from the console (or asking the AI Control chat to do it) is itself a demo of lifecycle orchestration.

Using the console

The Dashboard shows live stat tiles (pipelines, tables, sources, total rows/s) and one card per pipeline with in/out rates, a sparkline, and Start/Stop. Click any card to open it.

Pipelines & the SQL editor

A pipeline page has two panes: the definition on the left, live output on the right (results grid with flash-on-arrival, an SVG chart of any numeric column, and a metrics strip).

• Validation as you type. The editor calls the real compiler (debounced 500 ms) and shows positioned diagnostics — a red squiggle at the exact line:column, with the message below. Valid queries show the compiled plan summary.
• Autocomplete. Context-aware: after FROM/JOIN it suggests sources (and tables, where valid); after alias. it suggests that source's columns plus *; in expressions it offers columns (auto-qualified only when ambiguous), functions, aggregates, and keywords. After a Json column it offers ->/->>, and inside a path string it completes the declared nested keys level by level. Ctrl+Space forces the list; ↑↓ navigate, Enter/Tab accept, Esc dismisses.
• Visual builder. The Builder tab composes the same SQL from dropdowns — sources, join type/key/window, filters, group-by, projections — and round-trips into the editor.
• Save & start compiles, persists, and subscribes in one step. Compile failures mark the pipeline Failed with the error preserved.

Tables & search

A table page shows its definition on the left (same editor, table-mode validation, output schema preview) and the live materialized view on the right: rows update in place as deltas arrive — an aggregate row changes rather than appends. Above the grid sits row search (when enabled): type to swap the live view for matched rows, clear (×) to return. Editors control the index inline: an on/off switch and an Exact ↔ Fuzzy mode toggle (changes reindex the table automatically). See Row search.

Sources

Each source card shows its schema — Json fields expand into a drill-down tree — its rate, an enable switch, and a horizontally-scrollable live tape of raw events. In the editor, setting a field to Json reveals nested sub-field rows (recursively), so you can declare shapes like payload → user → id/tier. The generic profile generates values matching whatever schema you declare, including nested objects.

Users & themes

Admins manage users on the Users page (create, edit role, reset password; self-deletion is blocked). Three more admin pages sit below it in the sidebar: Access (roles, groups, per-user entitlements, approval templates, and a user's flattened effective permissions), Approvals (ungated — it is where a requester watches their own request as well as where approvers vote), and Audit. Controls you are not entitled to use are simply not rendered; the console re-reads its permissions on login, on tab focus, and every 60s while the tab is visible. See Roles, entitlements & approvals. The theme switch lives at the bottom of the sidebar — light is the default; the choice persists per browser and applies instantly with no flash on reload.

Streaming SQL reference

[WITH name AS (SELECT …), …]
SELECT expr [AS alias], … | * | alias.*
FROM source | (SELECT …) alias [, UNNEST(expr) AS l …]
  [ [INNER|LEFT [OUTER]|RIGHT [OUTER]|FULL [OUTER]] JOIN source|(SELECT …) [alias] WITHIN dur ON expr
  | CROSS JOIN source WITHIN dur | [CROSS] JOIN UNNEST(expr) AS l ] …
[WHERE expr]   -- may contain [NOT] IN (SELECT …), [NOT] EXISTS (SELECT …), scalar (SELECT agg…)
[GROUP BY expr, … | LATEST BY (col, …)]   -- LATEST BY: tables only
[WINDOW TUMBLING(SIZE dur) | HOPPING(SIZE dur, ADVANCE BY dur) | SESSION(GAP dur)]
[EMIT CHANGES | EMIT FINAL]
[UNION [ALL|DISTINCT] SELECT … ] …   -- top level or FROM (…) alias only; UNION (no ALL) is tables-only

One statement form exists outside that grammar: INSERT INTO <sink-name> SELECT … is pre-parse sugar handled above the Engine, which never learns about INSERT — it strips the target, stores the untouched SELECT, and enables the named sink. See the sink contract for what it is sugar for, including its one lossy edge on round-trip.

Durations: N MILLISECONDS | SECONDS | MINUTES | HOURS (singular accepted). Keywords are case-insensitive; comments start with --. SELECT * and alias.* expand to all columns (alias-prefixed like t_symbol when the query has multiple inputs); stars are not allowed with GROUP BY/aggregates.

UNION / UNION ALL combines two or more SELECTs of equal arity into one result (branch 0's column names win; kinds unify positionally — Long+Double → Double, everything else must match exactly, JSON/timestamp only with themselves). UNION ALL works in pipelines and tables alike (plain row concatenation — a row from two branches appears twice). Plain UNION/UNION ALL DISTINCT dedups and is tables-only — a pipeline has no Z-set weight to dedup with, so an unbounded distinct over an unbounded stream would be unbounded state; pipelines get a diagnostic naming UNION ALL as the fix. Accepted at the top level (optionally under WITH) and in derived-table position (FROM (… UNION …) alias) — not inside IN/EXISTS/scalar subqueries. A set-operation table runs single-partition only (Parallelism = 1).

Joins

Stream joins are interval joins (the Flink/ksqlDB model): each side buffers events for the WITHIN duration, matched on the equi-key from ON (extra conditions become a post-match filter).

SELECT t.symbol, t.price, q.bid, q.ask, t.price - q.bid AS above_bid
FROM trades t JOIN quotes q WITHIN 5 SECONDS ON t.symbol = q.symbol
WHERE t.side = 'BUY'

Windows, aggregates & EMIT

Aggregates (COUNT(*), COUNT(x), SUM, AVG, MIN, MAX) require a window in pipelines; non-aggregated select items must appear in GROUP BY. Closed windows add window_start/window_end columns. EMIT FINAL (default) emits one row per window+group when the watermark closes it; EMIT CHANGES additionally emits a running update per input event, flagged _final = false.

SELECT symbol, SUM(price * qty) / SUM(qty) AS vwap, COUNT(*) AS trades
FROM trades GROUP BY symbol WINDOW TUMBLING(SIZE 5 SECONDS)

JSON operators

Fields typed Json hold nested objects and arrays, accessed with Postgres syntax — -> returns the JSON value (chainable), ->> returns text. Keys are string literals; array indexes are 0-based integers. Missing keys and NULL propagate to NULL. A JSON path works anywhere an expression does — including join keys and GROUP BY:

SELECT e.eventType,
       e.payload -> 'user' ->> 'tier'    AS tier,
       e.payload -> 'order' ->> 'symbol' AS symbol,
       t.price
FROM app_events e JOIN trades t WITHIN 10 SECONDS
  ON e.payload -> 'order' ->> 'symbol' = t.symbol

Nested queries

Derived tables & CTEs — a FROM/JOIN item can be an inline subquery (alias required), and WITH names them for reuse. CTEs may reference only CTEs declared before them; recursion is not supported (by design — see Limitations). Windows may nest: an inner windowed query's rows enter the outer query timestamped at the inner window end.

WITH hot AS (SELECT symbol FROM trades GROUP BY symbol WINDOW TUMBLING(SIZE 10 SECONDS))
SELECT t.symbol, SUM(t.price * t.qty) / SUM(t.qty) AS vwap
FROM trades t WHERE t.symbol IN (SELECT symbol FROM hot)
GROUP BY t.symbol WINDOW TUMBLING(SIZE 5 SECONDS)

Subquery predicates — [NOT] IN (SELECT …) and [NOT] EXISTS (SELECT …) in WHERE (top-level AND positions) become semi/anti joins. Membership is by presence, so duplicates on the subquery side never fan out, and retracting the last matching row flips all affected rows. NOT IN ignores NULLs from the subquery (a documented deviation from three-valued SQL). Scalar subqueries — (SELECT AVG(price) FROM …) as an expression: single aggregate, no GROUP BY; in tables the result updates continuously with retract/assert cascades; correlated forms are supported for equality correlation only (decorrelated into a GROUP-BY join) — anything fancier gets a diagnostic suggesting a JOIN.

Streaming honesty rule — in pipelines, a subquery used for membership or a scalar must be windowed: its value is a rolling snapshot replaced at each inner window close, and events test against the snapshot current at their arrival. Tables have no such restriction (running Z-sets compose directly).

Arrays, UNNEST & LATEST BY

Sources can declare array fields (e.g. a multileg instrument's legs — see the seeded structures source). UNNEST flattens them: one row per element, in order; empty/NULL arrays produce zero rows. The element binds to its alias as a JSON value addressed with ->/->>:

SELECT l ->> 'ccy' AS ccy, SUM(l -> 'notional') AS notional
FROM structures s, UNNEST(s.legs) AS l
GROUP BY l ->> 'ccy'

⚠ Aggregate over -> (raw value), not ->> — ->> always returns text, and text contributes nothing to SUM/AVG. In tables, UNNEST is Z-set linear: retracting a 3-leg row retracts all 3 flattened rows. UNNEST arguments may reference only real sources (no UNNEST-of-UNNEST).

LATEST BY (key, …) (tables only, replaces GROUP BY) keeps the latest full row per key by event time, emitting retract/assert pairs as keys update — the natural way to materialize current state from an event stream (see the seeded order_states). Late (older-timestamped) arrivals are ignored; an upstream retraction of the retained row drops the key entirely (no fallback to older versions).

Functions, expressions & NULL

Scalar functions: ABS, ROUND(x[, digits]), UPPER, LOWER, COALESCE(a, b, …), IF(cond, then, else). Full boolean/comparison/arithmetic expressions with SQL three-valued NULL semantics (NULL op x is NULL; WHERE requires true). Numeric type promotion is automatic; division always yields a double.

Statistical aggregates: VAR_SAMP, VAR_POP, STDDEV_SAMP, STDDEV_POP and MEDIAN, alongside the original COUNT/SUM/AVG/MIN/MAX. VARIANCE and STDDEV are the bare spellings of the sample forms, which is what SQL means by them. All are subtractable, so they remain correct in a table where LATEST BY retracts a superseded row — an aggregate that only knew how to add would keep a replaced row's contribution forever, and the number would drift silently rather than fail. The result is always a double, even over integer input. VAR_SAMP of a single observation is NULL (undefined, not zero); VAR_POP of one is zero. MEDIAN interpolates linearly between the two middle observations, exactly as PERCENTILE_CONT(0.5) is defined to. PERCENTILE_CONT(p, x) takes the probability first, as SQL does, and p must be a constant in [0, 1] — it configures the aggregate for the whole group, so a per-row expression would silently mean "whichever row arrived first". MEDIAN(x) is PERCENTILE_CONT(0.5, x). COUNT(DISTINCT x) counts distinct values using SQL value equality, so the long 1 and the double 1.0 are one number, and it subtracts like the rest. The moments are accumulated around an offset taken from the first value seen, so a variance over prices near 1e8 with a spread of 1 is still accurate — the textbook Σx² form loses most of a double's mantissa there, and Welford's stable update cannot subtract.

Conditionals: searched CASE WHEN cond THEN expr [WHEN cond THEN expr]… [ELSE expr] END, plus the IF(cond, then, else) it desugars to — both build the identical expression, so there is one set of semantics. An omitted ELSE supplies NULL. Every branch must produce the same type (a Long/Double mix widens to Double, exactly as mixed arithmetic does); disagreeing branches are a compile diagnostic, not a runtime surprise, because a column whose type depends on the row is not something the planner can represent. The condition must be boolean — truthiness here is strict, so a non-boolean condition would silently always take the else-branch and is rejected instead. Only the searched form exists; there is no simple CASE expr WHEN value THEN …. Typical use is turning a computed number into a label in SQL rather than in every client: CASE WHEN d <= 0 THEN 'BREACHED' WHEN d <= 2 THEN 'WATCH' ELSE 'OK' END.

Type conversion: TO_LONG(x), TO_DOUBLE(x), TO_BOOL(x), TO_TIMESTAMP(x), TO_STRING(x) — every real feed arrives with some fields stringly-typed (CSV, form-encoded HTTP, a JSON producer that quoted its numbers), and these make a String-declared field usable numerically without re-declaring its schema. Also spelled as SQL-standard sugar: CAST(x AS type) — accepts the dialect's own type names (STRING/DOUBLE/LONG/BOOL/TIMESTAMP) and the common SQL spellings TEXT/BIGINT/INT/BOOLEAN/DOUBLE PRECISION (not JSON — there's no conversion function to desugar to). Both spellings build the identical expression, so there's exactly one set of semantics. All five are total: an unconvertible value is NULL, never an error — a bad field keeps the row flowing instead of killing the pipeline for everyone else; combine with COALESCE to supply a default. TO_TIMESTAMP accepts epoch-ms (a number, or a numeric string) or ISO-8601 text. TO_STRING is culture-invariant everywhere, and works on a JSON node from -> too — SUM(TO_DOUBLE(payload -> 'qty')) is exactly how to rescue a quoted-number field for aggregation. TO_BOOL's string rule is permissive, not a fixed spelling list: "true"/"false" (case-insensitive) and "0"/"" behave as expected, but any OTHER non-empty string is also true. See DESIGN.md §D11 for the full fine print, including why TO_STRING can't render a bare Timestamp-kind column as ISO-8601 (only TO_STRING(TO_TIMESTAMP(x))/TO_STRING(CAST(x AS TIMESTAMP)) can).

Materialized tables

A table's SQL is the same dialect with different rules — no windows, no EMIT, no WITHIN. Instead of window buckets, aggregates are running totals per group, and joins are relational — INNER (default), LEFT/RIGHT/FULL OUTER, and CROSS — over current state. When a group's value changes, the table emits a retraction of the old row and an assertion of the new one — the console applies both, so you see rows update in place. This is Z-set (DBSP-style) incremental view maintenance: correct under updates, not just appends.

SELECT symbol, COUNT(*) AS trades, SUM(qty) AS total_qty,
       AVG(price) AS avg_price, MIN(price) AS low, MAX(price) AS high
FROM trades GROUP BY symbol

Outer & cross joins. ON may combine several equalities into one composite key — ON a.x = b.x AND a.y = b.y uses both columns as the join key, not one key plus a filter. Every table join is incremental: an outer join's unmatched row is NULL-padded the instant it arrives, and that pad is retracted the instant a match shows up later (re-asserted if the last match disappears) — watching the raw delta stream during cold start shows pad → retract → product chatter, so only the consolidated state the console renders is meaningful. A genuinely non-equi extra condition (beyond the equi-key) stays a residual filter and still works on outer joins, at a cost: matching becomes per-row instead of per-key.

SELECT o.orderId, o.symbol, o.side, t.price AS fill_price
FROM orders o LEFT JOIN trades t ON o.symbol = t.symbol AND o.side = t.side

Tables can read tables. A table may FROM/JOIN another table by name; changes propagate down the chain as deltas (the seeded hot_symbols filters positions). Start order is enforced — a table's inputs must be running first — and stopping or deleting a table that a running table depends on is rejected with a clear error.

Persistence. A table's current rows survive restarts and are served immediately. Operator state (join indexes, aggregate accumulators) is rebuilt from live traffic after a restart — the table shows a Rebuilding badge until fresh data flows.

Row search

Big tables get a per-table reverse search index, toggleable by Editors (inline on the table page). Two modes:

The index updates incrementally with each delta, so results are always current. Search from the table page (results replace the live grid until cleared) or via GET /api/tables/{id}/search?q=…. Toggling the index or switching modes reindexes the table automatically.

Parallel execution

By default a table runs single-partition. Editors can switch it to parallel execution (2–16 partitions) in the table's Execution card — the compiled plan's stages then deploy as a grid of partitioned operator grains (hash-exchanged on join/group keys, epoch-batched), so a hot stage blocks one partition, not the whole table. Under sustained load this bounds the latency tail: in our soak, p99 at 4 partitions came in at 0.70× the single-partition baseline. Changing the setting restarts the table.

The dataflow panel on the table page visualizes the running graph: one column per stage (real operator names — Join, FilterProject, Reduce, LatestBy…), one cell per partition showing its frontier epoch and delta rate, colored by frontier lag. GET /api/tables/{id}/rows gains frontierEpoch for parallel tables: the returned rows reflect all deltas up to that epoch and none beyond it — a real consistency guarantee, not a best-effort snapshot.

Shared arrangements: when several parallel tables join the same source on the same key, the indexed input state is built once and shared (reference-counted, checkpointed, garbage-collected when the last consumer detaches) — inspect live sets at GET /api/meta/arrangements.

Late data: events are epoch-stamped at arrival — no retro-dating, no watermark drops at the table layer. A business-time-late event simply lands in the current epoch and is processed like any other delta. The one query-level exception: LATEST BY ignores rows older (by their own timestamp) than the currently retained row — that's the ordering the query asked for, not a dataflow drop.

Row history

Opt-in per table: keep the version trail of each row as its aggregates evolve. Enable it next to the search config (Editors), pick a retention mode, then click any row in the live grid to open its timeline — versions newest-first with changed fields highlighted, plus a retraction count.

An optional time window prunes versions older than the configured duration. History is fed by the table's own delta stream (the delta stream is the event log), so it never slows the table itself. API: POST /api/tables/{id}/history/lookup with {row} → versions; GET /api/tables/{id}/history/stats. Changing history config or the table's SQL resets the trail.

Retention: bounding a table

A materialized table is unbounded by default — it keeps whatever its query says the answer is, for as long as that stays true. Two opt-in bounds trim it, and both are set on the table definition (PUT /api/tables/{id}) or in its Execution card:

Persistence modes

How a table's materialized rows reach durable storage is a per-table choice (persistence), independent of the query. All four resume the same way — output rows are restored, operator internals (join indexes, aggregate accumulators) rebuild from live traffic, which is what the Rebuilding badge means.

flushMs sets the interval for Batched and FireAndForget (0 = the 2000 ms default; ignored by MemoryOnly), and changing it restarts the table. journalMaxEntries applies to Journaled only — too small and every flush degenerates into a full snapshot write, too large and activation spends its time replaying. On a large table, Journaled is what stops the flush interval from being a latency knob at all.

Sharded tables

Set shardBy to one or more of a table's output column names and every delta the table emits is additionally routed, by those columns' values, to a per-key shard holding just that key's rows and just that key's version history. The shard tier is a second materialization fed by the table's own delta stream — exactly like the row-history tier — so nothing about how the table is computed changes: the SQL path, the planner, the partitioned dataflow and every downstream table-over-table subscriber are untouched. Empty (the default) is today's behavior byte for byte.

The property that makes it worth having: a shard does not pin itself alive. An idle key deactivates, its state lives on disk, and the next lookup brings it back. --Shards:IdleSeconds (default 120) is that collection age; lowering it much below ~90 s also needs --Shards:QuantumSeconds, because Orleans requires the silo-wide collection scan interval to be strictly smaller than any collection age.

?fenced=true upgrades that scan from a set of observations into a genuine consistent cut: the router names the sequence it has forwarded up to and reads every shard without letting another batch through, so no delta at or before the fence is missing and none after it is present — checkably, since the shards' own deltasApplied must then sum to routedDeltasAtFence. It is off by default because the fence pauses the shard tier's ingest for the duration of the scan. Per-key reads need none of this and get none of it.

The rules a sharded table takes on, each because the tier is keyed by name and the shards are the durable copy:

• It keeps no persisted consolidated snapshot. GET /api/tables/{id}/rows is served live from the table's own executor and consults no shard — that route is polled every two seconds by the console, and a keyless listing that fanned out across the directory would wake every shard on every poll. The response carries a shardNote saying exactly that.
• Per-key history replaces the table-wide history grain — running both would double the memory the tier exists to save.
• searchEnabled is rejected outright: a table-wide inverted index keeps every row resident, which defeats the point.
• MemoryOnly is refused, and the table cannot be renamed — the tier is keyed by name.
• Columns are explicit and validated against the table's output fields at upsert. The textual GROUP BY/LATEST BY extraction the history tier falls back on is deliberately not used to pick them silently: best-effort matching is acceptable for "which versions belong together" and not for "which grain owns this row".
• Orleans only. The Dapr flavor stores shardBy and refuses to start such a table, the same shape as parallelism > 1.

Tags & metadata

Sources, pipelines, and tables carry free-form tags and key-value metadata (owner, desk, environment…), editable on each detail page (Editors). Tags show as badges in the lists and are click-to-filter on the pipelines and tables pages.

Roles, entitlements & approvals

Authorization has two layers. The three roles below are still the shape of a normal deployment, and they are now built-in role definitions — named bundles of grants — rather than a hard-coded ordering. On top of them sit per-resource entitlements, groups, an optional approval workflow and an audit log.

Authentication is JWT (12-hour expiry, PBKDF2-hashed passwords). The token does not carry permissions. Every request resolves the caller's grants server-side from the access document, so revoking an entitlement or disabling a login takes effect within Auth:PolicyCacheSeconds (default 10) — not at the user's next login. Auth:Mode=legacy rolls the whole feature back to plain role checks; it defaults to entitlements, and only the exact string legacy disables it, so a typo leaves enforcement on.

Grants: action × scope × effect

A grant is three fields. Grants attach to a role, to a group, or directly to a user; a decision flattens all three lists together.

{ "action": "pipeline.*", "scope": "prod-*", "effect": "Deny", "note": "no prod writes" }
{ "action": "table.read",  "scope": "tag:finance", "effect": "Allow" }
{ "action": "source.write","scope": "*", "effect": "Allow", "requiresApproval": true }

• Action is a flat dotted string with * globs: source.read|write|delete|ingest|run, pipeline.read|write|delete|control, table.read|write|delete|control, config.export|replace, user.read|write, access.read|write, audit.read, approval.request|decide|bypass, chat.use.
• Scope is *, an exact entity name, a prefix glob (prod-*), or tag:finance matching the entity's tags. It is the name, never the id — ids are random GUIDs, so a prod-* grant would match none of them.
• Deny overrides. Any matching Deny anywhere in the flattened list beats every Allow; there is no specificity ordering.
• A single-entity read is refused before the 404, so 403-vs-404 cannot be used to enumerate names. Lists filter rather than refuse: a caller entitled to nothing gets 200 [].
• requiresApproval is not an override — it applies only when no unconditional Allow matches. Adding it beside a role's broad Allow changes nothing; take the broad grant away (or Deny it) instead.

The console's Access page edits all of this — roles, groups, per-user entries and approval templates — and shows a user's flattened effective permissions. Same thing over REST: GET /api/access for the whole document, GET /api/access/effective/{user} for the flattened answer, and GET /api/auth/me for the caller's own.

Disabling a login

PUT /api/access/users/{username}/disabled with {"disabled": true} empties that user's grant set within the same cache TTL — every route, including /api/auth/me, then answers 403 and the console ends the session. It is a one-field call on purpose, so nobody has to re-send an entitlement list they don't know, and it seeds the user's roles from their credential record when no access entry exists yet (so disable → enable returns a working login, not a silently demoted one).

It is not token revocation. The JWT stays valid and signed; what changes is that it resolves to nothing. There is no JTI denylist. Auth:StrictViewer (default true) is what makes even the coarse read policy refuse a disabled principal; it deliberately fails open when the policy store is unreachable or the document has not been migrated yet.

Approvals

Off by default (Approvals:Enabled=false; every /api/approvals route answers 503 naming the key until you turn it on). An approval template matches an action/scope pattern and names the groups that decide it:

PUT /api/access/approval-templates/prod-deletes
{ "actionPattern": "pipeline.delete", "scopePattern": "prod-*", "requiredApprovals": 1,
  "approverGroups": ["prod-approvers"], "expiresAfterSeconds": 3600,
  "escalateAfterSeconds": 900, "escalationGroups": ["oncall"], "enabled": true }

A user files with POST /api/approvals ({action, scope, reason, payloadJson}), watches it on the console's Approvals page, and approvers vote with POST /api/approvals/{id}/approve or /reject; the requester can withdraw with /cancel. The rules, all enforced in one shared state machine:

• You cannot vote on your own request (compared case-insensitively), and the message says so rather than "you are not an approver".
• A vote needs both the approval.decide entitlement at the request's scope and membership of one of its approver groups. One rejection is decisive.
• An inbox shows a request to an administrator, to its requester, or to someone who is both in an approver group and entitled to decide at its scope.
• Template values are snapshotted when the request is filed, so editing a template cannot lower the bar under a request already collecting votes. A background sweeper expires and escalates; a past-deadline request also expires at vote time.
• origin: "chat" marks a request the AI control chat proposed on a human's behalf. With Chat:MayExecutePrivileged=true the model executes such actions itself instead — leave it false.

Give any automated caller its own login and keep it out of every approver group. The self-vote rule compares identities, so a script or an agent authenticating as an administrator files requests a human administrator can then approve — the store cannot tell them apart, because they are the same principal. The MCP server exposes request_approval and not approve for this reason, but that boundary lives in its tool list and nowhere else.

An approval is a record, not an execution. Approving does not perform the action, and an action matched by no template requires no approval at all — so a misspelled actionPattern is a control that silently isn't there. Anything that must fail closed belongs behind a Deny grant.

Audit log

Day-sharded and append-only; the console's Audit page and GET /api/audit/days + GET /api/audit/{yyyyMMdd} read it. Filters are exact actor, an action prefix, limit and offset.

• Every refusal is recorded, always. Allowed reads are not — otherwise "alice changed the prod pipeline" drowns under "alice listed the pipelines" — and neither is allowed source.ingest, one check per message on the hottest path. Outcomes are allowed, denied, requires-approval, executed, failed.
• truncated is on every page and matters. A day is capped at Audit:MaxEntriesPerDay (20 000, drop-oldest) and the dropped count is persisted, so silence is never mistaken for absence. The in-process write queue drops the newest under load — the onset of a burst is the useful end — and emits a real audit.dropped row.
• Before/after payloads are opt-in twice: ?includeChanges=true and access.read at *. Withholding is reported (changesIncluded, changesWithheld), never silent.
• They are always masked, with the same pass the config export uses. A credential rotation therefore reads "passwordHash": "***" → "***": the key's presence is the signal, the value never is (the diff is computed on the unmasked pair precisely so a rotation isn't rendered as "nothing changed").

Two rough edges worth knowing: GET /api/access/effective/{user} short-circuits on a disabled user (it reports the decision — no roles, no grants — so you can't see what the account would hold without re-enabling it), and every guard-level audit row is attributed origin: "rest" even when the decision was made on a gRPC call or a SignalR subscription. Only the chat's rows carry their own origin and onBehalfOf.

REST API

All endpoints are JSON under http://localhost:5199, authenticated with Authorization: Bearer <token> from login. Roles: V Viewer+ · E Editor+ · A Admin — that column is the coarse floor each route still carries; every handler additionally checks its own entitlement at the entity's name, so a 403 can come from either. An interactive OpenAPI reference (Scalar) lives at /scalar — authenticate there with the same Bearer token.

curl -s -X POST localhost:5199/api/auth/login \
  -H 'Content-Type: application/json' \
  -d '{"username":"editor","password":"editor123!"}'   # → { token, username, displayName, role }

Id-or-name resolution

Every read route under /api/{sources|pipelines|tables}/{key} accepts either the entity's ordinal Id or its exact Name — an id match wins outright, otherwise an exact name match. Matching is ordinal (case-sensitive, no prefix, no fuzzy) — the same rule the SQL namespace itself uses internally, so GET /api/tables/Trades and FROM Trades can never disagree about which table that is. Sources have no id at all — the name is the key, everywhere from the REST route to the stream to the SQL namespace — and a source can never be renamed.

curl localhost:5199/api/tables/positions      -H "Authorization: Bearer $TOKEN"   # by name -> 200
curl localhost:5199/api/tables/no-such-table  -H "Authorization: Bearer $TOKEN"   # -> 404
curl -X PUT localhost:5199/api/tables/positions -H "Authorization: Bearer $TOKEN" -d '{...}'  # -> 404

Zero or one match behaves as you'd expect (404 / 200). Two or more entities sharing a name is 409, not 404 (which would claim the entity doesn't exist) and not a silent pick of the first match — the body names every candidate id: {"error": "2 pipelines are named 'dup_name' — address one by id: id1, id2"}. gRPC maps the identical outcome to FailedPrecondition, never NotFound.

Renaming is asymmetric by design. A table's PUT may change name only while it is Stopped, carries no shardBy, and appears in no other table's inputs — the three ways a rename would otherwise strand something (a running query mid-flight, a sharded tier keyed by the old name, a dependent's FROM oldname); verified live — a Stopped, unsharded, unreferenced table renamed cleanly, and its old name 404s immediately afterward. Pipeline names are enforced unique going forward at create/rename time (409 "Name 'x' is already used by another pipeline"); a catalog with pre-existing duplicate pipeline names (legal before this) keeps working and surfaces in catalogWarnings on /api/meta/instance rather than refusing to boot.

Revisions & schema compatibility

Every source, pipeline and table carries two registry-assigned counters — read-only over the API, never client-settable. revision bumps on any definition change. schemaRevision — sources and tables only — bumps only when the field shape changes (a field added, removed, or its type changed); editing eventsPerSecond or a schedule leaves it untouched. That split is what makes a dependency pin useful at all: a cosmetic edit upstream must not invalidate every downstream pin.

Interactive editing stays permissive: PUT /api/sources/{name} allows a breaking field change by default (200). Pass ?allowBreaking=false to opt into the gate instead:

curl -X PUT "localhost:5199/api/sources/trades?allowBreaking=false" -H "Authorization: Bearer $TOKEN" -d '{... "qty" now String, was Double ...}'
# -> 409 {"error":"source 'trades': schema change is breaking",
#         "breakingReasons":["field 'qty' changed type: Double -> String"]}

Config import defaults the other way — gated unless the document opts out. On any upstream schema change (interactive or imported), every table that reads it has its persisted output schema and field numbers refreshed automatically, so reflection and downloaded .proto files stop serving a shape the source or table no longer produces — the dependent table itself is not restarted; it keeps running its already-compiled plan against the new shape. One gap: ConfigTable/ConfigPipeline export/import documents don't carry the purely operational persistence/flushMs/journalMaxEntries knobs, so editing only those through config import never bumps revision (a direct REST PUT does).

gRPC & typed clients

Alongside REST, StreamsForge serves a full gRPC surface on http://localhost:5299 (cleartext HTTP/2). Authentication is the same JWT, passed as gRPC metadata: Authorization: Bearer <token>. The static control plane (package streamsforge.v1) mirrors the REST API — SourceService, PipelineService, TableService with CRUD, lifecycle, and validate, plus StreamService whose server-streaming SubscribeSource / SubscribePipeline / SubscribeTable carry rows as google.protobuf.Struct. REST status semantics map onto gRPC codes: 400 → InvalidArgument, 404 → NotFound, 409 → FailedPrecondition, and role failures → Unauthenticated / PermissionDenied.

Dynamic reflection

The server implements gRPC reflection (grpc.reflection.v1alpha) from the live catalog: every source, every table, and every pipeline whose SQL currently compiles is published as a real protobuf message type in package streamsforge.dynamic.v1 — e.g. table positions becomes messages Positions, PositionsEvent, PositionsDelta. Tools like grpcurl, Postman, or Kreya discover them with zero setup:

grpcurl -plaintext localhost:5299 list
grpcurl -plaintext localhost:5299 describe streamsforge.dynamic.v1.Positions

Schema mapping: String→string, Double→double, Long→int64, Bool→bool, Timestamp→int64 (epoch millis), JSON fields with declared children → nested messages, schemaless JSON → google.protobuf.Struct. Field numbers are stable: the registry persists a per-entity numbering map, so editing a schema keeps existing numbers, gives new fields fresh ones, and marks removed fields' numbers reserved forever — previously generated clients keep working after schema edits.

Runtime-created entities are streamable without restarting anything via the generic typed stream (Viewer+):

service DynamicStreamService {
  rpc SubscribeEntity(EntitySubscribeRequest) returns (stream DynamicFrame);
}
message EntitySubscribeRequest { string entity_key = 1; }  // "source:{name}" | "pipeline:{id}" | "table:{id-or-name}"
message DynamicFrame { string entity_key = 1; bytes payload = 2; int64 seq = 3; }

payload holds the entity's own typed message bytes — {Entity}Event (row + seq + ts_ms) for sources and pipelines, {Entity}Delta (row + weight + seq) for tables, where negative weights are Z-set retractions. The schema is snapshotted when the subscription opens; re-subscribe after editing an entity's schema.

Typed client codegen

The console's API Explorer (sidebar → API Explorer) is the visual front door to all of this: browse every service and entity, read its generated proto definition, copy connect snippets, and watch a live decoded sample — the same rows the gRPC stream carries.

The endgame: a compile-time-typed client library for any entity, generated in one command. Each entity exposes a self-contained .proto (its messages + the DynamicStreamService contract) at GET /api/sources/{name}/proto, /api/pipelines/{id}/proto, /api/tables/{id}/proto (names work too). The bundled CLI turns it into a ready-built .NET library:

./tools/generate-client.sh table positions --user viewer --pass 'viewer123!'
# → downloads positions.proto, scaffolds a csproj (Google.Protobuf + Grpc.Net.Client + Grpc.Tools),
#   generates a StreamsForgeClient wrapper, and builds it.

The generated wrapper gives you login plus a typed IAsyncEnumerable subscription:

var jwt = await StreamsForgeClient.LoginAsync("http://localhost:5199", "viewer", "viewer123!");
await foreach (var delta in StreamsForgeClient.SubscribeAsync("http://localhost:5299", jwt))
    Console.WriteLine($"{delta.Row.Symbol}  avg={delta.Row.AvgPrice}  w={delta.Weight}");
// PositionsDelta — a real generated protobuf type; renames/typos are compile errors.

Because field numbers come from the same persisted registry map as reflection, a client generated today still decodes streams correctly after columns are added — new fields simply appear as unknown fields until you regenerate. For other languages, feed the downloaded .proto to protoc (Go, Python, TypeScript, Java…) — the file is standard proto3 with no custom options.

Client libraries

Six packages, all implementing the same shared design: subscribe → snapshot → replay, then fold the Z-set delta stream into current rows, with reconnect and backoff handled for you. A "live table" object in every language is a table whose rows are always current, without you writing a polling loop.

# Python
sf = streamsforge.connect(url="http://localhost:5199", grpc="localhost:5299", user="admin", password="admin123!")
t  = sf.table("trigger_monitor")     # subscribes, snapshots, replays; blocks until ready
t.df                                 # always-current pandas.DataFrame
sf.snapshot("mc_path_pnl")           # one-shot REST read: no subscription, no thread

// Kotlin
val sf = StreamsForge.connect(url = "http://localhost:5199", user = "admin", password = "admin123!")
val t  = sf.table("trigger_monitor")   // transport = AUTO by default
t.rowsFlow                             // StateFlow<List<Row>>

Need a typed, generated client for one specific entity instead of a generic one? That is the proto download + codegen path, which is a different tool for a different job: a compiled contract for one table's exact shape, rather than a general client for any of them.

Realtime events

The console receives live data over a SignalR hub at /hubs/stream (JWT via access_token query string). Subscriptions are per-entity; the client re-subscribes automatically after reconnects, retrying forever with capped backoff.

Payload formats

A source that fetches bytes — url, file, folder, nats — names the format those bytes are in. One parser per format turns the payload into the same "one JSON element per item" shape, and everything downstream — mapping, field coercion, dedup, the row path — is identical whichever one you picked. Structured sources (postgres, mssql, both CDC kinds, grpc, ingest, crdt) have no format: a result set is already rows.

CSV, and the files that are nearly CSV

The delimiter is sniffed from the header line when you don't name one — candidates , \t ; | counted outside quotes, highest count wins, a tie going to the earlier candidate (so a single-column file reads as comma-delimited, exactly as it did before sniffing existed). The header is the right sample: it is the one line whose cell count is definitional, and it rarely contains free text. That is why a TSV, an Excel export from a decimal-comma locale, and a pipe-separated dump all read through format: "csv" rather than through three more format constants nobody would find.

• Values are typed when they parse exactly (invariant culture) as long, double or bool, and stay strings otherwise — including the empty string for an empty cell.
• A repeated header name keeps the last occurrence and drops the earlier ones entirely, rather than overwriting in an unpredictable order.
• The failure mode is visible, not subtle. Sniff wrong and every row becomes one wide column with a header to match — obvious in the console's schema preview on the first poll, rather than a quiet mis-parse of individual values.

CSV out

The same writer serves every CSV the platform emits, so a download and a sink file are the same bytes:

Both are deliberately their own routes rather than a ?format=csv branch on the JSON ones: the per-entity OpenAPI document rewrites /rows' row shape into that table's real output schema, and a second content type on the same operation would have made that document describe two things at once. For a continuous file instead of a download, use the file sink.

Ingestion connectors

Alongside synthetic generator sources, a source's kind names how rows reach the platform. Every kind still declares a schema (fields) up front, same as a generator; a connector object carries the kind-specific configuration. Whatever a connector emits flows through the exact same door a generator does — onto the source's stream — so pipelines, tables, SignalR, gRPC and the console all work unchanged, whichever kind produced the row.

Three seams, not one. Push-shaped kinds implement IInboundTransport (something else decides when a message exists). Pull-shaped kinds implement IPolledTransport — PollAsync(def, cursor), so the cursor lives in the driver that persists it rather than inside a subscription object that dies with the activation. fix-duplex implements IDuplexTransport, the third seam, whose outbound half a sink reaches by name. Adding a kind is one class plus one registry line; the recipe is TRANSPORTS.md in the repository.

{
  "name": "weather",
  "kind": "url",
  "fields": [
    { "name": "city", "type": "String" },
    { "name": "tempC", "type": "Double" }
  ],
  "connector": {
    "schedule": { "intervalMs": 30000 },
    "url": {
      "url": "https://api.example.com/weather",
      "headers": { "Authorization": "Bearer <secret>" }
    },
    "mapping": {
      "itemsPath": "$.data[*]",
      "dedupKeyField": "city",
      "timestampField": "observedAt",
      "fields": [
        { "sourcePath": "city",   "field": { "name": "city",   "type": "String" } },
        { "sourcePath": "tempC",  "field": { "name": "tempC",  "type": "Double" } }
      ]
    }
  }
}

A connector with no hand-written mapping falls back to a pass-through: itemsPath: "$", one field per declared schema field of the same name — the shape an OpenAPI-derived schema (below) produces on its own.

Scheduling & backoff

connector.schedule is exactly one of cron (5- or 6-field, UTC, parsed by Cronos — the field count is detected by whitespace-token count, not guessed) or intervalMs (fixed interval; 1 s floor — no hot-looping a live endpoint). Omitting schedule entirely defaults to a 30 s interval. A shorthand "every 30s" / "every 5m" / "every 2h" parses to the equivalent intervalMs.

Failure backoff — after k consecutive failures, the next attempt is delayed min(30s × 2k−1, 15 min): 30s, 60s, 120s, … capped at 15 minutes. A cron source doesn't retry mid-interval — it skips straight to the first natural occurrence at or after the backoff point. Success resets the streak to zero immediately. Runtime status is polled by the SPA (2 s) and exposed at GET /api/sources/{name}/status → { nextRunMs, lastRunMs, lastStatus, lastError, consecutiveFailures, eventsEmittedTotal, lastBatchCount } (204 for generator-kind sources, which don't track this).

Mapping documents

A mapping document (JSON or YAML — one model, two encodings, one editor) tells the connector how to turn a fetched payload into rows. Paths use a deliberately tiny, closed JSONPath-lite subset:

Nothing else — recursive descent (..), key wildcards (.*), filters, slices, and negative indices are all rejected with a diagnostic naming the offending token. A missing segment (absent property, out-of-range index, wildcard on a non-array) yields no match rather than an error — only bad syntax throws.

{
  "itemsPath": "$.data.trades[*]",
  "dedupKeyField": "tradeId",
  "timestampField": "tradeTime",
  "fields": [
    { "sourcePath": "tradeId",   "field": { "name": "tradeId",   "type": "String" } },
    { "sourcePath": "price",     "field": { "name": "price",     "type": "Double" } },
    { "sourcePath": "user.tier", "field": { "name": "tier",      "type": "String" } }
  ]
}

dedupKeyField names an emitted field whose value dedups re-polled items — the tracker remembers up to 10,000 keys, FIFO; beyond that bound the oldest key is forgotten and could in principle be re-emitted (documented ceiling: at-least-once, not exactly-once — omit dedupKeyField and every poll re-emits everything it sees, deliberately, if that's not acceptable for the source, don't set one and dedupe downstream). timestampField names the emitted field holding the event time (epoch-ms or ISO-8601) that becomes the row's _ts; omit it and _ts is arrival time. Try a mapping against a sample payload before saving: POST /api/sources/schema/mapping-validate with { document, sample } → { ok, mapping, diagnostics, previewRows } (first 10 rows).

OpenAPI derivation

Rather than hand-write a schema, a url source can derive one from an OpenAPI v3 document (JSON or YAML, by URL or pasted inline): POST /api/sources/schema/derive-openapi with { openApi: { docUrl | docInline, operationId | schemaPointer } } → { fields, diagnostics } for the UI to review and confirm. Selection: an explicit schemaPointer ("#/…") wins; otherwise operationId picks an operation across every path/method, and its response schema is chosen by trying 200, 201, 2XX, default in order, then the first application/json* media type.

Resolution ceiling: only internal "#/…" $refs are followed (cycle-safe — a ref already on the current resolution path becomes schemaless Json + a diagnostic instead of recursing forever); any external or relative $ref is rejected outright the same way, never fetched. The response root, when itself an array, is treated as the items list (its items schema drives field derivation) — pair the derived fields with itemsPath: "$[*]" or an equivalent mapping. Derivation never throws for a malformed document; every failure becomes a diagnostic on a 200 response.

File & folder sources

File (connector.file: { path, format }) polls one file on the schedule; the whole file is re-parsed whenever its mtime changes — there is no tailing, no append-detection, just "did this file change since I last read it." Folder (connector.folder: { path, format, glob? }) polls a directory (no recursion); each file whose name+mtime isn't already in the ledger is parsed once and remembered. Formats: ndjson, json (a JSON array), csv (RFC-4180, header row required).

The name+mtime ledger is persisted per flavor (Orleans: grain [PersistentState]; Dapr: actor state in Redis) — restart-safe: a source that has already seen 9 files still shows 9 known files after a process restart, verified live. Like the dedup tracker, the ledger is bounded at 10,000 entries, FIFO. Because re-parse is whole-file and dedup is at-least-once beyond the bound, there is no cross-restart exactly-once guarantee — pair a file/folder source with mapping.dedupKeyField when the upstream can repeat content across restarts.

NATS subjects & client push

Two kinds where nothing is polled and the schedule is ignored.

nats subscribes to a subject (wildcards are the server's to interpret, not ours) and parses each message in the configured format. queueGroup is the important field with more than one instance: two replicas sharing one group split the subject between them instead of both ingesting every message. Credentials are token, username/password, or the contents of a .creds file (not a path — the catalog has to stay portable across hosts), all masked on read and export. Leaving JetStream unset is a core NATS subscribe: at-most-once, no cursor, nothing left behind on the server. Configuring a durable consumer opts into redelivery and acks — at the price of server-side state this platform then owns.

ingest has no connector at all: rows arrive at POST /api/sources/{name}/events with a body of {"events": [ … ]}. It is authorized two ways — an Editor JWT, or an X-SF-Ingest-Key header validated against that one source's configured keys, which is a machine credential scoped to a single source rather than a platform login. The knobs are about what happens when the buffer is full, and they are deliberately not one number:

Three reasons a row didn't land stay three counters — duplicate (dedup), dropped (capacity) and invalid (coercion) are never collapsed, because the operator's next action differs for each.

Database sources

postgres and mssql poll a table — or your own query — on a durable, monotonic cursor. The cursor is opaque to the platform: the transport mints it, the driver persists it once per cycle, and nothing in between parses it, which is why an LSN, a composite (ts,id) string and a plain bigint all fit without the platform learning what any of them mean.

A failed cycle keeps the old cursor. That rule is enforced once, in the shared polled-source driver, and not by each transport — a transport bug is exactly the case the rule has to protect against, and a rule enforced by the code that might be buggy protects nothing. Throwing from a poll is therefore a normal, expected outcome (a database is down far more often than a config is wrong); the driver reports the error and hands the incoming cursor back unchanged, so the same rows are re-read next cycle rather than skipped. Delivery is at-least-once.

Schema discovery is an optional capability rather than a second interface: POST /api/transports/{kind}/probe returns the fields and diagnostic notes a transport can discover, and 400s when that transport has no probe. The console renders its Discover button from the kind's descriptor, so it offers the button honestly instead of hopefully. There is deliberately no mapping editor for these kinds: for a row source the SELECT list is the mapping, and a JSONPath layer on top would be a second way to say the same thing, free to disagree with the first the moment a column is renamed in one and not the other.

Change data capture

postgres-cdc and mssql-cdc read the source database's own change log instead of polling a cursor column — Postgres logical replication over a slot + publication (pgoutput, via Npgsql.Replication), SQL Server's built-in capture tables (cdc.fn_cdc_get_all_changes_*, cursor is the binary(10) LSN). They are the same polled seam as their siblings above — CDC is still pull-shaped from this platform's point of view, one cycle at a time, even though the source database is doing the pushing internally. Extra fields: slotName/publicationName/createSlotIfMissing (Postgres), captureInstance (SQL Server), tables, maxPollMs.

A batch always ends on a transaction boundary in both dialects. batchSize is therefore a target for a cycle's read, never a hard ceiling on what one batch can emit — a transaction larger than batchSize is delivered whole, over budget, once a bounded re-read resolves it. Emitting a truncated transaction and advancing the cursor past it would be silent, permanent data loss.

Backfill is asymmetric between the dialects. postgres-cdc refuses snapshot outright — a replication slot carries no history from before its own creation, so there is nothing to snapshot; backfill with the polled postgres kind first, then switch the source to postgres-cdc to tail from where the backfill stopped. mssql-cdc's snapshot means "replay whatever the capture table still retains", not a full-table snapshot — bounded by that 3-day default.

For a database this connector does not speak natively — MySQL, Oracle, MongoDB — mapping.envelope: "debezium" is still the right route: Debezium Server emitting into a nats source, whose envelope this platform unwraps. The native kinds are an addition to that path, not a replacement: they exist because this connector already speaks Postgres and SQL Server, so a slot/capture-table reader was cheap. A standalone user guide with worked examples for both kinds lives at docs/cdc.md in the repository.

FIX: format, sessions, and order entry

Three separate things, useful separately and better together: a wire format any bytes-shaped source can name, a receive-only session kind, and a bidirectional session kind for order entry.

The fix format (alongside json/ndjson/csv) parses tag=value into the same shape the other parsers produce — so a FIX log on disk, a FIX message on a NATS subject and a live session all read identically.

{
  "name": "eurusd-md", "kind": "fix",
  "connector": {
    "fix": {
      "host": "fix.venue.example.com", "port": 9880,
      "senderCompId": "STREAMSFORGE", "targetCompId": "VENUE",
      "beginString": "FIX.4.4", "heartBtIntSeconds": 30,
      "resetOnLogon": true, "storePath": "", "msgTypes": "W",
      "onLogon": "35=V|262=1|263=1|264=1|265=1|267=2|269=0|269=1|146=1|55=EUR/USD"
    },
    "mapping": { "itemsPath": "$.MDEntries[*]", "fields": [
      { "sourcePath": "MDEntryType", "field": { "name": "side",  "type": "String" } },
      { "sourcePath": "MDEntryPx",   "field": { "name": "price", "type": "Double" } },
      { "sourcePath": "MDEntrySize", "field": { "name": "size",  "type": "Double" } }
    ] }
  }
}

fix-duplex — order entry. One live FIX session with both halves, declared as two entities that meet in the middle: a source of kind fix-duplex that owns the session (its inbound half behaves exactly like a fix source), and a duplex sink on any pipeline or table naming that source by sourceName. Publishing a row into an entity with that sink attached is sending an order.

Why two entities and not one. A pipeline's or table's sink list is re-signed and rebuilt on any field edit, including an unrelated one — a content hash of the active sink list, re-checked every 30 s. If the session lived in the sink, every such edit would cost a logon. It doesn't: the session lives in the source's connector driver, and the sink only ever asks it, by name, whether it is there.

• The message type comes from the row, not from config. A row needs a MsgType column (tag 35, e.g. "D" for NewOrderSingle), because one duplex source's outbound half plausibly carries several message types across its lifetime and a per-source default cannot express that. Every other column is looked up in a curated name → tag table, and a column with no known tag refuses the whole row rather than dropping just that column.
• Required fields are gated for exactly three message types — NewOrderSingle, OrderCancelRequest, OrderCancelReplaceRequest — refusing an incomplete row before it reaches the wire and naming both the field and its tag number. A MsgType outside those three is not gated at all: this is still not a real dictionary, so it sails through with whatever columns the row carried.
• TransactTime (tag 60) is stamped when absent on those three types and never overwritten when present. QuickFIX/n stamps SendingTime (52) but not 60, and FIX 4.4 requires it there — leaving it unstamped produced a wire message venues reject.
• generateClOrdId is opt-in and off by default, because a row missing ClOrdID is more useful refused than silently completed. A caller-supplied id is never overwritten — and never checked for uniqueness: there is no registry of previously-sent ids. Two rows with the same supplied ClOrdID are both sent as given, and the venue's reject for the duplicate is an ordinary inbound row.
• Reserved platform columns are skipped, not refused. Every real row carries _ts and _source, and a table-sourced row also carries _weight. Before the first live check that sent an order through a real table's SQL output rather than a hand-built row, the mapper refused every single one of them on the unknown-column rule. The generalizable lesson: a test that calls a session directly with a hand-built row cannot see anything the platform stamps onto a row before a sink ever touches it.

// the session itself
{ "name": "fixoe", "kind": "fix-duplex", "connector": { "fix": {
    "host": "oms.venue.example.com", "port": 9881,
    "senderCompId": "STREAMSFORGE", "targetCompId": "VENUE", "beginString": "FIX.4.4",
    "resetOnLogon": false, "storePath": "/var/lib/streamsforge/fix/fixoe.store",
    "generateClOrdId": false } } }

// orders flow in from "new_orders" and out over fixoe's session
{ "name": "orders_sent", "sql": "SELECT * FROM new_orders",
  "sinks": [ { "kind": "duplex", "duplex": { "sourceName": "fixoe" } } ] }

A row {"MsgType":"D","ClOrdID":"ORD-1","Symbol":"EUR/USD","Side":"1","OrdType":"2","OrderQty":1000000,"Price":1.2345} landing in new_orders sends a NewOrderSingle. The venue's ExecutionReport comes back on fixoe's inbound half — and correlates to the order that caused it in ordinary platform SQL, which is the whole trick:

SELECT o.ClOrdID, o.Symbol, e.OrdStatus, e.LastPx
FROM orders_sent o INNER JOIN execs e ON o.ClOrdID = e.ClOrdID   -- execs: SELECT * FROM fixoe

There is no FIX sink. The file sink's format list stays CSV and NDJSON, and writing FIX with no session to number the messages produces something no counterparty would accept.

gRPC federation

A grpc-kind source subscribes to another StreamsForge instance's DynamicStreamService/SubscribeEntity — the federation story. connector.grpc:

The remote schema is fetched once per (re)connect — a snapshot, not a live watch: an edit to the remote entity's schema is picked up only on the next reconnect. Reconnection is forever, with the same backoff formula as scheduled polling; an UNAUTHENTICATED response gets one immediate re-login-and-retry (when logging in, not a static token) before falling back to the normal backoff. A table-keyed subscription has insert-only source semantics: delta frames with weight ≤ 0 (Z-set retractions) are silently dropped — a source can only grow, retraction-aware federation is out of scope.

Try a target before saving: POST /api/sources/schema/from-remote with { grpc } → { fields, fieldNumbersJson, diagnostics } (Editor-only — it dials out). For table/pipeline entity keys, the ident in entityKey is an opaque id, not the remote's generated proto message name (which is derived from the entity's display name) — the reflection path resolves id→name first via a REST GET against restAddress before asking reflection for the symbol; a source:{name} key needs no such lookup, since the ident already is the name.

Named external endpoints (@name)

Any endpoint-shaped string on a connector — a url source's url, a NATS/Postgres/MSSQL/FIX host, an HTTP sink's url, a grpc connector's address/restAddress — can be written as @name instead of a literal, as the whole value: a URL that merely contains an @, like nats://user@host:4222, is left exactly as authored, since only a value that is entirely a reference counts. It resolves at connect time from Endpoints:<name> configuration — --Endpoints:primary-oltp=host:5432, Endpoints__PRIMARY_OLTP, or an Endpoints object in appsettings.json — never from the catalog, and the resolved value is never written back: an export still reads @primary-oltp.

The whole point: one catalog, authored once, imports byte-identical into every environment and connects to a different backend in each — verified live: the same document, never edited, ingested PROD-AAA under one --Endpoints:feed value and DEV-ZZZ under another, while its export still read "url":"@feed" and the resolved host appeared in it zero times.

GET /api/meta/endpoints (Viewer — not anonymous, unlike /api/meta/instance) lists every configured name and its value, for the operator who needs to know what a name resolves to. An unresolvable reference is an error at connect time (there's nothing to dial) but only a warning at import — a document destined for another environment must still be importable, which is the entire point of promotion.

Secrets-lite

The two secret slots on a connector — URL header values and gRPC password/token — follow a deliberately honest, minimal ceiling: stored plaintext at rest in each flavor's own store (Orleans JSON file / Dapr Redis — the same posture as JWT signing config today), no KMS, no encryption. Every read path (GET one, list, export) masks a non-empty secret as "***". Writing back a literal "***" means "keep the stored value" — this is what makes the SPA's GET (masked) → edit → PUT (whole object) cycle safe: it never clobbers a real secret with the mask string. GET /api/config/export omits real secret values unless the caller is Admin and passes includeSecrets=true.

Server plugins & out-of-tree kinds

A source or sink kind does not have to be built into StreamsForge. A library nobody here references can add one by installing two files: a .NET assembly in the host's plugins/ directory (next to the binaries, or wherever Plugins:Path points), and — only if the generic config form isn't enough — an ES module in ui-plugins/ (see below). No rebuild, no configuration beyond the files.

At startup the host loads every assembly in that directory, instantiates each IStreamsForgePlugin it finds and lets it register its kinds. Each outcome is one line on stdout:

[plugins] plugin 'Orion connector 1.2.0' (YourCompany.Orion.OrionPlugin) registered
[plugins] plugin assembly 'broken.dll' could not be loaded: Could not load file or assembly …

Once registered, the kind is a first-class citizen with no further wiring: it appears in GET /api/transports, the console renders its config form from the descriptor it declares, GET /api/meta/instance reports its version for requires: import gates, validation and secret masking apply to it, and both connector drivers run it. Its configuration lives in the platform's open settings bag (a string dictionary on the source/sink, stored, exported and imported like any other config) rather than in a typed field this repo would have to add — which is what lets such a kind add a field, or a whole extra dimension of them, without any change here. Fields the plugin declares as secrets mask as "***" and follow the usual "send *** back to keep the stored value" rule.

The contributor-facing recipe — the interface, the descriptor, the SettingsBag readers, and the one thing the bag cannot express — is in TRANSPORTS.md.

Console UI plugins

Every source and sink kind already gets a working config form for free — the console builds it from the kind's own descriptor (GET /api/transports), so a transport added in an outside library is configurable in the console without a line of front-end code. A UI plugin is the escape hatch for the kinds that form cannot express: a topic browser, a connection tester, a query builder. It replaces that one kind's config panel with the library author's own React component — everywhere the panel appears (the source modal and the sinks editor on pipelines and tables), and without rebuilding or reconfiguring StreamsForge.

Installing one — three steps, no build tooling required:

Checking what is installed: curl -s http://localhost:5199/api/ui-plugins lists the module URLs the console will import — anonymous, like the console's own bundle. If a kind still shows the generic form, look in the browser console: a plugin that fails to import is logged and skipped, never fatal.

Writing one — one ES module, no bundler, no React copy of its own:

// ui-plugins/rv.js
const { react, registerTransportEditor } = window.streamsforge
const h = react.createElement

function RvEditor({ value, onChange, disabled, idPrefix }) {
  return h('input', {
    id: `${idPrefix}-daemon`,
    value: value.daemon ?? '',
    disabled,
    // Spread the previous value: the config holds fields this editor never shows.
    onChange: (e) => onChange({ ...value, daemon: e.target.value }),
  })
}

registerTransportEditor('rv', RvEditor, 'inbound')   // omit 'inbound' to serve the sink half too

The component receives exactly what the built-in editor receives — { descriptor, value, onChange, isEdit, disabled, idPrefix, direction }. Alongside react (the console's own — bundling a second copy breaks hooks), window.streamsforge hands the plugin the rest of what the console already holds: api (authenticated REST, session token and environment header included), live (subscribeTable/subscribeSource/subscribePipeline on the console's single SignalR connection — no second socket for a live preview), and loadLiveTables(), which lazily resolves TanStack DB bound to a StreamsForge table for a plugin that wants query and join on top. apiVersion is 2; check it ((window.streamsforge?.apiVersion ?? 0) >= 2) rather than assuming.

A plugin that throws while rendering loses its own panel and nothing else — the surrounding form, and the stored configuration, are untouched. Registration is per kind and that is the only seam: a plugin cannot add pages, routes, or catalog behavior. The contributor-facing side (props, the registry, the example module) is in TRANSPORTS.md.

Egress: sinks

A sink is where a pipeline's result rows or a table's deltas leave the platform. Every pipeline and every table carries a sinks list; empty (the default) means nowhere. One entity may have several, and each is independently enabled.

"sinks": [
  { "kind": "file",     "enabled": true, "file": { "path": "/var/log/sf/{name}.csv", "format": "csv" } },
  { "kind": "postgres", "name": "warehouse", "db": { "host": "…", "table": "positions", "mode": "upsert",
                                                    "keyColumns": "symbol" } }
]

The contract, and what it deliberately does not do

Three more behaviors that surprise people, in the order they usually bite:

• {name} substitution. A path, subject, table or URL may contain {name}, replaced with the owning pipeline's id or the table's name — so one spec serves a whole catalog.
• The sink list is rebuilt on any field edit. The active sink list is content-hashed and re-checked every 30 s, so editing an unrelated field on the entity tears down and recreates its sink clients. Harmless for everything that opens its own connection lazily — and the reason the duplex sink owns no connection at all.
• A failing sink is visible only in the logs. Failures are counted internally and reported through a throttled callback; the entity's own status has no per-sink field. A sink whose @name endpoint fails to resolve likewise just drops out of its entity with a log line.

Pipelines and tables do not emit the same thing, and sinks that care say so. A table emits deltas: each row carries _weight — +1 for an assertion, -1 for a retraction — so a downstream store can mirror current state. A pipeline emits results: no identity, no weight. That is why mode: "upsert" is rejected outright on a pipeline sink — "mirror current state" means nothing where there is no identity to mirror on.

INSERT INTO <sink-name> SELECT … is pre-parse sugar for the same thing: give a sink a name, write INSERT INTO warehouse SELECT …, and the target is stripped, the untouched SELECT stored, and that named sink enabled. The Engine never learns about INSERT — a destination is already modelled as a sink spec. It is lossy on round-trip: reopening the entity shows the SELECT plus a sink row, and the console says so after save.

The file sink

Appends rows to a path on the host — the egress twin of the file source. path takes {name} and missing parent directories are created. format is csv (default) or ndjson; FIX is not accepted, deliberately (see FIX). columns gives an explicit comma-separated column order — e.g. symbol,qty,_weight — and empty means the first written row's key order.

The postgres and mssql sinks

Write rows into an existing database table. Connection fields mirror the database source's deliberately, so an operator configuring both directions fills in the same form twice rather than two different ones. This is the first sink whose natural unit is a transaction rather than a message, which is why it applies a whole batch at once.

upsert is how you mirror a table's current state into a database. Positive weights upsert on keyColumns, negative weights delete, and deletes are applied last within the one transaction — so a delete-then-reinsert of the same key inside a batch lands the way the caller meant it rather than the way the rows happened to be ordered.

The http and loopback sinks — and the cycle guard

Two ways to feed rows back into a source, one over the network and one not.

http POSTs each row to url. The body is always {"events": [ <row> ]} — exactly one event, wire-identical to this platform's own ingest request, because the destination this sink was built for is /api/sources/{name}/events on a StreamsForge host. A one-row batch is not an optimization left on the table: batching would mean holding rows before sending, which is precisely the unbounded buffering this whole family of sinks exists to avoid. Auth is one generic headerName/headerValue pair, not a bearer-token field — set X-SF-Ingest-Key for a StreamsForge ingest source, or Authorization: Bearer … for anything else, without this config needing to know which scheme the receiver expects. timeoutMs defaults to 3000.

loopback is the same idea with no network hop: it hands the row directly to an in-process hub that the target generator source drains on its own timer and republishes exactly as it would a synthetic tick. No URL, no header, no timeout — there is nothing to configure but targetSourceName (required, {name}-substitutable), which must name a started generator-kind source.

The duplex sink

The outbound half of a fix-duplex session, and the only sink that opens no connection of its own: sourceName names the source that owns the live session, every publish resolves it by name, and forwards. That is what makes the 30-second sink rebuild above free here — there is nothing to tear down. Everything about the row → FIX mapping, the required-field gate, ClOrdID and the sequence-store rules is in the FIX section, because it belongs to the session, not to the sink.

CRDT document sync

A crdt-kind source is an offline-tolerant twin, not a connector — it exists for a link that drops (an ERP integration across an unreliable network) or a document edited from more than one place (a digital twin whose connectivity isn't guaranteed). It is deliberately not an IInboundTransport: a Yjs update is a delta against durable, per-document state, not bytes through a named format parser, so it is driven by its own grain (CrdtDocGrain) rather than the connector driver every other kind shares. It never enters the dataflow — a document merges independently of pipelines/tables/SQL, and what crosses into the platform is the same thing every other source produces: ordinary rows on the source's stream. Orleans-only today; the Dapr flavor stores the kind and refuses to start it, the same shape as a sharded table.

Kind & document shape

Two config fields, both optional:

The document shape is a contract, not a convention. The root YMap's keys are entity keys; each value is either a nested YMap (the entity's attributes) or a scalar (a single-column entity). One key projects to one row. A nested YMap/YArray attribute flattens recursively into a dotted column path (address containing city becomes address.city; an array element's path segment is its index, so tags: ["a","b"] becomes tags.0/tags.1) — every dotted column must still be declared in fields. A document key that collides with a platform-reserved column name (_ts, _source, _weight, _op, _retract) is renamed rather than dropped or refused: strip the leading underscore(s) and prepend doc_, so a document's own _weight field becomes column doc_weight. Y.Text projects as its plain string — formatting is out of scope. A document key with no matching declared field is dropped, never guessed; a value that won't coerce to its declared type is set to null with a diagnostic, matching the platform's lenient coercion-failure default.

{
  "name": "inventory_twin",
  "kind": "crdt",
  "fields": [
    { "name": "sku",      "type": "String" },
    { "name": "qty",      "type": "Long" },
    { "name": "site",     "type": "String" },
    { "name": "_op",      "type": "String" },
    { "name": "_weight",  "type": "Long" },
    { "name": "_retract", "type": "Bool" }
  ],
  "connector": {
    "crdt": { "rootMap": "root", "keyField": "sku" }
  }
}

Declaring the three reserved/tombstone columns above (_op/_weight/_retract) as ordinary schema fields is what lets downstream SQL address them by name, the same as a CDC source's fields declares them — see Three stamps, three readers.

Three stamps, three readers

A source emits weight-1 asserts — there is no way for a row on an inbound stream to say "this key is gone." Deletion (a key removed from the root YMap) therefore has to be expressed as a convention on the row, and this platform reuses the one CDC already established rather than inventing a third spelling: a tombstone row carries only the entity's key plus three stamps, no stale attribute values.

All three are needed because _weight on an inbound row is just a column, not a real retraction — the Engine's Z-set weights are computed from table SQL, never carried in from ingress, so every source event is admitted as a +1 assert regardless of what it says. Without _retract, a deleted key reaches a LATEST BY table as one more create and the table ends up holding both the original row and a second, all-null row for the same key — found live during this feature's development. This is a property of the platform, not a defect this feature introduced: a Debezium/CDC delete has exactly the same limit, and WHERE _op <> 'd' is the identical working pattern used for CDC sources. _retract is what actually frees the key on a LATEST BY table; without it, the row is merely hidden by the WHERE clause, not gone.

Routes & replay

Status codes, in the order checked: 501 when this build has no CRDT document runtime at all (the Dapr flavor); then the usual guard-before-404 ordering so an unentitled caller can't use 403-vs-404 to enumerate names; 404 unknown source; 409 the source exists but isn't crdt-kind; a malformed base64 update is 400, checked last; 200 otherwise.

Why replay exists. A resuming table rebuilds its rows "purely from live traffic going forward" after a restart — correct for a generator or a broker, which keep producing. A CRDT document is not a stream of new events: its value is its current state, and D7 (below) makes re-delivering the update history that produced that state a no-op. So a consumer that lost its rows — a table resumed after a restart, or a table created over a document that was already populated — can never be refilled by re-sending updates; only by re-asserting current state, which is what replay does. It emits one _op = "c", _weight = +1 row per live entity; tombstoned keys don't enumerate, so a deleted entity is never resurrected, and calling it twice is harmless. The boot path issues this automatically for every enabled document once tables are running — this is what makes a silo recycle transparent. The route exists for the runtime case: a new table created over an already-populated document, or an operator who wants to force a refill without restarting anything.

Offline convergence

Re-applying an already-merged update changes no state, so the projection is unchanged and nothing is emitted downstream — a flaky link that redelivers the same batch four times costs four merges and zero downstream events (D7). This is what makes store-and-forward safe, and it has a sharper consequence than "no duplicates": an edge's whole offline session costs the platform exactly its net result, nothing more. Verified live (isolated instance): three transactions made entirely offline — create two entities, correct one's quantity, delete the other — drained as one batch on reconnect produced updatesApplied: 3, rowsEmitted: 1. One row reached the table, holding the corrected value, not the original; the created-then-deleted entity never appeared downstream at all, because create-then-delete inside one offline session has no net effect for anyone watching the projection.

Per-entity authorization & attribution

Per-document access needs nothing new. The two write routes check source.write scoped to the source name, so an entitlement grant already separates one document from another: an exact name, a prefix (prod-*), or a tag (tag:finance) each authorize their documents and no others, and an explicit Deny overrides an Allow as everywhere else. See Grants: action × scope.

requireEntityAuthorization adds a second check per entity the update touches, at scope {sourceName}/{entityKey}, decided before the update is merged — the frame is decoded, never applied, to find out what it would touch. A refused update is dropped from the batch and named in diagnostics; it does not turn the call into a 403 and does not strand the rest of the batch, the same rule a corrupt frame follows.

Note also that the granularity enforced is the entity key, not the field: the grant grammar has an action and a scope, and no field axis to plug a field name into. Field-level inspection is what makes the entity key recoverable for nested content — it is not itself a grantable level.

attributeChanges records the calling identity against each contributing client id, inside the document, so "who wrote this" survives replication the way the content does. Two things to weigh before enabling it. It writes into the document — which is the opposite of the choice made everywhere else here, where history lives outside in the delta journal and row history, both with retention — so it grows durable state that garbage collection will not reclaim. And it answers "who wrote this" but not "who deleted this": the underlying library records a deletion against a user only for a local transaction, and an update merged from an edge is by definition not local. That gap is real, is not worked around, and is stated here rather than left to be discovered.

Every merge and replay also writes one audit row recording what the call actually did — applied, emitted and diagnostic counts — separately from the authorization decision's own allow/deny row. See Roles, entitlements & approvals for reading the log.

The reconciliation pattern

A CRDT converges state. It has no place to reject a state. Two sites can each make an edit that is perfectly valid from where they're standing — a plausible local sale against a plausible local count — and the document that results from merging both is exactly as invalid as any single one of them, because nothing in the merge path ever asked whether the result makes domain sense. The merge always succeeds. A CRDT cannot enforce "stock never goes below zero": that's a global invariant, and a CRDT gives up global agreement by construction to get offline tolerance at all (D1/D2). So a domain invariant here is detected after the fact and compensated — never prevented.

The honest answer this platform has for that is two ordinary entities, built from nothing but streaming SQL over the projection — no new mechanism, because the projection already is queryable rows.

An exceptions/DLQ table, fed by streaming SQL straight off the source (never off a downstream table — see the callout below for why):

SELECT sku, qty, site, _ts AS detected_at
FROM inventory_twin
WHERE qty < 0 AND _op <> 'd'

Run live against a document with two entities — WIDGET-1, independently oversold by two sites that never saw each other's edit (one to qty: -4, the other to qty: -7; Yjs kept -7) — and a healthy GADGET-2 at qty: 12, this table (no LATEST BY — an exception log accumulates, it doesn't converge) held, after a replay of current state:

{
  "rows": [
    { "row": { "sku": "WIDGET-1", "qty": -7, "site": "dock-b", "detected_at": 1787237386729 }, "weight": 1 }
  ],
  "totalRows": 1
}

GADGET-2 never appears — the table is selective, not a mirror of everything. Every subsequent replay while the key stays broken appends another detection row (confirmed live by replaying twice: totalRows went to 2) — a repeat detection is logged, not deduplicated, which is the correct behavior for a DLQ that's supposed to answer "how long has this been broken."

Compensation reading those exceptions — a second table, chained onto the first by name, computing the corrective action. Nothing here re-states the detection predicate; the exceptions table already decided what is broken:

SELECT sku, site, (0 - qty) AS replenish_qty, detected_at
FROM inventory_exceptions

Created live, this comes back with "tableInputs": ["inventory_exceptions"] and "streamInputs": [] — it reads the table, not the document. After the two offline edits above land, the exceptions table holds both detections (it is a log, not a converged view: no LATEST BY), and the compensation table holds one corrective row for each:

{
  "rows": [
    { "row": { "sku": "WIDGET-1", "site": "dock-a", "replenish_qty": 4,
               "detected_at": 1787237911227, "_source": "inventory_exceptions" }, "weight": 1 },
    { "row": { "sku": "WIDGET-1", "site": "dock-b", "replenish_qty": 7,
               "detected_at": 1787237911278, "_source": "inventory_exceptions" }, "weight": 1 }
  ],
  "totalRows": 2
}

_source reads inventory_exceptions, which is the chain saying so in the row itself. From here compensation leaves the platform the ordinary way — a sink on the compensation table — because issuing the replenishment is somebody else's system's job, not this one's.

What this pattern does not give you. It is eventual — there is a real window, between the merge and whatever SQL cycle notices it, during which the invalid state is live and queryable by anyone reading the projection. It is after-the-fact — detection and compensation both run once the bad state already exists; nothing in this pattern, or in the CRDT layer beneath it, can refuse the merge that produced it. And compensating a state does not erase that it existed — the exceptions table is a permanent record, by design, not a queue that drains to empty.

The one case that can be prevented rather than merely compensated is a one-sided numeric bound — "this counter never goes negative," not general domain invariants. That's bounded counters on a plain YMap (Balegas et al., 2015): pre-allocate an allowance across named replicas so each can only ever spend what it locally holds, at the cost of the rebalance itself being an online, pairwise operation (a node that has spent its share stops until it reconnects) and a key count of O(replicas²) — a mechanism for named sites, not thousands of browser tabs. It's a distinct, narrower answer to a distinct, narrower problem, and worth knowing about precisely because it's easy to reach for it where this section's pattern is what's actually called for, or vice versa.

Escrow: bounded counters

A bounded counter (Balegas et al., 2015) recovers exactly one property without synchronous coordination: replica i may only ever spend what it locally holds, so the sum across every replica can never breach the bound K, even when every replica spends at once, entirely in ignorance of the others. State is two monotone maps — T[i][j], the allowance transferred from i to j, and D[i], what i has spent — with local allowance initiali + Σj T[j][i] − Σj T[i][j] − D[i].

No new Ycs type. Every key here has exactly one writer — replica i for its own d:i, and, for t:i:j, whoever owns i's allowance (see Rebalancing below) — so a plain YMap's last-writer-wins is never actually exercised: two replicas spending concurrently write to two different keys (d:a, d:b), and Yjs merges a map with two new keys by keeping both, never picking one. The counter lives as ordinary YMap keys — d:<replica> and t:<from>:<to> — inside a map named by connector.crdt.escrow.counterMap (default "escrow"), a sibling of rootMap: it is bookkeeping, not an entity, and never enumerates as a projected row.

{
  "connector": {
    "crdt": {
      "rootMap": "root",
      "keyField": "sku",
      "escrow": {
        "counterMap": "escrow",
        "reserveReplica": "reserve",
        "initialAllowance": { "reserve": 30, "dock-a": 40, "dock-b": 40, "dock-c": 40 }
      }
    }
  }
}

initialAllowance is the whole allocation policy — which replicas exist and what each starts with — and it is configured, never inferred: a replica name absent from it is not a silent zero, every operation naming an undeclared replica is refused. The bound K is not a separate field: it is exactly what initialAllowance sums to, because a transfer only ever moves allowance between replicas (see below), it never creates or destroys it.

Spending — offline, like any other edit

A spend is not a new route. It is an ordinary content edit — increment d:<replica> on the edge's own local document — shipped to the platform as Yjs update bytes through the existing POST /api/sources/{name}/crdt/updates, exactly like any other change. The edge decides locally, from whatever allowance its own (possibly stale) copy of the document currently shows, whether a spend is allowed at all — refusing it itself, before ever producing bytes to send, when it is not. Staleness only ever makes this more conservative: a transfer the edge hasn't yet seen simply isn't available to it yet, never the reverse. This is the whole mechanism that keeps the global sum bounded with zero synchronous coordination between replicas.

{
  "entityCount": 3,
  "updatesMerged": 9,
  "rowsEmitted": 6,
  "escrow": {
    "bound": 150,
    "totalSpent": 100,
    "replicas": [
      { "replica": "dock-a", "initialAllowance": 50, "spent": 50, "localAllowance": 0,  "exhausted": true  },
      { "replica": "dock-b", "initialAllowance": 50, "spent": 30, "localAllowance": 20, "exhausted": false },
      { "replica": "dock-c", "initialAllowance": 50, "spent": 20, "localAllowance": 30, "exhausted": false }
    ]
  }
}

Rebalancing — the online half

Moving allowance between two named replicas is pairwise coordination and cannot happen offline the way a spend can: deciding it honestly needs the current merged state of both replicas, which an edge working from its own last-synced copy cannot answer for a peer it hasn't seen. So it is the one operation in this feature with a route of its own:

Same 501 → guard → 404 → 409 ordering as the other two CRDT routes. Past that, a refusal — no escrow counter configured on this source, an undeclared replica name, a non-positive amount, or (the ordinary case) from not currently holding enough to give away what was asked — is a 200 with ok: false and a reason, the same "business refusal, not a protocol error" distinction /crdt/updates' own per-update diagnostics already draw. A refusal writes nothing to the document.

Moving allowance between two spending replicas is the giver's own operation, not the coordinator's. The giving replica performs the transfer on its own document and ships the result through POST /api/sources/{name}/crdt/updates like any other edit. That is sound for the same reason a spend is: the giver deducts before anyone else can use the allowance, so it can never later spend what it gave away. This is exactly what plan 020's escrow section means by "only replica i writes T[i][*]".

Rebalancing does not create or destroy allowance: within one converged document the bound K is unchanged by any transfer amount, because what is subtracted from from's formula is exactly what is added to to's. That algebra is not on its own a safety argument — it holds for the converged state, while a replica spends against its own view, which is precisely the gap the reserve rule above closes. The refusal above exists anyway, because an unchecked transfer would still let from's own local allowance go negative — a permanent local wrong answer (every future spend for that replica refuses, even once nothing more is asked of it) with no offsetting benefit. Refused for the same "refuse rather than oversell" reason a spend past allowance is.

The four limits

Awareness: presence/liveness

Awareness answers a different question than the rest of this section: not "what does the document say", but "who is looking at it right now". It is ephemeral by definition — a viewer's cursor, or simply the fact that someone has this document open, is not part of the document and must never become part of it. It lives entirely in a host-process registry (AwarenessRegistry, shared/StreamsForge.Api/Hubs/) and never reaches CrdtDocGrain, PendingUpdates, the delta journal, or any persisted state — restarting the host clears every presence entry, by design, the same way closing every browser tab would.

Off by default, and reached over SignalR only — there is no REST route for this feature at all. Setting connector.crdt.awareness is the opt-in itself (no separate boolean); its absence means StreamHub.SubscribeAwareness refuses to join anyone to the document's presence group.

{
  "connector": {
    "crdt": {
      "rootMap": "root",
      "keyField": "sku",
      "awareness": { "ttlSeconds": 30, "maxEntries": 50 }
    }
  }
}

Both numbers are deliberate operator decisions, not defaults to leave unexamined — this is the plan's own framing for the whole wave: awareness is "the thing most likely to flood a link", precisely because presence traffic scales with how many viewers a document has and how chatty their heartbeats are, not with anything the document's own content does.

The hub surface — not REST

Three StreamHub methods (shared/StreamsForge.Api/Hubs/StreamHub.cs), invoked over the same SignalR connection table/pipeline/source subscriptions already use:

Membership changes push as an awarenessUpdate event — (sourceName, peers[]) — to every other connection in the document's group: on join, on leave, on disconnect, and on any heartbeat whose own eviction pass discovered a stale peer. An ordinary heartbeat that only refreshes the caller's own expiry broadcasts nothing, which is what keeps steady-state traffic at roughly one small message per heartbeat interval per member rather than one per member per member.

Authorization: the same read grant a REST read would need

SubscribeAwareness asks AccessGuard for source.read at the exact scope a REST read of that source already asks for — presence reveals who is working on which document, so it is gated exactly like reading the document's own rows is. See StreamHub's own class doc (plan 015 wave 3-B) for why every subscribe method in this hub checks its own entitlement rather than relying on the class-level policy alone.

Heartbeat and UnsubscribeAwareness check nothing of their own — the same "checked at subscribe time only" rule the rest of this hub follows. That is safe specifically because AwarenessRegistry.Heartbeat can only refresh an entry SubscribeAwareness already created under a guard check; it can never create one, so calling it without ever having subscribed is a no-op, not a back door.

The cap: visible, not silent

The (maxEntries + 1)th distinct connection to join an already-full document is refused outright, naming the cap and the current count — never a silent drop, and never an eviction of someone already present to make room. Re-subscribing a connection that already holds an entry never counts against its own slot. This mechanism is sized for a bounded set of live editors/viewers on one document — tens, not thousands of browser tabs, the same scale escrow's own initialAllowance is sized for.

The TTL needs no background timer, and that is deliberate

Every call that touches a document's live set — join, heartbeat, leave, disconnect — evicts every expired entry in that document first, under the same lock, before doing anything else. So as long as at least one connection keeps heartbeating (which every live participant must do to keep its own entry alive), every other connection's expiry is discovered — and broadcast — within roughly one more heartbeat interval of that live participant. No hosted service, no per-document scheduled work sitting idle between calls.

The one typed client this wave ships is TypeScript's AwarenessSession (clients/typescript/src/awareness.ts), over its own dedicated SignalR connection — independent of whichever transport connect() chose for table deltas, since awareness has no gRPC equivalent in this platform. Client.awareness(sourceName, opts) returns a joined session exposing .peers and .onUpdate(listener); it runs its own heartbeat loop internally (roughly a third of the server's ttlSeconds, floored at 5s) and .close() leaves cleanly. The Python, Kotlin, .NET and React clients are unchanged by this wave — see the plan's own cut list for why awareness was scoped to one client rather than four.

Instance discovery & federation

Two layers, and the first is most of the value. Layer 1: every instance answers GET /api/meta/instance anonymously, like /healthz — enough for a peer, the admin app, or a human with just an address to learn what it is. Layer 2: a configured directory of peers, each probed via that same endpoint — which is what lets a grpc source federate by peer name instead of a hardcoded address.

Instance identity

GET /api/meta/instance is anonymous, deliberately: a peer has to learn what a deployment is before it holds any credential for it. It reports identity, capability, and catalog health, never entity names:

curl localhost:5199/api/meta/instance
{
  "instanceId": "7f44872c866646e2aeb4afa9831704f9",
  "name": "prod-1", "flavor": "orleans", "version": "1.0.0+85aa5d08",
  "endpoints": { "rest": "http://localhost:5199", "grpc": "http://localhost:5299" },
  "capabilities": ["grpc"],
  "plugins": ["generator","url","file","folder","grpc","nats","fix", "..."],
  "catalogCounts": { "sources": 6, "pipelines": 7, "tables": 5 },
  "catalogWarnings": [],
  "startedAtMs": 1787190620573
}

instanceId is persisted at {DataDir}/instance.json, so it survives a restart — deleting that directory mints a new identity, which is the documented reseed on the Orleans flavor and is merely arbitrary on Dapr, whose state otherwise lives entirely in Redis. catalogWarnings reports counts and kind names only — "1 pipeline(s) have a stale pin", "1 entit(ies) use kind 'x', which this instance has no connector registered for" — never an entity's own name, since this route has no credential to gate that on. There is no switch to turn the route off.

Peer directory

Configured peers come from Discovery:Peers — a section, so it binds as an array from any provider: CLI (--Discovery:Peers:0:Name producer --Discovery:Peers:0:RestEndpoint http://host:5199 --Discovery:Peers:0:GrpcEndpoint http://host:5299), env (DISCOVERY__PEERS__0__NAME), or an appsettings.json Discovery.Peers array. GET /api/meta/peers and POST /api/meta/peers/{name}/probe are Viewer-gated — unlike /instance above, this route is about this instance's own configuration, which does need a credential:

curl localhost:5199/api/meta/peers -H "Authorization: Bearer $TOKEN"
# [{"name":"producer","instanceId":"","restEndpoint":"http://localhost:7921",
#   "grpcEndpoint":"http://localhost:7922","lastSeenAtMs":0,"lastError":null,"info":null}]

curl -X POST localhost:5199/api/meta/peers/producer/probe -H "Authorization: Bearer $TOKEN"
# -> the same record, now with instanceId/lastSeenAtMs/info filled in from the peer's own /instance

Federated addressing

A grpc-kind source's connector.grpc.peer names a configured peer instead of an address/restAddress literal, resolved fresh at every (re)connect. When set, it wins over both address fields rather than filling them in when blank — a source naming a peer alongside a stale literal address must never silently connect to the stale one. Combined with id-or-name resolution, this is what lets an entityKey be authored as table:positions instead of table:<guid>: the consumer's existing id→name REST round trip (already needed to translate a remote reflection symbol) runs in reverse to canonicalize it.

{
  "connector": { "grpc": {
    "peer": "producer",
    "entityKey": "table:positions",
    "username": "admin", "password": "admin123!"
  } }
}

No address, no GUID, anywhere in the source's own definition — verified live: two instances, one federating a table from the other by peer name and table name alone. An unresolvable peer name takes the existing status-error path at the existing backoff, same as an unreachable literal address always has.

Environments

A named partition inside one running server — its own sources, pipelines and tables, its own name uniqueness, its own SQL namespace (FROM orders in staging can never resolve to default's orders), its own grain/actor keys. Every deployment already has one: default, which nothing has to create and which is exactly today's catalog under a new name.

Creating & listing

GET /api/environments (Viewer) always returns default first — synthesised, never stored, present even on an instance that predates this feature entirely — then every created environment, name-ordered, each with a live entityCount (sources + pipelines + tables, −1 if it could not be counted). Creating and deleting are Admin-only, deliberately: an environment is the one part of this feature that is genuinely destructive to get wrong, so a typo must never silently deploy into a brand-new empty catalog nobody meant to make.

curl localhost:5199/api/environments -H "Authorization: Bearer $TOKEN"
# [{"name":"default","description":"...","createdAtMs":0,"createdBy":"","entityCount":18}]

curl -X POST localhost:5199/api/environments -H "Authorization: Bearer $TOKEN" \
  -d '{"name":"staging","description":"pre-prod"}'
# -> 200 {"name":"staging","description":"pre-prod","createdAtMs":...,"createdBy":"admin","entityCount":-1}

curl -X POST localhost:5199/api/environments -H "Authorization: Bearer $TOKEN" -d '{"name":"default"}'
# -> 400 {"error":"'default' is not a valid environment name — ... and not one of the reserved names
#          (default, catalog, users, access, approvals, audit, events, metrics)."}
curl -X POST localhost:5199/api/environments -H "Authorization: Bearer $TOKEN" -d '{"name":"staging"}'
# -> 409 {"error":"Environment 'staging' already exists."}

Names are lower-case letters, digits and hyphens, 1–32 characters, and may not be one of the words already claimed elsewhere in this key space (default, catalog, users, access, approvals, audit, events, metrics). No environment can ever be renamed — there is no route for it — for the identical reason a sharded table can't be: the name is qualified into every grain key, actor id and stream id that environment's entities ever produced, and a rename would silently orphan all of it.

Selecting one

Every request that names an environment does so with the X-StreamsForge-Environment header; a ?env= query parameter overrides the header when both are present, for a browser navigation, an <img>/download link, or a curl that would rather not set a header. Naming default explicitly, an empty value, or naming nothing at all are the same thing and cost nothing extra — no lookup, no per-request state write. Naming anything else validates against the environment registry before the request reaches its route:

curl localhost:5199/api/tables -H "Authorization: Bearer $TOKEN" -H 'X-StreamsForge-Environment: staging'
# [] — staging's own (empty, until something is created in it) table list

curl localhost:5199/api/tables -H "Authorization: Bearer $TOKEN" -H 'X-StreamsForge-Environment: nope'
# -> 404 {"error":"environment 'nope' does not exist"} — refused before any catalog is touched

curl -X POST "localhost:5199/api/environments?env=staging" -H "Authorization: Bearer $TOKEN" \
     -H 'X-StreamsForge-Environment: nope'
# ?env= wins: the request resolves against staging, "nope" is never consulted

The 404 fires only after authentication and authorization run — an anonymous caller can never use this route to enumerate which environment names exist by probing the header against a protected path, because they never get past 401 to see it. /api/environments itself, /healthz, /api/meta/instance and every /api/auth/* route ignore the header entirely: the first is how a client recovers from having its own selected environment force-deleted out from under it (it must stay reachable to discover that, or an X-StreamsForge-Environment pointed at a name that no longer exists 404s on every route, including the one that would tell you why), and the rest describe the server process or a global user account, neither of which an environment partitions.

The console carries a matching environment picker next to the theme toggle: it stores the selection in the browser (not a server-side session), injects the header on every REST call, and passes ?env= to the SignalR connection — a stale selection pointing at a since-deleted environment falls back to default automatically rather than dead-ending on repeated 404s.

Force-delete, and exactly what it erases

DELETE /api/environments/{name} (Admin) refuses a non-empty environment outright; add ?force=true to delete it anyway. default refuses unconditionally, force or not — it always exists and this is the one route that can never touch that.

curl -X DELETE localhost:5199/api/environments/staging -H "Authorization: Bearer $TOKEN"
# -> 409 {"error":"Environment 'staging' is not empty (1 source(s), 0 pipeline(s), 1 table(s)) —
#          pass force=true to delete it and everything in it."}

curl -X DELETE "localhost:5199/api/environments/staging?force=true" -H "Authorization: Bearer $TOKEN"
# -> 204, and default's own tables/sources are provably untouched afterward

curl -X DELETE "localhost:5199/api/environments/default?force=true" -H "Authorization: Bearer $TOKEN"
# -> 409 {"error":"The default environment cannot be deleted."}

Force-delete is the one genuinely destructive operation this feature adds: it deletes the environment's catalog and the runtime state of everything in it — every source, pipeline and table is torn down through the same code path a manual one-at-a-time delete uses (so a Running table is stopped first, its history/shard tiers released), then the environment record itself is removed. The console's picker makes this a deliberate two-step confirmation — a plain delete first, which escalates to an explicit "force-delete erases catalog AND runtime state" dialog only after the server has confirmed the environment is non-empty.

Configuration import/export

The catalog — source, pipeline, and table definitions — can be exported to a portable document and imported back, including across the two runtime flavors: export from Orleans, import into Dapr (or the reverse), or version the document in git and promote it between environments. What travels is configuration only: no ids, no users/credentials, no runtime state (dedup ledgers, connector status, pipeline/table row data never leave the process this way).

Document format

One canonical shape, two encodings — canonical JSON or YAML, both accepted on import; export offers either. GET /api/config/export?format=json|yaml&includeSecrets=true (Viewer; includeSecrets additionally requires Admin, per the secrets-lite ceiling above).

{
  "version": 1,
  "sources": [
    {
      "name": "weather", "kind": "url",
      "fields": [ { "name": "city", "type": "String" }, { "name": "tempC", "type": "Double" } ],
      "connector": {
        "schedule": { "intervalMs": 30000 },
        "url": { "url": "https://api.example.com/weather", "headers": { "Authorization": "***" } }
      }
    }
  ],
  "pipelines": [
    { "name": "hot_weather", "sql": "SELECT city, tempC FROM weather WHERE tempC > 30", "running": true }
  ],
  "tables": [
    { "name": "latest_by_city", "sql": "SELECT * FROM weather LATEST BY (city)", "running": true,
      "searchEnabled": false, "searchMode": "Exact", "historyEnabled": false, "historyMode": "All",
      "historyLimit": 10, "historyWindowMs": 0, "parallelism": 1 }
  ]
}

Pipelines/tables carry running as desired state, not the live Status enum — a pipeline that's currently Failed exports as running: true (it was asked to run and never told to stop; only Stopped maps to false). A top-level include array names other documents (relative paths) to compose in before this one — see below. Canonical JSON export is deterministic: 2-space indent, camelCase, entities sorted by name, fixed property order, empty/null fields omitted — so export → reset → import → re-export is byte-identical (verified live, wave 3C and again end-to-end in W6).

Compose & merge rules

Multiple documents combine into one before anything is applied. Precise rules:

• A document's own include list resolves depth-first, in order, before the document itself — so "the includer always wins over everything it includes," recursively for nested includes.
• Across the resulting ordered sequence, later documents win per entity, matched by (kind, name).
• The merge is a shallow field override: a property present in the later entity replaces the earlier value entirely — arrays and objects replace whole, never element-merged or deep-merged.
• An explicit JSON null for an optional field clears it (distinct from the field being absent, which leaves the earlier value untouched).
• An include cycle or a missing include is fatal — a diagnostic naming the path/chain, no partial composition.

Includes resolve two ways depending on how the import arrives: an ordered multi-document JSON array composes directly (no filesystem involved — later element wins, same rules); a multipart file-set upload resolves each document's include entries by exact file name within the uploaded set only — the server never reads its own filesystem for an include, by construction (no path traversal possible). A single document with a non-empty include list is rejected outside a multipart upload — there's nothing on the server for it to resolve against.

Dependency pins

A pipeline or table's config document can carry "dependsOn": [{ "kind": "source", "name": "trades", "schemaRevision": 3 }] — a pin against the exact schemaRevision the query was authored against (schemaRevision: 0 declares the dependency edge with no compatibility claim — never stale, by design, which is what makes it safe to record a dependency nobody explicitly pinned a shape to). There is no SQL-level equivalent — FROM trades@3 does not exist and won't; the engine runs against live streams, not a versioned store — so pinning lives in the config document only.

A violated pin never stops anything: the entity's staleReason is set to a sentence naming what moved and from what, rolled into catalogWarnings on /api/meta/instance as a count, while the entity keeps running its already-compiled plan.

Connector version requirements

A config document can declare "requires": [{ "kind": "postgres-cdc", "version": "^2.0.0" }] — a SemVer-range subset (^, ~, >=a <b, */omitted = any) checked against this instance's own registered kinds (the same list plugins reports on /api/meta/instance) before anything else in the document is touched. An unsatisfied requirement refuses the whole import — not just the entities using that kind — identically under mode=validate and an apply mode:

curl -X POST "localhost:5199/api/config/import?mode=merge" -H "Authorization: Bearer $TOKEN" \
  -d '{"requires":[{"kind":"url","version":"^2.0.0"}], "sources":[...]}'
# -> HTTP 200 {"mode":"merge","ok":false,
#     "entries":[{"kind":"requires","name":"url","action":"error",
#       "diagnostics":["kind 'url' is installed at version '1.0.0', which does not satisfy the required range '^2.0.0'"]}]}
# nothing in the document was applied — every built-in kind reports version 1.0.0 until a plugin
# system exists to declare otherwise; a document that never sets "requires" is unaffected.

Import modes

POST /api/config/import?mode=validate|merge|replace — body is a single document, an ordered JSON array, or a multipart file set (detected by Content-Type).

Ordering matters and is fixed. Apply order: sources → tables (topologically sorted so a table's input table lands before it) → pipelines. replace-mode deletions run last, in reverse dependency order: pipelines → tables (reverse-topo) → sources — a running entity is stopped before it's deleted, mirroring the same stop-then-replace discipline as any other dependency-respecting shutdown. Every imported pipeline/table SQL is recompiled through the real Engine compiler against the composed post-import catalog; a compile failure marks that one entity error in the report without aborting the rest of the import (validate reports every entity's outcome the same way; apply modes simply skip the failed one).

The response is a per-entity report: { mode, ok, entries: [{ kind, name, action, diagnostics[] }] }, action one of created | updated | deleted | skipped | error.

# export (JSON, secrets masked)
curl -s localhost:5199/api/config/export -H "Authorization: Bearer $VIEWER_TOKEN" -o export.json

# validate first, then merge for real
curl -s -X POST "localhost:5199/api/config/import?mode=validate" \
  -H "Authorization: Bearer $EDITOR_TOKEN" -H 'Content-Type: application/json' --data @export.json
curl -s -X POST "localhost:5199/api/config/import?mode=merge" \
  -H "Authorization: Bearer $EDITOR_TOKEN" -H 'Content-Type: application/json' --data @export.json

# multipart include set — base.json includes overlay.json by file name
curl -s -X POST "localhost:5199/api/config/import?mode=merge" \
  -H "Authorization: Bearer $EDITOR_TOKEN" \
  -F "base.json=@base.json" -F "overlay.json=@overlay.json"

Auth summary: export is Viewer (Admin for includeSecrets=true); import is Editor; replace additionally requires Admin. Whole-import atomicity is not promised — each entity applies through the same serialized catalog registry (grain/actor turn-based), so per-entity writes are atomic but a failure partway through an import leaves earlier entities in this run already applied (the report shows exactly what happened, entity by entity).

Promotion is gated on schema shape by default: an entity whose incoming fields are a breaking change from what's stored — a type change, a removed field — is refused (action: "error", the field-level diagnostic naming what changed) unless the document sets "schemaPolicy": "any"; that string, and only that string, turns the gate off — a typo ("compatable") leaves it on. This is the opposite default from interactive editing, and deliberately: promoting a document is not the moment to discover a generated client three deployments away stopped decoding.

AI control chat

The AI Control page (Editor/Admin roles) is a conversational surface over the same catalog the REST API manages: ask it to list, create, pause, or delete sources, draft and validate SQL pipelines, or inspect table rows, search, and history. It is served by POST /api/chat on both runtimes and powered by Google Gemini function calling (default model gemini-3.6-flash); the model's reasoning for each turn is shown in a collapsible "Thought for…" block above the reply.

• Setup: export GEMINI_API_KEY=<key> (or config Gemini:ApiKey) before starting the host; without it the page shows a setup notice. Gemini:Model and Gemini:BaseUrl override the defaults.
• Audit trail: every tool call the model makes (name, input, result) is logged beneath the reply — the chat can do nothing the tools don't expose.
• Guardrails: destructive operations (deletes) require explicit in-conversation confirmation, and the tool loop is capped at 8 rounds per message.
• Rate limit: Chat:MaxRequestsPerSession (default 10) caps it per login session; past that the route answers 429. ≤0 disables the cap.
• It has no privilege of its own. The model acts as the calling user, through the same entitlement checks as any other caller, and its audit rows carry origin: "chat" and onBehalfOf. With approvals enabled it files a request rather than acting; Chat:MayExecutePrivileged=true lets it execute instead — leave it false.

Admin CLI & MCP server

Two front ends over the same REST surface, both in admin/, both with zero npm dependencies — one for a terminal, one for an agent, sharing one client so they cannot disagree.

SF_URL=http://localhost:5199 bun admin/sf.ts health
bun admin/sf.ts login admin admin123!
bun admin/sf.ts ls tables
bun admin/sf.ts get table positions
bun admin/sf.ts start pipeline <id>      # · stop · create · delete
bun admin/sf.ts rows positions --limit 20
bun admin/sf.ts results <pipeline-id>
bun admin/sf.ts validate "SELECT symbol, SUM(qty) FROM trades GROUP BY symbol"
bun admin/sf.ts config export > catalog.json
bun admin/sf.ts api GET /api/meta/instance   # escape hatch: any route, authenticated

SF_URL defaults to :5199 — point it at :5399 for the Dapr flavor, or at a container/Cloud Run URL. Everything the CLI can do is an ordinary REST call, so it is subject to the same entitlements as any other caller.

bun admin/mcp.ts serves the same operations as MCP tools over stdio, hand-written to the spec with no SDK and pinned by subprocess conformance tests. It is how an agent administers a running instance without being handed a shell.

Deployment & admin

Both flavors ship as self-contained containers under deploy/ — the same images run locally under docker compose and on Google Cloud Run:

• Orleans: one container (deploy/orleans/), compose on :6199. Single-silo topology → max-instances 1 on Cloud Run.
• Dapr: one four-container unit — app + daprd + placement + redis sharing a network namespace (deploy/dapr/), compose on :6399; the identical container set deploys as a single multi-container Cloud Run service.
• Cloud Run: deploy/<flavor>/deploy.sh builds, pushes, and applies the service manifest (parameterized by PROJECT_ID/REGION; --dry-run prints without executing). Data is ephemeral — a cold start reseeds the demo catalog.
• Cluster admin app (admin/, bun main.ts, :5599): start either stack on command, watch /healthz turn green, tail logs, and shut it down — against local compose or Cloud Run (MODE=cloudrun, start/stop = scale 1↔0). It never touches the dev servers on 5199/5399.

Running it inside your own process

StreamsForge's host is not a special executable — it is an ordinary ASP.NET Core application that co-hosts an Orleans silo and registers the shared API. Everything that makes it StreamsForge is behind extension methods, which is what makes the platform embeddable: your own service can be the StreamsForge instance rather than talk to one.

That matters in exactly two situations. If you ship software to customers, one process is one deployment artifact — no second service in the diagram, no broker for the customer to operate, no runtime they don't already have. And if you are building your own product on top, the SQL, the tables and the connectors are in-process: no network hop between your code and the engine, and your own types can be published straight into an ingest source.

The wiring

This is the whole of it — the same calls the shipped host makes, in the same order:

var builder = WebApplication.CreateBuilder(args);

builder.Host.UseOrleans(silo => { /* clustering, streams, storage — see the host's Program.cs */ });

builder.Services.AddStreamsForgeApi(builder.Configuration);   // endpoints, auth, access, audit, chat
builder.Services.AddOrleansFacades();                        // the runtime-specific facades
builder.Services.AddHostedService<GeneratorSupervisorService>();
builder.Services.AddHostedService<IngestDrainPumpService>();
builder.Services.AddHostedService<StreamBridgeService>();
builder.Services.AddHostedService<NatsPublisherService>();
DatabaseConnectors.RegisterAll();                            // out-of-core connectors opt in here

var app = builder.Build();
app.MapStreamsForgeApi(new StreamsForgeApiOptions(
    ProtosDir: …, GrpcPort: …, GrpcStaticServices: [ … ],
    DocsFilePath: …, SpaDistPath: …, Flavor: "orleans",
    DataDir: …, InstanceName: …, Version: …));
// your own controllers, endpoints, background work — same app, same DI container

StreamsForgeApiOptions is where a host states the facts only it knows — where the protos and the SPA bundle live, which gRPC port to advertise, which data directory carries its identity file. It exists so the shared endpoints answer byte-identically across both runtime flavors rather than each host resolving those inline.

Or just the engine

The narrower option: StreamsForge.Engine is a pure C# library with no Orleans, no Dapr and no ASP.NET Core types in it — tokenizer → parser → validator → planner → interpreted operators, plus the Z-set algebra. That is a deliberate architectural boundary, enforced because both runtime flavors depend on it, and it means you can compile and run this SQL dialect over your own data with no actor runtime at all. What you give up is everything the platform layer provides: persistence, the catalog, lifecycle, connectors, the API and the console. Take the engine alone if you want incremental SQL as a library; take the wiring above if you want the platform.

Performance & tuning

Every knob below has a measured reason to exist. Defaults are chosen so an untouched instance behaves exactly as it did before the knob was added.

Stream transport

The single largest latency lever, and the one worth checking first. Default --Streams:Transport pull is byte-identical stock Orleans memory streams, tunable with --Streams:PullPeriodMs (default 100). --Streams:Transport push swaps in an in-process push bus: tableDelta p50 1 ms versus 115 ms on stock pull, measured on this platform's own soak shapes. --Streams:PushCapacity (default 10 000) bounds it.

Table-level knobs

Memory is worth measuring rather than assuming: a flush amplifier found during plan 011 took a soak shape from 287 MB/min to 109 MB/min, and row retention is what makes a table plateau rather than climb. The soak harness itself is tools/soak/run-soak.sh --shape orders|instruments in the repository.

Everything else with a latency or cost story

• Auth:PolicyCacheSeconds (default 10) is how long a resolved permission set is cached — and therefore the worst-case lag on a revocation landing. Lower is safer and costs a resolve per request more often.
• Connector schedules have a 1 s floor on intervalMs, so nobody hot-loops a live endpoint, and failure backoff caps at 15 minutes.
• batchSize on a database or CDC source is a target per cycle, not a hard ceiling — a transaction larger than it is delivered whole.
• Chat:MaxRequestsPerSession (default 10) caps the AI control chat per login session; ≤0 disables the cap.
• Sinks never cost latency, by contract — they drop rather than block. If a sink appears to be slowing an entity down, the cause is somewhere else.

Architecture

One .NET process co-hosts an Orleans silo and the ASP.NET Core API. Orleans virtual actors (grains) own all state — single-threaded per entity, activated on demand:

The SQL engine (StreamsForge.Engine) is a pure C# library with no Orleans dependency: tokenizer → recursive-descent parser → validator (positioned diagnostics) → planner → interpreted operators. Persistence is a JSON-file grain store (one file per grain state under data/) — debuggable, restart-safe, and swappable for a database provider without touching grain code. The stack is Orleans 10.2.1 on .NET 10; the console is React 19 + Tailwind 4 + shadcn/ui with a bespoke design-token system.

Limitations & design notes

The product-level boundaries — what this platform is deliberately not — are in Where it fits. What follows is the engineering detail behind them.

• Delivery guarantees differ by ingress kind, and none of them is exactly-once. Polled database and CDC sources are at-least-once (a failed cycle keeps the old cursor, so rows are re-read rather than skipped). A FIX session is at-most-once into the platform: QuickFIX/n's sequence layer acknowledges a session, not a row landing in a table. Core NATS without a durable consumer is at-most-once. Idempotence belongs in your SQL — LATEST BY on a key — or downstream.
• Egress is fire-and-forget with no backpressure. A sink that cannot deliver counts the failure and drops the row; it never slows or stalls the entity it is attached to, and it does not retry from a queue. The entity's status carries no per-sink field, so a persistently failing sink is visible in the logs and nowhere else.
• A CDC delete does not retract anything. Ingress rows are events, and _weight on an inbound row is just a column — the Z-set weights that make a table a multiset come from table SQL. LATEST BY + WHERE _op <> 'd' hides a deleted key but does not free it.
• Environments are a namespace, not a security boundary. Until grants are scoped to one, any authenticated Editor can address any environment with a header. A source or table name may not contain . going forward, because that is the key separator.
• Three features are Orleans-only. shardBy, parallelism > 1 and the crdt source kind: the Dapr flavor stores each and refuses to start such an entity, rather than silently degrading. The Dapr flavor is feature-complete in code but behind on live verification — dapr/PARITY.md in the repository is the list of what is descoped, owed and unverified, and is the file to read before relying on it.
• Embedding means a project reference. No packages are published to a feed today, so running the platform in your own process means referencing the projects or packing them yourself.
• Single-process topology (by design, wherever it runs). Each flavor is one self-contained unit — locally as dotnet run, as a docker compose stack, or as a single Cloud Run service instance (max-instances=1; the Dapr flavor's app+daprd+placement+redis ride in one instance as sidecars). The Orleans programming model is production-shaped, but a multi-silo deployment would swap clustering, stream, and storage providers. Stream delivery is switchable: stock pull-based memory streams (default; 100 ms polling cadence, --Streams:PullPeriodMs) or the in-process push bus (--Streams:Transport push, tableDelta p50 1 ms vs 115 ms stock — see the comparison page for the measured scoreboard).
• Tables rebuild on restart. Snapshots persist and serve immediately, but operator internals rebuild from live traffic (the Rebuilding badge). Full operator-state checkpointing is future work.
• Materialized-table CROSS JOIN is single-partition only — it has no equi-key to hash rows on, so Parallelism above 1 is rejected at start with a clear error. INNER/LEFT/RIGHT/FULL OUTER table joins (plus SEMI/ANTI via EXISTS/IN subqueries) have no such restriction — see Materialized tables → Outer & cross joins for the incremental pad/retract semantics and NULL-key rules.
• Late events are dropped, not retroactively corrected — bounded-lateness semantics, not full differential dataflow.
• JSON path keys are literals (-> 'key', -> 0); computed keys aren't supported.
• No recursive CTEs — by design, not omission: recursion requires iterative dataflow (multitemporal timestamps à la Materialize), which would replace the epoch-based DBSP architecture. Bounded-depth traversals unroll into joins; "latest per chain" is LATEST BY + row history.
• Correlated subqueries beyond equality get a diagnostic ("rewrite as a JOIN"); NOT IN ignores subquery NULLs; no LEFT UNNEST; LATEST BY keeps a single row per key (no fallback history — that's what row history is for).