Real-Time Anticipation - Maximem Synap

Most applications talk to Synap over request-response: you call memories.create to write and fetch to read. That is complete and correct on its own. Real-time anticipation adds a second channel. 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.

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.

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: 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. 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:

required

Below the threshold, nothing is promoted. Short conversations — a two-turn support exchange, a one-shot question — never produce memories this way.

required

A turn missing user_id or customer_id is never persisted in the first place, so there is nothing to promote. It fails silently.

required

Promotion derives its scope from the conversation record. If that cannot be resolved, promotion is skipped.

note

If you ingested the conversation with 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.

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 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

Run it in a server

One shared stream, many tenants, and the failure modes worth alerting on.

The streaming API

listen, send_message, and stop_listening in full.

​Two channels, two jobs

​Opening the stream

​What you send

​What the stream does to memory

​What the SDK sends on its own

​What comes back

​The turn loop

​One stream per instance

​When the stream fails

​What anticipation does not do

Run it in a server

The streaming API

Two channels, two jobs

Opening the stream

What you send

What the stream does to memory

What the SDK sends on its own

What comes back

The turn loop

One stream per instance

When the stream fails

What anticipation does not do