> ## Documentation Index
> Fetch the complete documentation index at: https://docs.maximem.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Real-Time Anticipation in a Server
> Run one Listen stream for a whole multi-tenant process, scope every call per request, and alert on the failure that stays silent.
[Real-time anticipation](/concepts/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.
```python theme={null}
# 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.
```python theme={null}
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:
```python theme={null}
@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
The two-channel model, event types, and what the platform does with each one.