Streaming

import { defineDevframe, defineRpcFunction } from 'devframe'
import * as v from 'valibot' // npm i valibot

export default defineDevframe({
  id: 'my-tool',
  name: 'My Tool',
  async setup(ctx) {
    const my = ctx.scope('my-tool')

    const channel = my.rpc.streaming.create<string>('chat', { // -> my-tool:chat
      replayWindow: 256,
    })

    my.rpc.register(defineRpcFunction({
      name: 'start-chat', // -> my-tool:start-chat
      type: 'action',
      jsonSerializable: true,
      args: [v.object({ prompt: v.string() })],
      returns: v.object({ streamId: v.string() }),
      handler: async ({ prompt }) => {
        const stream = channel.start()
        ;(async () => {
          for await (const token of fakeLLM(prompt, { signal: stream.signal })) {
            stream.write(token)
          }
          stream.close()
        })()
        return { streamId: stream.id }
      },
    }))
  },
})

const stream = channel.start({ id: 'optional-explicit-id' })

// Imperative: minimal, hand-rolled producers
stream.write(chunk)
stream.error(err) // terminal failure
stream.close() // terminal success
stream.signal // AbortSignal: flips when consumers cancel
stream.id // string: what RPC clients subscribe to

// Web Streams: pipe any ReadableStream<T> in:
sourceReadable.pipeTo(stream.writable, { signal: stream.signal })

// Convenience: start + pipe in one call:
const stream = await channel.pipeFrom(sourceReadable)

import { Readable, Writable } from 'node:stream'

// Pipe a Node Readable into the streaming channel
sourceNodeReadable.pipe(Writable.fromWeb(stream.writable))

// Pipe the channel out to a Node Writable
Readable.fromWeb(reader.readable).pipe(targetNodeWritable)

import { connectDevframe } from 'devframe/client'

const my = (await connectDevframe()).scope('my-tool')
const { streamId } = await my.rpc.call('start-chat', {
  prompt: 'Hello',
})

const reader = my.rpc.streaming.subscribe<string>('chat', streamId) // -> my-tool:chat

// Async iterable: the simplest consumer pattern
for await (const token of reader)
  appendToken(token)

// Or pipe to a DOM-side WritableStream
await reader.readable.pipeTo(downloadWritable)

reader.cancel() // sends cancel upstream; the node-side stream.signal flips

// Node side: typically inside an action handler
ctx.rpc.register(defineRpcFunction({
  name: 'my-tool:upload-file',
  type: 'action',
  args: [v.object({ name: v.string() })],
  returns: v.object({ uploadId: v.string() }),
  handler: async ({ name }) => {
    const reader = channel.openInbound()

    // Process chunks asynchronously; the action returns immediately
    // so the browser side can start uploading.
    ;(async () => {
      const file = createWriteStream(name)
      for await (const chunk of reader)
        file.write(chunk)
      file.close()
    })()

    return { uploadId: reader.id }
  },
}))

// Browser side
const { uploadId } = await my.rpc.call('upload-file', {
  name: 'capture.bin',
})
const upload = my.rpc.streaming.upload<Uint8Array>('files', uploadId) // -> my-tool:files

// Imperative
upload.write(chunk1)
upload.write(chunk2)
upload.close()

// Or pipe a Web ReadableStream straight in:
fileReadable.pipeTo(upload.writable, { signal: upload.signal })

my.rpc.streaming.create<string>('chat', { // -> my-tool:chat
  replayWindow: 256, // chunks to retain per stream id
  closedStreamRetention: 30_000, // ms to hold closed streams for late subscribers
})

const reader = my.rpc.streaming.subscribe('chat', id, { // -> my-tool:chat
  highWaterMark: 1024, // raise if you expect bursts the consumer can recover from
})