feat: normalize stream_event contract + add live integration coverage

README.md

@@ -77,3 +77,38 @@ const session = createSession("agent-123", {
 ---
 
 Made with 💜 in San Francisco
+
+## Live integration tests (opt-in)
+
+The SDK includes live integration tests that hit real Letta Cloud endpoints and verify runtime contracts for:
+
+- session init shape
+- send/stream lifecycle (`assistant`, `reasoning`, `stream_event`, `result`)
+- `listMessages()` backfill/pagination shape
+- concurrent `listMessages()` during active stream
+- tool lifecycle (`tool_call` -> `tool_result`)
+
+These tests are opt-in and skipped by default.
+
+```bash
+# Required
+export LETTA_API_KEY=sk-let-...
+
+# Optional
+export LETTA_AGENT_ID=agent-...            # force a specific agent
+export LETTA_CONVERSATION_ID=conv-...      # force a specific conversation for init test
+export LETTA_BASE_URL=https://api.letta.com
+export LETTA_LIVE_TEST_TIMEOUT_MS=180000
+
+# Run live tests
+bun run test:live
+
+# Run and record sanitized fixtures to src/tests/fixtures/live/
+bun run test:live:record
+```
+
+Safety notes:
+
+- live tests create/use real conversations on the target account
+- fixture recording redacts obvious secrets/tokens and local home paths
+- keep fixture recording disabled in CI unless you explicitly want refreshed snapshots

package.json

@@ -19,7 +19,9 @@
     "build": "bun run build.ts",
     "dev": "bun run build.ts --watch",
     "check": "tsc --noEmit",
-    "test": "bun test"
+    "test": "bun test",
+    "test:live": "LETTA_LIVE_INTEGRATION=1 bun test src/tests/live.integration.test.ts",
+    "test:live:record": "LETTA_LIVE_INTEGRATION=1 LETTA_RECORD_FIXTURES=1 bun test src/tests/live.integration.test.ts"
   },
   "repository": {
     "type": "git",

src/index.ts

@@ -44,6 +44,10 @@ export type {
   SDKReasoningMessage,
   SDKResultMessage,
   SDKStreamEventMessage,
+  SDKStreamEventPayload,
+  SDKStreamEventDeltaPayload,
+  SDKStreamEventMessagePayload,
+  SDKUnknownStreamEventPayload,
   SDKErrorMessage,
   SDKRetryMessage,
   SkillSource,
@@ -74,6 +78,8 @@ export type {
 
 export { Session } from "./session.js";
 
+export { extractStreamTextDelta } from "./stream-events.js";
+
 // Tool helpers
 export {
   jsonResult,

src/session.test.ts

@@ -127,6 +127,36 @@ function createResultMessage(): WireMessage {
   } as WireMessage;
 }
 
+function createErrorWireMessage(): WireMessage {
+  return {
+    type: "error",
+    session_id: "session-1",
+    uuid: "error-1",
+    message: "Rate limit exceeded",
+    stop_reason: "llm_api_error",
+    run_id: "run-1",
+    api_error: {
+      error_type: "llm_api_error",
+      message: "429 from upstream provider",
+      message_type: "error_message",
+      run_id: "run-1",
+    },
+  } as WireMessage;
+}
+
+function createRetryWireMessage(): WireMessage {
+  return {
+    type: "retry",
+    session_id: "session-1",
+    uuid: "retry-1",
+    reason: "llm_api_error",
+    attempt: 2,
+    max_attempts: 4,
+    delay_ms: 1500,
+    run_id: "run-1",
+  } as WireMessage;
+}
+
 function createCanUseToolRequest(
   requestId: string,
   toolName: string,
@@ -402,6 +432,46 @@ describe("Session", () => {
     });
   });
 
+  describe("transformMessage error/retry mapping", () => {
+    test("maps error wire message to SDK error message", () => {
+      const session = new Session();
+      const wireMsg = createErrorWireMessage();
+
+      // @ts-expect-error - accessing private method for regression coverage
+      const transformed = session.transformMessage(wireMsg) as SDKMessage | null;
+
+      expect(transformed).toEqual({
+        type: "error",
+        message: "Rate limit exceeded",
+        stopReason: "llm_api_error",
+        runId: "run-1",
+        apiError: {
+          error_type: "llm_api_error",
+          message: "429 from upstream provider",
+          message_type: "error_message",
+          run_id: "run-1",
+        },
+      });
+    });
+
+    test("maps retry wire message to SDK retry message", () => {
+      const session = new Session();
+      const wireMsg = createRetryWireMessage();
+
+      // @ts-expect-error - accessing private method for regression coverage
+      const transformed = session.transformMessage(wireMsg) as SDKMessage | null;
+
+      expect(transformed).toEqual({
+        type: "retry",
+        reason: "llm_api_error",
+        attempt: 2,
+        maxAttempts: 4,
+        delayMs: 1500,
+        runId: "run-1",
+      });
+    });
+  });
+
   describe("background pump parity", () => {
     test("handles can_use_tool control requests before stream iteration starts", async () => {
       let callbackInvocations = 0;
@@ -507,5 +577,33 @@ describe("Session", () => {
         session.close();
       }
     });
+
+    test("emits error and retry messages instead of dropping them", async () => {
+      const session = new Session({
+        permissionMode: "default",
+      });
+      const transport = new MockTransport();
+      attachMockTransport(session, transport);
+
+      try {
+        transport.push(createInitMessage());
+        await session.initialize();
+
+        transport.push(createErrorWireMessage());
+        transport.push(createRetryWireMessage());
+        transport.push(createResultMessage());
+
+        const streamed: SDKMessage[] = [];
+        for await (const msg of session.stream()) {
+          streamed.push(msg);
+        }
+
+        expect(streamed.some((msg) => msg.type === "error")).toBe(true);
+        expect(streamed.some((msg) => msg.type === "retry")).toBe(true);
+        expect(streamed[streamed.length - 1]?.type).toBe("result");
+      } finally {
+        session.close();
+      }
+    });
   });
 });

src/session.ts

@@ -24,6 +24,7 @@ import type {
   ExecuteExternalToolRequest,
   ListMessagesOptions,
   ListMessagesResult,
+  SDKStreamEventPayload,
 } from "./types.js";
 import {
   isHeadlessAutoAllowTool,
@@ -780,17 +781,13 @@ export class Session implements AsyncDisposable {
     // Stream event (partial message updates)
     if (wireMsg.type === "stream_event") {
       const msg = wireMsg as WireMessage & {
-        event: {
-          type: string;
-          index?: number;
-          delta?: { type?: string; text?: string; reasoning?: string };
-          content_block?: { type?: string; text?: string };
-        };
+        event: unknown;
         uuid: string;
       };
+      const eventPayload = (msg.event ?? {}) as SDKStreamEventPayload;
       return {
         type: "stream_event",
-        event: msg.event,
+        event: eventPayload,
         uuid: msg.uuid,
       };
     }

src/stream-events.ts

@@ -0,0 +1,88 @@
+import type { SDKStreamEventPayload } from "./types.js";
+
+export type StreamTextKind = "assistant" | "reasoning";
+
+export interface StreamTextDelta {
+  kind: StreamTextKind;
+  text: string;
+}
+
+function extractTextFromContent(content: unknown): string | null {
+  if (typeof content === "string") {
+    return content;
+  }
+
+  if (Array.isArray(content)) {
+    const pieces: string[] = [];
+    for (const part of content) {
+      if (typeof part === "string") {
+        pieces.push(part);
+        continue;
+      }
+      if (!part || typeof part !== "object") {
+        continue;
+      }
+      const rec = part as Record<string, unknown>;
+      if (typeof rec.text === "string") {
+        pieces.push(rec.text);
+      }
+    }
+    const joined = pieces.join("");
+    return joined.length > 0 ? joined : null;
+  }
+
+  if (content && typeof content === "object") {
+    const rec = content as Record<string, unknown>;
+    if (typeof rec.text === "string") {
+      return rec.text;
+    }
+  }
+
+  return null;
+}
+
+/**
+ * Extract appendable assistant/reasoning text from a stream_event payload.
+ *
+ * Supports both shapes currently emitted by headless mode:
+ * 1) content_block style: { type, delta: { text|reasoning } }
+ * 2) message chunk style: { message_type: "assistant_message"|"reasoning_message", ... }
+ */
+export function extractStreamTextDelta(event: SDKStreamEventPayload): StreamTextDelta | null {
+  if (!event || typeof event !== "object") {
+    return null;
+  }
+
+  const rec = event as Record<string, unknown>;
+
+  const maybeDelta = rec.delta;
+  if (maybeDelta && typeof maybeDelta === "object") {
+    const delta = maybeDelta as Record<string, unknown>;
+
+    if (typeof delta.reasoning === "string" && delta.reasoning.length > 0) {
+      return { kind: "reasoning", text: delta.reasoning };
+    }
+
+    if (typeof delta.text === "string" && delta.text.length > 0) {
+      return { kind: "assistant", text: delta.text };
+    }
+  }
+
+  const messageType = rec.message_type;
+  if (messageType === "reasoning_message") {
+    const reasoningText =
+      typeof rec.reasoning === "string" ? rec.reasoning : extractTextFromContent(rec.content);
+    if (reasoningText && reasoningText.length > 0) {
+      return { kind: "reasoning", text: reasoningText };
+    }
+  }
+
+  if (messageType === "assistant_message") {
+    const assistantText = extractTextFromContent(rec.content);
+    if (assistantText && assistantText.length > 0) {
+      return { kind: "assistant", text: assistantText };
+    }
+  }
+
+  return null;
+}

src/tests/fixtures/live/init_contract.json

@@ -0,0 +1,39 @@
+{
+  "selectedAgentName": "letta-code-agent",
+  "init": {
+    "type": "init",
+    "agentId": "agent-d0ab297f-c180-4347-b3ab-20165fbe0f2f",
+    "sessionId": "agent-d0ab297f-c180-4347-b3ab-20165fbe0f2f",
+    "conversationId": "conv-d87b461c-e855-49d6-9688-f69e3864f44e",
+    "model": "claude-sonnet-4-5-20250929",
+    "tools": [
+      "AskUserQuestion",
+      "Bash",
+      "TaskOutput",
+      "Edit",
+      "EnterPlanMode",
+      "ExitPlanMode",
+      "Glob",
+      "Grep",
+      "TaskStop",
+      "Read",
+      "Skill",
+      "Task",
+      "TodoWrite",
+      "Write"
+    ],
+    "memfsEnabled": false,
+    "skillSources": [
+      "bundled",
+      "global",
+      "agent",
+      "project"
+    ],
+    "systemInfoReminderEnabled": true,
+    "sleeptime": {
+      "trigger": "step-count",
+      "behavior": "reminder",
+      "stepCount": 25
+    }
+  }
+}