# Workflow Quick Reference

> Copy-paste BHW patterns for AI assistants and humans: workflow skeleton, steps, transforms, fan-out, modules, agents, datasets, decisions, and config.

This page is the machine-readable cheat sheet for the BHW workflow language.
Every snippet below is canonical syntax. Prefer these shapes when suggesting
workflow edits. Details live on the linked pages.

## Skeleton

```bhw
discover = activity(
    name = "discover",
    triggers = [trigger.cron("0 0 * * *"), trigger.dispatch()],
    steps = [
        step.exec(
            command = ["subfinder", "-d", config.var("target_domain"), "-o", "subs.txt"],
            id = "enum",
        ),
    ],
)

workflow(name = "recon", activities = [discover])
```

- File must produce a `workflow(...)` value. Single-activity shorthand:
`workflow(name = ..., triggers = [...], steps = [...])`.
- Activities are wired by variable, not by `name` string:
`trigger.on(discover.succeeded())`, `discover.failed()`,
`discover.completed()`, `discover.artifacts.<name>`, `discover.when(cond)`.
- Events compose: `event.all(a, b)`, `event.any(a, b)`, nestable.

## Step identity and conditions

```bhw
scan = step.exec(command = ["nuclei", "-l", "live.txt"], id = "scan", name = "Nuclei scan")
recheck = step.exec(command = ["httpx", "-l", "live.txt"]).when(scan.succeeded())
cond = lambda run: run.steps.scan.conclusion == steps.Conclusion.SUCCEEDED
```

- `id` is the machine key (`run.steps.<id>` / `run.steps["<id>"]`); `name` is
the UI label. No `id` means no stable `run.steps` key.
- `run` fields: `run.ok` (bool), `run.inputs` (dict), `run.steps` (dict of
`step_result` with `status` / `conclusion`), `run.trigger`.
- Compare against `steps.Status` / `steps.Conclusion` (`SUCCEEDED`, `FAILED`,
`SKIPPED`). Default condition is `run.ok`.
- `.when(F)` is the fluent form of `cond = F`; `F` is a runtime predicate
`run -> bool` — a lambda or a step predicate such as `scan.succeeded()`.
- Most steps accept `id`, `name`, `cond`, `allow_failure`, `timeout`.
`step.spawn` accepts `name` and `timeout` only; its `allow_failure`
parameter is accepted but ignored.

## `step.script` vs `step.exec`

```bhw
step.script(content = "cat findings.txt | notify -provider slack")
step.script(content = lambda run: "subfinder -d {} -o subs.txt".format(run.inputs["domain"]))
step.exec(
    command = ["nuclei", "-l", "live.txt", "-severity", config.var("nuclei_severity")],
    env = {"GITHUB_TOKEN": config.secret("github_token")},
    stdin = path.file("subs-clean.txt"),
    stdout = path.file("live.txt"),
    stderr = path.file("errors.log"),
    cwd = path.directory("workdir"),
    timeout = duration.minutes(30),
)
```

- `script` = explicit shell (`shell = "bash"` default). `exec` = direct argv,
no shell; `command[0]` is the program.
- `exec` receives exactly what is declared: no implicit env, stdin, or prior
stdout. `stdin`/`stdout`/`stderr` must be `path.file(...)`;
`cwd` must be `path.directory(...)`. Non-zero exit fails unless
`allow_failure = True`.
- `command` elements: strings, step-output paths, `config.*` refs, module
input refs, lazy expressions. Nested list = optional argv group (expands
only when all referenced optional module inputs are bound). Bare optional
module input = dropped when unbound.

## Step outputs

A step declares what it publishes; consumers use the step result, never the
producer's file name. See [Step Outputs](/workflows/step-outputs).

```bhw
hosts = step.exec(
    command = ["subfinder", "-d", config.var("target")],
    stdout = path.file("hosts.txt"),
)
live = step.exec(command = ["httpx", "-l", hosts.output], stdout = path.file("live.txt"))

scan = step.exec(
    command = ["nmap", "-sV", "-oA", "scan", "example.com"],
    outputs = {"xml": path.file("scan.xml"), "gnmap": path.file("scan.gnmap")},
)
```

- `run.output` is the single declared output; `.outputs.<name>` addresses
several.
- `stdout = path.file(...)`, `step.transform(destination = ...)` and
`step.agent(output = ...)` expose that file as `run.output`.
- `outputs = {...}` declares files a tool writes itself; directory and glob
values capture trees and sets.
- Module outputs are read through the invocation, e.g.
`scan.outputs.findings` (or `scan.output` when the module declares one).
- `workflow(isolated_steps = True)` gives each data step its own private
filesystem; outputs are then the only data that crosses a step boundary.

## Durations

```bhw
duration.seconds(30)
duration.milliseconds(500)
duration.minutes(15)
duration.hours(1).minutes(30)
duration.days(7)
duration.parse("2h45m")
```

Strictly positive. Used for activity/step `timeout` and artifact `expires_in`.

## Dispatch inputs

```bhw
trigger.dispatch(inputs = {
    "domain": input.string(required = True, description = "Target domain"),
    "aggressive": input.bool(default = False, description = "Aggressive mode"),
})
```

Dispatch inputs are `input.string` / `input.bool` only. There is no
`input.file` dispatch input.

Read at runtime as `run.inputs["domain"]` (or
`lambda run: ...run.inputs["domain"]...`).

## Transforms

```bhw
step.transform(
    source = path.file("subs.txt").lines(),
    pipeline = transform.trim().lower().unique().sort(),
    destination = path.file("subs-clean.txt"),
)
```

- Sources: `.lines()` → `Items<string>`, `.text()` → `string`,
`.json()` → `json`, `.jsonl()` → `Items<record>`. Bare `path.file("x")`
without a parse is rejected ("raw file"). Explicit form:
`source = path.file("x")` + `pipeline = transform.lines()...`.
- String ops (element-wise): `trim`, `lower`, `upper`, `replace(from, to)`,
`regex(pattern)`, `regex_replace(pattern, replacement)`.
- Collection ops: `unique`, `take(n)`, `drop(n)`, `sort`, `reverse`,
`split(sep)`, `join(sep)`.
- Lambda ops (deferred, lambda or named `def`):
`map(lambda h: "{}:8080".format(h))`,
`filter(lambda row: row.severity == "high")`,
`flat_map(lambda row: row.urls)`.
- Memory model: element-wise/streaming/stateful are streaming;
`sort`/`reverse`/`join` buffer the whole sequence.
- Partitioning (terminal; only `partition.count(n)` / `partition.items(n)`,
bare ints rejected):
`transform.trim().unique().partition(partition.items(500))` with
`destination = path.file("batch-{partition_seq}.txt")`. Partitioned output
requires `{partition_seq}`; non-partitioned forbids it. Result types as
`Items<Batch<T>>`; materialize with `batch.as_file()`.
- Records: `row.field`, `row.host.lower()`, `row.urls[0]`.

## `step.for_each`

```bhw
step.for_each(
    items = path.file("subs-clean.txt").lines(),
    do = lambda sub: [step.exec(command = ["httpx", "-u", sub])],
    concurrency = 50,
)
step.for_each(
    items = path.glob(base = "batches", pattern = "scan-*.txt"),
    do = probe,   # def probe(target): return [...]
)
step.for_each(
    items = path.file("live.txt").lines().unique().partition(partition.count(20)),
    do = lambda batch: [step.exec(command = ["nuclei", "-l", batch.as_file()])],
)
```

- `items`: list, `path.glob(base, pattern)`, or any `Items<T>` expression.
- `do`: one-parameter function, deferred per item; no magic variable.
- `Batch<T>` is execution-local, not a file — call `.as_file()` for tools.
- Concurrency guide: CPU-bound ~5, network probes ~50, aggressive scanners
moderate.

## Concurrency

```bhw
step.parallel([[step.exec(command = ["subfinder", ...])], [step.exec(command = ["naabu", ...])]])
handle = step.spawn([step.exec(command = ["naabu", "-p", "-", ...])])
step.join([handle])
```

- `parallel` takes an outer list of branches (each an inner sequential list);
flat lists rejected; `parallel([])` is a no-op. Fails if any branch fails;
siblings are never cancelled. `cond`/`allow_failure` allowed on the barrier.
- `spawn` returns a handle (sync reference, not data); `join` blocks until all
succeed. `step.spawn` accepts `name` and `timeout`; its `allow_failure`
parameter is accepted but ignored. Handles never merge data — share files via workspace/artifacts.

## Modules

```bhw
domain = input.string(name = "domain", description = "Root domain")
subs = output.file(name = "subs", description = "Validated subdomains")

module(
    inputs = [domain],
    outputs = [subs],
    steps = [
        step.exec(command = ["subfinder", "-d", domain, "-o", "subs.txt"]),
        step.output(subs, path = path.file("subs.txt")),
    ],
)
```

```bhw
load("yourname/subdomain-enum", "subdomain_enum")
enum_run = step.module(subdomain_enum, inputs = {"domain": config.var("target_domain")})
step.exec(command = ["nuclei", "-l", enum_run.outputs.subs])
```

`step.module` only invokes modules imported with `load()` from the module
registry. A local `module(...)` value defined in the same file cannot be
passed to `step.module`.

- Module outputs are read through the **invocation value** using a declared
output name: `enum_run.outputs.subs`. A `load()` symbol is the module
definition, not an invocation, and has no `.outputs`; bind the invocation
first (`enum_run = step.module(...)`). Outputs are file references, usable as
module inputs, `step.exec` argv elements, and `artifact_entry(source = ...)`
sources. They are not `<activity>.artifacts.<name>`; that namespace only
exists after `step.upload_artifact`.
- A module runs in an **isolated workspace**: its `file`/`directory` inputs
are copied in, it cannot read the caller's other files, and only its declared
outputs are copied back (as `<module-path>.<name>`). Never read a producer's
internal file name; use the invocation's `.outputs.<name>`.
- `input.string/bool/file/directory(name, ...)` are public refs; `output.file`
likewise. `input.string(..., secret = True)` marks a credential, and callers
bind string inputs from literals or `config.var(...)`/`config.secret(...)`.
Each declared output assigned exactly once via `step.output`.
- Module files compile in module mode: no `workflow`/`activity` globals and no
project-context namespaces (`config`, `dataset`, `mcp`). Modules are
self-contained and bind such values through their declared inputs; agent
steps stay available (`provider.key`) but are explicit parameters.
Unbound optional inputs use their `default`.

## Agents

```bhw
step.agent(
    prompt = "Classify each finding as EXPLOITABLE, LIKELY_NOISE, or OUT_OF_SCOPE.",
    provider = provider.key("anthropic/personal"),
    model = model("default"),
    context = [
        agent.context.path(description = "Findings to triage.", path = path.file("findings.txt")),
        agent.context.text(description = "Target domain.", value = config.var("target_domain")),
        agent.context.artifact(description = "Prior results.", artifact = scan.artifacts.findings),
        agent.context.dataset(description = "Known hosts.", dataset = dataset.table("subdomains")),
    ],
    tools = [tool.shell(), mcp.server("github")],
    output = agent.output(path = path.file("agent-output.txt")),
)
```

- `provider` is required and must be enabled for the project
(`unknown_provider_key` otherwise). `model` is optional.
- `context` is a list of `agent.context.*` (dict form rejected):
`text(description, value)`, `path(description, path)`,
`artifact(description, artifact)`, `dataset(description, dataset)`.
Resources pass by reference, never injected wholesale.
- `tools`: `tool.shell()`, `mcp.server("name")` (preconfigured connections).
Output defaults to free-form text.

## Datasets

```bhw
step.update_dataset(
    table = "subdomains",
    source = path.file("subs-clean.txt").lines(),
    pipeline = transform.unique(),
    mapping = {"host": row.value, "source": "subfinder"},
    mode = "upsert",
)
step.download_dataset(
    table = "subdomains",
    destination = path.file("known.csv"),
    format = "csv",
)
```

- `mapping` keys are columns; values evaluate against `row` (`row.value` for
lines, `row.field` for JSONL). `table` names an existing project table (or
`dataset.table("name")` reference elsewhere).
- `step.download_dataset` exports the dataset's current logical state into the
workspace (`"csv"` default, or `"jsonl"`/`"json"`); the step exposes it as
`run.output`.
- `mode` is the string `"upsert"` or `"sync"`: `"upsert"` accumulates;
`"sync"` marks absent identities stale. Identity/reconciliation owned by
the dataset service. The `dataset.upsert()` / `dataset.sync()` values are
not accepted here.

## Decisions

```bhw
step.decision(
    state = path.file("findings.json"),
    questions = {
        "relevant": decision.noul("Is this finding relevant?"),
        "severity": decision.choice("Severity?", options = ["low", "medium", "high"]),
        "priority": decision.score("Priority?", min = 0, max = 3),
    },
    provider = "jevos",
    endpoint = config.var("jevos_endpoint"),
    api_key = config.secret("jevos_api_key"),
    output = path.file("decision.json"),
)
```

- Question types: `decision.noul(question)` (binary, optional confidence),
`decision.choice(question, options = [...])` (one of a fixed set),
`decision.score(question, min, max)` (bounded integer).
- `provider` is optional and selects the implementation: `"jevos"` (the
default, the built-in live jevos decision server reached at `endpoint`) or
`"mock"` (deterministic, no network, for tests). Any other name is rejected.
`endpoint`/`model`/`api_key` are optional and resolved at execution time;
`api_key` must be a `config.secret(...)` reference.
- jevos limits: a `choice` needs 2–26 options and a `score` needs 2–10 levels.
- The result is written to `output` as a typed JSON document keyed by question
name. It is model-produced data, not proof: answers are validated against the
declared questions before the result is written, and a self-reported
`probability` is not a calibrated probability.

## Config

```bhw
config.var("target_domain")
config.secret("github_token")
config.mcp("github")
"https://" + config.var("target_domain")
config.var("target_domain").trim()
```

- References only; resolved at execution time, never serialized. Secrets stay
sensitive through concat/trim/casing. Scope
(global/org/project/environment) hidden from the DSL.

## Artifacts and blobs

```bhw
step.upload_artifact(
    name = "subdomains",
    expires_in = duration.days(7),
    entries = [artifact_entry(source = "subs.txt", destination = "subs.txt")],
)
step.download_artifact(discover.artifacts.subdomains)
```