Threads

[TK]

1. Thread Definition

A thread is an instance of an agent—the runtime execution context where conversations happen. Each thread has isolated storage for messages and files, and maintains its own conversation history.

1.1 Agent Assignment

The agent assigned to a thread can change over time while preserving all history and context. This is called a handoff.

Handoffs enable workflows where specialized agents handle different parts of a conversation—a triage agent routes to support, support escalates to billing—all within the same thread with full context preserved.

1.2 ThreadState Interface

ThreadState is the unified interface for all thread operations. It provides a consistent API for interacting with threads regardless of context—whether from tools during execution, hooks, or HTTP endpoints.

1.2.1 Properties

1.2.2 Methods

Messages — See Messages for details

Logs

Resource Loading — See Resource Loading for details

Tool Invocation — See Tool Invocation for details

Effect Scheduling — See Effect Scheduling for details

Events — See Events for details

File System — See File System for details

1.3 Access Contexts

ThreadState is provided in different contexts:

2. Messages

2.1 Message Structure

2.2 Reading Messages

// Get recent messages
const { messages, total, hasMore } = await state.getMessages({
  limit: 50,
  order: 'desc',
});

// Get a single message
const message = await state.getMessage('msg-123');

2.3 Query Options

2.4 Injecting Messages

Messages can be injected into the conversation:

const message = await state.injectMessage({
  role: 'user',
  content: 'Additional context...',
  silent: true, // Hide from UI
  metadata: { source: 'tool' },
});

2.5 Updating Messages

Existing messages can be updated:

const updated = await state.updateMessage('msg-123', {
  content: 'Updated content',
  metadata: { edited: true },
});

3. Resource Loading

3.1 Loading Definitions

ThreadState provides access to registered definitions:

// Load a model definition
const model = await state.loadModel('gpt-4o');

// Load a prompt definition
const prompt = await state.loadPrompt('support-prompt');

// Load an agent definition
const agent = await state.loadAgent('customer-support');

3.2 Listing Resources

// Get available resource names
const prompts = state.getPromptNames();
const agents = state.getAgentNames();
const models = state.getModelNames();

4. Tool Invocation

4.1 Queuing Tools

Queue a tool for asynchronous execution:

// Queue a tool (non-blocking)
state.queueTool('send_email', {
  to: 'user@example.com',
  subject: 'Hello',
  body: 'Message content',
});

If the thread is currently executing, the tool is added to the queue. Otherwise, a new execution is started.

4.2 Direct Invocation

Invoke a tool and wait for the result:

// Invoke directly (blocking)
const result = await state.invokeTool('get_weather', {
  location: 'San Francisco',
});

if (result.status === 'success') {
  console.log(result.result);
}

5. Effect Scheduling

Effects are scheduled operations that run outside the normal tool execution context. Unlike tools which execute immediately during conversation, effects:

• Run independently of the conversation flow
• Can be scheduled with delays
• Execute with their own ThreadState context
• Are ideal for side effects that don’t need immediate results

5.1 Scheduling Effects

// Schedule an effect to run after 30 minutes
const effectId = await state.scheduleEffect(
  'send_reminder_email',
  { to: 'user@example.com', subject: 'Reminder' },
  30 * 60 * 1000  // delay in milliseconds
);

Returns a unique effect ID (UUID) that can be used to cancel the effect.

5.2 ScheduledEffect Record

5.3 Managing Scheduled Effects

// Get all pending effects
const effects = await state.getScheduledEffects();

// Get effects filtered by name
const reminders = await state.getScheduledEffects('send_reminder_email');

// Cancel a scheduled effect
const removed = await state.removeScheduledEffect(effectId);
if (removed) {
  console.log('Effect cancelled');
}

6. Events

6.1 Emitting Events

Send custom events to connected clients:

state.emit('progress', {
  step: 3,
  total: 10,
  message: 'Processing data...',
});

Events are delivered via WebSocket to any connected clients.

6.2 Event Guidelines

• Event names SHOULD be lowercase with underscores
• Event data MUST be JSON-serializable
• Events SHOULD NOT contain sensitive information

7. Context Storage

7.1 Per-Execution Context

The context property provides temporary storage:

// Store data during execution
state.context.userData = { id: 123, name: 'Alice' };

// Retrieve in another tool call
const user = state.context.userData;

7.2 Context Limitations

For persistent storage, use the file system or external storage solutions.

8. File System

Each thread has access to an isolated file system for storing files, images, and other binary data.

8.1 File Record

Files are represented by the FileRecord type:

8.2 Writing Files

// Write a text file
const record = await state.writeFile(
  '/data/config.json',
  JSON.stringify({ key: 'value' }),
  'application/json'
);

// Write a binary file with metadata
const imageRecord = await state.writeFile(
  '/images/photo.jpg',
  imageBuffer,
  'image/jpeg',
  { width: 1920, height: 1080 }
);

8.3 Reading Files

// Read file content
const data = await state.readFile('/data/config.json');
if (data) {
  const text = new TextDecoder().decode(data);
  const config = JSON.parse(text);
}

// Get file metadata without reading content
const info = await state.statFile('/images/photo.jpg');
if (info) {
  console.log(`Size: ${info.size} bytes`);
}

8.4 Directory Operations

// Create a directory
await state.mkdirFile('/documents');

// List directory contents
const { entries } = await state.readdirFile('/documents');
for (const entry of entries) {
  console.log(`${entry.isDirectory ? 'DIR' : 'FILE'}: ${entry.name}`);
}

// Remove an empty directory
await state.rmdirFile('/documents');

// Delete a file
await state.unlinkFile('/data/config.json');

8.5 Search Operations

// Search file contents
const grepResults = await state.grepFiles('error');
for (const result of grepResults) {
  console.log(`${result.path}:`);
  for (const match of result.matches) {
    console.log(`  Line ${match.line}: ${match.content}`);
  }
}

// Find files by glob pattern
const { paths } = await state.findFiles('**/*.json');
console.log('JSON files:', paths);

8.6 File Statistics

const stats = await state.getFileStats();
console.log(`Files: ${stats.fileCount}`);
console.log(`Directories: ${stats.directoryCount}`);
console.log(`Total size: ${stats.totalSize} bytes`);

8.7 Image Thumbnails

// Get thumbnail for an image
const thumbnail = await state.getFileThumbnail('/images/photo.jpg');
if (thumbnail) {
  // Use thumbnail data
}

8.8 Streaming Files

For large files, readFileStream provides memory-efficient access by yielding data in chunks.

Method

readFileStream(
  path: string,
  options?: ReadFileStreamOptions
): Promise<AsyncIterable<FileChunk> | null>

Options

FileChunk

Behavior

1. File Not Found: Returns null if the file does not exist.

2. External Files: Returns null for files stored externally (S3, R2, URL). External files must be fetched using their storage location.

3. Small Files: Files under the chunk threshold are yielded as a single chunk with index: 0, totalChunks: 1, and isLast: true.

4. Chunked Files: Large files yield multiple chunks in sequential order (0, 1, 2…).

5. Ordering Guarantee: Chunks are always yielded in sequential order. Implementations MUST NOT yield chunks out of order.

6. Abort Signal: When triggered, the stream stops yielding new chunks. Already-yielded chunks remain valid.

Example: Processing Large File

const stream = await state.readFileStream('/uploads/video.mp4');
if (!stream) {
  throw new Error('File not found');
}

for await (const chunk of stream) {
  console.log(`Progress: ${chunk.index + 1}/${chunk.totalChunks}`);
  // Process chunk.data without loading entire file into memory
}

Example: HTTP Streaming Response

const stream = await state.readFileStream('/uploads/large-file.bin');
if (!stream) {
  return new Response('Not Found', { status: 404 });
}

const readable = new ReadableStream({
  async start(controller) {
    for await (const chunk of stream) {
      controller.enqueue(chunk.data);
    }
    controller.close();
  }
});

return new Response(readable, {
  headers: { 'Content-Type': 'application/octet-stream' }
});

9. Execution State

9.1 Availability

The execution property is:

• Present during active execution (tools, hooks)
• Null when accessing at rest (endpoints)

if (state.execution) {
  console.log(`Step ${state.execution.stepCount}`);
}

9.2 Execution Properties

9.3 Controlling Execution

// Force next turn to a specific side (dual_ai only)
state.execution.forceTurn('b');

// Stop execution after current operation
state.execution.stop();

// Use abort signal for cancellation
fetch(url, { signal: state.execution.abortSignal });

10. Runtime Context (Non-Portable)

The _notPackableRuntimeContext property allows runtime implementations to inject platform-specific context.

10.1 Property Definition

10.2 Portability

Tools that access _notPackableRuntimeContext cannot be packed, shared, or published to tool registries.

10.3 Example: Accessing Cloudflare Bindings

import { defineTool } from '@standardagents/builder';
import { z } from 'zod';

export default defineTool({
  description: 'Query internal database',
  args: z.object({ userId: z.string() }),
  execute: async (state, args) => {
    const env = state._notPackableRuntimeContext?.env as Env | undefined;

    if (!env?.MY_DATABASE) {
      return { status: 'error', error: 'Database not available' };
    }

    const result = await env.MY_DATABASE
      .prepare('SELECT * FROM users WHERE id = ?')
      .bind(args.userId)
      .first();

    return { status: 'success', result: JSON.stringify(result) };
  },
});

Tip: Common implementations include { env: Env } for Cloudflare Workers bindings (KV, D1, R2), or { process: NodeJS.Process } for Node.js environments.

11. Thread Endpoints

Thread endpoints expose HTTP APIs for thread-specific operations. They automatically look up threads by ID and provide a ThreadState instance.

11.1 Defining Thread Endpoints

import { defineThreadEndpoint } from '@standardagents/spec';

export default defineThreadEndpoint(async (req, state) => {
  const { messages, total } = await state.getMessages({ limit: 10 });
  return Response.json({
    threadId: state.threadId,
    messageCount: total,
  });
});

11.2 Handler Signature

Thread endpoint handlers receive:

The handler MUST return a Response object.

11.3 Endpoint Context

When accessing a thread via an endpoint, state.execution is always null because the thread is not actively executing:

export default defineThreadEndpoint(async (req, state) => {
  // Execution is always null in endpoints
  if (state.execution === null) {
    // Thread is at rest, not executing
  }
  return Response.json({ threadId: state.threadId });
});

11.4 Error Handling

Thread endpoints automatically handle common errors:

11.5 Thread Endpoint Examples

Get Thread Messages:

import { defineThreadEndpoint } from '@standardagents/spec';

export default defineThreadEndpoint(async (req, state) => {
  const url = new URL(req.url);
  const limit = parseInt(url.searchParams.get('limit') || '50');
  const offset = parseInt(url.searchParams.get('offset') || '0');

  const result = await state.getMessages({ limit, offset, order: 'desc' });
  return Response.json(result);
});

Export Thread Data:

import { defineThreadEndpoint } from '@standardagents/spec';

export default defineThreadEndpoint(async (req, state) => {
  const { messages } = await state.getMessages({ order: 'asc' });

  return Response.json({
    exportedAt: new Date().toISOString(),
    thread: {
      id: state.threadId,
      agent: state.agentId,
      user: state.userId,
      createdAt: state.createdAt,
    },
    messages,
  });
});

List Thread Files:

import { defineThreadEndpoint } from '@standardagents/spec';

export default defineThreadEndpoint(async (req, state) => {
  const { entries } = await state.readdirFile('/');
  return Response.json({ files: entries });
});

12. Conformance Requirements

12.1 Thread Lifecycle

1. Creation: Thread created with metadata
2. Execution: Agent processes messages
3. At Rest: Thread accessible via endpoints
4. Resumption: New execution on next message

12.2 Platform Independence

ThreadState is an abstract interface. Implementations MUST NOT expose:

• Storage implementation details
• Platform-specific bindings
• Internal state management

12.3 Error Handling

Methods that return promises SHOULD throw on:

• Resource not found (loadModel, loadPrompt, loadAgent)
• Invalid parameters
• Storage errors

Methods SHOULD NOT throw on:

• Empty results (return empty arrays/null instead)
• Missing optional data

13. TypeScript Reference

/**
 * Thread state interface - the unified API for thread interactions.
 * Available to tools, hooks, and endpoints.
 */
interface ThreadState {
  // ─────────────────────────────────────────────────────────────────────────
  // Identity (readonly)
  // ─────────────────────────────────────────────────────────────────────────
  readonly threadId: string;
  readonly agentId: string;
  readonly userId: string | null;
  readonly createdAt: number;

  // ─────────────────────────────────────────────────────────────────────────
  // Messages
  // ─────────────────────────────────────────────────────────────────────────
  getMessages(options?: GetMessagesOptions): Promise<MessagesResult>;
  getMessage(messageId: string): Promise<Message | null>;
  injectMessage(input: InjectMessageInput): Promise<Message>;
  updateMessage(messageId: string, updates: MessageUpdates): Promise<Message>;

  // ─────────────────────────────────────────────────────────────────────────
  // Logs
  // ─────────────────────────────────────────────────────────────────────────
  getLogs(options?: GetLogsOptions): Promise<Log[]>;

  // ─────────────────────────────────────────────────────────────────────────
  // Resource Loading
  // ─────────────────────────────────────────────────────────────────────────
  loadModel<T = unknown>(name: string): Promise<T>;
  loadPrompt<T = unknown>(name: string): Promise<T>;
  loadAgent<T = unknown>(name: string): Promise<T>;
  getPromptNames(): string[];
  getAgentNames(): string[];
  getModelNames(): string[];

  // ─────────────────────────────────────────────────────────────────────────
  // Tool Invocation
  // ─────────────────────────────────────────────────────────────────────────
  queueTool(toolName: string, args: Record<string, unknown>): void;
  invokeTool(toolName: string, args: Record<string, unknown>): Promise<ToolResult>;

  // ─────────────────────────────────────────────────────────────────────────
  // Effect Scheduling
  // ─────────────────────────────────────────────────────────────────────────
  scheduleEffect(name: string, args: Record<string, unknown>, delay?: number): Promise<string>;
  getScheduledEffects(name?: string): Promise<ScheduledEffect[]>;
  removeScheduledEffect(id: string): Promise<boolean>;

  // ─────────────────────────────────────────────────────────────────────────
  // Events
  // ─────────────────────────────────────────────────────────────────────────
  emit(event: string, data: unknown): void;

  // ─────────────────────────────────────────────────────────────────────────
  // Context Storage
  // ─────────────────────────────────────────────────────────────────────────
  context: Record<string, unknown>;

  // ─────────────────────────────────────────────────────────────────────────
  // File System
  // ─────────────────────────────────────────────────────────────────────────
  writeFile(path: string, data: ArrayBuffer | string, mimeType: string, options?: WriteFileOptions): Promise<FileRecord>;
  readFile(path: string): Promise<ArrayBuffer | null>;
  readFileStream(path: string, options?: ReadFileStreamOptions): Promise<AsyncIterable<FileChunk> | null>;
  statFile(path: string): Promise<FileRecord | null>;
  readdirFile(path: string): Promise<ReaddirResult>;
  unlinkFile(path: string): Promise<void>;
  mkdirFile(path: string): Promise<FileRecord>;
  rmdirFile(path: string): Promise<void>;
  getFileStats(): Promise<FileStats>;
  grepFiles(pattern: string): Promise<GrepResult[]>;
  findFiles(pattern: string): Promise<FindResult>;
  getFileThumbnail(path: string): Promise<ArrayBuffer | null>;

  // ─────────────────────────────────────────────────────────────────────────
  // Execution State
  // ─────────────────────────────────────────────────────────────────────────
  execution: ExecutionState | null;

  // ─────────────────────────────────────────────────────────────────────────
  // Runtime Context (Non-Portable)
  // ─────────────────────────────────────────────────────────────────────────
  readonly _notPackableRuntimeContext?: Record<string, unknown>;
}

/**
 * Execution state available during active execution.
 */
interface ExecutionState {
  readonly flowId: string;
  readonly currentSide: 'a' | 'b';
  readonly stepCount: number;
  readonly sideAStepCount: number;
  readonly sideBStepCount: number;
  readonly stopped: boolean;
  readonly stoppedBy?: 'a' | 'b';
  readonly messageHistory: Message[];
  readonly abortSignal: AbortSignal;
  forceTurn(side: 'a' | 'b'): void;
  stop(): void;
}

/**
 * A scheduled effect pending execution.
 */
interface ScheduledEffect {
  id: string;
  name: string;
  args: Record<string, unknown>;
  scheduledAt: number;  // microseconds
  createdAt: number;    // microseconds
}