eastbournestreaming

Streaming inference MCP server for utility meter telemetry. It folds a stream of meter readings into fixed (tumbling) time windows with exactly-once semantics and runs an inference provider over each closed window.

Project codename: Eastbourne. Client: Quenby Metering.

The constraint that shapes the design

Regulated meter data cannot leave the customer's boundary. Two things follow, and both are enforced by tests:

• No network at runtime. The server transport is stdio (newline-delimited JSON-RPC 2.0), not a socket. tests/boundary.test.ts scans src/ and fails the build if any module imports node:net, node:http(s), node:tls, node:dgram, WebSocket, or calls fetch.

• In-process inference. All model behaviour goes through InferenceProvider (src/providers/base.ts). The bundled StubInferenceProvider (src/providers/stub.ts) is deterministic and runs offline with no API key. A production model ships on-premises behind the same interface.

Related MCP server: SEOSiri Data Pipeline MCP

Exactly-once semantics

For a fixed set of distinct eventIds the emitted aggregates are identical regardless of how often or in what order events are delivered, as long as a duplicate arrives before its window closes. This holds because:

• duplicates are dropped via an eventId dedup set scoped to open windows, and

• window closure is a deterministic function of event-time via a watermark (maxEventTime - allowedLateness), not wall-clock.

checkpoint() / WindowedAggregator.restore() carry the open windows and dedup set across a restart so the guarantee survives process recycling.

Architecture

stdin (JSON-RPC lines) ──► server.ts ──► WindowedAggregator ──► InferenceProvider
                              │                │                      │
                           Logger          watermark +            StubInferenceProvider
                           (stderr)        dedup + windows        (deterministic, offline)

The server is a thin envelope: it parses one request per line, routes to the aggregator, runs the provider over any windows that closed, and writes one response per line to stdout. State lives in WindowedAggregator; all model behaviour lives behind InferenceProvider. Nothing opens a socket — see the ADR on the stdio transport. Design decisions with a real alternative are recorded in docs/adr/.

Layout

• src/types.ts — domain types (Reading, WindowAggregate, ...)

• src/aggregator.ts — WindowedAggregator, the core algorithm

• src/providers/base.ts — InferenceProvider interface

• src/providers/stub.ts — deterministic offline provider (z-score anomaly)

• src/config.ts — env-var configuration, validated at startup

• src/logger.ts — structured JSON logging to stderr, with timed()

• src/errors.ts — ConfigError, WindowLimitError

• src/server.ts — stdio JSON-RPC entry point wiring the pieces together

• tests/*.test.ts — behavioural tests, node:test

Configuration

All configuration is read from environment variables at startup and validated; a malformed value aborts with exit code 78 (EX_CONFIG) and a logged reason rather than failing later on a bad window.

Operational behaviour

• Logs are newline-delimited JSON on stderr (stdout is the JSON-RPC channel). flush timing is logged at debug.

• Back-pressure: a reading that would open a window past EB_MAX_OPEN_WINDOWS is refused with JSON-RPC error -32000 (busy). The stream stays alive; the client retries once windows drain. Bounds memory against unbounded meters or lateness.

• Graceful degradation: if the inference provider throws, the affected window is still emitted with label: "degraded" and the error is logged. A model fault never stalls the pipeline or drops a computed aggregate.

• Graceful shutdown: on SIGINT/SIGTERM the server flushes open windows so no aggregate is lost, logs the closed windows, and exits 0.

• Bad input: malformed readings return JSON-RPC -32602 (invalid params); unparseable lines return -32700 and are dropped without crashing.

Requirements

Node 22+. Node runs the TypeScript sources directly; there is no build step and no dist/.

Install and test

npm ci
npm test          # runs tests/ with the built-in runner
npm run typecheck # tsc --noEmit

Running the server

npm run server

It reads one JSON-RPC request per line on stdin and writes one response per line on stdout. Pipe a session straight in (windows are 60s, allowed lateness 30s):

printf '%s\n' \
  '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' \
  '{"jsonrpc":"2.0","id":2,"method":"ingest","params":{"reading":{"eventId":"e1","meterId":"m1","timestampMs":0,"value":1.0}}}' \
  '{"jsonrpc":"2.0","id":3,"method":"flush"}' | npm run --silent server 2>/dev/null

The third line closes the window opened by the second and returns its inference:

{"jsonrpc":"2.0","id":3,"result":{"inferences":[{"key":{"meterId":"m1","windowStartMs":0,"windowEndMs":60000},"label":"normal","score":0.047426,"reason":"mean=1 z=0 threshold=3"}]}}

Logs go to stderr; the 2>/dev/null above hides them so only JSON-RPC responses show. Drop it (or set EB_LOG_LEVEL=debug) to see the structured log.

ingest returns { rejected: "duplicate" | "late" | null, inferences: [...] }. flush force-closes all open windows and returns their inferences. stats returns { openWindows }. Inferences are produced only for windows that close.

Known limitations / deferred work

• The dedup set is scoped to open windows; a duplicate that arrives after its window has closed is treated as a late reading and dropped rather than being recognised as a duplicate. For the supported lateness horizon this is equivalent, but a longer-horizon dedup would need a persistent seen-id log.

• The stub provider is a single-feature z-score detector — enough to exercise the boundary and the window pipeline, not a production anomaly model.

Quenby Metering is an illustrative client; this repository is a self-directed reference implementation built to work end to end.