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

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. 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:

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 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 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 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:

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 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:

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:

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:

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, ensuring type-safe consumption across the application.

Summary

• Type Safety: The SingleResult interface in 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, with independent system prompts and telemetry.
• Concurrency: mapWithConcurrencyLimit in 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 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 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, subagents must invoke the yield tool to signal completion. If a subagent neglects this requirement, the system dispatches reminders using 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. 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.

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