feat(flows): per-item fan-out — run an array of flow work concurrently - #5302
Conversation
…ss agents Bumps tinyflows to the per-item fan-out engine, so agent/tool_call/ http_request/memory/sub_workflow nodes now accept `concurrency` and `on_item_error`. The node contracts delegate to the engine catalog, so the new config surface reaches the builder agents automatically. Adds the host-side guard the OpenHumanAgentRunner doc had deferred: a process-wide semaphore (8, OPENHUMAN_FLOWS_MAX_PARALLEL_AGENTS) around harness turns. The engine's per-node bound is not sufficient — several nodes and several runs can fan out at once, and each harness turn is a full agent session. Over-wide fan-outs are throttled, never rejected.
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
There was a problem hiding this comment.
senamakel has reached the 50-credit limit for trial accounts. To continue receiving code reviews, upgrade your plan.
📝 WalkthroughWalkthroughThe PR adds a process-wide semaphore for harness agent turns. The limit defaults to eight and supports environment configuration. Invalid values use the default. Harness execution waits for permits, and tests cover configuration and throttling. ChangesHarness concurrency control
Estimated code review effort: 3 (Moderate) | ~20 minutes Sequence Diagram(s)sequenceDiagram
participant FlowNode
participant Semaphore
participant run_via_harness
participant Harness
FlowNode->>run_via_harness: start agent turn
run_via_harness->>Semaphore: acquire permit
Semaphore-->>run_via_harness: grant permit or wait
run_via_harness->>Harness: build and run harness agent
Harness-->>run_via_harness: complete turn and release permit
Possibly related PRs
Suggested labels: Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
Comment |
coderabbitai
Bot
added
feature
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 304f526dab
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| /// Default 8; override with `OPENHUMAN_FLOWS_MAX_PARALLEL_AGENTS`. Waiting on a | ||
| /// permit is *not* an error — an over-wide fan-out is throttled to this width | ||
| /// rather than rejected, so the workflow still completes, just more slowly. |
There was a problem hiding this comment.
When a per-item agent batch needs more than 600 seconds of aggregate work at eight slots—for example, 40 items averaging over two minutes, or when unrelated runs already occupy the process-wide permits—the queued calls cannot merely complete more slowly: run_flow_body wraps the entire engine future in the fixed 600-second FLOW_RUN_TIMEOUT_SECS timeout (src/openhuman/flows/ops.rs), so it cancels the remaining items and marks the run failed. Reconcile this queue with the overall deadline or reject an infeasible fan-out before starting rather than promising that throttling always completes.
Useful? React with 👍 / 👎.
| // session. Acquired before any work so waiters queue rather than pile | ||
| // up half-built agents. `_permit` is released on drop at end of scope, | ||
| // including on every early return below. | ||
| let _permit = HARNESS_AGENT_SLOTS.acquire().await.map_err(|_| { |
There was a problem hiding this comment.
When more than eight harness turns are active, or another flow occupies the global slots, this await can block before timeout_secs is read and before the tokio::time::timeout around run_single is created. Consequently, an agent node configured with a 10-second wall-clock timeout can wait for minutes and then receive an additional 10 seconds, violating the existing caller-chosen timeout contract; acquire the permit inside the same timeout budget or subtract the queue wait from the remaining duration.
Useful? React with 👍 / 👎.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/openhuman/tinyflows/caps.rs`:
- Around line 740-743: Update max_parallel_harness_agents to reject parsed
values greater than Semaphore::MAX_PERMITS, falling back to
DEFAULT_MAX_PARALLEL_HARNESS_AGENTS just like invalid or zero values. Add a test
covering an out-of-range override and verify the fallback prevents invalid
semaphore permit counts.
- Around line 3867-3880: Update
the_harness_ceiling_throttles_rather_than_rejects to test queued acquisition:
start a third acquire_owned while both semaphore permits are held, verify it
remains pending, release one held permit, then await the queued permit
successfully. Remove the try_acquire assertions and preserve the existing
two-permit setup.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 35532cef-7a6c-4eb4-9b4b-5b35473bfefb
📒 Files selected for processing (3)
src/openhuman/flows/node_contracts.rssrc/openhuman/tinyflows/caps.rsvendor/tinyflows
| fn max_parallel_harness_agents(raw: Option<&str>) -> usize { | ||
| raw.and_then(|v| v.trim().parse::<usize>().ok()) | ||
| .filter(|n| *n > 0) | ||
| .unwrap_or(DEFAULT_MAX_PARALLEL_HARNESS_AGENTS) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Confirm the repository's Tokio version before using Semaphore::MAX_PERMITS.
rg -n -A4 -B2 '^name = "tokio"$' Cargo.lock
rg -n -C3 'max_parallel_harness_agents|Semaphore::new' src/openhuman/tinyflows/caps.rsRepository: tinyhumansai/openhuman
Length of output: 2574
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- Tokio declarations and feature sources ---'
rg -n -C4 'tokio\s*=|tokio::sync::Semaphore|MAX_PERMITS' --glob 'Cargo.toml' --glob 'Cargo.lock' --glob '*.rs'
printf '%s\n' '--- Relevant implementation and tests ---'
sed -n '700,750p' src/openhuman/tinyflows/caps.rs
sed -n '3835,3885p' src/openhuman/tinyflows/caps.rsRepository: tinyhumansai/openhuman
Length of output: 207
🌐 Web query:
Tokio 1.52.3 Semaphore MAX_PERMITS Semaphore::new panics permits greater than MAX_PERMITS
💡 Result:
In Tokio 1.52.3, the tokio::sync::Semaphore struct explicitly defines a MAX_PERMITS constant, which is set to usize::MAX >> 3 [1][2][3]. As documented, both the Semaphore::new(permits: usize) constructor and the add_permits(n: usize) method will panic if the number of permits requested or added exceeds this MAX_PERMITS limit [1][2][4]. This design is intended to prevent overflow and ensure the internal state of the semaphore remains valid [1][3].
Citations:
- 1: https://docs.rs/tokio/latest/src/tokio/sync/semaphore.rs.html
- 2: https://docs.rs/tokio/1/tokio/sync/struct.Semaphore.html
- 3: https://docs.rs/tokio/latest/tokio/sync/struct.Semaphore.html
- 4: https://github.com/tokio-rs/tokio/blob/c637f6e7/tokio/src/sync/semaphore.rs
🏁 Script executed:
#!/bin/bash
set -u
printf '%s\n' '--- Tokio references ---'
rg -n -C3 'tokio|Semaphore|MAX_PERMITS' --glob 'Cargo.toml' --glob 'Cargo.lock' --glob '*.rs' . || true
printf '%s\n' '--- Concurrency parser and initialization ---'
sed -n '710,750p' src/openhuman/tinyflows/caps.rs
printf '%s\n' '--- Parser tests ---'
sed -n '3838,3870p' src/openhuman/tinyflows/caps.rsRepository: tinyhumansai/openhuman
Length of output: 50379
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
import sys
usize_max = (1 << (8 * 8)) - 1 # repository targets use 64-bit usize
tokio_max_permits = usize_max >> 3
assert usize_max > tokio_max_permits
assert int("18446744073709551615") > tokio_max_permits
print(f"usize::MAX = {usize_max}")
print(f"Semaphore::MAX_PERMITS = {tokio_max_permits}")
print("usize::MAX is out of range for Semaphore::new")
PYRepository: tinyhumansai/openhuman
Length of output: 285
Reject permit counts above Semaphore::MAX_PERMITS.
usize::MAX passes the current check, but Semaphore::new panics for values above Semaphore::MAX_PERMITS. Bound the override and add a test for an out-of-range value.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/openhuman/tinyflows/caps.rs` around lines 740 - 743, Update
max_parallel_harness_agents to reject parsed values greater than
Semaphore::MAX_PERMITS, falling back to DEFAULT_MAX_PARALLEL_HARNESS_AGENTS just
like invalid or zero values. Add a test covering an out-of-range override and
verify the fallback prevents invalid semaphore permit counts.
Source: Coding guidelines
| #[tokio::test] | ||
| async fn the_harness_ceiling_throttles_rather_than_rejects() { | ||
| // The contract the fan-out relies on: an over-wide batch waits for a | ||
| // slot and still completes. If this ever started returning an error | ||
| // instead, `concurrency: "all"` over a large array would fail the run | ||
| // rather than run it more slowly. | ||
| let slots = tokio::sync::Semaphore::new(2); | ||
| let held = slots.acquire().await.expect("first permit"); | ||
| let held2 = slots.acquire().await.expect("second permit"); | ||
| assert_eq!(slots.available_permits(), 0); | ||
| // A third acquirer is pending, not refused. | ||
| assert!(slots.try_acquire().is_err(), "no permits left"); | ||
| drop(held); | ||
| assert!(slots.try_acquire().is_ok(), "a released slot is reusable"); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect whether this test creates a waiting acquisition.
rg -n -C4 'the_harness_ceiling_throttles|try_acquire|acquire_owned' src/openhuman/tinyflows/caps.rsRepository: tinyhumansai/openhuman
Length of output: 1222
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- target test and surrounding module ---'
sed -n '3825,3905p' src/openhuman/tinyflows/caps.rs
printf '%s\n' '--- semaphore usage in the module ---'
rg -n -C5 'Semaphore|acquire|try_acquire|concurrency' src/openhuman/tinyflows/caps.rs
printf '%s\n' '--- Tokio dependency declarations ---'
rg -n -C3 'tokio' Cargo.toml app/Cargo.toml app/src-tauri/Cargo.toml 2>/dev/null || trueRepository: tinyhumansai/openhuman
Length of output: 18456
🏁 Script executed:
python3 - <<'PY'
from pathlib import Path
import re
path = Path("src/openhuman/tinyflows/caps.rs")
text = path.read_text()
match = re.search(
r'async fn the_harness_ceiling_throttles_rather_than_rejects\(\) \{(.*?)\n \}',
text,
re.S,
)
assert match, "target test not found"
body = match.group(1)
print("test_has_try_acquire:", "try_acquire" in body)
print("test_has_queued_acquire:", bool(re.search(r"(acquire_owned|spawn|acquire\(\))", body)))
print("test_releases_before_final_try:", body.find("drop(held)") < body.rfind("try_acquire"))
print("production_acquires_waiting_permit:", bool(re.search(
r'HARNESS_AGENT_SLOTS\.acquire\(\)\.await', text
)))
PYRepository: tinyhumansai/openhuman
Length of output: 295
🏁 Script executed:
python3 - <<'PY'
from pathlib import Path
import re
text = Path("src/openhuman/tinyflows/caps.rs").read_text()
match = re.search(
r'async fn the_harness_ceiling_throttles_rather_than_rejects\(\) \{(.*?)\n \}',
text,
re.S,
)
assert match, "target test not found"
body = match.group(1)
lines = body.splitlines()
exact_acquire_calls = [
line.strip() for line in lines
if re.search(r'\bslots\.acquire\s*\(', line)
]
owned_acquire_calls = [
line.strip() for line in lines
if re.search(r'\b(?:slots\.)?acquire_owned\s*\(', line)
]
try_acquire_calls = [
line.strip() for line in lines
if re.search(r'\bslots\.try_acquire\s*\(', line)
]
zero_index = next(
i for i, line in enumerate(lines)
if "available_permits(), 0" in line
)
post_zero_exact_acquires = [
line.strip() for line in lines[zero_index + 1:]
if re.search(r'\bslots\.acquire\s*\(', line)
]
print("exact_slots_acquire_calls:", exact_acquire_calls)
print("acquire_owned_calls:", owned_acquire_calls)
print("try_acquire_calls:", try_acquire_calls)
print("exact_slots_acquires_after_zero_permits:", post_zero_exact_acquires)
assert len(exact_acquire_calls) == 2
assert not owned_acquire_calls
assert len(try_acquire_calls) == 2
assert not post_zero_exact_acquires
PYRepository: tinyhumansai/openhuman
Length of output: 531
Test a queued acquire, not try_acquire.
try_acquire() returns immediately and does not create a waiter. Start a third acquire_owned() while both permits are held, assert it remains pending, release a permit, then await it. The production path uses HARNESS_AGENT_SLOTS.acquire().await, so the test must cover this blocking behavior.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/openhuman/tinyflows/caps.rs` around lines 3867 - 3880, Update
the_harness_ceiling_throttles_rather_than_rejects to test queued acquisition:
start a third acquire_owned while both semaphore permits are held, verify it
remains pending, release one held permit, then await the queued permit
successfully. Remove the try_acquire assertions and preserve the existing
two-permit setup.
Source: Coding guidelines
1 participant
What
Adopts the tinyflows per-item fan-out engine (tinyhumansai/tinyflows#26) and adds the host-side guard it needs.
A flow node can now multiply an array of input into N concurrent units of work — array in, array out — with the width driven by data rather than by how many sibling nodes were hand-authored:
execution: per_itemalready existed; every implementation was a sequential loop. See the engine PR for the full config surface (concurrency,on_item_error) and its defaults.Changes here
vendor/tinyflowsgitlink bump. Depends on tinyhumansai/tinyflows#26 — merge that first.src/openhuman/flows/node_contracts.rs— no field duplication needed: this module already delegates totinyflows::catalogand only appends host notes, so the newexecution/concurrency/on_item_errorfields reachlist_node_kinds,get_node_kind_contract, and theworkflow_builderagent automatically. Adds one host note warning that a per-item agent node is a full harness turn each (own model context, own tool loop) and is therefore far costlier than a per-itemtool_call— fan out over an already-narrowed list, not a raw fetch.src/openhuman/tinyflows/caps.rs—HARNESS_AGENT_SLOTS, a process-wide semaphore (default 8,OPENHUMAN_FLOWS_MAX_PARALLEL_AGENTS) held across eachrun_via_harnessturn.This is the "future host-side per-item guard" the
OpenHumanAgentRunnerdoc already anticipated. The engine's per-nodeconcurrencyis not sufficient on its own: several nodes, and several concurrent runs, can each be fanning out at the same time, so without a shared ceiling one workflow could open hundreds of full agent sessions and exhaust memory or the provider rate limit. An over-wide fan-out waits for a slot rather than failing —concurrency: "all"over a large array runs more slowly, it does not error.A malformed or zero override falls back to the default rather than producing a zero-permit semaphore, which would deadlock every flow agent node in the process.
Concurrency safety review
run_agentwas already re-entrant and this PR documents why, so a future refactor does not quietly break it:Agent::from_config_for_agentbuilds a fresh agent per call, and the model override is stamped on a clonedConfig— concurrent calls never mutate shared state.escalated_origin_for_nested_harness, Flow agent nodes run nested harness tool loops without HITL approval #4595) andAPPROVAL_FLOW_RUN_CONTEXTare task-locals. The engine'sbuffer_unorderedpolls every item on the caller's task, so they propagate and HITL gating still applies to each fanned-out turn. A future change totokio::spawnper item would silently break this — noted in the code.Validation
GGML_NATIVE=OFF cargo check --lib— cleancargo test --lib openhuman::tinyflows::— 181 passedcargo test --lib openhuman::flows::node_contracts— 8 passedcargo fmt— cleanLive per-item progress
Adds
DomainEvent::FlowRunItemProgress(running, thensuccess/errorper item), bridged asflow:run_item_progress/flow_run_item_progress, anduseFlowRunProgressDetailedfolding it into per-node{ total, running, succeeded, failed }.A node with
execution: "per_item"and aconcurrencyis one step running N units of work, so the existingFlowRunProgressfires once for the whole node — the canvas could show it as busy but never how far through the batch it was.Two deliberate choices:
flow_run_stepsrow per item would multiply run-history storage by the fan-out width to record something no run view reads back. The durable row stays the source of truth for what the node did.A node that runs once emits no item frames, so consumers need no special case.
Frontend validation
pnpm typecheck— cleanpnpm test— 8985 passed, 2 skippedpnpm lint— 0 errors (the 100 warnings are pre-existing; none in the changed file)