{"record":{"id":"26f0f6b9f6da33c1","repo":"can1357/oh-my-pi","slug":"auth-broker-stream-did-not-start-with-snapshot","errorCode":null,"errorMessage":"Auth broker stream did not start with snapshot","messagePattern":"Auth broker stream did not start with snapshot","errorType":"exception","errorClass":"AuthBrokerError","httpStatus":null,"severity":"error","filePath":"packages/ai/src/auth-broker/client.ts","lineNumber":256,"sourceCode":"\t\t\ttry {\n\t\t\t\tparsed = JSON.parse(sse.data);\n\t\t\t} catch (err) {\n\t\t\t\tthrow new AuthBrokerError(\"Auth broker stream returned malformed JSON\", {\n\t\t\t\t\tbody: sse.data,\n\t\t\t\t\tcause: err,\n\t\t\t\t});\n\t\t\t}\n\t\t\tconst validated = snapshotStreamEventSchema(parsed);\n\t\t\tif (validated instanceof type.errors) {\n\t\t\t\tthrow new AuthBrokerError(\"Auth broker stream event failed schema validation\", {\n\t\t\t\t\tbody: validated.summary,\n\t\t\t\t});\n\t\t\t}\n\t\t\tconst event = validated as SnapshotStreamEvent;\n\t\t\tif (!sawFirstEvent) {\n\t\t\t\tsawFirstEvent = true;\n\t\t\t\tif (event.kind !== \"snapshot\") {\n\t\t\t\t\tthrow new AuthBrokerError(\"Auth broker stream did not start with snapshot\", { body: sse.data });\n\t\t\t\t}\n\t\t\t}\n\t\t\tyield event;\n\t\t}\n\t\tif (!opts.signal?.aborted) {\n\t\t\tthrow new AuthBrokerError(\n\t\t\t\tsawFirstEvent\n\t\t\t\t\t? \"Auth broker stream ended unexpectedly\"\n\t\t\t\t\t: \"Auth broker stream ended before initial snapshot\",\n\t\t\t\t{ status: response.status },\n\t\t\t);\n\t\t}\n\t}\n\n\t/**\n\t * Fetch aggregate broker usage with a timeout sized for serialized\n\t * same-provider account probes.\n\t */","sourceCodeStart":238,"sourceCodeEnd":274,"githubUrl":"https://github.com/can1357/oh-my-pi/blob/969062200754ea02cfac922e5ebb8c608c079e15/packages/ai/src/auth-broker/client.ts#L238-L274","documentation":"openSnapshotStream consumes the broker's SSE snapshot stream and enforces the protocol contract that the first event frame must be a full `snapshot` event (client.ts:253-257). If the first parsed, schema-valid SnapshotStreamEvent has kind !== \"snapshot\", the client cannot establish a baseline credential state, so it throws AuthBrokerError immediately rather than yielding events on top of a missing baseline. This protects consumers like RemoteAuthCredentialStore from applying incremental `entry`/`removed` deltas against a state that was never initialized.","triggerScenarios":"Calling AuthBrokerClient.openSnapshotStream() against a broker that emits its first SSE data frame with kind \"entry\", \"refresh\", or \"removed\" instead of kind \"snapshot\" — i.e. a server violating the documented stream protocol where frame #1 must be a full snapshot.","commonSituations":"Running a broker server built from a different or older/newer version of the protocol than the client expects; a custom or proxy SSE endpoint that forwards mid-stream deltas from an already-active session (joining the stream after the initial snapshot was emitted); a misconfigured reverse proxy or load balancer that drops the first event frame; testing against a mock broker that only emits upserts.","solutions":["Verify the broker is running a version whose /v1/snapshot/stream endpoint emits a full snapshot as the first event; upgrade or redeploy the auth-broker server to match the client.","Check that no proxy/middleware between client and broker strips or delays the first SSE frame (some proxies buffer or drop the initial event).","Catch AuthBrokerError from openSnapshotStream and fall back to the long-polling path (AuthBrokerClient.fetchSnapshot), which fetches the snapshot over plain JSON.","If you run a mock/custom broker, make its first emitted frame `{ kind: \"snapshot\", ... }` matching snapshotStreamEventSchema."],"exampleFix":"// before\nfor await (const event of client.openSnapshotStream({ signal })) {\n  applyEvent(event);\n}\n\n// after\ntry {\n  for await (const event of client.openSnapshotStream({ signal })) {\n    applyEvent(event);\n  }\n} catch (err) {\n  if (err instanceof AuthBrokerError && err.message.includes(\"did not start with snapshot\")) {\n    logger.warn(\"snapshot stream protocol mismatch, falling back to polling\", { body: err.body });\n    await pollSnapshotLoop(signal);\n  } else {\n    throw err;\n  }\n}","handlingStrategy":"try-catch","validationCode":null,"typeGuard":"function isSnapshotFirstProtocolError(err: unknown): err is AuthBrokerError {\n  return err instanceof AuthBrokerError && err.message === \"Auth broker stream did not start with snapshot\";\n}","tryCatchPattern":"try {\n  for await (const event of client.openSnapshotStream({ signal })) {\n    applyEvent(event);\n  }\n} catch (err) {\n  if (isSnapshotFirstProtocolError(err)) {\n    logger.warn(\"broker stream protocol violation, falling back to polling\", { body: err.body });\n    await startPollingFallback(signal);\n  } else {\n    throw err;\n  }\n}","preventionTips":["Pin broker and client to matching versions; protocol changes ship in lockstep.","Test SSE integrations against a mock that emits `{ kind: \"snapshot\" }` as frame 1.","Add an integration test that asserts the first streamed event kind is \"snapshot\".","Avoid proxies that transform or buffer SSE traffic on the broker path."],"tags":["sse","protocol-violation","auth-broker","streaming"],"backgroundTag":"sse-stream-protocol-mismatch","analyzedSha":"969062200754ea02cfac922e5ebb8c608c079e15","analyzedAt":"2026-08-31T10:29:35.737Z","schemaVersion":2},"datasetVersion":"2026-08-31T14:17:45.589Z"}