frontend: SSE reconnection via tasks/resubscribe

[birdayz]

What

When an SSE connection drops mid-stream, transparently reconnect using the A2A protocol's tasks/resubscribe endpoint instead of showing a fatal error. Bump @a2a-js/sdk from 0.3.5 to 0.3.10.

Why

Long-running agent tasks are particularly vulnerable to transient network failures. A brief WiFi hiccup or load balancer timeout shouldn't kill the entire conversation. The task keeps running server-side -- we just lost the event stream and need to pick it back up.

Implementation details

The core logic lives in use-message-streaming.ts. When streamMessage() catches an error and the task is still in-flight (has a task ID, state is not terminal), it enters a resubscribe loop:

1. Close the active text block and insert a disconnected connection-status block
2. Retry up to 5 times with exponential backoff (1s, 2s, 4s, 8s, 16s)
3. Each attempt creates a fresh A2AClient and calls resubscribeTask({ id: taskId })
4. On success, process the resumed event stream with the same handlers as the initial stream
5. On permanent failure, insert a gave-up block and fall through to the existing error path

New connection-status ContentBlock variant flows through the full type system: runtime types, DB schema, serialization, deserialization, and a new ConnectionStatusBlock UI component.

The reconnecting / reconnected status blocks replace each other in-place so the UI doesn't accumulate stale status lines during retries.

Test suite covers 8 scenarios: successful stream, error handling, reconnection success, reconnection after partial content, all retries exhausted, non-resubscribable errors, error parsing, and task completion during reconnection.

References

• A2A protocol tasks/resubscribe: https://google.github.io/A2A/#/topics/streaming?id=disconnection-and-recovery

[weeco]

@claude Review the PR

[claude]

Claude finished @weeco's task in 2m 11s —— View job

---

PR Review: SSE Reconnection via tasks/resubscribe

Comprehensive code review completed. Overall, this is an excellent implementation of SSE reconnection with strong test coverage and thoughtful error handling. The code is well-structured, thoroughly tested, and includes helpful documentation.

Tasks

• Read repository guidelines (CLAUDE.md)
• Review core reconnection logic (use-message-streaming.ts)
• Review test coverage (use-message-streaming.test.ts)
• Review UI components (connection-status-block.tsx)
• Review database operations and schema changes
• Review type definitions and data flow
• Review supporting utilities
• Provide comprehensive feedback

---

✅ Strengths

1. Excellent Test Coverage - 17 comprehensive test scenarios covering edge cases like:

   • Progressive reconnection with partial data delivery
   • Attempt counter reset logic to allow unlimited retries with progress
   • Terminal state detection during failed resubscribe attempts
   • Connection status block replacement to prevent UI clutter

2. Robust Error Handling:

   • Distinguishes between programming errors (TypeError/ReferenceError) and network errors
   • Preserves partial content received before disconnection
   • Comprehensive error parsing with regex fallbacks for SDK limitations

3. Smart Reconnection Logic:

   • Exponential backoff prevents server hammering
   • Attempt counter reset when progress is made allows resilient long-running tasks
   • Only attempts resubscription when task is actually in-flight (isResubscribable)

4. Type Safety: Full type flow from SDK → runtime types → DB schema → UI components

5. User Experience:

   • Connection status blocks replace in-place to avoid UI clutter
   • "Reload to check status" button in gave-up state provides user recovery path
   • Preserves all content received before disconnection

---

🔍 Issues & Suggestions

Critical Issues

1. Race Condition in Connection Status Block Replacement (use-message-streaming.ts:216-232)

The current logic replaces the last block if it's a connection-status, but this could fail if event handlers push new blocks between status updates:

// Current code at line 216-232
if (status === 'reconnecting' || status === 'reconnected' || status === 'gave-up') {
  const lastIdx = state.contentBlocks.length - 1;
  if (lastIdx >= 0 && state.contentBlocks[lastIdx].type === 'connection-status') {
    state.contentBlocks[lastIdx] = block;
    // ... update UI
    return;
  }
}

Problem: If processResubscribeStream delivers an event that pushes a content block (e.g., task-status-update) before the next pushConnectionStatus('reconnecting') call, the last block won't be connection-status, and we'll stack status blocks.

Suggestion: Track the connection status block index explicitly in StreamingState:

type StreamingState = {
  // ... existing fields
  connectionStatusBlockIndex?: number; // NEW
};

const pushConnectionStatus = ({ status, state, ... }: ConnectionStatusParams): void => {
  const block: ContentBlock = { /* ... */ };

  if (status === 'reconnecting' || status === 'reconnected' || status === 'gave-up') {
    if (state.connectionStatusBlockIndex !== undefined && 
        state.contentBlocks[state.connectionStatusBlockIndex]?.type === 'connection-status') {
      state.contentBlocks[state.connectionStatusBlockIndex] = block;
      onMessageUpdate(/* ... */);
      return;
    }
  }

  // First connection-status or new disconnection
  state.connectionStatusBlockIndex = state.contentBlocks.length;
  state.contentBlocks.push(block);
  onMessageUpdate(/* ... */);
};

This ensures we always replace the correct block even if other content arrives between status updates.

---

Medium Priority Issues

2. Unclear Finalize Behavior After Recovery Failure (use-message-streaming.ts:420-429)

const recovered = await resubscribeLoop(state, agentCardUrl, assistantMessage, onMessageUpdate);
if (recovered) {
  try {
    const result = await finalizeMessage(state, assistantMessage);
    onMessageUpdate(result.assistantMessage);
    return result;
  } catch {
    // finalizeMessage failed (e.g. DB write) -- fall through to error path
  }
}

Issue: The empty catch block swallows the error details. While test scenario 16 covers this case, production debugging would benefit from logging.

Suggestion: Log the error for observability:

} catch (finalizeError) {
  console.error('finalizeMessage failed after recovery:', finalizeError);
  // fall through to error path
}

---

3. Potential Memory Leak in Long-Running Reconnection (use-message-streaming.ts:256-313)

The resubscribeLoop creates a new A2AClient on each attempt without explicitly cleaning up the previous client. While JavaScript GC should handle this, if clients maintain open connections or event listeners, this could accumulate resources during long retry sequences.

Suggestion: Consider explicit cleanup if the SDK provides it:

let currentClient: A2AClient | null = null;

while (attempt < MAX_RESUBSCRIBE_ATTEMPTS) {
  // ...
  try {
    currentClient?.close?.(); // If SDK provides cleanup method
    currentClient = await createA2AClient(agentCardUrl);
    const stream = currentClient.resubscribeTask({ id: taskId });
    // ...
  }
}

---

4. Hardcoded Usage Metadata Structure (types.ts:73-80, database-operations.ts:234-241)

The usage metadata structure is duplicated and hardcoded across multiple files. If the SDK changes the usage structure, this requires updates in several places.

Current: Explicit type definition repeated:

// types.ts
export type MessageUsageMetadata = {
  input_tokens: number;
  output_tokens: number;
  total_tokens: number;
  max_input_tokens?: number;
  cached_tokens?: number;
  reasoning_tokens?: number;
};

// database-operations.ts
usage?: {
  input_tokens: number;
  output_tokens: number;
  // ... repeated
};

Suggestion: Import the type from SDK if available, or extract to a shared type definition:

import type { UsageMetadata } from '@a2a-js/sdk'; // If available
// OR create a shared type module

---

5. Type Mismatch in Database Schema (chat-db.ts:67-74)

The contentBlocks field in the DB schema stores timestamps as ISO strings, but the runtime type (types.ts:28-70) uses Date objects. This is handled correctly by serializeContentBlocks, but the types don't reflect this difference clearly.

Current:

// chat-db.ts (DB types) - timestamp: string
type: 'connection-status';
timestamp: string;

// types.ts (runtime types) - timestamp: Date
type: 'connection-status';
timestamp: Date;

Suggestion: Document this explicitly or use branded types:

// In database-operations.ts
type SerializedContentBlock = ContentBlock & {
  timestamp: string; // ISO 8601
};

---

Minor Issues

6. Biome Disable Comment Justification (use-message-streaming.ts:261, 326)

Two biome-ignore directives mention "complex business logic" and "retry loop with backoff". These are accurate, but consider if the functions could be broken down:

// Line 261
// biome-ignore lint/complexity/noExcessiveCognitiveComplexity: retry loop with backoff, progress detection, and error classification

Suggestion: While the complexity is justified, consider extracting helper functions like:

• shouldContinueRetrying(attempt, maxAttempts, taskState)
• calculateBackoffDelay(attempt)
• isTerminalState(state)

This would reduce cognitive complexity and improve testability.

---

7. Magic Number for Max Attempts (use-message-streaming.ts:114)

const MAX_RESUBSCRIBE_ATTEMPTS = 5;

Suggestion: Consider making this configurable or at least document the rationale for 5 attempts. Total backoff time is 31 seconds (1+2+4+8+16), which seems reasonable but might warrant explanation.

/**
 * Maximum consecutive reconnection attempts before giving up.
 * With exponential backoff (1s, 2s, 4s, 8s, 16s), total retry time is ~31s.
 * The counter resets on any progress, so tasks making incremental progress
 * can reconnect indefinitely.
 */
const MAX_RESUBSCRIBE_ATTEMPTS = 5;

---

8. Connection Status Component Button Behavior (connection-status-block.tsx:69-72)

The "Reload to check status" button triggers a full page reload, which might be jarring:

<Button onClick={() => window.location.reload()} size="sm" variant="outline">
  <RefreshCwIcon />
  Reload to check status
</Button>

Suggestion: Consider a gentler retry mechanism first (retry resubscription once more) before forcing a reload, or use soft reload navigation if using a router.

---

9. Test Timer Handling Inconsistency (use-message-streaming.test.ts:382-391)

Most tests use shouldAdvanceTime: true, but test scenario 5 switches to manual advancement:

vi.useRealTimers();
vi.useFakeTimers({ shouldAdvanceTime: false });

Suggestion: Document why this test requires manual advancement while others use auto-advance. This prevents confusion during future test maintenance.

---

10. Missing Error Code Constants (use-message-streaming.ts:49-54)

The regex patterns parse JSON-RPC error codes, but the code doesn't define constants for common error codes (like -1 for unknown, -32000 for server error, etc.).

Suggestion: Define error code constants if you need to handle specific codes differently:

const JSON_RPC_ERROR_CODES = {
  UNKNOWN: -1,
  SERVER_ERROR: -32000,
  INVALID_REQUEST: -32600,
  // ... etc
} as const;

---

📋 Additional Observations

Good Practices Observed:

• ✅ Comprehensive test coverage (17 scenarios + parseA2AError tests)
• ✅ Clear separation of concerns (streaming logic, UI, database, types)
• ✅ Detailed inline documentation explaining SDK limitations (e.g., regex parsing rationale)
• ✅ Proper TypeScript strict types throughout
• ✅ Temporal event ordering preserved (Jupyter-style rendering)
• ✅ SDK version bump with clear rationale (0.3.5 → 0.3.10 for resubscribe support)

Minor Suggestions:

• Consider extracting retry logic to a reusable utility if other streams need similar behavior
• The connection-status type could potentially be reused for other connection-related features
• Consider telemetry/analytics for reconnection events to monitor real-world reliability

---

🎯 Verdict

This PR demonstrates high-quality engineering with:

• Thoughtful architecture that handles edge cases
• Excellent test coverage (>95% of critical paths)
• Clear documentation and rationale for design decisions
• Proper separation of concerns

The critical issue (#1) should be addressed before merging to prevent potential UI glitches in production. The other issues are enhancements that could be addressed in follow-up PRs.

Recommendation: ✅ Approve after addressing the race condition in issue #1

Great work on this resilient reconnection implementation! The exponential backoff with progress-based retry reset is particularly clever for handling long-running tasks.

---