Get started with Temporal

{
  "scripts": {
    "dev": "next dev",
    "worker": "tsx workflow/worker.ts"
  }
}

/** @type {import('next').NextConfig} */
export default {
  serverExternalPackages: ['@temporalio/client'],
};

import type { InvocationData } from '@ably/ai-transport';

export interface RunIds {
  runId: string;
  invocationId: string;
  triggerEventId: string;
}

export interface ChatWorkflowInput {
  invocation: InvocationData;
  invocationId: string;
}

export interface ToolCallInfo {
  toolCallId: string;
  toolName: string;
  input: unknown;
}

// The inference step either finishes the turn or asks to run a server tool.
export type InferenceOutcome =
  | { kind: 'done' }
  | { kind: 'server-tools'; toolCalls: ToolCallInfo[] };

export const TASK_QUEUE = 'ai-transport-demo';

import { proxyActivities } from '@temporalio/workflow';
import type { ChatWorkflowInput } from './shared.js';
import type * as activities from './activities.js';

const { openRun, runInferenceStep } = proxyActivities<typeof activities>({
  startToCloseTimeout: '5 minutes',
  retry: { maximumAttempts: 3 },
});

// getStockPrice throws on odd prices (~half the time), so give the tool
// activity enough attempts to show Temporal retrying it.
const { runToolStep } = proxyActivities<typeof activities>({
  startToCloseTimeout: '5 minutes',
  retry: { maximumAttempts: 5 },
});

export async function chatWorkflow(input: ChatWorkflowInput): Promise<void> {
  const ids = await openRun({
    invocation: input.invocation,
    invocationId: input.invocationId,
  });

  let outcome = await runInferenceStep({ ids, invocation: input.invocation });
  while (outcome.kind === 'server-tools') {
    for (const toolCall of outcome.toolCalls) {
      await runToolStep({ ids, invocation: input.invocation, toolCall });
    }
    outcome = await runInferenceStep({ ids, invocation: input.invocation });
  }
}

import { z } from 'zod';
import type { Tool } from 'ai';

export const tools: Record<string, Tool> = {
  getStockPrice: {
    description: 'Get the current stock price for a ticker symbol.',
    inputSchema: z.object({
      symbol: z.string().describe('The ticker symbol, for example "AAPL"'),
    }),
    execute: async ({ symbol }: { symbol: string }) => {
      // Intentionally flaky: throws on an odd price (~half the time) and
      // succeeds on an even one. The retry re-rolls the price.
      const priceUSD = Math.round(50 + Math.random() * 500);
      if (priceUSD % 2 !== 0) {
        throw new Error(`stock price service returned an odd price (${priceUSD}), retry me`);
      }
      return { symbol, priceUSD };
    },
  },
};

import { Context } from '@temporalio/activity';
import Ably from 'ably';
import { streamText, convertToModelMessages, stepCountIs, type UIMessage } from 'ai';
import { anthropic } from '@ai-sdk/anthropic';
import { Invocation, type AgentRun, type InvocationData } from '@ably/ai-transport';
import {
  createAgentSession,
  pendingToolCalls,
  stripToolExecutes,
  vercelRunOutcome,
} from '@ably/ai-transport/vercel';
import type { VercelOutput, VercelProjection } from '@ably/ai-transport/vercel';
import { stepIdFor } from '@ably/ai-transport/temporal';
import type { InferenceOutcome, RunIds, ToolCallInfo } from './shared.js';
import { tools } from './tools.js';

type VercelAgentRun = AgentRun<VercelOutput, VercelProjection, UIMessage>;

const makeAbly = () => new Ably.Realtime({ key: process.env.ABLY_API_KEY! });

interface StepInput {
  ids: RunIds;
  invocation: InvocationData;
}

export async function openRun(input: {
  invocation: InvocationData;
  invocationId: string;
}): Promise<RunIds> {
  const ably = makeAbly();
  const session = createAgentSession({ client: ably, channelName: input.invocation.sessionName });
  try {
    await session.connect();
    const run = session.createRun(Invocation.fromJSON(input.invocation), {
      invocationId: input.invocationId,
      // Pin the Run id to the Temporal workflowId so a fresh-process retry of
      // openRun re-enters the same Run instead of opening a parallel one.
      runId: input.invocationId,
      signal: Context.current().cancellationSignal,
    });
    while (run.view.hasOlder()) await run.view.loadOlder();
    await run.start();

    // Leave the Run open (detach, not end) for the first runInferenceStep to adopt.
    await session.detach();

    return {
      runId: run.runId,
      invocationId: run.invocationId,
      triggerEventId: input.invocation.inputEventId,
    };
  } finally {
    ably.close();
  }
}

export async function runInferenceStep(input: StepInput): Promise<InferenceOutcome> {
  const ably = makeAbly();
  const session = createAgentSession({ client: ably, channelName: input.invocation.sessionName });
  try {
    await session.connect();
    const run = session.adoptRun(input.ids, { signal: Context.current().cancellationSignal });
    await run.load();
    while (run.view.hasOlder()) await run.view.loadOlder();

    const outcome = await runInference(run, stepIdFor(input.ids.invocationId));
    await session.detach();
    return outcome;
  } finally {
    ably.close();
  }
}

export async function runToolStep(input: StepInput & { toolCall: ToolCallInfo }): Promise<void> {
  const ably = makeAbly();
  const session = createAgentSession({ client: ably, channelName: input.invocation.sessionName });
  try {
    await session.connect();
    const run = session.adoptRun(input.ids, { signal: Context.current().cancellationSignal });
    await run.load();

    const step = run.createStep({ stepId: stepIdFor(input.ids.invocationId) });
    await step.start();
    const tool = tools[input.toolCall.toolName] as { execute: (input: unknown) => Promise<unknown> };
    const output = await tool.execute(input.toolCall.input);
    await step.send({
      type: 'tool-output-available',
      toolCallId: input.toolCall.toolCallId,
      output,
    });
    await step.end();
    await session.detach();
  } finally {
    ably.close();
  }
}

async function runInference(run: VercelAgentRun, stepId: string): Promise<InferenceOutcome> {
  const step = run.createStep({ stepId });
  await step.start();

  const conversation = run.view.getMessages().map((m) => m.message);
  const result = streamText({
    model: anthropic('claude-sonnet-4-20250514'),
    messages: await convertToModelMessages(conversation),
    tools: stripToolExecutes(tools),
    abortSignal: run.abortSignal,
    // The workflow drives the loop; this call runs one step only.
    stopWhen: stepCountIs(1),
  });

  const pipeResult = await step.pipe(result.toUIMessageStream());
  const outcome = await vercelRunOutcome(pipeResult, result.finishReason);
  await step.end();

  if (outcome.reason === 'complete' || outcome.reason === 'cancelled') {
    await run.end({ reason: outcome.reason });
    return { kind: 'done' };
  }
  if (outcome.reason === 'error') {
    await run.end({
      reason: 'error',
      error: new Ably.ErrorInfo(outcome.error.message, 104000, 500),
    });
    return { kind: 'done' };
  }

  // The model asked for a server tool: leave the Run open for runToolStep to adopt.
  const toolCalls = pendingToolCalls(run.messages)
    .filter((call) => typeof tools[call.toolName]?.execute === 'function')
    .map((call) => ({ toolCallId: call.toolCallId, toolName: call.toolName, input: call.input }));
  return { kind: 'server-tools', toolCalls };
}

import path from 'node:path';
import { config as loadDotenv } from 'dotenv';
import { NativeConnection, Worker } from '@temporalio/worker';
import * as activities from './activities.js';
import { TASK_QUEUE } from './shared.js';

// tsx does not auto-load .env.local the way `next dev` does.
loadDotenv({ path: path.resolve(__dirname, '../.env.local') });

async function main() {
  const connection = await NativeConnection.connect({ address: 'localhost:7233' });
  const worker = await Worker.create({
    connection,
    namespace: 'default',
    taskQueue: TASK_QUEUE,
    workflowsPath: require.resolve('./workflows'),
    activities,
  });
  console.log(`worker listening on ${TASK_QUEUE}`);
  await worker.run();
}

main().catch((err) => {
  console.error(err);
  process.exit(1);
});

import { Client, Connection } from '@temporalio/client';
import type { InvocationData } from '@ably/ai-transport';
import type { ChatWorkflowInput } from '../../../workflow/shared';
import { TASK_QUEUE } from '../../../workflow/shared';

let cachedTemporal: Client | undefined;
async function temporalClient() {
  if (cachedTemporal) return cachedTemporal;
  const connection = await Connection.connect({ address: 'localhost:7233' });
  cachedTemporal = new Client({ connection, namespace: 'default' });
  return cachedTemporal;
}

export async function POST(req: Request) {
  const invocation = (await req.json()) as InvocationData;
  const invocationId = crypto.randomUUID();

  const client = await temporalClient();
  const args: [ChatWorkflowInput] = [{ invocation, invocationId }];
  await client.workflow.start('chatWorkflow', {
    workflowId: invocationId,
    taskQueue: TASK_QUEUE,
    args,
  });

  return Response.json({ invocationId });
}