Scenarios and Runbooks — Real-Time AI Interaction

Timeline:
t=0.0s   User sends question via WebSocket
t=0.3s   Bedrock TTFT, first token arrives
t=0.3-1.2s  40 tokens streamed to client successfully
t=1.2s   ──── NETWORK INTERRUPTION ────
         │  User's mobile switches from WiFi to LTE
         │  TCP connection resets
         │  API Gateway receives TCP RST
         │  API Gateway fires $disconnect route
         └──────────────────────────────────
t=1.2s   Bedrock continues generating (unaware of disconnect)
t=1.3s   ECS tries POST @connections/{connectionId}
         → GoneException (connection no longer exists)
t=1.3s   ECS catches GoneException, stops reading Bedrock stream
t=1.3s   Partial response "Based on your interest in..." is lost
         (only existed in ECS task memory, not persisted)
t=3.0s   Client reconnects with new connectionId
t=3.1s   Client sends resume request with messageId
t=3.2s   Server has no record of partial response → must regenerate

┌──────────┐     ┌─────────────┐     ┌──────────┐     ┌─────────┐
│  Client  │────>│ API Gateway │────>│   ECS    │────>│ Bedrock │
│ (Mobile) │     │  WebSocket  │     │ Fargate  │     │ Claude3 │
└──────┬───┘     └──────┬──────┘     └────┬─────┘     └────┬────┘
       │                │                 │                 │
       │  40 tokens OK  │                 │                 │
       │<═══════════════│<════════════════│<════════════════│
       │                │                 │                 │
    ╔══╧════════════╗   │                 │                 │
    ║ WiFi → LTE    ║   │                 │                 │
    ║ TCP RST       ║   │                 │                 │
    ╚══╤════════════╝   │                 │                 │
       │                │                 │                 │
       X  Connection    │── $disconnect──>│                 │
          LOST          │                 │── GoneException │
                        │                 │   on next POST  │
                        │                 │                 │
                        │                 │── STOP stream──>│
                        │                 │                 │
                        │                 │  Partial text   │
                        │                 │  LOST (not saved)│

# ECS Fargate application logs
filter @message like /GoneException/
| stats count() as gone_exceptions by bin(5m)

# Identify mid-stream disconnects specifically
filter @message like /GoneException/ and @message like /chunk_index > 0/
| stats count() as mid_stream_drops by bin(5m)

# In the ECS streaming orchestrator, after each token:
async def handle_stream_token(self, message_id: str, chunk: StreamChunk):
    """Process each streaming token with Redis persistence."""
    # 1. Append token to Redis list (survives disconnect)
    redis_key = f"stream:{message_id}:tokens"
    self._redis.rpush(redis_key, json.dumps({
        "text": chunk.text,
        "index": chunk.index,
        "timestamp": time.time(),
    }))
    # Set TTL for cleanup (5 minutes — enough for reconnection)
    self._redis.expire(redis_key, 300)

    # 2. Try to send to client via WebSocket
    try:
        await self.ws_handler.send_token(connection_id, chunk.text, chunk.index)
    except GoneException:
        # Connection is gone — but tokens are safe in Redis
        logger.info(f"Connection gone for {message_id}, tokens preserved in Redis")
        # Do NOT stop reading from Bedrock — continue generating
        # The full response will be available for reconnection

# Modified stream reading loop
async def stream_with_persistence(self, message_id: str, prompt: str):
    connection_alive = True

    async for chunk in self.fm_client.stream_response(prompt, model="sonnet"):
        # Always persist to Redis
        await self.persist_token(message_id, chunk)

        # Try to deliver if connection is still alive
        if connection_alive:
            try:
                await self.ws_handler.send_token(connection_id, chunk.text)
            except GoneException:
                connection_alive = False
                logger.info(f"Continuing generation for {message_id} (client disconnected)")

    # Save final response to DynamoDB session
    full_response = self.assemble_from_redis(message_id)
    await self.save_session(message_id, full_response)

// Client reconnection handler
websocket.onclose = async (event) => {
    if (event.code === 1006 && currentMessageId) {
        // Abnormal closure during active stream
        savePartialResponse(currentMessageId, accumulatedText);

        await reconnectWithBackoff();

        // Request resume from server
        websocket.send(JSON.stringify({
            action: "resume",
            data: {
                messageId: currentMessageId,
                lastChunkIndex: lastReceivedChunkIndex,
            }
        }));
    }
};

async def handle_resume(self, connection_id: str, message_id: str, last_index: int):
    """Resume streaming from where the client left off."""
    redis_key = f"stream:{message_id}:tokens"
    total_tokens = self._redis.llen(redis_key)

    if total_tokens == 0:
        # No cached tokens — must regenerate
        await self.ws_handler.send_error(
            connection_id, "STREAM_EXPIRED", "Regenerating response..."
        )
        return await self.regenerate(connection_id, message_id)

    # Replay tokens from last_index + 1 onwards
    for i in range(last_index + 1, total_tokens):
        token_json = self._redis.lindex(redis_key, i)
        token = json.loads(token_json)
        await self.ws_handler.send_token(
            connection_id, token["text"], token["index"]
        )

    # Check if stream is still generating
    stream_complete_key = f"stream:{message_id}:complete"
    if self._redis.exists(stream_complete_key):
        await self.ws_handler.send_complete(connection_id)
    else:
        # Stream still in progress — attach this connection to receive new tokens
        await self.attach_to_active_stream(connection_id, message_id)

# 1. Simulate mid-stream disconnect
# Connect via wscat, send a message, then kill the connection after 1 second

# 2. Check Redis for cached tokens
redis-cli LRANGE stream:msg-test-001:tokens 0 -1

# 3. Reconnect and request resume
# Send resume message, verify tokens are replayed

# 4. Monitor CloudWatch metrics
# Verify mid_stream_disconnect_with_recovery metric counts up
# Verify stream_data_loss metric stays at 0

Token Production vs Consumption Rate:

Production (Bedrock → ECS → WebSocket → Client):
  80 tokens/sec ════════════════════════════════>

Consumption (Client JS parse + render + scroll):
  ~30 tokens/sec ═══════════════>
                                  ↑
                        50 tokens/sec deficit
                        accumulates in browser
                        WebSocket message queue

Memory over time (2000 token response, ~25 seconds):
  RAM (MB)
  500 ┤                                          ╭──── Tab killed
      │                                     ╭────╯
  400 ┤                                ╭────╯
      │                           ╭────╯
  300 ┤                      ╭────╯
      │                 ╭────╯
  200 ┤            ╭────╯
      │       ╭────╯
  100 ┤  ╭────╯
      │──╯
   50 ┤─ Baseline
      └────┬────┬────┬────┬────┬────┬────┬────┬──
           5   10   15   20   25   30   35   40  seconds

┌─────────────────────────────────────────────────────────┐
│                    Client (Browser)                      │
│                                                         │
│  WebSocket        Message Queue        JS Main Thread   │
│  Receiver         (Growing!)           (Overloaded)     │
│  ┌──────┐        ┌──────────────┐     ┌──────────────┐ │
│  │ WS   │──────> │ token 41     │     │ Render token │ │
│  │ onmsg│        │ token 42     │     │ #15          │ │
│  │      │        │ token 43     │     │              │ │
│  │ 80/s │        │ ...          │     │ DOM update   │ │
│  │      │        │ token 120    │     │ MD parse     │ │
│  │      │        │ (80 queued!) │     │ Scroll       │ │
│  └──────┘        └──────────────┘     │ ~30/s        │ │
│                   ▲                    └──────────────┘ │
│                   │ 50 tokens/sec accumulating          │
│                   │ ~300 KB/sec memory growth            │
└─────────────────────────────────────────────────────────┘

class StreamFlowController {
    constructor(websocket, options = {}) {
        this.ws = websocket;
        this.maxQueueSize = options.maxQueueSize || 30;
        this.resumeThreshold = options.resumeThreshold || 10;
        this.isPaused = false;
        this.queueSize = 0;
    }

    onTokenReceived() {
        this.queueSize++;

        if (!this.isPaused && this.queueSize > this.maxQueueSize) {
            this.ws.send(JSON.stringify({
                action: "flow_control",
                data: { command: "pause" }
            }));
            this.isPaused = true;
        }
    }

    onTokenRendered() {
        this.queueSize = Math.max(0, this.queueSize - 1);

        if (this.isPaused && this.queueSize <= this.resumeThreshold) {
            this.ws.send(JSON.stringify({
                action: "flow_control",
                data: { command: "resume" }
            }));
            this.isPaused = false;
        }
    }
}

async def adaptive_send(self, connection_id: str, tokens: list):
    """Batch multiple tokens into single WebSocket frame when client is slow."""
    if self.backpressure.state == FlowState.BUFFERING:
        # Client has signaled pause — accumulate tokens
        return

    if self.backpressure.state == FlowState.DRAINING:
        # Send accumulated tokens in batches
        batch = tokens[:10]  # 10 tokens per frame
        combined_text = "".join(t.text for t in batch)
        await self.ws_handler.send_token(
            connection_id,
            combined_text,
            batch[0].index,
        )

// BEFORE: O(n^2) — re-render all text on each token
onToken(text) {
    this.fullText += text;
    this.element.innerHTML = markdownToHtml(this.fullText); // EXPENSIVE
}

// AFTER: O(1) — append only new text, batch DOM updates
class StreamRenderer {
    constructor(container) {
        this.container = container;
        this.pendingText = "";
        this.rafScheduled = false;
    }

    onToken(text) {
        this.pendingText += text;
        if (!this.rafScheduled) {
            this.rafScheduled = true;
            requestAnimationFrame(() => this.flush());
        }
    }

    flush() {
        this.rafScheduled = false;
        if (!this.pendingText) return;

        // Append a new text node (not re-render everything)
        const textNode = document.createTextNode(this.pendingText);
        this.container.appendChild(textNode);
        this.pendingText = "";

        // Scroll to bottom (use scrollIntoView, not scrollTop assignment)
        this.container.lastChild.scrollIntoView({ behavior: "instant" });
    }
}

// Client-side performance test
const metrics = {
    tokensReceived: 0,
    tokensRendered: 0,
    maxQueueDepth: 0,
    peakMemoryMB: 0,
};

// After a 2000-token stream:
// tokensReceived === tokensRendered (no backlog)
// maxQueueDepth < 50 (backpressure working)
// peakMemoryMB < 100 (no memory bloat)

Timeline:
t=0.0s   Client sends complex query via WebSocket
t=0.0s   API Gateway routes to ECS Fargate (sendMessage route)
t=0.1s   ECS begins RAG retrieval
         ──── NO TOKENS FLOWING ────
t=0.5s   OpenSearch embedding query sent
t=2.0s   OpenSearch returns 15 context documents
t=2.5s   DynamoDB product detail lookups complete
t=3.5s   Prompt construction (3000 tokens) complete
t=4.0s   Bedrock InvokeModelWithResponseStream called
t=4.0-4.8s  Bedrock processing large prompt (no output yet)
         ──── STILL NO TOKENS FLOWING ────
t=4.8s   First token generated by Bedrock
t=5.0s   ⚠ Client timeout fires! Client shows "Request timed out"
         Client may attempt to reconnect or re-send
t=5.1s   Server pushes first token via @connections
         Client receives token but has already shown error

Client          API Gateway       ECS Fargate        OpenSearch      Bedrock
  │                 │                 │                   │              │
  │── sendMessage──>│── Route ───────>│                   │              │
  │                 │                 │── embed query ───>│              │
  │   ⏳ waiting     │   ⏳ waiting     │                   │              │
  │                 │                 │<── 15 docs ───────│              │
  │   ⏳ waiting     │   ⏳ waiting     │── product lookup ─┤              │
  │                 │                 │                   │              │
  │   ⏳ 5s timeout  │                 │── build prompt ──>│              │
  │   approaching   │                 │                   │              │
  │                 │                 │── stream call ────┤────────────>│
  │                 │                 │                   │   ⏳ TTFT     │
  │   ⚠ TIMEOUT!   │                 │                   │   800ms      │
  │   "Request      │                 │                   │              │
  │    failed"      │                 │<─ first token ────┤──────────── │
  │                 │<─ @connections──│                   │              │
  │<── token ───────│  (but client    │                   │              │
  │   (too late)    │   already       │                   │              │
  │                 │   timed out)    │                   │              │

async def handle_send_message(self, connection_id: str, message: dict):
    """Handle incoming chat message with progress signals."""
    message_id = generate_message_id()

    # IMMEDIATELY acknowledge receipt (before any processing)
    await self.ws_handler._post_to_connection(connection_id, {
        "action": "message_received",
        "data": {
            "messageId": message_id,
            "status": "processing",
            "timestamp": int(time.time() * 1000),
        }
    })

    # Client now knows the server received the message
    # Client should reset its timeout to a longer "processing" timeout

async def rag_with_progress(self, connection_id: str, message_id: str, query: str):
    """RAG pipeline that sends progress updates to the client."""

    # Progress: searching knowledge base
    await self.send_progress(connection_id, message_id, "searching", 0.2)
    context_docs = await self.opensearch.vector_search(query)

    # Progress: retrieving product details
    await self.send_progress(connection_id, message_id, "retrieving", 0.4)
    products = await self.dynamodb.batch_get_products(context_docs)

    # Progress: building context
    await self.send_progress(connection_id, message_id, "preparing", 0.6)
    prompt = self.build_prompt(query, context_docs, products)

    # Progress: generating response
    await self.send_progress(connection_id, message_id, "generating", 0.8)
    # Now call Bedrock streaming — first token will arrive shortly

async def send_progress(self, connection_id, message_id, stage, progress):
    await self.ws_handler._post_to_connection(connection_id, {
        "action": "progress",
        "data": {
            "messageId": message_id,
            "stage": stage,
            "progress": progress,
            "timestamp": int(time.time() * 1000),
        }
    })

const TIMEOUT_CONFIG = {
    ack_timeout: 3000,        // 3s to receive message_received
    processing_timeout: 30000, // 30s for RAG + generation to start
    streaming_timeout: 10000,  // 10s between tokens during streaming
    total_timeout: 120000,     // 2 min absolute maximum
};

class TimeoutManager {
    constructor(ws) {
        this.ws = ws;
        this.currentPhase = "waiting_ack";
        this.timer = null;
    }

    startAckTimeout() {
        this.currentPhase = "waiting_ack";
        this.timer = setTimeout(() => {
            this.onTimeout("No acknowledgment received");
        }, TIMEOUT_CONFIG.ack_timeout);
    }

    onMessageReceived() {
        clearTimeout(this.timer);
        this.currentPhase = "processing";
        this.timer = setTimeout(() => {
            this.onTimeout("Processing taking too long");
        }, TIMEOUT_CONFIG.processing_timeout);
    }

    onProgressUpdate() {
        clearTimeout(this.timer);
        // Reset processing timeout on each progress update
        this.timer = setTimeout(() => {
            this.onTimeout("Processing stalled");
        }, TIMEOUT_CONFIG.processing_timeout);
    }

    onFirstToken() {
        clearTimeout(this.timer);
        this.currentPhase = "streaming";
        this.resetStreamingTimeout();
    }

    onToken() {
        this.resetStreamingTimeout();
    }

    resetStreamingTimeout() {
        clearTimeout(this.timer);
        this.timer = setTimeout(() => {
            this.onTimeout("Streaming stalled");
        }, TIMEOUT_CONFIG.streaming_timeout);
    }
}

# Simulate a complex query with long RAG pipeline
# 1. Send complex query via WebSocket
# 2. Verify ACK received within 1 second
# 3. Verify 2+ progress updates received
# 4. Verify first token received within 6 seconds
# 5. Verify NO client timeout events in telemetry

# CloudWatch query for timeout incidents
filter @message like /client_timeout/
| stats count() by bin(1h)
# Should drop to near zero after fix

Normal SSE Lifecycle:
  Client ──GET /stream──> ECS ──stream tokens──> Client ──event:done──> Client closes
  ✓ Clean                                                               ✓ Resources freed

Broken SSE Lifecycle (this scenario):
  Client ──GET /stream──> ECS ──stream tokens──> Client ──event:done──> ???
  │                        │                                             │
  │  Connection stays open │  Handler stays alive                        │ EventSource does
  │  HTTP keep-alive       │  File descriptor held                       │ NOT auto-close on
  │                        │  Memory allocated                           │ "done" event
  │                        │  Redis refs active                          │
  │                        │                                             │
  │  ◄──── ZOMBIE CONNECTION ────────────────────────────────────►      │

Over time:
  ┌─────────────────────────────────────────────────────────┐
  │              ECS Fargate Task                            │
  │                                                         │
  │  Active SSE connections:     15                         │
  │  Zombie SSE connections:    247    ← growing hourly     │
  │  File descriptors used:    524/1024                     │
  │  Memory consumed by zombies: ~124 MB                    │
  │                                                         │
  │  ⚠ At this rate, FD limit reached in 4 hours           │
  └─────────────────────────────────────────────────────────┘

async def handle_sse_stream(request):
    """SSE endpoint that properly closes after stream completes."""
    session_id = request.query.get("session_id")
    sse_manager = SSEResponseManager()
    session = sse_manager.create_session(session_id)

    async def generate():
        # Send retry directive
        yield f"retry: 3000\n\n"

        # Stream tokens from Bedrock
        async for chunk in bedrock_stream:
            event = sse_manager.create_token_event(session, chunk.text)
            yield event.serialize()

        # Send done event
        done_event = sse_manager.create_done_event(session)
        yield done_event.serialize()

        # CRITICAL: Close the response stream after done
        # This terminates the HTTP response, signaling the client
        # that no more data will be sent on this connection.
        # Without this, the connection stays open as a zombie.

    response = web.StreamResponse()
    response.headers.update(sse_manager.get_sse_headers())
    await response.prepare(request)

    try:
        async for data in generate():
            await response.write(data.encode("utf-8"))
    finally:
        # Ensure response is finalized (sends zero-length chunk for chunked encoding)
        await response.write_eof()
        sse_manager.cleanup_session(session_id)
        logger.info(f"SSE connection closed for session {session_id}")

const eventSource = new EventSource(`/stream?session_id=${sessionId}`);

eventSource.addEventListener("done", (event) => {
    const data = JSON.parse(event.data);
    console.log(`Stream complete: ${data.total_chunks} chunks`);

    // CRITICAL: Explicitly close the EventSource
    // Without this, the browser will attempt to reconnect
    // when the server closes the connection
    eventSource.close();

    // Update UI to show response is complete
    updateStreamStatus("complete");
});

eventSource.addEventListener("error", (event) => {
    if (eventSource.readyState === EventSource.CLOSED) {
        // Server closed the connection — this is expected after "done"
        console.log("SSE connection closed by server (expected)");
        return;
    }
    // Genuine error
    handleSSEError(event);
});

class SSEConnectionReaper:
    """
    Periodically scans for and terminates zombie SSE connections
    that should have been closed after stream completion.
    """

    MAX_SSE_DURATION = 300  # 5 minutes absolute max for any SSE connection
    REAP_INTERVAL = 60      # Check every 60 seconds

    def __init__(self, sse_manager: SSEResponseManager):
        self.sse_manager = sse_manager
        self._active_connections: Dict[str, float] = {}  # session_id -> start_time

    def register(self, session_id: str):
        self._active_connections[session_id] = time.time()

    def deregister(self, session_id: str):
        self._active_connections.pop(session_id, None)

    async def reap_loop(self):
        """Run in background to clean up zombie connections."""
        while True:
            await asyncio.sleep(self.REAP_INTERVAL)
            now = time.time()
            zombies = [
                sid for sid, start in self._active_connections.items()
                if now - start > self.MAX_SSE_DURATION
            ]
            for sid in zombies:
                logger.warning(f"Reaping zombie SSE connection: {sid}")
                self.sse_manager.cleanup_session(sid)
                self.deregister(sid)
                # Force-close the response (implementation depends on web framework)

# Set ALB idle timeout to 120 seconds for SSE target group
# This ensures connections that are truly idle get cleaned up
# Normal SSE responses complete in <30 seconds
aws elbv2 modify-target-group-attributes \
    --target-group-arn arn:aws:elasticloadbalancing:... \
    --attributes Key=idle_timeout.timeout_seconds,Value=120

# 1. Trigger 10 SSE streaming responses
# 2. After all complete, check open connections on ECS task:
ss -s  # Show socket summary
lsof -i -P | grep ESTABLISHED | wc -l

# Expected: Only baseline connections (Redis, DDB, etc.)
# NO lingering SSE connections

# 3. CloudWatch metric: sse_zombie_connection_count should be 0

UTF-8 encoding of "東京グール" (Tokyo Ghoul):
  東 = E6 9D B1     (3 bytes)
  京 = E4 BA AC     (3 bytes)
  グ = E3 82 B0     (3 bytes)
  ー = E3 83 BC     (3 bytes)
  ル = E3 83 AB     (3 bytes)
  Total: 15 bytes

Chunked encoding with BAD chunk boundary:

  Chunk 1 (10 bytes):          Chunk 2 (5 bytes):
  E6 9D B1  E4 BA AC  E3 82   B0  E3 83 BC  E3 83 AB
  ├─ 東 ──┤ ├─ 京 ──┤ ├─ ??   ├──────────────────────┤
                       ↑
                  SPLIT HERE!
                  E3 82 is the start of グ (needs E3 82 B0)
                  but B0 is in the next chunk

  Client receives Chunk 1:
  "東京" + INVALID BYTE SEQUENCE (E3 82) → TextDecoder error or U+FFFD (�)

  Client receives Chunk 2:
  B0 E3 83 BC E3 83 AB → starts with continuation byte (B0) → MORE ERRORS

ECS Fargate                  ALB                      Client
    │                         │                         │
    │── HTTP Response ──────>│                         │
    │   Transfer-Encoding:   │                         │
    │   chunked              │                         │
    │                         │                         │
    │── chunk: "東京" + E382 ─>│── forward chunk 1 ────>│
    │   (10 bytes)           │                         │── TextDecoder
    │                         │                         │   ERROR on E3 82
    │                         │                         │   Shows: 東京�
    │── chunk: B0 + "ール" ──>│── forward chunk 2 ────>│
    │   (5 bytes)            │                         │── TextDecoder
    │                         │                         │   ERROR on B0
    │                         │                         │   Shows: 東京��ール
    │                         │                         │
    │                         │                         │   Expected: 東京グール
    │                         │                         │   Got:      東京��ール

const decoder = new TextDecoder("utf-8", { fatal: true }); // strict mode
try {
    const text = decoder.decode(chunk);
} catch (e) {
    // TypeError: The encoded data was not valid UTF-8
    reportUnicodeError(chunk);
}

class Utf8SafeChunkedWriter:
    """
    Writes chunked HTTP responses with guaranteed UTF-8 character
    boundary alignment.
    """

    def __init__(self, response, target_chunk_size: int = 4096):
        self.response = response
        self.target_chunk_size = target_chunk_size
        self._buffer = bytearray()

    async def write(self, text: str) -> None:
        """Write text to the chunked response, UTF-8-safe."""
        encoded = text.encode("utf-8")
        self._buffer.extend(encoded)

        while len(self._buffer) >= self.target_chunk_size:
            # Find the last safe UTF-8 boundary at or before target_chunk_size
            boundary = self._find_utf8_boundary(
                self._buffer, self.target_chunk_size
            )
            # Flush up to the boundary
            chunk = bytes(self._buffer[:boundary])
            self._buffer = self._buffer[boundary:]
            await self.response.write(chunk)

    async def flush(self) -> None:
        """Flush remaining buffer (guaranteed complete UTF-8)."""
        if self._buffer:
            await self.response.write(bytes(self._buffer))
            self._buffer.clear()

    @staticmethod
    def _find_utf8_boundary(data: bytearray, max_pos: int) -> int:
        """
        Find the last valid UTF-8 character boundary at or before max_pos.

        UTF-8 byte patterns:
          0xxxxxxx  -> ASCII, single byte (always a boundary)
          110xxxxx  -> Start of 2-byte sequence
          1110xxxx  -> Start of 3-byte sequence
          11110xxx  -> Start of 4-byte sequence
          10xxxxxx  -> Continuation byte (NOT a boundary)
        """
        pos = min(max_pos, len(data))

        # Walk backward from pos until we find a character start
        while pos > 0:
            byte = data[pos - 1]

            # If this byte is ASCII (0xxxxxxx), boundary is after it
            if byte < 0x80:
                return pos

            # If this byte is a continuation byte (10xxxxxx), keep going back
            if (byte & 0xC0) == 0x80:
                pos -= 1
                continue

            # This byte is a start byte (11xxxxxx)
            # Determine the expected sequence length
            if (byte & 0xE0) == 0xC0:
                seq_len = 2
            elif (byte & 0xF0) == 0xE0:
                seq_len = 3
            elif (byte & 0xF8) == 0xF0:
                seq_len = 4
            else:
                # Invalid byte, treat as boundary
                return pos

            # Check if the full sequence fits before max_pos
            seq_start = pos - 1
            if seq_start + seq_len <= max_pos:
                # Full sequence fits, boundary is after it
                return seq_start + seq_len
            else:
                # Sequence would be split, boundary is before it
                return seq_start

        return 0

async def stream_chunked_response(request, bedrock_stream):
    """Stream Bedrock response with UTF-8-safe chunked encoding."""
    response = web.StreamResponse(
        headers={
            "Content-Type": "text/plain; charset=utf-8",
            "Transfer-Encoding": "chunked",
        }
    )
    await response.prepare(request)

    writer = Utf8SafeChunkedWriter(response, target_chunk_size=4096)

    async for chunk in bedrock_stream:
        if chunk.text:
            await writer.write(chunk.text)

    # Flush any remaining bytes (guaranteed complete)
    await writer.flush()
    await response.write_eof()

class ResilientStreamDecoder {
    constructor() {
        // Use non-fatal mode: replaces invalid sequences with U+FFFD
        // instead of throwing
        this.decoder = new TextDecoder("utf-8", { fatal: false });
        this.pendingBytes = new Uint8Array(0);
    }

    decode(chunk) {
        // Combine any pending bytes from previous chunk
        const combined = this._concat(this.pendingBytes, new Uint8Array(chunk));

        // Find the last safe UTF-8 boundary
        const boundary = this._findUtf8Boundary(combined);

        if (boundary === combined.length) {
            // All bytes form complete characters
            this.pendingBytes = new Uint8Array(0);
            return this.decoder.decode(combined, { stream: true });
        }

        // Split at boundary: decode complete chars, save incomplete
        const complete = combined.slice(0, boundary);
        this.pendingBytes = combined.slice(boundary);

        return this.decoder.decode(complete, { stream: true });
    }

    flush() {
        // Decode any remaining bytes at end of stream
        if (this.pendingBytes.length > 0) {
            const text = this.decoder.decode(this.pendingBytes);
            this.pendingBytes = new Uint8Array(0);
            return text;
        }
        return this.decoder.decode();
    }

    _findUtf8Boundary(bytes) {
        let pos = bytes.length;
        // Walk backward to find the start of the last character
        while (pos > 0 && pos > bytes.length - 4) {
            pos--;
            const byte = bytes[pos];
            if ((byte & 0x80) === 0) {
                // ASCII — boundary is after this byte
                return pos + 1;
            }
            if ((byte & 0xC0) !== 0x80) {
                // Start byte found — check if sequence is complete
                let expectedLen;
                if ((byte & 0xE0) === 0xC0) expectedLen = 2;
                else if ((byte & 0xF0) === 0xE0) expectedLen = 3;
                else if ((byte & 0xF8) === 0xF0) expectedLen = 4;
                else return pos; // invalid, skip

                if (pos + expectedLen <= bytes.length) {
                    return pos + expectedLen; // Sequence complete
                }
                return pos; // Sequence incomplete — boundary before it
            }
        }
        return bytes.length;
    }

    _concat(a, b) {
        const result = new Uint8Array(a.length + b.length);
        result.set(a, 0);
        result.set(b, a.length);
        return result;
    }
}

def create_token_event(self, session, text):
    """Create a token event, ensuring text contains only complete UTF-8."""
    # Validate the text is valid UTF-8 (should be, since it came from Bedrock)
    try:
        text.encode("utf-8").decode("utf-8")
    except UnicodeError:
        logger.warning(f"Invalid UTF-8 in token text, cleaning: {repr(text)}")
        text = text.encode("utf-8", errors="replace").decode("utf-8")

    # Rest of event creation unchanged
    event_id = f"{session.session_id}-{session.chunk_index}"
    session.chunk_index += 1
    session.accumulated_text += text
    session.last_event_id = event_id
    session.last_sent_at = time.time()

    return SSEEvent(
        event_type=SSEEventType.TOKEN,
        data={"text": text, "index": session.chunk_index - 1},
        event_id=event_id,
    )

# Server-side test
def test_utf8_safe_chunking():
    """Verify chunks never split multi-byte characters."""
    test_texts = [
        "東京グール",                    # All 3-byte CJK
        "Hello 東京 World",             # Mixed ASCII and CJK
        "🔥 ファイア 🔥",               # 4-byte emoji + 3-byte katakana
        "Berserk ベルセルク Vol.1",      # Mixed with punctuation
        "進撃の巨人 Attack on Titan",    # Full manga title
    ]

    for text in test_texts:
        writer = Utf8SafeChunkedWriter(mock_response, target_chunk_size=5)
        # Use tiny chunk size to force many splits
        writer.write(text)
        writer.flush()

        # Verify each chunk is valid UTF-8
        for chunk in mock_response.chunks:
            try:
                chunk.decode("utf-8")
            except UnicodeDecodeError:
                assert False, f"Chunk contains incomplete UTF-8: {chunk.hex()}"

        # Verify reassembled text matches original
        reassembled = b"".join(mock_response.chunks).decode("utf-8")
        assert reassembled == text

// Client-side test
function testResilientDecoder() {
    const decoder = new ResilientStreamDecoder();
    // Simulate "東" (E6 9D B1) split across two chunks
    const chunk1 = new Uint8Array([0xE6, 0x9D]);      // Incomplete "東"
    const chunk2 = new Uint8Array([0xB1, 0xE4, 0xBA, 0xAC]); // Complete "東" + "京"

    const text1 = decoder.decode(chunk1);
    console.assert(text1 === "", "Should buffer incomplete char");

    const text2 = decoder.decode(chunk2);
    console.assert(text2 === "東京", "Should decode complete chars including buffered");
}