Inkdown
Start writing

Edward

3 files·0 subfolders

Shared Workspace

Edward
Orchestration Layer

Stream Continuation

Shared from "Edward" on Inkdown

Edward Stream Continuation Architecture - Deep Dive

Principal Engineer's Technical Reference

Easy to digest HLD

alt

Table of Contents

  1. Executive Summary
  2. Architecture Principles
  3. Core Components Deep Dive
  4. Stream Continuation Flow
Overview
Stream Continuation
  • Checkpoint Mechanism
  • Event Sourcing & Replay
  • Frontend Reconnection Strategy
  • Worker Resumption
  • Failure Modes & Recovery
  • Infrastructure Cross-Questions
  • Key Files Reference

    • 1. Executive Summary

    What is Stream Continuation?

    Stream Continuation is Edward's resilience architecture that enables:

    • Seamless reconnection after client disconnects (page refresh, network blip, tab switch)
    • Worker crash recovery without losing in-progress code generation
    • Exact state replay from any point in the stream using cursor-based tracking
    • Multi-turn agent loop checkpointing for long-running operations
    Why This Matters

    Without stream continuation:

    • Page refresh = lost work, user frustration
    • Worker restart = orphaned runs, inconsistent state
    • Network hiccup = incomplete code generation
    • Long operations = no recovery from mid-execution failures

    With stream continuation:

    • Durable execution: Every event persisted, every state checkpointed
    • Resumable UX: Users can refresh/reconnect without losing progress
    • Operational resilience: Workers can restart without data loss
    • Audit trail: Complete event history for debugging and compliance

    2. Architecture Principles

    2.1 Event Sourcing

    All stream events are persisted as an immutable log:

    Plain text

    Benefits:

    • Exact replay from any sequence number
    • Audit trail for debugging
    • Supports multiple reconnections
    • Enables time-travel debugging
    2.2 Dual-Channel Delivery

    Events delivered via two channels simultaneously:

    Plain text

    Why Both?

    • PostgreSQL: Durable, queryable, supports historical replay
    • Redis Pub/Sub: Low-latency, push-based, supports many subscribers
    2.3 Cursor-Based Resumption

    Frontend tracks last processed event:

    TypeScript

    Resumption Flow:

    1. Page loads → read cursor from sessionStorage
    2. Open SSE stream with ?lastEventId=seq:12345
    3. Backend replays events from PostgreSQL where seq > 12345
    4. Subscribe to Redis for live events
    5. Update cursor on each new event
    2.4 Checkpoint-Based Worker Resumption

    Agent loop state checkpointed on continuation turns:

    TypeScript

    Persisted to: runs.metadata.resumeCheckpoint

    3. Core Components Deep Dive

    3.1 Event Persistence Layer

    File: apps/api/services/runs/runEvents.service.ts

    TypeScript

    Key Design Decisions:

    DecisionRationale
    Sequential seq numbersEnables cursor-based replay, ordering guarantee
    Dual write (Postgres + Redis)Durability + real-time delivery
    Envelope patternDecouples storage format from event schema
    Optional publisherAllows testing without Redis dependency
    3.2 SSE Streaming with Replay

    File: apps/api/services/run-event-stream-utils/service.ts

    Core Function: streamRunEventsFromPersistence()

    TypeScript

    Replay Logic:

    Plain text

    Buffering During Replay:

    TypeScript

    Why Buffer?

    • Live events may arrive during replay
    • Need to prevent duplicates
    • Need to maintain ordering (seq-based)
    3.3 SSE Backpressure Handling

    File: apps/api/services/sse-utils/service.ts

    Problem: Slow clients can cause memory buildup

    Solution: Queue-based backpressure with graceful degradation

    TypeScript

    Backpressure Configuration:

    TypeScript

    Graceful Degradation:

    • Queue builds up → monitor size
    • Exceeds threshold → trigger onSlowClient
    • Close stream gracefully → client can reconnect
    3.4 Worker Event Capture

    File: apps/api/services/runs/agent-run-worker/processor.helpers.ts

    Problem: Worker needs to capture SSE stream and persist events

    Solution: Create mock Response object that intercepts writes

    TypeScript

    Why This Pattern?

    • Decouples stream session from persistence mechanism
    • Enables testing without actual HTTP response
    • Serializes event persistence (prevents race conditions)
    • Tracks failures for finalization

    4. Stream Continuation Flow

    4.1 Normal Flow (No Interruption)
    Plain text
    4.2 Client Disconnect + Reconnect
    Plain text
    4.3 Worker Crash + Restart
    Plain text

    5. Checkpoint Mechanism

    5.1 Checkpoint Structure

    File: apps/api/services/runs/runMetadata.ts

    TypeScript

    Stored in: runs.metadata.resumeCheckpoint (JSONB column)

    5.2 When Checkpoints Are Created

    File: apps/api/services/chat/session/loop/agentLoop.turnOutcome.ts

    Checkpoints created on continuation turns (when agent loop continues):

    TypeScript

    Continuation Scenarios:

    ScenarioWhenWhy Checkpoint
    Tool Results ContinuationTools called but no code outputNeed to send tool results back to LLM
    No-Progress NudgeNo tools called, no outputNeed to send nudge prompt

    NOT Checkpointed:

    • Initial turn (turn 0)
    • Final turn (when loop exits)
    • Turns that produce code output
    5.3 Checkpoint Persistence Flow

    File: apps/api/services/runs/agent-run-worker/processor.session.ts

    TypeScript
    5.4 Checkpoint Usage on Worker Restart

    File: apps/api/services/chat/session/loop/agentLoop.runner.ts

    TypeScript
    5.5 Checkpoint Cleanup

    File: apps/api/services/runs/agent-run-worker/processor.finalize.ts

    On successful completion:

    TypeScript

    Why Clear?

    • Run is complete, no need to resume
    • Reduces metadata size
    • Prevents accidental replay of completed runs

    6. Event Sourcing & Replay

    6.1 Event Schema

    File: packages/shared/src/streamEvents.ts

    TypeScript

    Event Types:

    CategoryEvents
    Lifecyclemeta (SESSION_START, TURN_START, TURN_COMPLETE, SESSION_COMPLETE)
    Contenttext, thinking_*, file_*
    Sandboxsandbox_*, install_*, command
    Toolsweb_search, url_scrape
    Errorserror (fatal, recoverable)
    Metricsmetrics, rate_limit_status, preview_url, build_status
    6.2 Event Persistence

    Database Schema:

    Sql

    Append Function:

    TypeScript
    6.3 Event Replay Query

    File: apps/api/services/run-event-stream-utils/service.ts

    TypeScript

    Query Pattern:

    Sql
    6.4 Replay Batching

    Why Batch?

    • Large runs may have thousands of events
    • Don't overwhelm client with single massive response
    • Allow progressive rendering

    Batching Logic:

    TypeScript

    Multiple Batches:

    • First batch: 500 events
    • If more events exist, loop continues
    • Next iteration: next 500 events
    • Continues until all events replayed

    7. Frontend Reconnection Strategy

    7.1 Cursor Persistence

    File: apps/web/stores/chatStream/cursorPersistence.ts

    TypeScript

    Storage Strategy:

    LayerPurposeLifetime
    In-memory MapFast access during sessionTab lifetime
    sessionStoragePersistence across refreshTab lifetime (survives refresh)

    Why Not localStorage?

    • Cursors are session-specific
    • Don't persist across browser restarts
    • Avoid stale cursors for old runs
    7.2 Stream Resumption

    File: apps/web/stores/chatStream/resumeRunStream.ts

    TypeScript
    7.3 API Client

    File: apps/web/lib/api/chat.ts

    TypeScript

    Request Format:

    Plain text
    7.4 Page-Level Orchestration

    File: apps/web/hooks/chat/useChatPageOrchestration.ts

    Active Run Lookup:

    TypeScript

    Lookup Strategy:

    ModeAttemptsUse Case
    Aggressive6Page load, user expects active run
    Single1Background check
    Defer0Streaming already in progress
    7.5 Stream Processor with Replay

    File: apps/web/lib/streaming/processors/chatStreamProcessor.ts

    Built-in Replay:

    TypeScript

    Why Client-Side Replay?

    • Backend stream may end prematurely
    • Client detects incomplete session
    • Automatic retry with backoff
    • Merges results seamlessly

    8. Worker Resumption

    8.1 Worker Lifecycle

    File: apps/api/services/runs/agent-run-worker/processor.ts

    Startup:

    TypeScript
    8.2 Checkpoint Detection
    TypeScript
    8.3 Resumption Flow
    TypeScript
    8.4 Finalization

    File: apps/api/services/runs/agent-run-worker/processor.finalize.ts

    On Success:

    TypeScript

    On Failure:

    TypeScript

    Why Keep Checkpoint on Failure?

    • Allows manual retry from checkpoint
    • Preserves state for debugging
    • Can be cleared by explicit retry logic

    9. Failure Modes & Recovery

    9.1 Failure Mode Matrix
    Failure ModeDetectionRecoveryData Loss
    Client disconnectreq.close eventReconnect with cursorNone
    Slow clientBackpressure queueGraceful closeNone (events persisted)
    Worker crashJob timeoutQueue retry + checkpointMinimal (since last checkpoint)
    Redis unavailableConnection errorFallback to PostgreSQL pollingNone
    PostgreSQL unavailableQuery errorRetry with backoffTemporary (events buffered)
    Stream timeoutGuard timerTerminate with reasonNone (events persisted)
    Network blipSSE connection dropAuto-reconnectNone
    9.2 Client Disconnect Handling

    File: apps/api/services/run-event-stream-utils/service.ts

    TypeScript

    What Happens:

    1. Client closes connection (refresh, navigate away)
    2. Backend receives req.close event
    3. Unsubscribe from Redis pub/sub
    4. Stop heartbeats
    5. Close response gracefully
    6. Worker continues execution (independent of client connection)
    9.3 Slow Client Handling

    File: apps/api/services/run-event-stream-utils/service.ts

    TypeScript

    Detection:

    • Write queue exceeds threshold
    • res.write() returns false (backpressure)
    • Queue not draining fast enough

    Recovery:

    1. Close stream gracefully
    2. Client can reconnect with cursor
    3. Events already persisted → replay on reconnect
    9.4 Worker Crash Recovery

    File: apps/api/services/runs/agent-run-worker/processor.ts

    Crash Detection:

    • Worker process dies (OOM, panic, etc.)
    • Job queue detects timeout
    • Run remains in "RUNNING" state

    Recovery:

    1. Queue retries job (configurable retries)
    2. New worker picks up job
    3. Load run from DB
    4. Check for checkpoint in metadata
    5. Resume from checkpoint.turn

    What's Lost:

    • Work since last checkpoint
    • Typically: partial turn execution
    • Events persisted before crash are safe
    9.5 Stream Timeout

    File: apps/api/services/chat/session/orchestrator/streamGuards.ts

    TypeScript

    On Timeout:

    1. Abort controller triggered
    2. Agent loop stops
    3. Termination reason set to STREAM_TIMEOUT
    4. Session finalized with timeout reason
    5. Client receives error event
    9.6 Redis Unavailable

    File: apps/api/lib/redisPubSub.ts

    TypeScript

    Fallback Strategy:

    • Events still persisted to PostgreSQL
    • Client reconnects → replay from PostgreSQL
    • Live events missed during Redis outage
    • Mitigation: Poll PostgreSQL for new events

    10. Infrastructure Cross-Questions

    10.1 PostgreSQL

    Q: What's the event volume per run?

    Typical run:

    • Meta events: 5-10 (session start, turn start/complete x5, session complete)
    • Text events: 50-200 (narrative, explanations)
    • File events: 10-50 (file_start, file_content xN, file_end)
    • Command events: 5-20 (command execution results)
    • Total: 70-300 events per run

    Q: What's the storage requirement?

    Per run estimate:

    • Event envelope: ~200 bytes overhead
    • Event payload: ~500 bytes average (varies widely)
    • Per event: ~700 bytes
    • Per run (200 events): ~140 KB

    At 10,000 runs/day:

    • Daily: 1.4 GB
    • Monthly: 42 GB
    • Yearly: 500 GB

    Q: Should we partition the run_events table?

    Recommendation: Yes, partition by created_at (monthly partitions)

    Sql

    Benefits:

    • Faster cleanup (drop old partitions)
    • Improved query performance (partition pruning)
    • Easier backup/restore

    Q: What indexes are needed?

    Sql
    10.2 Redis

    Q: What's the Redis memory footprint?

    Pub/sub channels:

    • One channel per active run: edward:run-events:{runId}
    • Channel name: ~50 bytes
    • Subscriber overhead: ~100 bytes per subscriber
    • Per active run: ~150 bytes

    At 1,000 concurrent runs:

    • Channel memory: ~150 KB (negligible)

    Q: What about event buffering?

    Events are NOT buffered in Redis (only published):

    • Publisher sends event → all subscribers receive
    • No persistence in Redis
    • PostgreSQL is the source of truth

    Q: Redis cluster vs standalone?

    Recommendation: Standalone for pub/sub

    Why:

    • Pub/sub doesn't benefit from clustering (messages not persisted)
    • Simpler deployment
    • Lower latency

    Failover:

    • Sentinel for automatic failover
    • Clients reconnect on failover
    • Events persisted to PostgreSQL during outage

    Q: What's the pub/sub latency?

    Typical latency:

    • Publish to subscriber: <1ms (same region)
    • 99th percentile: <5ms

    Impact:

    • Live events arrive within milliseconds
    • Replay from PostgreSQL is the bottleneck (not Redis)
    10.3 SSE Infrastructure

    Q: How many concurrent SSE connections?

    Estimate:

    • Active runs: 1,000
    • Connections per run: 1-3 (user may have multiple tabs)
    • Concurrent connections: 1,000-3,000

    Q: Can a single server handle this?

    Node.js SSE capacity:

    • Memory per connection: ~10-50 KB
    • CPU per connection: minimal (event-driven)
    • Single server: 10,000+ connections feasible

    Q: What about horizontal scaling?

    Challenge: SSE connections are sticky (can't load balance mid-stream)

    Solutions:

    1. Sticky sessions (recommended)

      • Load balancer routes by cookie/session
      • Same server handles entire stream
      • Simple, effective

    2. Connection migration

      • On server shutdown, notify clients
      • Clients reconnect to new server
      • Complex, rarely needed

    3. WebSocket + migration

      • Use WebSocket instead of SSE
      • Supports connection migration
      • More complex protocol

    Q: What's the heartbeat strategy?

    TypeScript

    Why Heartbeats?

    • Keep connection alive (prevent timeout)
    • Detect dead connections
    • Load balancers may close idle connections

    Q: How to handle connection limits?

    Browser limits:

    • Chrome: 6 connections per domain
    • Firefox: 6 connections per domain
    • Safari: 6 connections per domain

    Mitigation:

    • Single SSE connection per chat
    • Reuse connection for multiple runs
    • Use domain sharding if needed (not recommended)
    10.4 Worker Scaling

    Q: How many workers?

    Formula:

    Plain text

    Typical:

    • Concurrent runs: 1,000
    • Runs per worker: 10-50 (depends on LLM latency)
    • Workers needed: 20-100

    Q: Worker autoscaling?

    Metrics to track:

    • Queue depth (Redis)
    • Average run duration
    • Worker CPU/memory

    Scaling policy:

    • Scale up: Queue depth > threshold for 2 minutes
    • Scale down: Queue depth = 0 for 10 minutes

    Q: What about worker affinity?

    No affinity needed:

    • Workers are stateless
    • State in PostgreSQL + checkpoint
    • Any worker can process any job
    10.5 Data Retention

    Q: How long to keep events?

    Recommendation: 30 days

    Why:

    • Supports debugging recent issues
    • Allows replay for active users
    • Compliance (audit trail)

    Cleanup strategy:

    Sql

    Q: What about checkpoint retention?

    Recommendation: Clear on completion, keep on failure

    Why:

    • Completed runs don't need checkpoint
    • Failed runs may need manual retry
    • Checkpoints are small (few KB)
    10.6 Monitoring

    Q: What metrics to track?

    Stream Health:

    • stream_reconnect_count: Reconnections per run
    • stream_replay_events: Events replayed per reconnect
    • stream_duration_ms: Total stream duration
    • stream_termination_reason: Distribution of termination reasons

    Checkpoint Health:

    • checkpoint_created_count: Checkpoints created
    • checkpoint_resume_count: Runs resumed from checkpoint
    • checkpoint_turn_distribution: Turns per checkpoint

    Event Volume:

    • events_persisted_count: Events per run
    • events_replay_lag_ms: Time to replay events
    • redis_pubsub_latency_ms: Pub/sub latency

    Q: What alerts to set?

    MetricThresholdAction
    stream_reconnect_count > 5Per runInvestigate network issues
    checkpoint_resume_count > 10%Of runsInvestigate worker stability
    events_replay_lag_ms > 5000P99Optimize replay queries
    redis_pubsub_latency_ms > 100P99Check Redis health

    11. Key Files Reference

    Core Orchestration
    FilePurposeLines
    apps/api/services/runs/runEvents.service.tsEvent persistence~40
    apps/api/services/run-event-stream-utils/service.tsSSE streaming + replay~250
    apps/api/services/sse-utils/service.tsSSE backpressure handling~200
    apps/api/services/runs/agent-run-worker/processor.tsWorker execution~350
    apps/api/services/runs/agent-run-worker/processor.helpers.tsEvent capture~150
    apps/api/services/runs/agent-run-worker/processor.finalize.tsRun finalization~150
    Checkpoint & Continuation
    FilePurposeLines
    apps/api/services/runs/runMetadata.tsCheckpoint schema~100
    apps/api/services/runs/agent-run-worker/processor.session.tsCheckpoint persistence~100
    apps/api/services/chat/session/loop/agentLoop.runner.tsAgent loop with checkpoint~200
    apps/api/services/chat/session/loop/agentLoop.turnOutcome.tsContinuation logic~250
    apps/api/services/chat/session/shared/checkpoint.types.tsCheckpoint types~15
    apps/api/services/chat/session/shared/continuation.tsContinuation prompts~250
    Frontend
    FilePurposeLines
    apps/web/stores/chatStream/cursorPersistence.tsCursor storage~50
    apps/web/stores/chatStream/resumeRunStream.tsStream resumption~150
    apps/web/lib/api/chat.tsSSE API client~100
    apps/web/lib/streaming/processors/chatStreamProcessor.tsStream processing~400
    apps/web/hooks/chat/useChatPageOrchestration.tsPage orchestration~250
    Shared Types
    FilePurposeLines
    packages/shared/src/streamEvents.tsEvent type definitions~250
    Infrastructure
    FilePurposeLines
    apps/api/lib/redisPubSub.tsRedis pub/sub~80
    apps/api/lib/redis.tsRedis client creation~20
    Tests
    FilePurpose
    apps/api/tests/services/runs/agentRun.processor.session.test.tsCheckpoint tests
    apps/api/tests/services/runs/runMetadata.test.tsMetadata parsing tests
    apps/api/tests/controllers/chat/streamSession.shared.test.tsContinuation prompt tests
    apps/api/tests/services/chat/runEventStream.utils.service.test.tsStream replay tests

    Appendix A: Glossary

    TermDefinition
    CheckpointSnapshot of agent loop state for resumption
    CursorLast processed event ID (client-side)
    Event SourcingPattern of persisting all state changes as events
    ReplayRe-emitting historical events on reconnection
    SSEServer-Sent Events (one-way streaming from server to client)
    TurnSingle iteration of agent loop (LLM call + tool execution)
    WorkerBackground process that executes agent runs

    Appendix B: Sequence Diagrams

    B.1 Normal Stream Flow
    Plain text
    B.2 Reconnection Flow
    Plain text

    Appendix C: Configuration Reference

    Environment Variables
    VariableDefaultDescription
    STREAM_GUARD_TIMEOUT_MS1200000Stream timeout (20 min)
    MAX_AGENT_CONTINUATION_PROMPT_CHARS20000Continuation prompt limit
    MAX_AGENT_TURNS10Max agent loop iterations
    MAX_AGENT_TOOL_CALLS_PER_TURN10Tool budget per turn
    MAX_AGENT_TOOL_CALLS_PER_RUN50Tool budget per run
    MAX_REPLAY_BATCH500Events per replay batch
    HEARTBEAT_INTERVAL_MS15000SSE heartbeat interval
    Redis Channels
    ChannelPurpose
    edward:run-events:{runId}Live event pub/sub
    edward:run-cancel:{runId}Cancel signal
    agent-runsWorker job queue
    Storage Keys
    KeyPurpose
    sse_cursor:{chatId}:{runId}Frontend cursor

    Document Version: 1.0
    Last Updated: March 25, 2026
    Author: Principal Engineering Review