Progress Events

title: Progress Events

SmartMemory publishes real-time progress for long-running operations (pipeline ingests, evolution runs, grounding passes, Studio-scheduled jobs) through a single, workspace-scoped event bus on Redis Streams. Consumers subscribe over Server-Sent Events at /memory/progress/stream.

The contract is defined once at docs/features/PLAT-PROGRESS-1/progress-event-contract.json and used verbatim across producer, transport, and SDK layers.

Event shape

{
  "run_id":  "uuid-of-parent-operation",
  "scope":   "workspace:<workspace-id>",
  "seq":     42,
  "ts":      1713096000.123,
  "kind":    "pipeline.stage",
  "status":  "progress",
  "stage":   "extract",
  "payload": { "progress": 50, "message": "Extracting entities" }
}

Fields per the v1.5.0 contract:

• run_id — UUID of the parent operation; required for replay queries.
• scope — "workspace:<id>". Derived server-side from the caller's JWT workspace; never accepted as a client-supplied override.
• seq — monotonic per-run sequence. Clients order and dedupe on this.
• ts — server-side epoch seconds at emission.
• kind — dotted, producer-owned tag. Consumers filter on this. No transport-level registry.
• status — one of started | progress | ok | warn | error.
• stage — pipeline stage name when applicable.
• payload — kind-specific object. Validation happens at producer and consumer; the transport is untyped.

Common kind values

• pipeline.stage — core ingest pipeline per-stage progress
• evolver.result — evolution stage outcomes (when CORE-EVO-REPORT-1 lands)
• graph.node / graph.edge — graph viewer drip feed
• sweep.iter — parameter sweep iteration
• batch.item — batch ingest per-item
• search.hop — multi-hop retrieval trace
• studio.job — Studio-scheduled stage jobs

Producing events

Server-side Python producers use the three helpers in smartmemory.progress:

from smartmemory.progress import bind, unbind, bound_progress, emit

# Bind run + scope once per operation (reads contextvars), emit freely.
with bound_progress(run_id="run-42", scope="workspace:team-alpha"):
    emit(kind="pipeline.stage", status="progress", stage="extract",
         payload={"progress": 50, "message": "Extracting entities"})
    # ...more emits inside the same run

The context-managed form is required inside any path that runs in a fresh context (FastAPI BackgroundTasks, asyncio.create_task, threading.Thread) — contextvars set by a calling request handler do not propagate into those.

Environment:

• SMARTMEMORY_PROGRESS_MAXLEN — approximate ring-buffer length per workspace stream (default 10000). Bump to 50k-100k for batch-heavy deployments where a single run can emit tens of thousands of events before any client reconnects.

Subscribing — SSE endpoint

GET /memory/progress/stream

Three modes depending on query params:

from_seq requires run_id. from_seq and since are mutually exclusive.

Auth: Authorization: Bearer <JWT> or X-API-Key: <key>. Workspace membership required. The scope is server-derived; no client query param can override it. Browser clients must use a fetch-based SSE transport (native EventSource cannot attach custom headers).

Frame format:

id: 1713096000123-0
data: { "run_id": "...", "scope": "workspace:...", ... }

: keepalive

Keepalive comments arrive every 15 seconds to prevent idle-connection timeouts.

Errors:

JS SDK subscriber (@internal)

The JS SDK provides a reconnect-aware subscriber at @smartmemory/sdk-js/progress. It is marked @internal — consumed by the first-party UIs (smart-memory-graph, smart-memory-studio, smart-memory-insights) and not listed in the public API index.

import { subscribeProgress } from '@smartmemory/sdk-js/progress';

const sub = subscribeProgress({
  runId: 'run-42',
  fromSeq: 0,        // replay from start of the run
  token: '<jwt>',    // or apiKey
  onEvent: (event) => {
    if (event.kind !== 'studio.job') return;
    console.log(event.payload.progress, event.payload.message);
    if (event.status === 'ok' || event.status === 'error') sub.close();
  },
  onError:     (err) => console.error('SSE fatal', err),
  onReconnect: () => console.log('reconnecting…'),
});

The SDK uses @microsoft/fetch-event-source under the hood for header-based auth. Reconnect + Last-Event-ID resume is automatic.

Related

• Feature: PLAT-PROGRESS-1 — design, plan, contract
• Follow-up audit: STUDIO-PROGRESS-AUDIT-1 — boundary inventory of Studio's progress surface