fix(harness): detach memory flush and maintenance from agent call response - #2617
Open
birdie7761 wants to merge 2 commits into
Open
fix(harness): detach memory flush and maintenance from agent call response#2617birdie7761 wants to merge 2 commits into
birdie7761 wants to merge 2 commits into
Conversation
…ponse MemoryFlushMiddleware and MemoryMaintenanceMiddleware appended their LLM-backed work via concatWith onto the returned Flux. Because ReActAgent.callInternal ends with takeLast(1), which can't emit until the upstream signals onComplete, callers consuming the agent response to completion (blockLast, takeLast(1), WebFlux controllers awaiting Mono<Msg>) ended up waiting for the full memory flush LLM call (19-27s per call) and consolidation LLM call (~44s first run). The original implementation used doOnComplete (fire-and-forget). A later commit (PR agentscope-ai#1802, RC4) swapped it for concatWith, introducing this regression. This fix restores the fire-and-forget behavior: both middlewares now subscribe their work independently of the returned Flux via doOnComplete().subscribe(), so the Flux completes as soon as the underlying agent call does. Additional engineering safeguards: - Both middlewares implement AutoCloseable and track pending Disposables in a Set guarded by synchronized(pending); a pre-subscribe closed check prevents new work, a post-subscribe check disposes any subscription that raced with close(). HarnessAgent.close() drains them (bounded by 5s) before workspace teardown to prevent races with temp dir deletion in tests and CLI shutdown. - captureFlushRequest is wrapped in try/catch so exceptions (e.g. from List.copyOf or resolveAgentState) cannot escape into the doOnComplete callback and turn an already-completed Flux into an error. - MemoryFlushMiddleware: flush LLM call is capped by a 5-minute timeout (after subscribeOn, measuring execution not queue wait) so a hung model provider cannot tie up a boundedElastic worker. - MemoryMaintenanceMiddleware: consolidate LLM call is capped by a 5-minute timeout. - A FlushRequest record snapshots messages (List.copyOf) on the complete thread before background execution, avoiding races with mutable AgentState context. New regression tests (TDD): - onAgent_completesBeforeSlowFlushFinishes / completesBeforeSlowConsolidationFinishes: mock a slow LLM, assert Flux completes in <1s while the LLM is still running in the background. - onAgent_flushError_doesNotPropagateToFlux / maintenanceError_doesNotPropagateToFlux: verify errors in detached work never reach the caller. - onAgent_afterClose_doesNotScheduleNewFlush / doesNotScheduleNewMaintenance: verify close() prevents new work; uses polling (not fixed sleep) for deterministic verification on slow CI. - close_waitsForPendingFlush_thenReturns / waitsForPendingMaintenance_thenReturns: verify close() drains in-flight work before returning. - close_disposesHungFlush_andReturnsPromptly: verifies close() disposes a hanging model (Flux.never) and returns within CLOSE_AWAIT_TIMEOUT. Fixes agentscope-ai#2276 Fixes agentscope-ai#2225
Codecov Report❌ Patch coverage is 📢 Thoughts on this report? Let us know! |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
AgentScope-Java Version
Description
Fixes #2276, #2225
MemoryFlushMiddlewareandMemoryMaintenanceMiddlewareappended their LLM-backed work viaconcatWithonto the returnedFlux. BecauseReActAgent.callInternalends withtakeLast(1), which cannot emit until the upstream signalsonComplete, callers that consume the agent response to completion (blockLast(),takeLast(1), WebFlux controllers awaitingMono<Msg>) ended up waiting for the full memory flush LLM call (19–27s per call) and the consolidation LLM call(~44s on first run).
The original implementation used
doOnComplete(fire-and-forget). PR #1802 (RC4) inadvertently swapped it forconcatWith, introducing this regression.This PR restores the fire-and-forget behavior: both middlewares now subscribe their work independently of the returned
FluxviadoOnComplete().subscribe(), so theFluxcompletes as soon as the underlying agent call does.Additional safeguards
AutoCloseableand track pendingDisposables in aSetguarded bysynchronized(pending). A pre-subscribeclosedcheck prevents new work, and a post-subscribe check disposes any subscription that raced withclose().HarnessAgent.close()drains them (bounded by 5s) before workspace teardown to prevent races with temp directory deletion in tests and CLI shutdown.captureFlushRequestis wrapped in try/catch so exceptions (e.g. fromList.copyOforresolveAgentState) cannot escape into thedoOnCompletecallback and turn an already-completedFluxinto an error.MemoryFlushMiddleware: flush LLM call is capped by a 5-minute timeout (aftersubscribeOn, measuring execution not queue wait) so a hung model provider cannot tie up aboundedElasticworker.MemoryMaintenanceMiddleware: the entire maintenance run (file I/O + consolidation LLM) is capped by a 5-minuteMAINTENANCE_TIMEOUT; the consolidation LLM itself is capped by a slightly shorter 4m30sCONSOLIDATION_TIMEOUTso logs candistinguish "consolidation timed out" from "entire maintenance timed out".
RuntimeContextviaRuntimeContext.builder().from(rc).build()before background execution, preserving all fields that filesystem operations orNamespaceFactoryimplementations may depend on whileisolating attribute maps from concurrent mutations.
Regression tests
onAgent_completesBeforeSlowFlushFinishes/completesBeforeSlowConsolidationFinishes: mock a slow LLM, assert the returnedFluxcompletes in <1s while the LLM is still running in the background.onAgent_flushError_doesNotPropagateToFlux/maintenanceError_doesNotPropagateToFlux: verify errors in detached work never reach the caller.onAgent_afterClose_doesNotScheduleNewFlush/doesNotScheduleNewMaintenance: verifyclose()prevents new work; uses polling instead of fixed sleep for deterministic verification on slow CI.close_waitsForPendingFlush_thenReturns/waitsForPendingMaintenance_thenReturns: verifyclose()drains in-flight work before returning.close_disposesHungFlush_andReturnsPromptly/close_disposesHungConsolidation_andReturnsPromptly: verifyclose()disposes a hanging model and returns withinCLOSE_AWAIT_TIMEOUT.Checklist
Code has been formatted with
mvn spotless:applyAll tests are passing (
mvn test)Javadoc comments are complete and follow project conventions
Related documentation has been updated
Code is ready for review
Fixes [Bug]:harnessAgent编排模式,Memory 的
concatWith后置钩子阻塞了call()Flux complete处理,导致编排长时间等待?是否有其他方案 #2276Fixes [Bug]:MemoryConsolidator blocks agent call via
concatWith+.block()#2225