How PrimeAgent Manages Asynchronous Operations: Event Streams and Cancellation

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. 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, 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, packages/ai/src/providers/anthropic.ts, or 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 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 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:

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:

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 provides the core async iterator implementation, buffering partial messages and exposing them via typed events.
• stream() and streamSimple() in 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 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. 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 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}