> ## 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, at compaction |
| **Timing** | Immediate | Deferred, up to \~5 minutes |
| **Effect** | Durable memory you control | Prefetched context, plus durable memory |
| **Use for** | Documents, backfills, anything urgent | Live conversation turns |
**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.
For an agent reporting conversation turns, this is sufficient on its own; it is the basis of the [Agent Integration](/setup/agent-integration). What it costs you is immediacy (a turn is retrievable once its conversation compacts, up to about five minutes) and control over how content is typed and 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, and your application sees no error and no exception. On a B2C instance send `user_id` only: `customer_id` is not accepted there, and an event without one is persisted normally.
## 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.
When a conversation compacts, Synap summarizes it 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.
Compaction fires on any of three triggers, so every conversation gets there:
| Trigger | Default | Catches |
| ------------------- | ----------------------- | ------------------ |
| **Token threshold** | 3,000 tokens | Long conversations |
| **Message count** | 10 messages | Busy conversations |
| **Idle period** | 5 minutes of inactivity | Everything else |
The first two are configurable per Instance. The idle trigger is the backstop that makes streaming self-sufficient: a two-turn exchange still compacts a few minutes after the user stops, and its turns still become memory. This is why the [Agent Integration](/setup/agent-integration) has no ingestion call in its loop.
Two conditions still apply:
A turn missing `user_id` or `customer_id` is never persisted, so there is nothing to promote. It fails silently.
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.
**Do not stream a turn and also ingest that same turn.** If you call `memories.create()` on text you already reported as `user_message` / `assistant_message`, it is extracted twice, once immediately and once at compaction, costing double and producing overlapping memories the deduplication stage then reconciles.
Stream conversation turns; ingest everything else.
Promotion is automatic but not instant: a turn becomes retrievable when its conversation compacts, which for a quiet conversation means about five minutes. When something must be retrievable sooner than that, ingest it explicitly. See [when to still call `memories.create()`](/setup/agent-integration#when-to-still-call-memories-create).
## 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 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,
)
# No ingestion call. These turns become long-term memory
# when the conversation compacts.
return reply
```
```javascript JavaScript theme={null}
import { randomUUID } from 'node:crypto';
let conversation_id = randomUUID(); // must be a valid UUID, reused across the conversation
async function handle_turn(user_text, user_id, customer_id) {
// 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.
const context = await sdk.user.context.fetch({
user_id: user_id,
customer_id: customer_id,
conversation_id: conversation_id,
search_query: [user_text],
});
const 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,
});
// No ingestion call. These turns become long-term memory
// when the conversation compacts.
return reply;
}
```
```typescript TypeScript theme={null}
import { randomUUID } from 'node:crypto';
let conversation_id = randomUUID(); // must be a valid UUID, reused across the conversation
async function handle_turn(user_text, user_id, customer_id) {
// 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.
const context = await sdk.user.context.fetch({
user_id: user_id,
customer_id: customer_id,
conversation_id: conversation_id,
search_query: [user_text],
});
const 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,
});
// No ingestion call. These turns become long-term memory
// when the conversation compacts.
return reply;
}
```
This is the complete loop. See [Agent Integration](/setup/agent-integration) for the full walkthrough including startup and shutdown. Reported turns become long-term memory when the conversation compacts; add [`memories.create`](/sdk-reference/memories/create) only for content that is not a conversation turn, or that must be retrievable sooner.
## 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, up to about five minutes later
* 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.