# How PrimeAgent Manages Asynchronous Operations: Event Streams and Cancellation

> Discover how PrimeAgent manages asynchronous operations using event streams and cancellation. Gain fine-grained control over LLM responses with cancellable, retry-aware async iterators.

- Repository: [Prime Intellect/prime-agent](https://github.com/PrimeIntellect-ai/prime-agent)
- Tags: internals
- Published: 2026-08-09

---

**PrimeAgent handles asynchronous LLM operations through a unified event-stream abstraction centered on `AssistantMessageEventStream`, which provides cancellable, retry-aware async iterators that normalize streaming responses across all providers while exposing fine-grained control via `StreamOptions`.**

PrimeAgent, developed by PrimeIntellect-ai, implements a robust asynchronous architecture to orchestrate streaming responses from diverse LLM providers. The system unifies disparate APIs—ranging from OpenAI to Anthropic to Amazon Bedrock—under a consistent event-stream interface that supports cancellation, automatic retries, and incremental message assembly. Understanding how PrimeAgent manages asynchronous operations reveals a type-safe design pattern that isolates provider-specific I/O while offering consumers either granular event access or simple promise-based resolution.

## The Core Abstraction: AssistantMessageEventStream

At the heart of PrimeAgent’s async handling lies the **`AssistantMessageEventStream`** class, implemented in [`packages/ai/src/utils/event-stream.ts`](https://github.com/PrimeIntellect-ai/prime-agent/blob/main/packages/ai/src/utils/event-stream.ts). This concrete async iterator emits a strictly typed sequence of events including `start`, `text_delta`, `thinking_delta`, `toolcall_*`, `done`, and `error`.

The stream internally buffers partial `AssistantMessage` objects, forwarding incremental updates to listeners while maintaining an accumulator for the final result. Consumers retrieve the completed message via the **`result()`** method, which returns a promise resolving to the full `AssistantMessage` once the stream emits a `done` event.

## Streaming Entry Points in stream.ts

The primary entry points for asynchronous operations reside in **[`packages/ai/src/stream.ts`](https://github.com/PrimeIntellect-ai/prime-agent/blob/main/packages/ai/src/stream.ts)**, which exports **`stream()`** and **`streamSimple()`**. These functions resolve the appropriate provider via `resolveApiProvider` and delegate the request to the provider’s specific implementation.

Both functions return an `AssistantMessageEventStream`, establishing a consistent contract regardless of the underlying LLM service. The `stream()` method accepts a `Model`, a `Context`, and optional **`StreamOptions`**, immediately returning the event stream without blocking the caller.

## Provider Implementation Pattern

Each LLM provider—such as those in **[`packages/ai/src/providers/openai-completions.ts`](https://github.com/PrimeIntellect-ai/prime-agent/blob/main/packages/ai/src/providers/openai-completions.ts)**, **[`packages/ai/src/providers/anthropic.ts`](https://github.com/PrimeIntellect-ai/prime-agent/blob/main/packages/ai/src/providers/anthropic.ts)**, or **[`packages/ai/src/providers/amazon-bedrock.ts`](https://github.com/PrimeIntellect-ai/prime-agent/blob/main/packages/ai/src/providers/amazon-bedrock.ts)**—creates an HTTP request or SDK call and feeds raw response chunks into an `AssistantMessageEventStream` instance.

Providers translate vendor-specific streaming formats into the unified event protocol. They check **`options.signal?.aborted`** before processing each chunk to honor cancellation requests, and they handle retryable errors according to the `maxRetries` and `maxRetryDelayMs` parameters defined in the options.

## Configuration and Control via StreamOptions

Fine-grained asynchronous control is defined in **[`packages/ai/src/types.ts`](https://github.com/PrimeIntellect-ai/prime-agent/blob/main/packages/ai/src/types.ts)** through the **`StreamOptions`** interface. This configuration object includes:

- **`signal`**: An `AbortSignal` enabling immediate cancellation of the underlying HTTP request or SDK call.
- **`timeoutMs`**: Client-side timeout thresholds that trigger automatic abortion.
- **`maxRetries`** and **`maxRetryDelayMs`**: Retry logic parameters that re-issue requests on transient failures.
- **`sessionId`**: Enables prompt caching for providers like OpenAI and Anthropic, allowing the remote service to reuse cached prompts and reduce latency.
- **`onPayload`** and **`onResponse`**: Callbacks for intercepting raw request payloads and HTTP response headers.

When a caller aborts via `signal`, the stream emits an **`error`** event with `stopReason: "aborted"`. If the provider exhausts its retry limit, the stream reports `stopReason: "error"` with detailed `errorMessage` metadata.

## Convenience Methods for Non-Streaming Use

For consumers requiring only the final response without incremental updates, **[`packages/ai/src/stream.ts`](https://github.com/PrimeIntellect-ai/prime-agent/blob/main/packages/ai/src/stream.ts)** exposes **`complete()`** and **`completeSimple()`**. These convenience wrappers invoke `stream()` and immediately await `result()`, returning a single `Promise<AssistantMessage>`.

This abstraction hides the event-stream plumbing when the caller does not need live token streaming, effectively treating the LLM call as a standard async function while retaining all cancellation and retry capabilities.

## Practical Implementation Examples

The following examples demonstrate streaming versus one-shot asynchronous patterns using PrimeAgent’s API.

**Streaming with Cancellation:**

```typescript
import { stream } from "packages/ai/src/stream.js";
import { getModel } from "packages/ai/src/models.js";

async function demoStreaming() {
  const model = await getModel("gpt-4o-mini");
  const ctx = { 
    messages: [{ 
      role: "user", 
      content: "Write a haiku.", 
      timestamp: Date.now() 
    }] 
  };

  const s = stream(model, ctx, {
    temperature: 0.7,
    signal: AbortSignal.timeout(30_000), // Auto-cancel after 30s
  });

  for await (const ev of s) {
    if (ev.type === "text_delta") {
      process.stdout.write(ev.delta); // Live token output
    }
  }

  const finalMsg = await s.result();
  console.log("\n---\nFull response:", finalMsg.content);
}

```

**One-Shot Request with Retry Configuration:**

```typescript
import { complete } from "packages/ai/src/stream.js";
import { getModel } from "packages/ai/src/models.js";

async function demoOneShot() {
  const model = await getModel("claude-3-5-sonnet");
  const ctx = { 
    messages: [{ 
      role: "user", 
      content: "Explain async/await in 2 sentences.", 
      timestamp: Date.now() 
    }] 
  };

  const assistantMsg = await complete(model, ctx, {
    maxTokens: 200,
    maxRetries: 3,
    onResponse: (resp) => console.log("Headers received:", resp.headers),
  });

  console.log("Assistant reply:", assistantMsg.content);
}

```

## Summary

- **`AssistantMessageEventStream`** in [`packages/ai/src/utils/event-stream.ts`](https://github.com/PrimeIntellect-ai/prime-agent/blob/main/packages/ai/src/utils/event-stream.ts) provides the core async iterator implementation, buffering partial messages and exposing them via typed events.
- **`stream()`** and **`streamSimple()`** in [`packages/ai/src/stream.ts`](https://github.com/PrimeIntellect-ai/prime-agent/blob/main/packages/ai/src/stream.ts) serve as the unified entry points, resolving providers and returning controllable event streams.
- **`StreamOptions`** in [`packages/ai/src/types.ts`](https://github.com/PrimeIntellect-ai/prime-agent/blob/main/packages/ai/src/types.ts) configures cancellation via `AbortSignal`, timeout handling, automatic retries, and session-aware caching.
- Provider implementations in `packages/ai/src/providers/` translate vendor-specific streaming protocols into the unified event format while respecting cancellation signals.
- **`complete()`** offers a promise-based interface for non-streaming use cases, internally managing the event stream lifecycle.

## Frequently Asked Questions

### How does PrimeAgent handle request cancellation?

PrimeAgent accepts an `AbortSignal` through `StreamOptions`, which providers check via `options.signal?.aborted` before processing each response chunk. When triggered, the underlying fetch or SDK call terminates immediately, and the `AssistantMessageEventStream` emits an error event with `stopReason: "aborted"`.

### What is the difference between `stream()` and `complete()`?

**`stream()`** returns an `AssistantMessageEventStream` that yields incremental events like `text_delta` and `toolcall_*`, allowing real-time UI updates. **`complete()`** returns a `Promise<AssistantMessage>` that resolves only when the full response is assembled, hiding the streaming internals for simpler async/await patterns.

### Where are retry and timeout configurations defined?

Retry logic and timeouts are specified in the **`StreamOptions`** interface located in [`packages/ai/src/types.ts`](https://github.com/PrimeIntellect-ai/prime-agent/blob/main/packages/ai/src/types.ts). Parameters include `maxRetries`, `maxRetryDelayMs`, and `timeoutMs`, which providers implement to govern HTTP request resilience and client-side deadline management.

### How do providers integrate with the event stream system?

Each provider creates an `AssistantMessageEventStream` instance and pushes parsed response chunks into it as typed events. Files like [`packages/ai/src/providers/openai-completions.ts`](https://github.com/PrimeIntellect-ai/prime-agent/blob/main/packages/ai/src/providers/openai-completions.ts) demonstrate how raw SSE (Server-Sent Events) or SDK streams convert into the standardized `text_delta` and `done` events consumed by the rest of the application.