# Flows, tasks, and subtasks

> Execution hierarchy: Flow lifecycle and StatusType states, Task objectives, Generator and Refiner subtask plans, Assistant mode, and putUserInput, stopFlow, finishFlow boundaries.

- Repository: vxcontrol/pentagi
- GitHub: https://github.com/vxcontrol/pentagi
- Human docs: https://grok-wiki.com/public/docs/vxcontrol-pentagi-eca0cf4de5f5
- Complete Markdown: https://grok-wiki.com/public/docs/vxcontrol-pentagi-eca0cf4de5f5/llms-full.txt

## Source Files

- `backend/docs/flow_execution.md`
- `backend/pkg/controller/flow.go`
- `backend/pkg/controller/task.go`
- `backend/pkg/controller/subtask.go`
- `backend/pkg/controller/assistant.go`
- `backend/pkg/graph/schema.graphqls`
- `backend/pkg/server/services/flows.go`

---

---
title: "Flows, tasks, and subtasks"
description: "Execution hierarchy: Flow lifecycle and StatusType states, Task objectives, Generator and Refiner subtask plans, Assistant mode, and putUserInput, stopFlow, finishFlow boundaries."
---

PentAGI runs autonomous pentests as a **Flow → Task → Subtask** hierarchy under `backend/pkg/controller`. A `FlowWorker` owns a Docker-backed session and queues user input; each user objective becomes a `TaskWorker`; the Generator and Refiner agents decompose and replan ordered `SubtaskWorker` steps that the Primary Agent executes until barrier tools `done` or `ask` fire. Shared lifecycle values use GraphQL `StatusType` and Postgres enums `FLOW_STATUS`, `TASK_STATUS`, and `SUBTASK_STATUS`.

## Execution hierarchy

| Level | Runtime type | Persisted as | Role |
| --- | --- | --- | --- |
| Flow | `FlowWorker` | `flows` | Session boundary: provider, primary container, assistants, task queue, data under the flow work dir |
| Task | `TaskWorker` | `tasks` | One user objective (`input`, LLM-derived `title`, final `result`) |
| Subtask | `SubtaskWorker` | `subtasks` | One planned step (`title`, `description`, `result`) run via a Primary Agent message chain |
| Assistant | `AssistantWorker` | `assistants` | Interactive side channel on a flow; not in the Task/Subtask stack |

```text
FlowWorker  ── owns ──► TaskController
    │                      │
    │                      ├── TaskWorker (user objective)
    │                      │      └── SubtaskController
    │                      │             ├── GenerateSubtasks  (Generator agent)
    │                      │             ├── PopSubtask → SubtaskWorker.Run
    │                      │             └── RefineSubtasks   (Refiner agent)
    │
    └── AssistantWorker[]  (optional; independent msg chain)
```

Workers coordinate status **upward**: subtask → task → flow. Completing the last task does not finish the flow; it returns the flow to `waiting` so another task can start.

## StatusType lifecycle

GraphQL and the API expose one status enum for flows, tasks, subtasks, and assistants:

```graphql
enum StatusType {
  created
  running
  waiting
  finished
  failed
}
```

Database enums match (`FLOW_STATUS`, `TASK_STATUS`, `SUBTASK_STATUS`).

```mermaid
stateDiagram-v2
  [*] --> created: row inserted
  created --> running: worker starts work
  running --> waiting: ask / stop / interrupt / await input
  waiting --> running: putUserInput or resume
  running --> finished: success / Finish
  running --> failed: unrecoverable error
  waiting --> finished: Finish (force complete)
  finished --> [*]
  failed --> [*]
```

### Status meaning by level

| Status | Flow | Task | Subtask |
| --- | --- | --- | --- |
| `created` | Row just inserted; title/model may still be filled by provider setup | Row after `CreateTask`, before run | Planned by Generator/Refiner; not yet popped |
| `running` | Active task or generation in progress; UI should treat input as busy | Subtask loop executing | Primary Agent chain in `PerformAgentChain` |
| `waiting` | Ready for new task or answer to `ask` | Paused on a waiting subtask, or ready after prior completion | Barrier `ask`, cancel, or chain error reset |
| `finished` | Session closed via `finishFlow` / `Finish` | Reporter path marked success, or forced finish | Barrier `done` success |
| `failed` | Terminal failure path on flow row | Reporter `Success=false` | `PerformResultError` |

### Status back-propagation

- **Subtask → task**: `SubtaskStatusRunning` sets task `running`; `waiting` sets task `waiting`; terminal subtask statuses do not auto-finish the task (the task loop continues or reports).
- **Task → flow**: task `running` / `waiting` map to flow `running` / `waiting`; task `finished` or `failed` sets flow to **`waiting`** (accept next input), not flow `finished`.
- **Interrupt** (`context.Canceled` / `DeadlineExceeded`): incomplete task/subtask reset to `waiting` so the session can resume.

### Load and restart rules

| Entity | Loadable statuses | Notes |
| --- | --- | --- |
| Flow | `running`, `waiting` only | Other statuses return `ErrNothingToLoad` |
| Task | not `created` | Finished/failed load as completed; waiting/running reload subtasks |
| Subtask | finished, failed, waiting; running demoted | Mid-run subtasks are forced back to `created` then reloaded; pure `created` without chains is skipped |

On process start, `FlowController.LoadFlows` restores only active flows and resumes incomplete non-waiting tasks.

## Flow lifecycle

### Create

1. GraphQL `createFlow(modelProvider, input, resourceIds?)` or REST `POST /api/v1/flows/` with `CreateFlow` (`input`, `provider`, optional `functions`, `resource_ids`).
2. `FlowController.CreateFlow` → `NewFlowWorker`: insert flow (`status=created`, title `untitled`), bind provider, build tools executor, prepare Docker primary container, start `worker()` goroutine.
3. Unless assistant-only `dryRun`, first `PutInput` runs immediately and creates the first task.

Provider choice is BYOK: any configured provider name for the user (OpenAI, Anthropic, Gemini, Bedrock, Ollama, custom, DeepSeek, GLM, Kimi, Qwen, and so on). No hosted connector is required beyond keys and optional server URLs.

### Worker input loop

`flowWorker.worker` continues incomplete tasks after load, then reads from channel `input` (`flowInputTimeout` = 1s for early error reporting).

`processInput`:

1. If any task is incomplete and `waiting` → `task.PutInput` (answers `ask`) and `runTask`.
2. Else set flow `running`, cancel-safe context for generation, `CreateTask` (includes Generator), then `execTask` / `Run`.

Only one task cancel scope is active at a time (`taskST`, `taskWG`). `Stop` cancels that scope; stop during GenerateSubtasks returns the flow to `waiting` without fatal error.

### Finish vs stop

| Operation | GraphQL | REST `PUT /flows/{id}` | Effect |
| --- | --- | --- | --- |
| Stop current work | `stopFlow(flowId)` | `{ "action": "stop" }` | Cancels task context; waits up to `stopTaskTimeout` (5s); worker stays registered; status tends to `waiting` |
| End session | `finishFlow(flowId)` | `{ "action": "finish" }` | Cancels flow, closes input channel, finishes incomplete tasks/assistants, releases executor, sets flow `finished`, removes from controller map |
| Delete | `deleteFlow` | `DELETE /flows/{id}` | Attempts `Finish` if worker exists, deletes DB row and flow-scoped memory docs |

`Stop` does **not** set `finished`. After stop, use `putUserInput` / REST `action: input` for a new or resumed objective. `Finish` is terminal for the live worker.

## Tasks

A task is one user objective inside a flow.

### Creation path

1. `TaskWorker` via `NewTaskWorker`: `Provider.GetTaskTitle` → `CreateTask` (`status=created`) → publish `taskCreated` → log `MsglogTypeInput` → **`GenerateSubtasks`** (must return at least one subtask) → publish `taskUpdated`.
2. `Run` loops: `PopSubtask` → `SubtaskWorker.Run` → unless waiting, `RefineSubtasks` → repeat until no planned subtasks or soft cap.
3. `Provider.GetTaskResult` (Reporter agent) writes `result` and sets `finished` or `failed`; message log type `report`.

### Soft cap

`providers.TasksNumberLimit = 15`. The run loop continues while `len(ListSubtasks) < TasksNumberLimit+3` (in-memory workers for completed/refined steps). Generator/Refiner prompts receive `N` as max planned slots (refiner: `max(15 - completedCount, 0)`). Empty refine plan means “objectives met; stop planning.”

### Input while waiting

`TaskWorker.PutInput` requires `IsWaiting()` and forwards input to the first incomplete waiting subtask. Calling put input when the task is running returns an error at the task layer; flow-level input while no task is waiting **creates a new task**.

## Subtasks: Generator and Refiner

`SubtaskController` owns the plan.

| Method | Agent | Behavior |
| --- | --- | --- |
| `GenerateSubtasks` | Generator | LLM plan → insert all rows as `created`; empty plan is an error |
| `RefineSubtasks` | Refiner | After each successful subtask: delete remaining `created` rows, insert new plan; empty plan = no change / done |
| `PopSubtask` | — | First planned DB row; create `SubtaskWorker` + `PrepareAgentChain` if needed |
| `ListSubtasks` / `GetSubtask` | — | In-memory workers only |

### Subtask run results

`Provider.PerformAgentChain` returns:

| Result | Status | Cause |
| --- | --- | --- |
| `PerformResultDone` | `finished` | Primary Agent called barrier `done` |
| `PerformResultWaiting` | `waiting` | Primary Agent called barrier `ask` |
| `PerformResultError` | `failed` | Unrecoverable agent/tool failure |

`PutInput` on a subtask requires waiting, injects into the msg chain via `PutInputToAgentChain`, logs input, clears waiting, then a new `Run` continues the same chain (`EnsureChainConsistency` first).

### Planned vs completed

- **Planned**: `status=created` (and queue order from DB planned query).
- **Active**: `running` or `waiting` with a live worker and msg chain.
- **Done**: `finished` / `failed` with `result` text.

Assistant tool `patch_flow_subtasks` can replace the planned list via delta ops (add/remove/modify/reorder) while the flow is not mid-run in a conflicting way; obtain `task_id` from `get_flow_status`.

## Assistant mode

Assistants attach to a **flow**, not to a task/subtask. They use a separate message chain and log stream (`AssistantLog`).

### Create and call

| GraphQL | Behavior |
| --- | --- |
| `createAssistant(flowId, modelProvider, input, useAgents, resourceIds?)` | Create assistant; may create/load flow worker with `dryRun` if needed; inject `FlowWorker` into provider |
| `callAssistant(...)` | Queue input (`assistantInputTimeout` = 2s) |
| `stopAssistant` / `deleteAssistant` | Cancel run or remove assistant |

`useAgents`:

- `true` — delegation tools (`search`, `pentester`, `coder`, `advice`, `memorist`, `maintenance`) plus direct tools.
- `false` — direct tools and search/memory only; no specialist agent delegation.

Assistants use `nil` task/subtask IDs for many agent handlers and return natural language to the user (not Primary-Agent barrier semantics).

### Flow-management tools (assistant only)

When `AssistantProvider.SetFlowWorker` is set:

| Tool | Purpose |
| --- | --- |
| `get_flow_status` | Flow/tasks/subtasks snapshot; optional verbose agent messages |
| `stop_flow` | Cancel running task (handler timeout ~15s); reports post-stop state |
| `submit_flow_input` | Deliver text when flow is `waiting` (answer `ask` or start new task); rejects when still running |
| `patch_flow_subtasks` | Rewrite planned subtasks for a task ID |

These mirror user control surfaces but run from the assistant tool loop.

## Control boundaries: putUserInput, stopFlow, finishFlow

### putUserInput / input

**GraphQL:** `putUserInput(flowId, input, modelProvider?, resourceIds?)`  
**REST:** `PUT /api/v1/flows/{flowID}` with `"action":"input"`, `"input":"..."`, optional provider and `resource_ids`.

Requires live worker (`GetFlow`); optional mid-session provider switch; copies user resources into flow FS and container when permitted.

| Flow state | Effect |
| --- | --- |
| Incomplete task `waiting` | Resume that task (answer `ask`) |
| No waiting incomplete task | Create new task from input (Generator runs first) |
| Worker finished / not in map | Error `flow not found` / not editable |

Permission: `flows.edit` (scoped by flow ownership unless admin).

### stopFlow

Cancels the **current task context** only. Does not finish assistants, does not release Docker, does not set flow `finished`. Use when the agent is looping or you need to change direction; then `putUserInput` or assistant `submit_flow_input`.

### finishFlow

Terminal session close: stop input loop, complete unfinished tasks/assistants as finished, release executor (containers), persist `FlowStatusFinished`, drop worker from controller. Further automation needs a new flow (or reload paths that only accept `running`/`waiting` will not attach).

### Permission summary

| Mutation | Privilege |
| --- | --- |
| `createFlow` | `flows.create` |
| `putUserInput`, `stopFlow`, `finishFlow`, `renameFlow` | `flows.edit` |
| `deleteFlow` | `flows.delete` |
| Assistants create/call/stop | assistant/flow edit privileges (see auth docs) |

## Human-in-the-loop: `ask`

Barrier tool `ask` is available on the Primary Agent when `ASK_USER=true` (`config.AskUser`, default **false**). Flow:

1. Agent calls `ask` → `PerformResultWaiting`.
2. Subtask/task/flow become `waiting`.
3. User answers via `putUserInput` / REST `input` / assistant `submit_flow_input`.
4. Input is applied to the agent chain; subtask `Run` continues.

With `ASK_USER=false`, agents cannot pause for user Q&A; they must complete or fail without that barrier.

## API surface

### GraphQL (primary control plane)

| Operation | Kind | Notes |
| --- | --- | --- |
| `createFlow` | Mutation | Returns `Flow` |
| `putUserInput` | Mutation | `ResultType` success/error |
| `stopFlow` / `finishFlow` / `deleteFlow` / `renameFlow` | Mutation | |
| `flow` / `flows` / `tasks(flowId)` | Query | Tasks include nested `subtasks` |
| `flowCreated` / `flowUpdated` / `taskCreated` / `taskUpdated` | Subscription | Real-time UI |

### REST under `/api/v1`

| Method | Path | Role |
| --- | --- | --- |
| `GET` | `/flows/`, `/flows/{id}`, `/flows/{id}/graph` | List, detail, tasks+subtasks graph |
| `POST` | `/flows/` | Create (`CreateFlow`) |
| `PUT` | `/flows/{id}` | Patch: `stop` \| `finish` \| `input` \| `rename` |
| `DELETE` | `/flows/{id}` | Delete |
| `GET` | `/flows/{id}/tasks/`, `.../tasks/{taskID}`, `.../graph` | Task views |
| `GET` | `/flows/{id}/subtasks/`, `.../tasks/{taskID}/subtasks/` | Subtask views |

Swagger: `/api/v1/swagger`.

### Example: create and drive a flow (GraphQL)

```graphql
mutation Create($provider: String!, $input: String!) {
  createFlow(modelProvider: $provider, input: $input) {
    id
    status
    title
  }
}

mutation Input($flowId: ID!, $input: String!) {
  putUserInput(flowId: $flowId, input: $input)
}

mutation Stop($flowId: ID!) {
  stopFlow(flowId: $flowId)
}

mutation Finish($flowId: ID!) {
  finishFlow(flowId: $flowId)
}
```

### Example: REST patch input

```json
{
  "action": "input",
  "input": "Continue with credential stuffing on the staging API only.",
  "provider": "openai"
}
```

## Timing and operational limits

| Constant / env | Value | Where |
| --- | --- | --- |
| `TasksNumberLimit` | 15 | Max planned subtask budget per task |
| `flowInputTimeout` | 1s | Early error wait after queueing flow input |
| `assistantInputTimeout` | 2s | Assistant input queue |
| `stopTaskTimeout` | 5s | `FlowWorker.Stop` wait for task WG |
| Flow mgmt tool ops | ~15s | Assistant `stop_flow` / `submit_flow_input` handlers |
| `ASK_USER` | default `false` | Enables Primary Agent `ask` barrier |

Provider/model selection is per flow (and optionally switched on later input). Specialist agent models come from provider YAML / UI profiles; see provider configuration docs.

## Troubleshooting

| Symptom | Likely cause | Action |
| --- | --- | --- |
| `putUserInput` fails with flow not found | Worker finished, never created, or process restarted without loadable status | Check flow status; only `running`/`waiting` reload; create a new flow if `finished` |
| Input ignored / creates extra task | No task in `waiting`; input starts a new objective | Use input only when status is `waiting` if you mean to answer `ask` |
| Stuck `running` | Agent loop, long terminal tool, or hung LLM | `stopFlow`, inspect toolcall/msg logs, then new input |
| `stop` timeout | Task did not exit within 5s | Retry stop; check container/terminal; finish only if abandoning session |
| No subtasks generated | Generator returned empty / provider error | Fix provider keys/models; inspect agent logs for generator chain |
| `ask` never appears | `ASK_USER` false | Set `ASK_USER=true` and recreate flow |
| Subtask restarts after process crash | Running subtask demoted to `created` on load | Expected; chain may re-prepare |
| Assistant cannot `submit_flow_input` | Flow still `running` | `stop_flow` first, then submit |

## Next

<CardGroup>
  <Card title="Agents and supervision" href="/agents-and-supervision">
    Primary, Generator, Refiner, Reporter, specialists, execution monitor, and tool-call limits.
  </Card>
  <Card title="Tools and sandbox" href="/tools-and-sandbox">
    Barrier tools done/ask, Docker terminal and file tools, timeouts, and default images.
  </Card>
  <Card title="GraphQL API" href="/graphql-api">
    Full Query, Mutation, and Subscription surface for flows, tasks, and live logs.
  </Card>
  <Card title="REST API" href="/rest-api">
    Gin routes under /api/v1 for flows, tasks, subtasks, and patch actions.
  </Card>
  <Card title="Tools reference" href="/tools-reference">
    Named registry including assistant flow-management tools and argument shapes.
  </Card>
  <Card title="Sample pentest prompts" href="/sample-pentest-prompts">
    Ready flow inputs from examples/prompts for first successful engagements.
  </Card>
</CardGroup>