Chat Controller

Overview

createChatController is the central orchestrator of rxjs-ai. It wires together a model adapter, a message history, and a reactive status machine into a single, easy-to-use handle. Every interaction — sending a message, retrying, or cancelling — is expressed as an RxJS stream, so you can compose it with any other reactive logic in your application.

import { createChatController } from "rxjs-ai";

const chat = createChatController(myAdapter);

ChatModelAdapter interface

Before creating a controller you need an adapter — an object with a single method that bridges rxjs-ai to your LLM of choice.

interface ChatModelAdapter {
  complete(request: ChatModelRequest): Observable<ChatChunk>;
}

interface ChatModelRequest {
  messages: ChatMessage[];
  signal: AbortSignal;
}

The complete method receives the full conversation history and an AbortSignal for cancellation, and must return an Observable that emits one or more chunks as the model responds.

ChatChunk

A chunk is either a plain string or a structured object:

type ChatChunk = string | { content: string; done?: boolean };

Form	When to use
string	Quick prototyping — every emission is appended to the assistant message.
{ content, done? }	Production adapters — set done: true on the final chunk to signal completion explicitly.

Basic usage

import { createChatController } from "rxjs-ai";

const adapter: ChatModelAdapter = {
  complete({ messages, signal }) {
    // Return an Observable<ChatChunk>
    return myLlmStream(messages, signal);
  },
};

const chat = createChatController(adapter);

// Subscribe to state changes
chat.messages$.subscribe((msgs) => console.log("Messages:", msgs));
chat.status$.subscribe((s) => console.log("Status:", s));

// Send a user message — the adapter is called automatically
chat.send("Hello, AI!");

API reference

Arguments

Argument	Type	Description
model	ChatModelAdapter	Required. The model adapter to use for completions. Passed as the first positional argument.
options	CreateChatControllerOptions	Optional configuration object (see below).

Options (CreateChatControllerOptions)

Option	Type	Default	Description
now	() => number	Date.now	Clock function used for message timestamps.
idGenerator	() => string	internal nanoid	Custom ID generator for message IDs.
initialMessages	ChatMessage[]	[]	Pre-populate the conversation history.

Returned handle — ChatController

Observable properties

Property	Type	Description
messages$	Observable<ChatMessage[]>	Current message list. Emits after every mutation.
status$	Observable<ChatStatus>	Lifecycle status of the controller.
error$	Observable<unknown | null>	The latest error, or null when cleared.
state$	Observable<ChatState>	Combined snapshot of messages, status, and error.

Methods

Method	Signature	Description
getState	() => ChatState	Synchronously read the current state snapshot.
send	(content: string, meta?: Record<string, unknown>) => void	Append a user message and trigger a completion.
retryLast	() => void	Remove the last assistant message and re-run the completion.
cancel	() => void	Abort the in-flight completion. Status moves to cancelled.
destroy	() => void	Tear down all internal subscriptions. The controller must not be reused.

Status lifecycle

The controller follows a deterministic state machine:

idle ──▶ loading ──▶ streaming ──▶ idle
                          │
                          ├──▶ error
                          │
                          └──▶ cancelled

• idle — No request in flight. Ready to send.
• loading — A request has been dispatched to the adapter but no chunk has arrived yet.
• streaming — At least one chunk has been received; the assistant message is being built up.
• error — The adapter Observable errored. Inspect error$ for details. Call retryLast() to try again.
• cancelled — The user called cancel() while loading or streaming.

After error or cancelled, the status returns to idle once the user calls send() or retryLast().

Types

type ChatRole = "system" | "user" | "assistant" | "tool";

interface ChatMessage {
  id: string;
  role: ChatRole;
  content: string;
  createdAt: number;
  meta?: Record<string, unknown>;
}

type ChatStatus = "idle" | "loading" | "streaming" | "error" | "cancelled";

interface ChatState {
  messages: ChatMessage[];
  status: ChatStatus;
  error: unknown | null;
}

type ChatChunk = string | { content: string; done?: boolean };

interface ChatModelRequest {
  messages: ChatMessage[];
  signal: AbortSignal;
}

interface ChatModelAdapter {
  complete(request: ChatModelRequest): Observable<ChatChunk>;
}

interface ChatController {
  messages$: Observable<ChatMessage[]>;
  status$: Observable<ChatStatus>;
  error$: Observable<unknown | null>;
  state$: Observable<ChatState>;
  getState(): ChatState;
  send(content: string, meta?: Record<string, unknown>): void;
  retryLast(): void;
  cancel(): void;
  destroy(): void;
}