Session Projection

De-duplicated, gap-aware, memory-bounded session view for gateway clients

Turn a raw gateway event stream into a correct, resumable session view — no per-client de-dup, gap-healing or bounded retention to re-implement.

from praisonai.gateway import GatewayClient

client = GatewayClient(url="ws://localhost:8765", agent_id="assistant")
await client.connect()

async for state in client.project():
    ui.render(state)   # de-duped, gap-aware, memory-bounded

Quick Start

1. Simplest usage

   Iterate the projected state and render it — the reducer folds every event for you.

   from praisonai.gateway import GatewayClient
   
   client = GatewayClient(url="ws://localhost:8765", agent_id="assistant")
   await client.connect()
   
   async for state in client.project():
       ui.render(state)

2. With a resume snapshot

   Seed the projection with a snapshot on reconnect so the view converges instead of drifting.

   from praisonai.gateway import GatewayClient
   
   client = GatewayClient(url="ws://localhost:8765", agent_id="assistant")
   await client.connect()
   
   snapshot = {"messages": [...], "cursor": 42}
   async for state in client.project(snapshot=snapshot):
       ui.render(state)

3. Bound retention on a long-lived session

   Cap how many finished runs stay tracked; active streams are never evicted.

   async for state in client.project(max_tracked_runs=50):
       ui.render(state)

4. Pure-core usage without the wrapper

   Drive the reducer directly for third-party clients or tests — no transport required.

   from praisonaiagents.gateway import SessionProjection
   
   proj = SessionProjection(max_tracked_runs=200)
   state = proj.apply_snapshot({"messages": [...], "cursor": 5})
   
   for event in my_event_source():
       state = proj.apply(event)
       render(state)

How It Works

The reducer folds a snapshot plus live events into an immutable state, reconciling every duplicate along the way.

De-duplication Behaviours

Three distinct reconciliations keep the view correct, each keyed on a different identifier.

Configuration Options

The reducer constructor takes a single option.

SessionProjectionState — the immutable value returned by every apply call:

RunView — the immutable per-run streaming view:

GatewayClient.project()

Yields a SessionProjectionState after the initial snapshot (if any) and after every subsequent event.

TypeScript Parity

The praisonai-ts SDK ships the same reducer, importing from the package root. It accepts both camelCase (SDK) and snake_case (wire) keys — messageId/message_id, requestId/request_id, senderId/sender_id, runId/run_id — so wire frames from the Python gateway drop straight in with identical de-dup, reconciliation, gap and retention semantics.

import { SessionProjection } from 'praisonai';

const proj = new SessionProjection({ maxTrackedRuns: 200 });

for (const event of myEventSource) {
  const state = proj.apply(event);
  render(state);   // state.entries / state.runs / state.hasTransportGap
}

import {
  SessionProjection,
  type RunView,
  type ProjectionEntry,
  type SessionProjectionState,
  type SessionSnapshot,
} from 'praisonai';

const proj = new SessionProjection({ maxTrackedRuns: 200 });
const s0 = proj.applySnapshot({ messages: [...], cursor: 5 });

for (const event of myEventSource) {
  const state = proj.apply(event);
  render(state);
}

Common Patterns

from praisonai.gateway import GatewayClient

client = GatewayClient(url="ws://localhost:8765", agent_id="assistant")
await client.connect()

async for state in client.project():
    dashboard.render(state.entries, state.runs)

from praisonai.gateway import GatewayClient

client = GatewayClient(url="ws://localhost:8765", agent_id="assistant")
await client.connect()

snapshot = {"messages": stored_transcript, "cursor": last_cursor}
async for state in client.project(snapshot=snapshot):
    if state.has_transport_gap:
        await client.resync()
    ui.render(state)

from praisonaiagents.gateway import SessionProjection

proj = SessionProjection(max_tracked_runs=100)
state = proj.apply_snapshot(my_snapshot)

for event in my_event_source():
    state = proj.apply(event)
    render(state)

Choose Your Entry Point

Pick the wrapper when you use the bundled client; drive the pure core everywhere else.

Best Practices

Related