Proposal: Generalize River’s core for broader SSE use cases

[justsomelegs]

I know the v2 rewrite of River is still in the very early stages, and this maybe this is more of a v3 idea.

While the project is currently focused on AI streams, I think it would be really interesting if the core primitives were designed for any kind of SSE stream, with the AI-specific features built on top of those.

That approach could open the library up to a wider range of use cases:

• Live Data Feeds
• Collaborative Apps
• Durable Event Logs
• Notification Streams
• Sync-engines

and of course, AI and LLM Streams

I imagine River could become to SSE/streams what better-auth is to authentication, providing a strong foundation that can be extended with plugins for AI and other specialized features.

Just thinking out loud here. I’ll be using the library whichever direction it goes as, but thought this might be worth considering.

EDIT:

very crude and inspired idea but I was thinking something like this for the end developer:

export const server = riverServer({
	plugins: [AI(), auth()] // wouldn't matter the order just provides the functions to the server (server.ai / sever.auth)
	streams: {
		notifications: {
			schema: z.object({ userId: z.string(), text: z.string() }),
				runner: async (args) => { }
		},
		livePrices: {
			schema: z.object({ symbol: z.string(), price: z.number(), ts: z.number() }),
			runner: async (args) => { }
		},
		ai:  {
			schema: z.object({ role: z.enum(['assistant', 'user']), content: z.string() }),
			runner: async (args) => { }
		}
	},
	//examples below if you wanted to go past streams
	queries: {},
	mutations: {},
	sockets: {},
});

[bmdavis419]

This is a WIP from v2 and while the syntax is a bit different, I'm 100% in agreement here that it should be more generic:

import { V2_DEV } from '$lib/index.js';
import z from 'zod';

const myFirstV2AgentStream = V2_DEV.RIVER_STREAMS.createRiverStream(
	'my-first-v2-agent',
	V2_DEV.RIVER_STREAMS.riverStorageDefaultProvider
);

const myFirstV2Agent = V2_DEV.RIVER_AGENTS.createRiverAgent({
	inputSchema: z.object({
		prompt: z.string()
	}),
	stream: myFirstV2AgentStream,
	runner: async (args) => {
		const { input, abortSignal, meta, stream, agentRunId } = args;

		console.log('STARTING A V2 AGENT');

		// TODO: keep grinding this out...
		// also need to figure out the typesafety on the stream, probably need to attach a type to the stream itself?
	}
});

the agents are now just basically wrappers around streams with typesafty and RPC calling. Have a ton more to do on this and more experimenting, will put up a PR when it's closer to ready

[justsomelegs]

didn't make a PR as it might not be the direction you want to go but had a go at making a rough demo for a server side implementation today as had the day off work, though didn't have enough time to work on client side stuff today:

changes:

• changed primitive focus from agents to streams
• added plugin system
• plugins can extends context and inject their own helpers and data into stream runners
• ai sdk moved into the plugin format (very early implementation and probably needs completley reworking as its more there to demonstrate what the plugin system could do)

main...justsomelegs:river:legs/v2-dev

types and the AI plugin implementation still need a lot of work they're kinda just there as "functional" placeholders.
more wanted to show off how stuff would be defined and how types could work with a plugin system.

durability/storage providers, custom agents and a bunch of other stuff still need to be implemented as well.

const server = riverServer({
	plugins: [
		AI({
			models: {
				'gpt-4o-mini': openrouter('openai/gpt-4o-mini'),
				'claude-3.5-sonnet': openrouter('anthropic/claude-3.5-sonnet')
			},
			defaultModel: 'gpt-4o-mini'
		})
	],

	streams: (stream) => ({
		aiChat: stream({
			chunkSchema: z.object({
				delta: z.string()
			}),
			inputSchema: z.object({
				query: z.string(),
				model: z.enum(['gpt-4o-mini', 'claude-3.5-sonnet']).default('gpt-4o-mini'),
				system: z.string().optional()
			}),
			runner: async ({ input, appendChunk, ai, abortSignal }) => {
				const result = ai.streamText({
					model: input.model,
					prompt: input.query,
					system: input.system || 'You are a helpful assistant.',
					temperature: 0.7
				});

				await ai.pipeTextStream(
					result,
					(delta) => {
						appendChunk({ delta });
					},
					abortSignal
				);
			}
		}),
		notifications: stream({
			chunkSchema: z.object({
				type: z.literal('note'),
				text: z.string(),
				i: z.number()
			}),
			runner: async ({ appendChunk, abortSignal }) => {
				let i = 0;

				while (!abortSignal.aborted && i < 10) {
					appendChunk({ type: 'note', text: `Notification ${i}`, i });
					await new Promise((r) => setTimeout(r, 500));
					i++;
				}
			}
		})
	})
});

[justsomelegs]

This is a WIP from v2 and while the syntax is a bit different, I'm 100% in agreement here that it should be more generic:

import { V2_DEV } from '$lib/index.js';
import z from 'zod';

const myFirstV2AgentStream = V2_DEV.RIVER_STREAMS.createRiverStream(
'my-first-v2-agent',
V2_DEV.RIVER_STREAMS.riverStorageDefaultProvider
);

const myFirstV2Agent = V2_DEV.RIVER_AGENTS.createRiverAgent({
inputSchema: z.object({
prompt: z.string()
}),
stream: myFirstV2AgentStream,
runner: async (args) => {
const { input, abortSignal, meta, stream, agentRunId } = args;

  console.log('STARTING A V2 AGENT');

  // TODO: keep grinding this out...
  // also need to figure out the typesafety on the stream, probably need to attach a type to the stream itself?

}
});

the agents are now just basically wrappers around streams with typesafty and RPC calling. Have a ton more to do on this and more experimenting, will put up a PR when it's closer to ready

had a peak around the v2 when trying out the thing i suggested above and your v2 is for sure a lot more cleaner then v1.

[justsomelegs]

thinking about it more if i was to continue with this api design id probably allow plugins to define a scope "global" or "stream" then on the stream there would be a use field where you could enable the scoped plugins - so for example the ai() plugin could be scoped then on a stream you want the ai functionality you could put use: ["ai"].

would probably have an overrideScope property on the plugin so users could make scoped functionality available to all streams if they so desired

[bmdavis419]

just got back home and am taking a look at this for the first time for real. Like the idea of plugins, I was actually just about to write some utilities for making it easier to work with the AI sdk. The biggest hurdle in my mind right now is the type safety on the streams. I have to attach the generics on the stream, which makes defining stuff kinda annoying... Gonna do some more testing today and tmrw and see what I come up with

[justsomelegs]

@bmdavis419

no problem at all i never expect a reply.

the branch i made on my fork is 100% types safe at the moment. in theory all you would have to pass is the type of the server to the client.

client<server>({})

everything should have autocomplete on the branch too, including i added scopes to plugins yesterday too

the types and what gets injected into the streams is scoped and 100% typesafe. so if a plug-in is scoped to "stream" and you define it it in the use array the context gets updated with the methods from the plugin.

again a lot of work to get it working but gotta say not having to work about types as a user could be a big selling point

[justsomelegs]

still wanna figure out a way to programmatically infer the endpoints from the streams but shouldn't be too hard, probably just a key'd map with the runners as the value and catchall routes in whatever framework adapters get added.

in theory you could make the plugins define their own streams for a no setup use-case too.

the only problem at the moment is that i need to figure out a way to allow people to define streams wherever they want instead of in the riverStream config and still get the same level of type safety.

[bmdavis419]

Just did some scratch work stuff and came up with this. I think u were very much onto something with just making everything streams. For now at least I want to be more of a primitive then anything so I don't really want to deal with plugins, gonna work on making this:

import { defaultStorageProvider } from "river"

export const myRiverStream = createRiverStream({
	inputSchema: z.object({
		prompt: z.string()	
	}),
	streamProvider: defaultStorageProvider,
	runner: async (stuff) => {
		const { input, initStream, runId, meta, abortSignal } = stuff
		
		const { fullStream } = streamText({...})
		
		type StreamChunkType = InferAiSdkStream<typeof fullStream>
		
		const stream = initStream<StreamChunkType>()
		
		return stream.sendData(({
			appendAiSdkStream,
			appendChunk	
		}) => {
			appendAiSdkStream(
				fullStream,
				// intercept, optional
				(curChunk, writeChunk) => {
					writeChunk(curChunk)
				}
			)
			
			appendChunk(...)
		})
	}
})

[justsomelegs]

i really like the separation of concerns of defining a stream anywhere in any .ts file for the server like in a streams.ts file.

very convex like in that way.

const { createRiverStream } = riverConfig({..configHere}) // config for plugins, provider etc.

const stream = createRiverStream({}) // stream logic

const server = riverServer({...stream}) // server and stream handling

[justsomelegs]

i think it depends if you want the types to just work with minimal concerns, then it might have to be a config style like what i made. other wise i think there will have to be a bit of forcing, codegen or type gymnastics on the clients end because of how typescripts inference and stuff works.

[justsomelegs]

hey i know you're already working on v3 but I couldn't help myself and kept messing around with my 2.5 idea to see what I could do with types and such. ended up rewriting a bunch of the server code just to see if the generalized approach with end-to-end type saftey actually works.

heres a simplified look at what I have so far.

server:

const river = riverConfig({
  plugins: [ai()],
  ai: { /* config */ }
});

const vowelCounter = river.stream({
  chunkSchema: z.object({ letter: z.string(), isVowel: z.boolean() }),
  runner: async ({ input, appendChunk }) => { /* ... */ }
});

const resumableStream = river.stream({
  storageProvider: inMemoryStorage,  // pluggable storage
  runner: async ({ input, appendChunk, runId }) => { /* ... */ }
});

const aiStream = river.stream({
  use: ['ai'],  // only pulls in AI when needed
  runner: async ({ input, ai, appendChunk }) => { /* ... */ }
});

const server = riverServer(river, {
    vowelCounter,
    questionAsker,
    simpleChat,
    resumableStream
});

export type River = typeof server.streams;

client

const client = createRiverClient<River>({
  endpoint: '/stream',
  retry: { attempts: 3, delayMs: 1000, exponentialBackoff: true }
});

// Fully typed, same API for all stream types
const stream = client.stream.vowelCounter({
  onStart() { /* ... */ },
  onChunk(chunk) { /* chunk is typed! */ },
  onComplete(summary) { /* ... */ }
});

await stream.start({ yourName: "Hello" });

// Resumability works the same way
const resumable = client.stream.resumableStream({
  onChunk(chunk) { /* ... */ }
});

await resumable.start({ count: 20 }, { runId: "stable-id" });

currently i've got these things working:

• generalized streams with the ability to add and use plugins (scoped and global)
• scoped plugin usage per stream
• pluggable storage for resumability (though needs some work to make the API for storage providers nicer)
• lifecycle hooks on server and client
• full type inference end-to-end the only type that needs to be passed / typed is the River type from the sever so the client knows the stream schemas and such.
• standard schema validation so can use things other then zod
• some demos of it working

still rough around the edges and needs a lot of work to become finalized but i'm really impressed with how the type safety came out considering im not using a traditional builder pattern. just thought it might be interesting to see where the idea went.

link to the branch if anyone is interested: https://github.com/justsomelegs/river/tree/legs/v2-dev