fix(agent): make progress forwarding non-blocking#4356
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughThe CI workflows update feature-gated test and link-check validation. Learning and Composio coverage tests now exercise direct subscriber handling and action-tool argument validation followed by a successful retry. ChangesValidation and testing updates
Estimated code review effort: 2 (Simple) | ~10 minutes Possibly related PRs
Suggested labels: Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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/agent/harness/engine/progress.rs`:
- Around line 432-435: The progress forwarder test timeout is too tight and can
flake under slow CI scheduling. Update the timeout used around the forwarder in
the progress test assertions (the one using timeout(Duration::from_millis(...))
and the matching check in the other occurrence) to a slightly larger bound while
keeping the same expectation that the forwarder completes promptly.
🪄 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: 6c82366b-eaee-4f27-bca5-98b3345d8e21
📒 Files selected for processing (1)
src/openhuman/agent/harness/engine/progress.rs
sanil-23
left a comment
sanil-23
left a comment
There was a problem hiding this comment.
Thanks for digging into this @YOMXXX. I spent time reproducing #4269 end-to-end against a staging build, and I don't think this PR addresses the actual failure mode — sharing the evidence so we can redirect the fix.
Reproduction
Drove the real research agent via CDP against a staging build (staging-api.tinyhumans.ai), looping deep-research queries (arxiv paper + web search — the issue's own repro shape), watching the core [stream] breadcrumbs and the delta pipeline.
What actually happens
The hang is an upstream SSE stall on the read side, not delta backpressure. Representative trace:
21:40:40 [stream] OpenHuman POST .../chat/completions (stream=true, tools=14)
21:42:40 [stream] streaming chat failed, falling back to non-streaming: error decoding response body
Exactly 120s between the POST and the failure: the SSE body goes silent mid-response, the reader parks on bytes_stream.next().await (compatible_stream_native.rs), and it is only cut when the 120s total request timeout fires (surfacing as error decoding response body), after which the existing fallback-to-non-streaming recovers. I saw this stall class on 2 of the first 3 research runs, plus a sibling "Request timed out." backend error frame — same ~120s silent-stall regime.
Crucially, during the stall there are zero ProviderDeltas in flight — the provider is blocked reading bytes, not sending deltas. So making the delta forwarders non-blocking (this PR) has nothing to act on in that state.
Why this PR doesn't fix it
- It changes only
progress.rs(the delta forwarders). The stall is incompatible_stream_native.rs's read loop, which is untouched here. - The wedge it targets requires deltas flooding a stalled sink; the reproduced hang is the opposite — zero output, a silent read.
- Tracing the consumer chain end to end, the terminal consumer (
progress_bridge.rs) always drains (bounded synchronous ledger writes + a non-blockingbroadcastpublish), so progress backpressure is transient and self-clearing — it can't produce the indefinite hang #4269 describes. - The two added tests construct a capacity-1 sink that is pre-filled and never drained — a synthetic condition that doesn't occur in the running system (the bridge drains). "timeout before / pass after" demonstrates the mechanism, but not that it is #4269's cause.
One tradeoff to flag: switching text deltas to drop-on-Full means live streaming can drop token frames under load (final content is still correct via the aggregated response, but streaming fidelity regresses) — a real cost for a path that isn't the observed failure.
Why users see an indefinite hang
On default config the 120s total request timeout rescues the stall (a ~2-minute freeze — "cursor blinks, no output"). But #3856 advises operators to raise OPENHUMAN_INFERENCE_TIMEOUT_SECS up to 3600s for long research turns; with that raised, this exact stall hangs for up to an hour = the reported indefinite hang.
Suggested direction
A per-token inactivity watchdog on the SSE read — reset on every received chunk, ~60–90s, independent of the total request timeout, classified retryable so it falls back cleanly. That bounds the stall regardless of backend behavior and independent of the total-timeout knob operators are told to raise — which the total timeout can't do and this PR doesn't address. Happy to share the fix I'm testing.
Net: I'd suggest not closing #4269 with this PR. The non-blocking forwarder change may be reasonable defensive hardening in its own right, but it doesn't resolve the reproduced hang.
Resolve conflict in the streaming-delta forwarder after upstream moved it from engine/progress.rs to session/tool_progress.rs and removed SubagentProgress (sub-agent runs now share TurnProgress + spawn_delta_forwarder). Re-apply the tinyhumansai#4269 fix on the new structure: the forwarder pushes provider deltas with a non-blocking try_send (emit_stream_delta / emit_nonblocking) instead of a blocking send().await, so a saturated UI/progress channel drops transient delta frames rather than backpressuring the provider stream and wedging the turn in RESPONSE. Keep the spawn_delta_forwarder regression test; drop the SubagentProgress variant (type removed upstream, coverage subsumed).
YOMXXX
changed the title
|
Updated this PR after the latest main merge and sanil's reproduction notes. It no longer claims to close #4269: the title/body now scope it to defensive progress-forwarder hardening only, and #4269 remains open for the SSE read-side inactivity-watchdog fix. Also merged current upstream/main so CI can re-evaluate against the current tree. |
coderabbitai
Bot
added
agent
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
.github/workflows/ci-lite.yml (1)
813-844: 🎯 Functional Correctness | 🔴 Critical | ⚡ Quick winMissing dependency: add
rust-feature-gate-smoketo the final CI gate.The
rust-feature-gate-smokejob is missing from theneedsarray and theresultsvalidation matrix. Because it is orphaned frompr-ci-gate, a failure in the feature-gate contract checks (or the ungated asserts) will not block the PR from being merged.Please add the missing job to ensure the feature-gate smoke test acts as a strict guard.
🛠️ Proposed fix
needs: - changes - frontend-checks - rust-quality + - rust-feature-gate-smoke - rust-core-coverage - rust-tauri-coverage - scripts-tests - test-inventory - pester-install - tinycortex-tests - orch-ip-gate - feature-forwarding-gate if: always() runs-on: ubuntu-latest timeout-minutes: 15 steps: - name: Require PR CI jobs to pass run: | set -euo pipefail declare -A results=( ["Detect Changed Areas"]="${{ needs.changes.result }}" ["Frontend Checks"]="${{ needs['frontend-checks'].result }}" ["Rust Quality"]="${{ needs['rust-quality'].result }}" + ["Rust Feature-Gate Smoke"]="${{ needs['rust-feature-gate-smoke'].result }}" ["Rust Core Coverage"]="${{ needs['rust-core-coverage'].result }}" ["Rust Tauri Coverage"]="${{ needs['rust-tauri-coverage'].result }}" ["Scripts Self-Tests"]="${{ needs['scripts-tests'].result }}" ["Test Inventory"]="${{ needs['test-inventory'].result }}" ["PowerShell Install Test"]="${{ needs['pester-install'].result }}" ["TinyCortex Memory Tests"]="${{ needs['tinycortex-tests'].result }}" ["Orchestration IP Gate"]="${{ needs['orch-ip-gate'].result }}" ["Feature Forwarding Gate"]="${{ needs['feature-forwarding-gate'].result }}" )🤖 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 @.github/workflows/ci-lite.yml around lines 813 - 844, Add rust-feature-gate-smoke to the final CI gate’s needs list and to the results matrix in the “Require PR CI jobs to pass” step, using the job’s result for validation so feature-gate smoke failures block the PR.
🤖 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.
Outside diff comments:
In @.github/workflows/ci-lite.yml:
- Around line 813-844: Add rust-feature-gate-smoke to the final CI gate’s needs
list and to the results matrix in the “Require PR CI jobs to pass” step, using
the job’s result for validation so feature-gate smoke failures block the PR.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: c048908d-2e6f-4c2f-88f2-aae0338f4a70
📒 Files selected for processing (2)
.github/workflows/ci-lite.yml.github/workflows/pr-quality.yml
|
Clarifying the latest CodeRabbit pre-merge warnings: that review appears to be scoped to the final CI-maintenance commit only, not the full PR diff. The PR still contains the progress-forwarder hardening in The title/body no longer claim to close #4269. That issue remains open for the separate SSE read-side inactivity-watchdog fix called out in sanil-23’s reproduction notes. |
There was a problem hiding this comment.
🧹 Nitpick comments (1)
src/openhuman/learning/startup.rs (1)
274-298: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueUpdate test name, assertion, and avoid redundant polling.
Now that the test invokes the subscriber directly and bypasses the event bus, the name and assertion message no longer reflect its behavior. Furthermore, since the
handle(...).awaitcall completes synchronously rather than relying on async event bus delivery,wait_for_candidatesis no longer necessary and introduces an artificial 40ms settling delay.Consider renaming the test, updating the assertion, and checking the candidate count directly.
♻️ Proposed refactor
- #[tokio::test] - async fn learning_subscriber_fires_with_no_channel_configured() { + #[tokio::test] + async fn email_signature_subscriber_handles_canonicalized_event() { let source_id = unique_source_id("e2e"); let body = signature_body(); let expected = parse_signature(&body, &source_id, &source_id).len(); assert!( expected > 0, "signature body must yield at least one identity candidate" ); let event = DomainEvent::DocumentCanonicalized { source_id: source_id.clone(), source_kind: "email".to_string(), chunks_written: 1, chunk_ids: vec![format!("{source_id}-c1")], canonicalized_at: 0.0, body_preview: Some(body), }; crate::core::event_bus::EventHandler::handle(&EmailSignatureSubscriber, &event).await; - let got = wait_for_candidates(&source_id, expected).await; + let got = candidates_for(&source_id); assert_eq!( got, expected, - "email-signature subscriber must push the parsed identity candidates \ - with no channel configured anywhere (`#5003`)" + "email-signature subscriber must push the parsed identity candidates" ); }🤖 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/learning/startup.rs` around lines 274 - 298, Update learning_subscriber_fires_with_no_channel_configured to reflect direct subscriber invocation rather than event-bus delivery, revise the assertion message accordingly, and replace wait_for_candidates with a direct candidate-count check after handle(...).await so no polling delay remains.
🤖 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.
Nitpick comments:
In `@src/openhuman/learning/startup.rs`:
- Around line 274-298: Update
learning_subscriber_fires_with_no_channel_configured to reflect direct
subscriber invocation rather than event-bus delivery, revise the assertion
message accordingly, and replace wait_for_candidates with a direct
candidate-count check after handle(...).await so no polling delay remains.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 88a3d30d-6cc4-40c6-9418-9bf91608d98a
📒 Files selected for processing (1)
src/openhuman/learning/startup.rs
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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 `@tests/raw_coverage/composio_credentials_state_raw_coverage_e2e.rs`:
- Around line 310-311: Update the initial action_tool.execute call in the
contract prompt test to pass an empty JSON object, ensuring the first execution
omits the query argument before the existing valid-argument retry.
🪄 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: 3b49a97c-2cca-464d-a3d8-6d4c9d39c446
📒 Files selected for processing (1)
tests/raw_coverage/composio_credentials_state_raw_coverage_e2e.rs
2 participants
Summary
Scope note
This PR is defensive hardening for progress-channel backpressure only. It no longer claims to fix #4269. Sanil's reproduction points at an upstream SSE read-side silent stall, not delta forwarding backpressure, so #4269 should stay open for an SSE inactivity-watchdog fix in the provider stream path.
Validation
Previous branch validation before the main merge:
cargo fmt --manifest-path Cargo.toml --checkgit diff --checkCARGO_TARGET_DIR=/Users/liguanchen/Desktop/lgc/openhuman-gh4352/target GGML_NATIVE=OFF cargo test --manifest-path Cargo.toml --lib agent::harness::engine::progress::tests:: -- --nocaptureCARGO_TARGET_DIR=/Users/liguanchen/Desktop/lgc/openhuman-gh4352/target GGML_NATIVE=OFF cargo test --manifest-path Cargo.toml --lib openhuman::agent::harness::subagent_runner::ops::tests::typed_mode_forwards_child_text_and_thinking_deltas -- --exactCARGO_TARGET_DIR=/Users/liguanchen/Desktop/lgc/openhuman-gh4352/target GGML_NATIVE=OFF cargo check --manifest-path Cargo.tomlAfter the latest
upstream/mainmerge, PR CI should be the source of truth for the full updated tree.Impact
Related
AI Authored PR Metadata
Branch
fix/GH-4269-response-watchdogBehavior Changes
Duplicate / Superseded PR Handling
Summary by CodeRabbit