jsonlines

Files (1)

• SKILL.md

SKILL.md (first lines)

# Vovk.ts JSON Lines streaming

JSON Lines is Vovk's streaming model: server yields one JSON object per line, client consumes as async iterable. Works for chat completions, progress updates, long-running jobs, anything where data arrives over time.

Two authoring styles:

- **Generator handlers** — `function*` / `async function*` with `yield`. Common case.
- **`JSONLinesResponder`** — low-level manual `send` / `close` / `throw`. For orchestration, fan-in, or when stream shape doesn't fit a generator.

## Generator handler

Simplest streaming handler: `async function*` yielding objects. Each yield → one JSON line on wire.

```ts
import { post, procedure, operation } from 'vovk';
import { z } from 'zod';

export default class StreamController {
  @operation({ summary: 'Stream messages' })
  @post('completions')
  static getJSONLines = procedure({
    iteration: z.object({ message: z.string() }),
  }).handle(async function* () {
    yield { message: 'Hello,' };
    yield { message: ' World' };
  });
}
```

**`iteration`**: validation schema for each yielded object. Same role `output` plays for non-streaming procedures.

### Delegating to a service with `yield*`

Keep controller thin, let service do work:

```ts
static getJSONLines = procedure({
  iteration: z.object({ message: z.string() }),
}).handle(function* () {
  yield* StreamService.getJSONLines();
});
```

### Validation cadence

Default: **only first yielded item validated** — cheap, catches shape bugs early. Opt into every-iteration validation when shape varies:

```ts
procedure({
  iteration: z.union([
    z.strictObject({ users: z.array(UserSchema) }),
    z.strictObject({ tasks: z.array(TaskSchema) }),
  ]),
  validateEachIteration: true,
})
```

Use sparingly — every-item validation adds per-yield cost.

## `JSONLinesResponder` — manual control

For cases generator can't express: fan-in from multiple concurrent sources, conditional close, error signaling outside generator protocol. Still a `procedure` — handler is plain `async` function building responder, kicking off producer, returning responder. With `iteration` defined, Vovk wires `responder.onBeforeSend` automatically → every `send()` runs through same schema (and same `validateEachIteration` rules) as generator handler.

```ts
import { JSONLinesResponder, procedure, get } from 'vovk';
import { z } from 'zod';

class StreamController {
  @get('tokens')
  static streamTokens = procedure({
    iteration: z.object({ message: z.string() }),
  }).handle(async (req) => {
    const responder = new JSONLinesResponder<{ message: string }>(req);
    void StreamService.streamTokens(responder);
    return responder;
  });
}
```

Handler returns synchronously; service runs as floating promise (`void ...`), feeds responder over time. Don't `await` service — would block response.

**Constructor**:

```ts
new JSONLinesResponder<T>(
  request?: Request | null,              // checks Accept: application/jsonl
  getResponse?: (responder) => Response, // custom Response builder
);
```

**Methods**:

| Method | Purpose |
|---|---|
| `send(item: T): Promise<void>` | Serialize + send one JSON line. Validates per `validateEachIteration`. |
| `close(): void` | End stream. |
| `throw(err: Error): void` | Send error frame + close. |

**Properties**: `response`, `headers`, `readableStream`.

### Service-owned streaming

Service accepts responder + orchestrates:

```ts
export default class StreamService {
  static async streamTokens(responder: JSONLinesResponder<Token>) {
    const tokens = [{ message: 'Hello,' }, { message: ' World' }];
    for (const t of tokens) await responder.send(t);
    responder.close();
  }
}
```

Controller returns responder immediately; service populates via floating promise (`void service.run(responder)`).

## Client consumption

Generated RPC method returns async iterable.

From finom/vovk, MIT.

More from finom/vovk