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

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, 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 has no ingestion call in its loop. Two conditions still apply:

required

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

note

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

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

    # No ingestion call. These turns become long-term memory
    # when the conversation compacts.
    return reply

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

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 for the full walkthrough including startup and shutdown. Reported turns become long-term memory when the conversation compacts; add 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 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

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