Skip to content

fix(harness): detach memory flush and maintenance from agent call response - #2617

Open
birdie7761 wants to merge 2 commits into
agentscope-ai:mainfrom
birdie7761:fix/harness-memory-flush-detach
Open

fix(harness): detach memory flush and maintenance from agent call response#2617
birdie7761 wants to merge 2 commits into
agentscope-ai:mainfrom
birdie7761:fix/harness-memory-flush-detach

Conversation

@birdie7761

@birdie7761 birdie7761 commented Aug 8, 2026

Copy link
Copy Markdown
Contributor

AgentScope-Java Version

 2.0.1

Description

Fixes #2276, #2225

MemoryFlushMiddleware and MemoryMaintenanceMiddleware appended their LLM-backed work via concatWith onto the returned Flux. Because ReActAgent.callInternal ends with takeLast(1), which cannot emit until the upstream signals
onComplete, callers that consume 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 the consolidation LLM call
(~44s on first run).

The original implementation used doOnComplete (fire-and-forget). PR #1802 (RC4) inadvertently swapped it for concatWith, introducing this regression.

This PR 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 safeguards

  • Both middlewares implement AutoCloseable and track pending Disposables in a Set guarded by synchronized(pending). A pre-subscribe closed check prevents new work, and 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 directory 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: the entire maintenance run (file I/O + consolidation LLM) is capped by a 5-minute MAINTENANCE_TIMEOUT; the consolidation LLM itself is capped by a slightly shorter 4m30s CONSOLIDATION_TIMEOUT so logs can
    distinguish "consolidation timed out" from "entire maintenance timed out".
  • Both middlewares snapshot the full RuntimeContext via RuntimeContext.builder().from(rc).build() before background execution, preserving all fields that filesystem operations or NamespaceFactory implementations may depend on while
    isolating attribute maps from concurrent mutations.

Regression tests

  • onAgent_completesBeforeSlowFlushFinishes / completesBeforeSlowConsolidationFinishes: mock a slow LLM, assert the returned 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 instead of fixed sleep for deterministic verification on slow CI.
  • close_waitsForPendingFlush_thenReturns / waitsForPendingMaintenance_thenReturns: verify close() drains in-flight work before returning.
  • close_disposesHungFlush_andReturnsPromptly / close_disposesHungConsolidation_andReturnsPromptly: verify close() disposes a hanging model and returns within CLOSE_AWAIT_TIMEOUT.

Checklist

…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

codecov Bot commented Aug 8, 2026

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 81.30841% with 20 lines in your changes missing coverage. Please review.

Files with missing lines Patch % Lines
...arness/agent/middleware/MemoryFlushMiddleware.java 79.31% 11 Missing and 1 partial ⚠️
.../agent/middleware/MemoryMaintenanceMiddleware.java 83.67% 7 Missing and 1 partial ⚠️

📢 Thoughts on this report? Let us know!

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

1 participant