# How Pentagi Handles Background Jobs and Asynchronous Tasks

> Discover how Pentagi manages background jobs and async tasks using a generic queue, configurable worker pools, and a done-context chain for ordered results. Optimize your pentesting operations.

- Repository: [VXControl/pentagi](https://github.com/vxcontrol/pentagi)
- Tags: internals
- Published: 2026-03-21

---

**Pentagi processes long-running penetration-testing operations through a lightweight generic queue system that uses configurable worker pools and a done-context chain to guarantee ordered results.**

The open-source penetration testing platform **vxcontrol/pentagi** executes time-consuming security assessments—such as Docker-based tool runs and stream processing—through a robust background job architecture. This system decouples HTTP/GraphQL request handling from actual computation, allowing the application to manage asynchronous tasks efficiently while maintaining strict result ordering for tool output streams.

## Core Queue Abstraction

At the heart of Pentagi's concurrency model sits a generic queue implementation defined in [`backend/pkg/queue/queue.go`](https://github.com/vxcontrol/pentagi/blob/main/backend/pkg/queue/queue.go). The system abstracts background processing through a minimal interface that supports lifecycle management and state inspection.

### The Queue Interface

The `queue.Queue` interface defines the contract for any background processor. It requires four methods: `Instance()` for identification, `Running()` for state checks, and `Start()` and `Stop()` for lifecycle control. This interface appears at lines 20–25 of the queue implementation file.

### Typed Queue Creation

The `queue.NewQueue` function instantiates a typed queue with configurable concurrency. It accepts an input channel, an output channel, a worker count (defaulting to 10), and a user-supplied processing function. This factory function, located at lines 48–69, wires together the communication channels and worker pool that drive all asynchronous execution in the system.

## Worker Pool Architecture

When `Start()` is invoked, the queue initializes a cancellable **root context** and launches a pool of worker goroutines. These workers read from an internal buffered channel, execute the supplied `process` function against each incoming message, and push results to the output channel.

Each worker logs its lifecycle events using **logrus** for observability, as implemented in the `worker` method (lines 35–44). The spawning logic and goroutine management reside in the `Start` method (lines 94–102), which creates the specified number of workers that concurrently consume tasks while maintaining thread-safe operations.

## Ordered Processing Guarantee

Unlike standard worker pools that complete jobs in arbitrary order based on execution time, Pentagi implements a **done-context chain** to preserve submission order. This is critical for penetration testing workflows where tool output streams must appear in sequence.

The `reader` goroutine (lines 78–87) creates a new `context.Context` for every incoming message, passing the previous message's `doneCtx` as `lastDoneCtx`. Workers block on `msg.doneCtx.Done()` before transmitting results (lines 53–56), ensuring each message finishes processing only after its predecessor has been fully handled.

## Integration Points

Pentagi's queue system integrates with higher-level controllers to handle distinct asynchronous workflows across the application.

### Task and Sub-task Controllers

The **task controller** ([`backend/pkg/controller/task.go`](https://github.com/vxcontrol/pentagi/blob/main/backend/pkg/controller/task.go)) receives incoming HTTP and GraphQL requests, creates channels for incoming work, and wires them to dedicated queue instances. Similarly, the **sub-task controller** ([`backend/pkg/controller/subtask.go`](https://github.com/vxcontrol/pentagi/blob/main/backend/pkg/controller/subtask.go)) employs identical queue wiring for nested job execution, enabling concurrent tool runs while preserving ordering within individual flows.

### Tool Execution

Individual security tools—such as Traversaal, DuckDuckGo, Browser, and Executor—run inside the queue's `process` function. The processing function receives tool-specific requests and executes them within Docker sandboxes, returning structured results back through the queue's output channel. The [`backend/pkg/tools/executor.go`](https://github.com/vxcontrol/pentagi/blob/main/backend/pkg/tools/executor.go) file contains the implementation details for this Docker-orchestrated execution logic.

## Lifecycle Management

The queue provides explicit control over background job lifecycles through coordinated start and stop operations.

Calling `Start()` spawns the worker pool and reader goroutine, returning an error only if the queue is already running. The `Stop()` method cancels the root context, closes the internal communication channel, and waits for all workers to exit using `wg.Wait()`. This graceful shutdown logic appears in the queue implementation's lifecycle management section (lines 11–24), ensuring no goroutines leak when flows or sub-tasks complete.

## Implementation Example

The following pattern demonstrates how Pentagi creates and manages a queue for tool execution within a flow:

```go
// Input channel receives tool requests (type ToolRequest)
// Output channel receives processed results (type ToolResult)
input := make(chan ToolRequest)
output := make(chan ToolResult)

// Process function executes a tool inside Docker and returns the result
process := func(req ToolRequest) (ToolResult, error) {
    // … Docker execution logic …
}

// Create a queue with 5 workers
q := queue.NewQueue(input, output, 5, process)

// Start background processing
if err := q.Start(); err != nil {
    log.Fatalf("cannot start queue: %v", err)
}

// Feed work
go func() {
    for _, r := range pendingRequests {
        input <- r
    }
    close(input) // signals end of work
}()

// Consume results (preserves order)
for res := range output {
    handleResult(res)
}

// When the flow finishes
q.Stop()

```

This pattern appears throughout [`backend/pkg/controller/task.go`](https://github.com/vxcontrol/pentagi/blob/main/backend/pkg/controller/task.go) and [`backend/pkg/controller/subtask.go`](https://github.com/vxcontrol/pentagi/blob/main/backend/pkg/controller/subtask.go), providing a consistent model for handling background jobs across the codebase.

## Summary

- Pentagi uses a generic **queue abstraction** ([`backend/pkg/queue/queue.go`](https://github.com/vxcontrol/pentagi/blob/main/backend/pkg/queue/queue.go)) with a configurable worker pool to decouple request handling from long-running security tool execution.
- The **done-context chain** mechanism guarantees that results are delivered in the same order they were submitted, essential for ordered tool output streams.
- Worker pools default to **10 concurrent workers** but can be tuned per queue instance via `queue.NewQueue`.
- **Logrus integration** provides observability into worker lifecycle events for debugging and monitoring.
- The **root context** pattern enables graceful cancellation and shutdown through the `Stop()` method, preventing resource leaks.
- Controllers in [`task.go`](https://github.com/vxcontrol/pentagi/blob/main/task.go) and [`subtask.go`](https://github.com/vxcontrol/pentagi/blob/main/subtask.go) wire HTTP/GraphQL endpoints to queue instances, while [`executor.go`](https://github.com/vxcontrol/pentagi/blob/main/executor.go) implements the Docker-based processing functions.

## Frequently Asked Questions

### How does Pentagi ensure tool output streams remain ordered?

Pentagi implements a **done-context chain** where each message carries its own `doneCtx`. Workers block on the previous message's context before emitting results, as seen in [`backend/pkg/queue/queue.go`](https://github.com/vxcontrol/pentagi/blob/main/backend/pkg/queue/queue.go) (lines 53–56). This ensures that even if later tools finish faster, their results wait for preceding operations to complete, preserving stream order.

### What is the default worker pool size in Pentagi queues?

By default, Pentagi initializes queues with **10 workers**. This value is configurable through the worker count parameter in `queue.NewQueue` (lines 48–69), allowing developers to tune concurrency based on expected workload and resource constraints.

### How does Pentagi handle cancellation of background jobs?

Cancellation propagates through a **root context** created during `Start()`. When `Stop()` is called, it cancels this context and closes the internal channel, signaling all workers to exit. The method then waits for the worker wait group (`wg.Wait()`) to drain, ensuring graceful shutdown without leaking goroutines.

### Which Pentagi components use the queue system for asynchronous tasks?

The **task controller** ([`backend/pkg/controller/task.go`](https://github.com/vxcontrol/pentagi/blob/main/backend/pkg/controller/task.go)) and **sub-task controller** ([`backend/pkg/controller/subtask.go`](https://github.com/vxcontrol/pentagi/blob/main/backend/pkg/controller/subtask.go)) instantiate queues to handle HTTP and GraphQL requests asynchronously. Additionally, the **tool executor** ([`backend/pkg/tools/executor.go`](https://github.com/vxcontrol/pentagi/blob/main/backend/pkg/tools/executor.go)) implements the processing functions that queues invoke to run security tools inside Docker containers.