# How the omp Task Subagent System Works: Typed Results and Parallel Execution

> Explore the omp task subagent system for isolated subprocess execution and typed results. Learn how it parallelizes work items with configurable concurrency limits for efficient processing.

- Repository: [Can Bölük/oh-my-pi](https://github.com/can1357/oh-my-pi)
- Tags: internals
- Published: 2026-05-21

---

**The omp task subagent system executes agent definitions in isolated subprocesses, returning strongly typed results through structured schemas while parallelizing work items across configurable concurrency limits.**

The **OH-My-Pi** (`omp`) command-line framework orchestrates complex coding workflows through its sophisticated task execution architecture. When you invoke the `omp task` command, the system spawns specialized **subagents** to process work items concurrently, enforcing strict type safety through TypeScript interfaces while managing execution flow through configurable worker pools.

## Task Definitions and Typed Result Schemas

The foundation of the system rests on strict TypeScript interfaces defined in **[`packages/coding-agent/src/task/types.ts`](https://github.com/can1357/oh-my-pi/blob/main/packages/coding-agent/src/task/types.ts)**. This file establishes the contracts that govern how tasks parameterize subagents and how results return to the caller.

The **`SingleResult`** interface serves as the primary return type for every subagent execution, capturing exit codes, output streams, token usage, and extracted tool data:

```typescript
export interface SingleResult {
  index: number;
  id: string;
  agent: string;
  task: string;
  exitCode: number;
  output: string;
  stderr: string;
  truncated: boolean;
  durationMs: number;
  tokens: number;
  // Strongly typed fields ensure downstream reliability
}

```

These type definitions enable the framework to validate results at compile time and provide IntelliSense for downstream UI components consuming task output.

## Executing Subagents with runSubprocess

The **`runSubprocess`** function in **[`packages/coding-agent/src/task/executor.ts`](https://github.com/can1357/oh-my-pi/blob/main/packages/coding-agent/src/task/executor.ts)** manages the lifecycle of individual subagent executions. This asynchronous function creates isolated **`AgentSession`** instances that inherit parent configuration while maintaining separation of concerns.

**Session initialization** involves several critical steps:

- **Model registry inheritance**: The subagent receives access to the parent's model configurations and telemetry settings.
- **System prompt injection**: The framework renders [`subagent-system-prompt.md`](https://github.com/can1357/oh-my-pi/blob/main/subagent-system-prompt.md) from **`packages/coding-agent/src/prompts/system/`** and injects it into the new session context.
- **Progress streaming**: Real-time events flow through `TASK_SUBAGENT_PROGRESS_CHANNEL` and `TASK_SUBAGENT_EVENT_CHANNEL` via the `EventBus`, enabling live UI updates during execution.

**Yield enforcement** ensures subagents complete properly. If a subagent fails to call the `yield` tool, the system sends reminders using [`subagent-yield-reminder.md`](https://github.com/can1357/oh-my-pi/blob/main/subagent-yield-reminder.md) up to three times before forcing termination.

**Result finalization** occurs through **`finalizeSubprocessOutput`**, which validates output against optional JSON schemas and injects warning constants (`SUBAGENT_WARNING_NULL_YIELD`, `SUBAGENT_WARNING_MISSING_YIELD`) when appropriate:

```typescript
export async function runSubprocess(options: ExecutorOptions): Promise<SingleResult> {
  const done = await runSubagent();
  const finalized = finalizeSubprocessOutput({
    rawOutput,
    exitCode,
    stderr,
    doneAborted: Boolean(done.aborted),
    outputSchema,
  });
  // Artifact writing and SingleResult construction...
}

```

## Parallel Execution and Concurrency Control

When tasks contain multiple items, the system leverages **`mapWithConcurrencyLimit`** from **[`packages/coding-agent/src/task/parallel.ts`](https://github.com/can1357/oh-my-pi/blob/main/packages/coding-agent/src/task/parallel.ts)** to manage worker pools efficiently. This generic utility accepts a concurrency limit and distributes work across a fixed number of parallel workers.

The implementation creates a shared index counter that workers increment atomically:

```typescript
export async function mapWithConcurrencyLimit<T, R>(
  items: T[],
  concurrency: number,
  fn: (item: T, index: number) => Promise<R>,
  signal?: AbortSignal,
): Promise<ParallelResult<R>> {
  const worker = async () => {
    while (true) {
      if (workerSignal.aborted) return;
      const index = nextIndex++;
      if (index >= items.length) return;
      try {
        results[index] = await fn(items[index], index);
      } catch (error) {
        if (!workerSignal.aborted) {
          abortController.abort(); // Fail-fast on errors
          rejectFirst(error);
        }
      }
    }
  };
  // Worker pool orchestration...
}

```

**Abort handling** provides graceful degradation. When an external `AbortSignal` triggers or any worker throws a non-abort error, the pool immediately stops scheduling new work while preserving results from completed tasks.

## Practical Implementation Examples

### Running a Single Subagent

For programmatic use, import `runSubprocess` to execute individual agents with full type safety:

```typescript
import { runSubprocess } from "@oh-my-pi/coding-agent";

const opts = {
  cwd: process.cwd(),
  agent: { name: "reviewer", systemPrompt: "...", source: "bundled" },
  task: "Review this PR",
  assignment: "Summarize the changes",
  index: 0,
  id: "sub-001",
  modelOverride: "gpt-4o-mini",
};

const result = await runSubprocess(opts);
console.log("Exit:", result.exitCode);
console.log("Typed data:", result.extractedToolData?.report_finding);

```

### Parallel Task Execution

Batch processing uses `mapWithConcurrencyLimit` to respect resource constraints:

```typescript
import { mapWithConcurrencyLimit, runSubprocess } from "@oh-my-pi/coding-agent";

async function runBatch(items: TaskItem[]) {
  const wrapper = (item: TaskItem, idx: number) => 
    runSubprocess({
      cwd: process.cwd(),
      agent: item.agent,
      task: item.task,
      assignment: item.assignment,
      index: idx,
      id: `sub-${idx}`,
    });

  const { results, aborted } = await mapWithConcurrencyLimit(
    items,
    4, // Maximum 4 concurrent subagents
    wrapper,
    abortController.signal,
  );

  return { results, aborted };
}

```

### Consuming Typed Results in the UI

The terminal UI (TUI) consumes `TaskProgress` events to render live status:

- **`currentTool`** and **`currentToolArgs`** display active tool invocations.
- **`tokens`** and **`contextTokens`** track resource utilization.
- **`extractedToolData`** provides structured access to findings like `report_finding` objects.

All progress fields derive from the `AgentProgress` interface defined in **[`types.ts`](https://github.com/can1357/oh-my-pi/blob/main/types.ts)**, ensuring type-safe consumption across the application.

## Summary

- **Type Safety**: The `SingleResult` interface in **[`types.ts`](https://github.com/can1357/oh-my-pi/blob/main/types.ts)** guarantees structured, predictable data returns from every subagent execution.
- **Isolation**: Each subagent runs in a dedicated `AgentSession` spawned by **`runSubprocess`** in **[`executor.ts`](https://github.com/can1357/oh-my-pi/blob/main/executor.ts)**, with independent system prompts and telemetry.
- **Concurrency**: **`mapWithConcurrencyLimit`** in **[`parallel.ts`](https://github.com/can1357/oh-my-pi/blob/main/parallel.ts)** implements worker pools with configurable limits and graceful abort semantics.
- **Observability**: Real-time progress streams through dedicated `EventBus` channels enable live monitoring of subagent states.
- **Yield Enforcement**: Automatic reminders and forced termination ensure subagents complete properly via the `finalizeSubprocessOutput` validation.

## Frequently Asked Questions

### What is the SingleResult interface in the omp task system?

**`SingleResult`** is the primary return type defined in **[`packages/coding-agent/src/task/types.ts`](https://github.com/can1357/oh-my-pi/blob/main/packages/coding-agent/src/task/types.ts)** that structures subagent execution outcomes. It contains metadata including `exitCode`, `output`, `stderr`, `durationMs`, and `tokens`, along with `extractedToolData` for typed tool outputs. This interface ensures that callers receive predictable, strongly typed data regardless of the specific agent implementation.

### How does the omp task system handle parallel execution limits?

The system uses **`mapWithConcurrencyLimit`** from **[`packages/coding-agent/src/task/parallel.ts`](https://github.com/can1357/oh-my-pi/blob/main/packages/coding-agent/src/task/parallel.ts)** to enforce concurrency constraints. This utility creates a worker pool of the specified size that atomically increments a shared index to claim work items. It supports **fail-fast** behavior where any worker error immediately aborts remaining tasks while preserving completed results, and respects external `AbortSignal` instances for cancellation.

### What happens when a subagent forgets to call yield?

According to the implementation in **[`executor.ts`](https://github.com/can1357/oh-my-pi/blob/main/executor.ts)**, subagents must invoke the `yield` tool to signal completion. If a subagent neglects this requirement, the system dispatches reminders using **[`subagent-yield-reminder.md`](https://github.com/can1357/oh-my-pi/blob/main/subagent-yield-reminder.md)** up to three times. After exceeding the retry threshold, **`finalizeSubprocessOutput`** injects `SUBAGENT_WARNING_MISSING_YIELD` into the result and forces termination, ensuring the parent task receives structured output despite the subagent failure.

### How are task schemas defined in oh-my-pi?

Task schemas derive from simple-mode configurations stored as JSON schemas (`taskSchema`, `taskSchemaNoIsolation`) in **[`packages/coding-agent/src/task/types.ts`](https://github.com/can1357/oh-my-pi/blob/main/packages/coding-agent/src/task/types.ts)**. These schemas define **`TaskParams`** for input validation and optional output schemas for result verification. The **`TaskToolSchemaInstance`** validates CLI payloads before spawning subagents, while **`finalizeSubprocessOutput`** enforces output schema compliance during result finalization.