Speak session stream protocol v2

src/commands/session.tsx

@@ -26,17 +26,20 @@ import {
 import { sessionUrl } from '../lib/urls'
 import {
   resolveWsBase,
+  sessionStatusWord,
   streamSession,
   StreamUnavailableError,
   type StreamFrame,
   type StreamOutcome,
 } from '../lib/ws'
+import { isConnectVisibleRecord, recordToItems } from '../lib/events'
 import type {
   AgentSession,
   AgentSessionSource,
   AgentSessionStatus,
   GithubAccountSnippet,
   ReplayAgentSessionRequest,
+  SessionRecord,
   SessionSearchResult,
   SessionSearchScope,
   SessionTranscript,
@@ -807,19 +810,26 @@ export async function watchSessionStreaming(
   const token = requireToken()
   const wsBase = resolveWsBase(resolveApiBase())
 
-  // The server sends a `status` frame as its keepalive, so collapse unchanged
-  // statuses — both to keep the human log quiet and the NDJSON stream clean.
+  // Session frames are LWW snapshots resent on any change (cost ticks
+  // included), so collapse to status-word transitions — both to keep the
+  // human log quiet and the NDJSON stream clean of near-duplicates. Heartbeats
+  // are liveness only; deltas are ephemeral partials the committed record
+  // supersedes — a line-oriented log skips both.
   let lastStatus: string | undefined
   const onFrame = (frame: StreamFrame) => {
-    if (frame.type === 'status') {
-      if (frame.status === lastStatus) return
-      lastStatus = frame.status
+    if (frame.type === 'session' || frame.type === 'snapshot') {
+      const word = sessionStatusWord(
+        (frame as { session: AgentSession }).session,
+      )
+      if (word === lastStatus) return
+      lastStatus = word
     }
+    if (frame.type === 'heartbeat' || frame.type === 'delta') return
     if (json) {
       console.log(JSON.stringify(frame))
       return
     }
-    renderFrameHuman(frame)
+    renderFrameHuman(frame, lastStatus)
   }
 
   let outcome: StreamOutcome
@@ -851,30 +861,37 @@ export async function watchSessionStreaming(
   if (exitCodeForStatus(outcome.status) !== 0) process.exitCode = 1
 }
 
-function renderFrameHuman(frame: StreamFrame): void {
+function renderFrameHuman(frame: StreamFrame, statusWord?: string): void {
   switch (frame.type) {
-    case 'status':
-      console.log(`${nowClock()}  ${frame.status}`)
-      break
-    case 'stdout':
-      writeChunk(process.stdout, frame.data)
+    case 'snapshot':
+    case 'session':
+      console.log(`${nowClock()}  ${statusWord ?? ''}`)
       break
-    case 'stderr':
-      writeChunk(process.stderr, frame.data)
+    case 'records_append': {
+      // Raw records, rendered client-side (the semantic-relay philosophy):
+      // one line per transcript item, same shaping as `session connect`.
+      const records = (frame as { records: SessionRecord[] }).records
+      for (const record of records) {
+        if (!isConnectVisibleRecord(record)) continue
+        for (const item of recordToItems(record, `w${record.feed_seq}`)) {
+          const line = item.detail ? `${item.text}  ${item.detail}` : item.text
+          if (line.trim()) console.log(line)
+        }
+      }
       break
+    }
+    case 'messages':
+      break // queued-chip bookkeeping; nothing to log line-orientedly
     case 'error':
-      console.error(`error: ${frame.message ?? frame.data ?? 'stream error'}`)
+      console.error(`error: ${(frame as { message?: string }).message ?? 'stream error'}`)
       break
     case 'done':
       break // handled by the caller
+    default:
+      break // unknown frame types are ignored (protocol §3.6)
   }
 }
 
-function writeChunk(stream: NodeJS.WriteStream, data?: string): void {
-  if (!data) return
-  stream.write(data.endsWith('\n') ? data : data + '\n')
-}
-
 // Exit 0 for a successful terminal status, non-zero otherwise (spec §4.1).
 export function exitCodeForStatus(status: string): number {
   return status === 'completed' ? 0 : 1

src/lib/api.ts

@@ -46,6 +46,7 @@ import type {
   SandboxVariableSummary,
   SearchSessionsQuery,
   SearchSessionsResponse,
+  SessionMessage,
   SessionRecord,
   SyncAgentSessionRequest,
   SyncAgentSessionResponse,
@@ -251,9 +252,17 @@ export class ApiClient {
   // inbox delivers it to the agent's Claude Code stdin at the next turn
   // boundary, or wakes the session when idle. 409 for single-shot / closed
   // sessions (no inbox loop to attend it).
-  sendSessionMessage(sessionId: string, message: string): Promise<AgentSession> {
+  // Returns the CREATED SessionMessage (protocol v2 §4.2) so callers key
+  // their optimistic chip on its id. `idempotencyKey` makes retries safe:
+  // the server dedupes per (session, key) and returns the original message.
+  sendSessionMessage(
+    sessionId: string,
+    message: string,
+    idempotencyKey?: string,
+  ): Promise<SessionMessage> {
     return this.request('POST', `/v1/sessions/${encodeURIComponent(sessionId)}/messages`, {
       message,
+      idempotency_key: idempotencyKey ?? null,
     } satisfies SendSessionMessageRequest)
   }
 

src/lib/types.ts

@@ -105,6 +105,8 @@ export interface AgentSession {
   updated_at: string
   status: AgentSessionStatus
   status_reason: string | null
+  // Why the session ended, finer than status; null until terminal.
+  exit_status?: string | null
   source?: AgentSessionSource
   agent_config_id: string | null
   // Durable-conversation identity (stateful sessions): a keyed session runs
@@ -158,6 +160,9 @@ export type AgentConfig = Record<string, unknown>
 // message appended to a durable session's inbox.
 export interface SendSessionMessageRequest {
   message: string
+  // Retry-safety key, unique per (session, key): a retried POST returns the
+  // original message instead of double-queueing a turn (protocol v2 §4.2).
+  idempotency_key?: string | null
 }
 
 // Laptop -> cloud handoff params: start a fresh session on the built-in
@@ -353,19 +358,30 @@ export type LifecycleRecordType =
 export interface SessionRecord {
   id: string
   agent_session_id: string
-  session_execution_id: string
   created_at: string
   feed_seq: number
   stream_seq: number
-  source: RecordSource
+  // Free TEXT server-side; unknown sources must be ignored, not crashed on.
+  source: RecordSource | string
   record_type: string
   record_format: string
+  // The turn this record belongs to (stream protocol v2 §3.3).
+  agent_turn_id?: string | null
+  // Correlates a user-echo record back to the inbox message it echoes, so
+  // queued chips retire by id (§4.2).
+  session_message_id?: string | null
   payload: Record<string, unknown>
   [key: string]: unknown
 }
 
 export interface ListSessionRecordsResponse {
   records: SessionRecord[]
+  // The OPEN inbox slice (pending rows only) — the server-side queued signal.
+  messages?: SessionMessage[]
+  // Whether records past this page remain (only meaningful with ?limit=).
+  has_more?: boolean
+  // The retention head: lowest stored feed_seq, or null when no records.
+  earliest_feed_seq?: number | null
 }
 
 // One exchange within a keyed session (a single Claude Code turn), from
@@ -389,6 +405,9 @@ export interface SessionMessage {
   body: string
   status: 'pending' | 'delivered'
   feed_seq: number | null
+  // The sender's display name for single-human-utterance messages; null for
+  // event-rendered messages.
+  author?: string | null
   created_at: string
   delivered_at: string | null
   delivered_turn_id: string | null