React streaming preview

// eslint-disable-next-line @nx/enforce-module-boundaries -- The isolated consumer copies this internal example layout.
import { ExampleFooter, ExampleLayout } from '../../../../shared/react-example';
import { useEffect, useRef, useState } from 'react';
import { useAgent } from '@threadplane/react';
import { ChatInput, MessageList } from '@threadplane/react/chat';
import { createStreamingInspection } from './inspection';
import type { InspectionFrame } from './inspection';
import {
  createConnectedApplication,
  type streamingConnection,
} from './connection';
 
export function StreamingDemo({
  connection,
  onReady,
  onInspection,
}: {
  readonly connection: ReturnType<typeof streamingConnection>;
  readonly onReady: () => void;
  readonly onInspection?: (frame: InspectionFrame) => void;
}) {
  const publisher = useRef(onInspection);
  publisher.current = onInspection;
  const [inspection] = useState(() =>
    createStreamingInspection((frame) => publisher.current?.(frame))
  );
  const [application, setApplication] = useState(() =>
    createConnectedApplication(connection)
  );
  const [resetting, setResetting] = useState(false);
  const [generation, setGeneration] = useState(0);
  const live = useRef(true);
  const snapshot = useAgent(application);
  useEffect(() => inspection.attach(application), [application, inspection]);
  useEffect(() => {
    onReady();
  }, [onReady]);
  useEffect(
    () => () => {
      void application.dispose();
    },
    [application]
  );
  useEffect(
    () => () => {
      live.current = false;
      inspection.dispose();
    },
    [inspection]
  );
 
  async function reset() {
    if (snapshot.busy || resetting) return;
    setResetting(true);
    inspection.reset();
    await application.dispose();
    if (!live.current) return;
    setApplication(createConnectedApplication(connection));
    setGeneration((value) => value + 1);
    setResetting(false);
  }
 
  const status = snapshot.busy
    ? snapshot.creation === 'pending'
      ? 'Creating conversation…'
      : 'Receiving response…'
    : snapshot.outcome === 'success'
    ? 'Response complete.'
    : snapshot.outcome === 'aborted'
    ? 'Stopped. Start a new conversation to continue.'
    : !snapshot.canSubmit
    ? 'Start a new conversation to continue.'
    : 'Ready.';
 
  return (
    <ExampleLayout className="streaming-demo">
      <header>
        <div>
          <h1>LangGraph streaming</h1>
        </div>
        <button
          type="button"
          disabled={snapshot.busy || resetting}
          onClick={() => void reset()}
        >
          New conversation
        </button>
      </header>
      <p>Watch an answer arrive as the agent writes it.</p>
      <MessageList rows={snapshot.rows} label="Conversation" />
      <ExampleFooter>
        {snapshot.error && <p role="alert">{snapshot.error}</p>}
        <ChatInput
          key={generation}
          label="Message"
          placeholder="Ask about LangGraph…"
          disabled={resetting || (!snapshot.canSubmit && !snapshot.busy)}
          busy={snapshot.busy}
          onStop={snapshot.busy ? () => void application.stop() : undefined}
          onSubmit={(text) => {
            if (!application.getSnapshot().canSubmit) return false;
            void inspection.submit(application, text);
          }}
        />
        <p role="status">{status}</p>
      </ExampleFooter>
    </ExampleLayout>
  );
}

import { Client } from '@langchain/langgraph-sdk';
import { createSession } from '@threadplane/langgraph';
// eslint-disable-next-line @nx/enforce-module-boundaries -- The isolated pilot build copies this neutral internal bridge; it is not a published package.
import type { RuntimeBridgeConfiguration } from '../../../../../libs/cockpit-runtime-bridge/src/index';
import { createStreamingApplication } from './application';
 
export function streamingConnection(
  configuration: RuntimeBridgeConfiguration
): {
  apiUrl: string;
  headers: Record<string, string>;
} {
  if (configuration.status === 'error')
    throw new Error('Runtime configuration failed.');
  if (
    configuration.status === 'configured' &&
    configuration.target.kind === 'ag-ui'
  ) {
    throw new Error('Runtime configuration failed.');
  }
  const target =
    configuration.status === 'configured' &&
    configuration.target.kind === 'langsmith'
      ? configuration.target
      : null;
  return {
    apiUrl: target?.apiUrl ?? new URL('/api', window.location.origin).href,
    headers: target ? { 'x-api-key': target.apiKey } : {},
  };
}
 
export function createConnectedApplication(
  connection: ReturnType<typeof streamingConnection>
) {
  const client = new Client({
    apiUrl: connection.apiUrl,
    apiKey: null,
    defaultHeaders: connection.headers,
    callerOptions: { maxRetries: 0 },
  });
  return createStreamingApplication({
    async createThread(signal) {
      const threadId = crypto.randomUUID();
      const created = await client.threads.create({
        threadId,
        ifExists: 'raise',
        signal,
      });
      if (created.thread_id !== threadId)
        throw new Error('Conversation creation was not confirmed.');
      return threadId;
    },
    sessionFactory: (threadId) =>
      createSession({
        assistantId: 'streaming',
        threadId,
        apiUrl: connection.apiUrl,
        clientOptions: { maxRetries: 0, defaultHeaders: connection.headers },
      }),
  });
}