> ## 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
> Synap can push context to your agent before it asks. The Listen stream is a long-lived gRPC connection that carries activity signals up and anticipated context bundles down, turning the next fetch() into a local cache read.
Most applications talk to Synap over request-response: you call [`memories.create`](/sdk-reference/memories/create) to write and [`fetch`](/sdk-reference/context/fetch) to read. That is complete and correct on its own.
**Real-time anticipation adds a second channel.** [`instance.listen()`](/sdk-reference/instance/listen) opens a long-lived gRPC stream. Your app reports what the agent is doing; Synap predicts what it will need next and pushes context bundles down the stream *before* the agent asks. The next `fetch()` then resolves from a local cache instead of a network round-trip.
## Two channels, two jobs
The two channels do different work, and the difference is mostly about **when** memory happens.
| | Ingestion | Listen stream |
| ---------------------------- | ---------------------------------------------------- | ------------------------------------------- |
| **Transport** | REST | gRPC (bidirectional) |
| **You call** | `memories.create` / `conversation.ingest_transcript` | `instance.listen` + `instance.send_message` |
| **Carries** | Full turns and documents | Activity signals |
| **Creates long-term memory** | Yes — on every call | Yes — but only at compaction |
| **Timing** | Immediate, per call | Deferred and batched |
| **Effect** | Durable memory you control | Prefetched context, plus deferred memory |
| **Required** | Yes, for reliable memory | No — a latency optimization |
**The stream is not memory-neutral.** Conversation turns you send with `send_message()` are persisted to conversation history, and when the conversation compacts, those raw turns are promoted into the same ingestion pipeline `memories.create()` uses.
This is real memory, but it is **not a substitute for ingesting**. It only fires once a conversation is long enough to compact, it lands well after the turn it came from, and you have no control over how the content is typed or scoped. See [What the stream does to memory](#what-the-stream-does-to-memory).
## Opening the stream
`listen()` requires an initialized SDK. It resolves your `client_id` and `instance_id`, authenticates the stream with the same API key your REST calls use, and holds the connection open until you close it.
```python theme={null}
from maximem_synap import MaximemSynapSDK
sdk = MaximemSynapSDK(api_key="synap_your_key_here")
await sdk.initialize()
await sdk.instance.listen(
on_reconnect=lambda attempt: logger.info("stream reconnected (attempt %d)", attempt),
on_disconnect=lambda reason: logger.warning("stream lost: %s", reason),
)
```
Both callbacks take one argument — `on_reconnect` receives the attempt count, `on_disconnect` receives the reason. A zero-argument callback raises `TypeError` at the moment the stream drops, which is precisely when you need the diagnostic.
`on_context` is optional. Arriving bundles are written to the SDK's anticipation cache automatically, so `fetch()` finds them whether or not you supply a callback. Use it only when you want to react to bundles directly.
Connection targets are configurable via [`SDKConfig`](/sdk/configuration): `grpc_host`, `grpc_port`, and `grpc_use_tls`. Defaults point at Synap Cloud with TLS on.
## What you send
Report agent activity with [`send_message`](/sdk-reference/instance/send-message). The `event_type` determines how the platform treats it.
| `event_type` | When to send it | What the platform does |
| ------------------- | -------------------------------------------- | ------------------------------------------------------------------------- |
| `user_message` | The user's turn arrives, before you retrieve | Persists the turn to conversation history; signals a new turn is starting |
| `assistant_message` | After your agent produces its reply | Persists the turn; **triggers anticipation for the next turn** |
| `tool_call` | Your agent decides to call a tool | Observed for situational awareness — informative, not a trigger |
| `context_request` | Your agent plans a retrieval | Supplies `search_queries` / `context_types` as direct anticipation hints |
A `user_message` or `assistant_message` event is only persisted when **both** `user_id` and `customer_id` are present. If either is missing the platform skips persistence and logs server-side — your application sees no error and no exception. On B2C, pass your user identifier as both.
## What the stream does to memory
A `user_message` or `assistant_message` event does more than feed anticipation. The platform writes it to the conversation's history, and that history has a second consumer.
Once a conversation crosses its compaction threshold — by default **3,000 tokens or 10 messages**, configurable per Instance — Synap compacts it. Compaction summarizes the conversation for short-term context, and then **promotes the raw turns into the long-term ingestion pipeline**: the same extraction stages `memories.create()` runs, producing memories of the same quality.
So turns sent over the stream do become durable memory. Four things decide whether that actually happens for a given conversation:
Below the threshold, nothing is promoted. Short conversations — a two-turn support exchange, a one-shot question — never produce memories this way.
A turn missing `user_id` or `customer_id` is never persisted in the first place, so there is nothing to promote. It fails silently.
Promotion derives its scope from the conversation record. If that cannot be resolved, promotion is skipped.
If you ingested the conversation with [`conversation.ingest_transcript`](/sdk-reference/conversation/ingest-transcript), promotion skips it deliberately — the full transcript was already ingested at higher fidelity, and extracting again would duplicate it.
**Streaming turns *and* ingesting the same turns extracts the content twice.** The guard above covers `ingest_transcript` only. If you call `memories.create()` per turn on the same text you are streaming as `user_message` / `assistant_message`, that content is extracted once on your call and again at compaction, costing double and creating overlapping memories the deduplication stage then has to reconcile.
Pick one owner for durable memory: either ingest explicitly and treat the stream as signal, or ingest with `ingest_transcript` so the guard applies.
Because promotion is deferred and conditional, treat it as a backstop rather than your memory strategy. If your agent needs a fact to be retrievable on the next turn, ingest it explicitly — promotion will not have run yet.
## What the SDK sends on its own
Once the stream is live, the SDK instruments `fetch()` for you. You do not write this code, but you should know it exists:
* **`context_fetch`** — a retrieval was requested
* **`context_used`** — the retrieval was served from the anticipation cache
* **`context_assembled`** — what the SDK actually composed for the model
These drive the platform's learning loop: prefetch outcome scoring, per-pattern hit rates, and the Requests page audit trail. They carry ids and counts only — never raw prompt content.
## What comes back
The platform pushes **context bundles** down the stream. Each is written into the SDK's in-process anticipation cache. When your agent then calls `fetch()`:
* **Cache hit** → served locally in roughly a millisecond, no network call
* **Cache miss** → falls through to the normal REST retrieval path
This is why anticipation is safe to adopt incrementally. A cold or dead stream costs you nothing but the latency you already had.
## The turn loop
The mechanism that makes this work is easy to miss: **the `assistant_message` you send at the end of turn N is what warms the cache for turn N+1.** Anticipation happens *between* turns, not during them. By the time your next turn calls `fetch()`, the bundle has already landed.
So the ordering matters. Emit `assistant_message` after your agent replies — not before.
```python theme={null}
import uuid
conversation_id = str(uuid.uuid4()) # must be a valid UUID, reused across the conversation
async def handle_turn(user_text: str, user_id: str, customer_id: str) -> str:
# 1. Report the user's turn.
await sdk.instance.send_message(
content=user_text,
role="user",
event_type="user_message",
conversation_id=conversation_id,
user_id=user_id,
customer_id=customer_id,
)
# 2. Retrieve. Warm cache → local hit; cold → REST fallback, same API.
context = await sdk.user.context.fetch(
user_id=user_id,
customer_id=customer_id,
conversation_id=conversation_id,
search_query=[user_text],
)
reply = await your_llm(context, user_text)
# 3. Report the reply. THIS is what anticipates turn N+1.
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,
)
# 4. Ingest the turn, so it is retrievable now rather than
# whenever this conversation next compacts.
await sdk.memories.create(
document=f"User: {user_text}\n\nAssistant: {reply}",
user_id=user_id,
customer_id=customer_id,
document_type="ai-chat-conversation",
mode="fast",
)
return reply
```
Step 4 is what makes the turn retrievable immediately. Drop it and the turn still reaches long-term memory eventually, via compaction promotion — but not until this conversation crosses its threshold, and never at all if it stays short.
This example ingests every turn *and* streams every turn, which is the double-extraction case described above. It is written this way to show both calls in one place. In production, pick one owner for durable memory.
## One stream per instance
In a multi-tenant server, open **one** stream for the whole process and distinguish users through the `user_id` / `customer_id` on each `send_message` and `fetch`. Do not open a stream per user session.
The platform enforces per-instance and per-client stream quotas (currently 100 concurrent streams per instance, 200 per client). A stream-per-session design saturates that quota under real concurrency, and the surplus connections enter a `RESOURCE_EXHAUSTED` reconnect loop.
Streams also have a maximum lifetime of one hour, after which the server closes them. This is normal: the SDK reconnects with exponential backoff and your `on_reconnect` callback fires. A long-lived server should expect periodic reconnects, not a single permanent connection.
See [Real-Time Anticipation in a Server](/patterns/real-time-anticipation-server) for the full pattern, including the health check worth alerting on.
## When the stream fails
Failure is designed to be non-fatal — retrieval falls back to REST — but the two SDKs differ in how loudly they say so.
| | Python | JavaScript / TypeScript |
| ---------------- | ------------------------------------------------------------ | ------------------------------------------- |
| `listen()` fails | **Raises** (`AuthenticationError`, `SDKNotInitializedError`) | **Warns to console** and falls back to HTTP |
| Reconnect | Exponential backoff, 10 attempts; counter resets on success | Exponential backoff with jitter |
| After exhaustion | Stream stays down; `fetch()` keeps working over REST | Same |
In both SDKs, a permanently dead stream is **silent** at the application level — `fetch()` still returns correct results, just without the latency benefit. Do not assume streaming is working because your agent works. Monitor `instance.is_listening` and log from `on_disconnect`.
## What anticipation does not do
* It does **not** ingest a turn at the moment you send it — promotion happens at compaction, or not at all if the conversation stays short
* It does **not** give you control over how promoted content is typed, scoped, or tagged; `memories.create()` does
* It does **not** replace `memories.create` or `conversation.ingest_transcript` as your memory strategy
* It does **not** change what `fetch()` returns — only how fast it returns
* It is **not** required; every Synap feature works without it
One shared stream, many tenants, and the failure modes worth alerting on.
`listen`, `send_message`, and `stop_listening` in full.