Real-Time Anticipation in a Server

Real-time anticipation is straightforward in a script: open a stream, send events, close it. In a long-lived server it needs three decisions that are easy to get wrong, and one of the wrong answers only shows up under production concurrency.

One stream for the process

The instinct is to open a stream per user session. Don’t. Synap enforces per-instance and per-client stream quotas — currently 100 concurrent streams per instance and 200 per client. A stream-per-session design saturates that quota at real concurrency and surplus connections enter a RESOURCE_EXHAUSTED reconnect loop, which degrades the sessions that did connect. Instead, open one stream at process startup and distinguish users with the user_id / customer_id you already pass on every call.

# app_state.py — module scope, one per process
from maximem_synap import MaximemSynapSDK, SDKConfig

sdk = MaximemSynapSDK(api_key=os.environ["SYNAP_API_KEY"])
_stream_healthy = False


async def startup() -> None:
    global _stream_healthy
    await sdk.initialize()

    def on_reconnect(attempt: int) -> None:
        global _stream_healthy
        _stream_healthy = True
        log.info("synap_stream_reconnected attempt=%d", attempt)

    def on_disconnect(reason: str) -> None:
        global _stream_healthy
        _stream_healthy = False
        log.warning("synap_stream_disconnected reason=%s", reason)

    try:
        await sdk.instance.listen(
            on_reconnect=on_reconnect,
            on_disconnect=on_disconnect,
        )
        _stream_healthy = True
    except Exception as exc:
        # Anticipation is an optimization. A failed stream must not
        # stop the server from booting — fetch() still works over REST.
        log.error("synap_stream_unavailable error=%s", exc)


async def shutdown() -> None:
    await sdk.instance.stop_listening()

Scope lives on the request, not the stream. One process-wide stream serves every tenant because send_message() and fetch() each carry their own user_id / customer_id.

Scope every call, per request

With a shared stream, correctness depends entirely on passing the right identifiers on each call. Nothing about the stream itself is tenant-specific.

async def handle_turn(user_id: str, customer_id: str, conversation_id: str, text: str) -> str:
    await sdk.instance.send_message(
        content=text,
        role="user",
        event_type="user_message",
        conversation_id=conversation_id,
        user_id=user_id,
        customer_id=customer_id,
    )

    context = await sdk.user.context.fetch(
        user_id=user_id,
        customer_id=customer_id,
        conversation_id=conversation_id,
        search_query=[text],
    )

    reply = await your_llm(context, text)

    await sdk.instance.send_message(
        content=reply,
        role="assistant",
        event_type="assistant_message",
        conversation_id=conversation_id,
        user_id=user_id,
        customer_id=customer_id,
    )

    # Makes the turn retrievable now. Streamed turns also reach long-term
    # memory on their own at compaction, but not before then — and calling
    # both on the same text extracts it twice. Pick one owner per tenant.
    await sdk.memories.create(
        document=f"User: {text}\n\nAssistant: {reply}",
        user_id=user_id,
        customer_id=customer_id,
        document_type="ai-chat-conversation",
        mode="fast",
    )
    return reply

A user_message or assistant_message missing either user_id or customer_id is dropped server-side with no client-visible error. On B2C, pass your user identifier as both. In a multi-tenant server, treat a missing customer_id as a bug in your request handler, not an optional field.

Expect reconnects

Streams have a maximum lifetime of one hour, after which the server closes them. This is normal operation, not an error: the SDK reconnects with exponential backoff (10 attempts, and the counter resets on every successful connect), and on_reconnect fires. A server that runs for a week will reconnect well over a hundred times. Log reconnects at INFO, not WARNING, or you will train yourself to ignore them.

The failure that stays silent

This is the one worth alerting on. If the stream dies permanently — quota exhaustion, a network partition longer than the backoff window, revoked credentials — your application keeps working. fetch() falls through to REST and returns correct results. Latency rises, and nothing else changes. No exception reaches your handler. That is the intended design, and it is why streaming is safe to adopt. It also means you will not notice unless you look:

@app.get("/health/synap")
async def synap_health() -> dict:
    return {
        "stream_connected": sdk.instance.is_listening,
        "degraded": not sdk.instance.is_listening,
    }

Alert on is_listening going false and staying false. Do not alert on individual disconnects — those are routine.

The JavaScript/TypeScript SDK never raises from listen() at all: it logs a console warning and falls back to HTTP. There, checking synap.isListening is the only way to know the stream is up.

Checklist

One stream per process, opened at startup, closed at shutdown
listen() failure logged, never fatal to boot
user_id and customer_id on every send_message and fetch
assistant_message emitted after the reply, so it warms the next turn
One owner for durable memory: either ingest explicitly, or rely on compaction promotion — not both on the same text
is_listening exported to your health endpoint and alerted on

How anticipation works

The two-channel model, event types, and what the platform does with each one.

Guides

Patterns

Cookbook

Real-Time Anticipation in a Server

One stream for the process

Scope every call, per request

Expect reconnects

The failure that stays silent

Checklist

How anticipation works

​One stream for the process

​Scope every call, per request

​Expect reconnects

​The failure that stays silent

​Checklist

How anticipation works

One stream for the process

Scope every call, per request

Expect reconnects

The failure that stays silent

Checklist