diff --git a/apps/website/content/docs/streaming/api-reference.mdx b/apps/website/content/docs/streaming/api-reference.mdx index 221377fe..0f66497a 100644 --- a/apps/website/content/docs/streaming/api-reference.mdx +++ b/apps/website/content/docs/streaming/api-reference.mdx @@ -10,12 +10,12 @@ See the generated `stream-adapter.api.md` for complete declarations. Putting the ## Types -| Type | Purpose | -| ------------------------------------ | --------------------------------------------------------------------------------------- | -| `RowModelLike` | Structural atomic-transaction target with string or number IDs. | -| `StreamConnection` | `{ done: Promise; dispose(): void }`. | -| `TransactionBatcher` | RAF-batched `add`, `{ id, changes }` update, remove, flush, error, and dispose methods. | -| `PartialStreamOptions` | Fixed `rowId`, optional `onIssue`, and optional complete-row `createRow`. | +| Type | Purpose | +| ------------------------------------ | ------------------------------------------------------------------------------------------------------- | +| `RowModelLike` | Structural atomic-transaction target with string or number IDs. | +| `StreamConnection` | `{ done: Promise; dispose(): void }`. | +| `TransactionBatcher` | RAF-batched `add`, `{ id, changes }` update, remove, flush, dispose, plus `error` and `subscribeError`. | +| `PartialStreamOptions` | Fixed `rowId`, optional `onIssue`, and optional complete-row `createRow`. | ## `createBatcher(rowModel)` @@ -27,7 +27,9 @@ batcher.remove(["old-row"]); batcher.flush(); ``` -Scheduled work coalesces into one transaction per animation frame. `batcher.error` rejects with an asynchronous transaction failure. +Scheduled work coalesces into one transaction per animation frame. Calls are appended, not merged: two `update`s for the same row become two `{ id, changes }` entries inside that one transaction, applied in call order. + +`batcher.error` rejects with an asynchronous transaction failure. It is a promise, so it only ever reports the first failure; `batcher.subscribeError(listener)` delivers the same failure to a callback and returns an unsubscribe function. Subscribing after a failure invokes the listener immediately, so there is no race between wiring it up and the failure landing. ## Connectors and parsers diff --git a/apps/website/content/docs/streaming/index.mdx b/apps/website/content/docs/streaming/index.mdx index a2114827..d09231bf 100644 --- a/apps/website/content/docs/streaming/index.mdx +++ b/apps/website/content/docs/streaming/index.mdx @@ -6,7 +6,7 @@ nav: Streaming `@pretable/stream-adapter` connects an async source to an explicit row model. It coalesces producer events into at most one atomic row-model transaction per animation frame, no matter how fast the source emits. -There are two source shapes, and they behave nothing alike. An element stream appends a whole new row per event; a partial stream never adds a row at all — it keeps patching the same row ID. Watch both: +There are two source shapes, and they behave nothing alike. An element stream appends a whole new row per event; a partial stream patches one fixed row ID over and over. A partial stream adds a row only if you hand it a `createRow` factory, and never more than the one row it targets. Watch both: ## New rows arrive @@ -16,7 +16,7 @@ Each yielded value is a complete row. `connectElementStream` appends one as it a ## One row grows -Each yielded value is a `Partial` patch to a single, fixed row ID — nothing is ever appended. Row `msg-1` below is seeded before its stream connects; `msg-2` isn't seeded, so `createRow` builds it from the first partial that targets it: +Each yielded value is a `Partial` patch to a single, fixed row ID. Row `msg-1` below is seeded before its stream connects; `msg-2` isn't seeded, so its first partial is reported through `onIssue` and `createRow` builds the row from it: diff --git a/apps/website/content/docs/streaming/partial-streams.mdx b/apps/website/content/docs/streaming/partial-streams.mdx index f111ab11..929f2528 100644 --- a/apps/website/content/docs/streaming/partial-streams.mdx +++ b/apps/website/content/docs/streaming/partial-streams.mdx @@ -44,7 +44,7 @@ The connector never asserts that a partial is a complete row. Supply a factory w ```ts connectPartialStream(rowModel, partials, { - rowId: "msg-1", + rowId: "msg-2", createRow(partial, id) { return { id, @@ -56,7 +56,9 @@ connectPartialStream(rowModel, partials, { }); ``` -Without `createRow`, an unknown target is reported through `onIssue` and no row is fabricated — that's the warning `msg-2`'s connection above logs for its first partial, before `createRow` builds the row. +`onIssue` fires whether or not `createRow` is supplied — the connector reports the unknown target first, then builds the row if it has a factory. That ordering is what `msg-2` shows above: it logs the `unknown-update-id` warning for its first partial and _then_ gains its row. Without `createRow`, the report is all that happens and no row is fabricated. + +`createRow` receives the changes **accumulated** across that frame, not just the one partial that triggered it: partials arriving in the same animation frame are batched into a single transaction, so a row created on the first frame already carries every field that landed in it. ## Lifecycle diff --git a/apps/website/content/examples/partial-row-stream/PartialRowGrid.tsx b/apps/website/content/examples/partial-row-stream/PartialRowGrid.tsx index bdacb66c..ba1129eb 100644 --- a/apps/website/content/examples/partial-row-stream/PartialRowGrid.tsx +++ b/apps/website/content/examples/partial-row-stream/PartialRowGrid.tsx @@ -1,7 +1,7 @@ "use client"; import { connectPartialStream } from "@pretable/stream-adapter"; -import { PretableSurface } from "@pretable/react"; +import { PretableSurface, useDisposeOnUnmount } from "@pretable/react"; import { createLocalRowModel } from "@pretable/core"; import { useEffect, useMemo } from "react"; @@ -52,7 +52,11 @@ export function PartialRowGrid() { }; }, [rowModel]); - useEffect(() => () => rowModel.dispose(), [rowModel]); + // NOT `useEffect(() => () => rowModel.dispose())`: StrictMode rehearses an + // unmount in dev, `useMemo` hands the same model back to the remount, and the + // grid then renders nothing at all. `useDisposeOnUnmount` defers the disposal + // by a microtask so a remount can cancel it. + useDisposeOnUnmount(rowModel); return ( { + beforeEach(() => { + vi.useFakeTimers(); + }); + afterEach(() => { + vi.useRealTimers(); + }); + + it("still seeds, creates and grows its rows when effects are rehearsed", async () => { + vi.spyOn(console, "warn").mockImplementation(() => {}); + + render( + + + , + ); + + await act(async () => { + await vi.advanceTimersByTimeAsync(FULL_DURATION_MS); + }); + + // Header + msg-1 (seeded) + msg-2 (built by createRow). Asserting the rows, + // not just that it mounted: the failure mode renders a header and no data. + expect(screen.getAllByRole("row")).toHaveLength(3); + expect( + screen.getAllByRole("gridcell").map((cell) => cell.textContent ?? ""), + ).toContain(FIRST_REPLY); + }); +}); diff --git a/apps/website/content/examples/streaming-chat-grid/ChatGrid.tsx b/apps/website/content/examples/streaming-chat-grid/ChatGrid.tsx index 193d2479..2ee046a3 100644 --- a/apps/website/content/examples/streaming-chat-grid/ChatGrid.tsx +++ b/apps/website/content/examples/streaming-chat-grid/ChatGrid.tsx @@ -1,7 +1,7 @@ "use client"; import { connectElementStream } from "@pretable/stream-adapter"; -import { PretableSurface } from "@pretable/react"; +import { PretableSurface, useDisposeOnUnmount } from "@pretable/react"; import { createLocalRowModel } from "@pretable/core"; import { useEffect, useMemo } from "react"; @@ -47,7 +47,11 @@ export function ChatGrid({ }; }, [openResponseEvents, prompt, rowModel]); - useEffect(() => () => rowModel.dispose(), [rowModel]); + // NOT `useEffect(() => () => rowModel.dispose())`: StrictMode rehearses an + // unmount in dev, `useMemo` hands the same model back to the remount, and the + // grid then renders nothing at all. `useDisposeOnUnmount` defers the disposal + // by a microtask so a remount can cancel it. + useDisposeOnUnmount(rowModel); return ( { + beforeEach(() => { + vi.useFakeTimers(); + }); + afterEach(() => { + vi.useRealTimers(); + }); + + it("still streams rows when effects are rehearsed", async () => { + render( + + + , + ); + + await act(async () => { + await vi.advanceTimersByTimeAsync(RESPONSE_DURATION_MS * 3); + }); + + // Header row + 3 scripted assistant responses. Asserting the ROWS, not just + // that the component mounted: the failure mode is a grid that renders its + // header and no data, which a "did it mount" check passes. + expect(screen.getAllByRole("row")).toHaveLength(4); + expect( + screen.getAllByRole("gridcell").map((cell) => cell.textContent), + ).toContain("assistant"); + }); +}); diff --git a/packages/stream-adapter/src/parse-partial-stream.ts b/packages/stream-adapter/src/parse-partial-stream.ts index cea9d76a..4194c184 100644 --- a/packages/stream-adapter/src/parse-partial-stream.ts +++ b/packages/stream-adapter/src/parse-partial-stream.ts @@ -3,10 +3,15 @@ import type { StreamState } from "@cacheplane/json-stream"; /** * Parse a UTF-8 string stream into an `AsyncIterable>`. - * Emits incremental partial rows as a streaming JSON parse fills out - * each top-level array element — useful when an LLM is streaming - * partial JSON and you want field-by-field updates instead of waiting - * for each row to complete. + * + * The root must be a single JSON **object**, not an array — a non-object root + * throws. Each yielded value is the cumulative snapshot of that object as more + * keys resolve, not a delta, so the last value yielded is the complete row. + * Useful when an LLM is streaming partial JSON for one row and you want + * field-by-field updates instead of waiting for the object to close. + * + * For a stream of many complete rows, use {@link parseElementStream}, which + * does take a top-level array. * * Pair with {@link connectPartialStream} for end-to-end partial-stream * → grid wiring.