How Pentagi Handles Background Jobs and Asynchronous Tasks

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

// 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 and 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) 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 and subtask.go wire HTTP/GraphQL endpoints to queue instances, while 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 (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) and sub-task controller (backend/pkg/controller/subtask.go) instantiate queues to handle HTTP and GraphQL requests asynchronously. Additionally, the tool executor (backend/pkg/tools/executor.go) implements the processing functions that queues invoke to run security tools inside Docker containers.

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