fix(harness): support asynchronous memory flush

agentscope-harness/src/main/java/io/agentscope/harness/agent/HarnessAgent.java

@@ -2439,7 +2439,8 @@ public HarnessAgent build() {
                                 effectiveFlushPrompt,
                                 memoryConfig.flushTrigger(),
                                 effectiveIsolationScope,
-                                periodicGate));
+                                periodicGate,
+                                memoryConfig.asyncFlush()));
 
                 String effectiveConsolidationPrompt =
                         memoryConfig.consolidationPrompt() != null

agentscope-harness/src/main/java/io/agentscope/harness/agent/memory/MemoryConfig.java

@@ -32,7 +32,7 @@
  *   <li><b>Flush</b> — extracts long-term memories from a conversation window into today's
  *       daily ledger ({@code memory/YYYY-MM-DD.md}). Prompt: {@link #flushPrompt()},
  *       defaults to {@link MemoryFlushManager#DEFAULT_FLUSH_PROMPT}. Trigger:
- *       {@link #flushTrigger()}.</li>
+ *       {@link #flushTrigger()}. Completion mode: {@link #asyncFlush()}.</li>
  *   <li><b>Consolidation</b> — periodically merges daily ledgers into the curated
  *       {@code MEMORY.md}. Prompt: {@link #consolidationPrompt()}, defaults to
  *       {@link MemoryConsolidator#DEFAULT_CONSOLIDATION_PROMPT}. Run cadence:
@@ -147,6 +147,7 @@ public String toString() {
     private final int dailyFileRetentionDays;
     private final int sessionRetentionDays;
     private final FlushTrigger flushTrigger;
+    private final boolean asyncFlush;
 
     private MemoryConfig(Builder b) {
         this.model = b.model;
@@ -157,6 +158,7 @@ private MemoryConfig(Builder b) {
         this.dailyFileRetentionDays = b.dailyFileRetentionDays;
         this.sessionRetentionDays = b.sessionRetentionDays;
         this.flushTrigger = b.flushTrigger;
+        this.asyncFlush = b.asyncFlush;
     }
 
     /**
@@ -206,6 +208,14 @@ public FlushTrigger flushTrigger() {
         return flushTrigger;
     }
 
+    /**
+     * Whether the per-call memory flush runs asynchronously after the response stream completes.
+     * Defaults to {@code false} so memory persistence retains its historical completion semantics.
+     */
+    public boolean asyncFlush() {
+        return asyncFlush;
+    }
+
     /** Returns a config equivalent to the harness's historical defaults. */
     public static MemoryConfig defaults() {
         return new Builder().build();
@@ -225,6 +235,7 @@ public static final class Builder {
         private int dailyFileRetentionDays = DEFAULT_DAILY_FILE_RETENTION_DAYS;
         private int sessionRetentionDays = DEFAULT_SESSION_RETENTION_DAYS;
         private FlushTrigger flushTrigger = FlushTrigger.always();
+        private boolean asyncFlush = false;
 
         /**
          * Sets a dedicated model for memory operations (flush + consolidation),
@@ -331,6 +342,16 @@ public Builder flushTrigger(FlushTrigger flushTrigger) {
             return this;
         }
 
+        /**
+         * Sets whether the per-call memory flush should run asynchronously after the response
+         * stream completes. Background failures are logged and do not fail the completed response.
+         * Disabled by default so the response waits for persistence to finish.
+         */
+        public Builder asyncFlush(boolean asyncFlush) {
+            this.asyncFlush = asyncFlush;
+            return this;
+        }
+
         public MemoryConfig build() {
             return new MemoryConfig(this);
         }

agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/MemoryFlushMiddleware.java

@@ -34,7 +34,9 @@
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
+import reactor.util.context.ContextView;
 
 /**
  * Middleware that triggers memory flush and message offload at the end of each agent call.
@@ -57,6 +59,13 @@
  * which runs independently of memory flush so history stays complete even when flush is
  * disabled.
  *
+ * <p>By default the response stream waits for the per-call flush to finish. When asynchronous
+ * flush is enabled, the middleware copies the completed call's messages and schedules a detached
+ * flush on a bounded single-worker scheduler. Failures are logged without changing the completed
+ * response. The scheduler accepts at most three queued tasks; excess flushes are rejected and
+ * logged rather than accumulating without bound. Detached tasks are not awaited during agent
+ * shutdown.
+ *
  * <p>The throttle window is tracked per <em>isolation key</em>, which matches the memory data
  * isolation in use:
  * <ul>
@@ -69,13 +78,17 @@
 public class MemoryFlushMiddleware implements HarnessRuntimeMiddleware {
 
     private static final Logger log = LoggerFactory.getLogger(MemoryFlushMiddleware.class);
+    // Keep fire-and-forget flushes serial and bound their backlog when the memory model is slow.
+    private static final Scheduler ASYNC_FLUSH_SCHEDULER =
+            Schedulers.newBoundedElastic(1, 3, "memory-flush");
 
     private final WorkspaceManager workspaceManager;
     private final Model model;
     private final String flushPrompt;
     private final MemoryConfig.FlushTrigger flushTrigger;
     private final IsolationScope isolationScope;
     private final PeriodicGate periodicGate;
+    private final boolean asyncFlush;
 
     public MemoryFlushMiddleware(WorkspaceManager workspaceManager, Model model) {
         this(
@@ -84,7 +97,8 @@ public MemoryFlushMiddleware(WorkspaceManager workspaceManager, Model model) {
                 MemoryFlushManager.DEFAULT_FLUSH_PROMPT,
                 MemoryConfig.FlushTrigger.always(),
                 IsolationScope.USER,
-                new LocalPeriodicGate());
+                new LocalPeriodicGate(),
+                false);
     }
 
     public MemoryFlushMiddleware(
@@ -98,7 +112,8 @@ public MemoryFlushMiddleware(
                 flushPrompt,
                 flushTrigger,
                 IsolationScope.USER,
-                new LocalPeriodicGate());
+                new LocalPeriodicGate(),
+                false);
     }
 
     public MemoryFlushMiddleware(
@@ -113,7 +128,8 @@ public MemoryFlushMiddleware(
                 flushPrompt,
                 flushTrigger,
                 isolationScope,
-                new LocalPeriodicGate());
+                new LocalPeriodicGate(),
+                false);
     }
 
     public MemoryFlushMiddleware(
@@ -123,6 +139,24 @@ public MemoryFlushMiddleware(
             MemoryConfig.FlushTrigger flushTrigger,
             IsolationScope isolationScope,
             PeriodicGate periodicGate) {
+        this(
+                workspaceManager,
+                model,
+                flushPrompt,
+                flushTrigger,
+                isolationScope,
+                periodicGate,
+                false);
+    }
+
+    public MemoryFlushMiddleware(
+            WorkspaceManager workspaceManager,
+            Model model,
+            String flushPrompt,
+            MemoryConfig.FlushTrigger flushTrigger,
+            IsolationScope isolationScope,
+            PeriodicGate periodicGate,
+            boolean asyncFlush) {
         this.workspaceManager = workspaceManager;
         this.model = model;
         this.flushPrompt =
@@ -131,6 +165,7 @@ public MemoryFlushMiddleware(
                 flushTrigger != null ? flushTrigger : MemoryConfig.FlushTrigger.always();
         this.isolationScope = isolationScope != null ? isolationScope : IsolationScope.USER;
         this.periodicGate = periodicGate != null ? periodicGate : new LocalPeriodicGate();
+        this.asyncFlush = asyncFlush;
     }
 
     @Override
@@ -140,28 +175,69 @@ public Flux<AgentEvent> onAgent(
             AgentInput input,
             Function<AgentInput, Flux<AgentEvent>> next) {
         final RuntimeContext rc = ctx != null ? ctx : RuntimeContext.empty();
-        return next.apply(input)
-                .concatWith(
-                        Mono.defer(() -> doFlush(agent, rc))
-                                .subscribeOn(Schedulers.boundedElastic())
-                                .onErrorResume(
-                                        e -> {
-                                            log.warn("Memory flush failed: {}", e.getMessage());
-                                            return Mono.empty();
-                                        })
-                                .then(Mono.<AgentEvent>empty()));
+        Flux<AgentEvent> response = next.apply(input);
+        if (asyncFlush) {
+            return response.transformDeferredContextual(
+                    (events, contextView) ->
+                            events.doOnComplete(() -> startAsyncFlush(agent, rc, contextView)));
+        }
+        return response.concatWith(
+                Mono.defer(() -> doFlush(agent, rc))
+                        .subscribeOn(Schedulers.boundedElastic())
+                        .onErrorResume(this::handleFlushError)
+                        .then(Mono.<AgentEvent>empty()));
     }
 
-    private Mono<Void> doFlush(Agent agent, RuntimeContext rc) {
-        AgentState state = RuntimeContext.resolveAgentState(rc, agent);
-        if (state == null) {
-            return Mono.empty();
+    private void startAsyncFlush(Agent agent, RuntimeContext rc, ContextView contextView) {
+        // Capture AgentState's defensive copy before detaching, so a later call cannot change
+        // the conversation window this flush is responsible for.
+        List<Msg> messages;
+        try {
+            messages = snapshotMessages(agent, rc);
+        } catch (RuntimeException e) {
+            logFlushError(e);
+            return;
+        }
+        if (messages.isEmpty()) {
+            return;
+        }
+        try {
+            Mono.defer(() -> doFlush(rc, messages))
+                    .subscribeOn(ASYNC_FLUSH_SCHEDULER)
+                    .contextWrite(context -> context.putAll(contextView))
+                    .onErrorResume(this::handleFlushError)
+                    .subscribe();
+        } catch (RuntimeException e) {
+            logFlushError(e);
         }
-        List<Msg> messages = state.getContext();
+    }
+
+    private Mono<Void> handleFlushError(Throwable error) {
+        logFlushError(error);
+        return Mono.empty();
+    }
+
+    private void logFlushError(Throwable error) {
+        log.warn("Memory flush failed: {}", error.getMessage());
+    }
+
+    private Mono<Void> doFlush(Agent agent, RuntimeContext rc) {
+        List<Msg> messages = snapshotMessages(agent, rc);
         if (messages.isEmpty()) {
             return Mono.empty();
         }
+        return doFlush(rc, messages);
+    }
+
+    private List<Msg> snapshotMessages(Agent agent, RuntimeContext rc) {
+        AgentState state = RuntimeContext.resolveAgentState(rc, agent);
+        if (state == null) {
+            return List.of();
+        }
+        return state.getContext();
+    }
 
+    private Mono<Void> doFlush(RuntimeContext rc, List<Msg> messages) {
         MemoryFlushManager flushManager =
                 new MemoryFlushManager(workspaceManager, model, flushPrompt);
 

agentscope-harness/src/test/java/io/agentscope/harness/agent/HarnessAgentAsyncMemoryFlushTest.java

@@ -0,0 +1,128 @@
+/*
+ * Copyright 2024-2026 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.agentscope.harness.agent;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import io.agentscope.core.agent.RuntimeContext;
+import io.agentscope.core.message.Msg;
+import io.agentscope.core.message.MsgRole;
+import io.agentscope.core.message.TextBlock;
+import io.agentscope.core.model.ChatResponse;
+import io.agentscope.core.model.GenerateOptions;
+import io.agentscope.core.model.Model;
+import io.agentscope.core.model.ToolSchema;
+import io.agentscope.harness.agent.memory.MemoryConfig;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Sinks;
+
+class HarnessAgentAsyncMemoryFlushTest {
+
+    @TempDir Path workspace;
+
+    @Test
+    void asyncFlushConfig_isWiredToMemoryFlushMiddleware() throws Exception {
+        Files.createDirectories(workspace);
+        Sinks.Many<ChatResponse> memoryResponse = Sinks.many().unicast().onBackpressureBuffer();
+        CountDownLatch memoryModelStarted = new CountDownLatch(1);
+        CountDownLatch memoryModelFinished = new CountDownLatch(1);
+        Model memoryModel =
+                new Model() {
+                    @Override
+                    public Flux<ChatResponse> stream(
+                            List<Msg> messages, List<ToolSchema> tools, GenerateOptions options) {
+                        memoryModelStarted.countDown();
+                        return memoryResponse
+                                .asFlux()
+                                .doFinally(ignored -> memoryModelFinished.countDown());
+                    }
+
+                    @Override
+                    public String getModelName() {
+                        return "slow-memory-model";
+                    }
+                };
+
+        try (HarnessAgent agent =
+                HarnessAgent.builder()
+                        .name("async-memory-agent")
+                        .model(responseModel("answer"))
+                        .workspace(workspace)
+                        .memory(MemoryConfig.builder().model(memoryModel).asyncFlush(true).build())
+                        .build()) {
+            try {
+                Msg reply =
+                        agent.call(
+                                        userMessage("Remember this"),
+                                        RuntimeContext.builder()
+                                                .userId("user")
+                                                .sessionId("session")
+                                                .build())
+                                .block(Duration.ofSeconds(5));
+
+                assertNotNull(reply);
+                assertEquals("answer", reply.getTextContent());
+                assertTrue(
+                        memoryModelStarted.await(5, TimeUnit.SECONDS),
+                        "configured memory model should start after the response completes");
+            } finally {
+                memoryResponse.tryEmitComplete();
+            }
+            assertTrue(
+                    memoryModelFinished.await(5, TimeUnit.SECONDS),
+                    "background flush should terminate before test cleanup");
+        }
+    }
+
+    private Model responseModel(String text) {
+        return new Model() {
+            @Override
+            public Flux<ChatResponse> stream(
+                    List<Msg> messages, List<ToolSchema> tools, GenerateOptions options) {
+                return Flux.just(
+                        new ChatResponse(
+                                "primary-response",
+                                List.of(TextBlock.builder().text(text).build()),
+                                null,
+                                Map.of(),
+                                "stop"));
+            }
+
+            @Override
+            public String getModelName() {
+                return "primary-model";
+            }
+        };
+    }
+
+    private Msg userMessage(String text) {
+        return Msg.builder()
+                .role(MsgRole.USER)
+                .content(TextBlock.builder().text(text).build())
+                .build();
+    }
+}