diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/HarnessAgent.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/HarnessAgent.java
index bf31afe8db..4cc0ec2e09 100644
--- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/HarnessAgent.java
+++ b/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
diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/memory/MemoryConfig.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/memory/MemoryConfig.java
index dad15e6cb3..f931a2caad 100644
--- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/memory/MemoryConfig.java
+++ b/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);
         }
diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/MemoryFlushMiddleware.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/MemoryFlushMiddleware.java
index 39c6f8c5cb..dd4de527d5 100644
--- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/middleware/MemoryFlushMiddleware.java
+++ b/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,6 +78,9 @@
 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;
@@ -76,6 +88,7 @@ public class MemoryFlushMiddleware implements HarnessRuntimeMiddleware {
     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);
 
diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/HarnessAgentAsyncMemoryFlushTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/HarnessAgentAsyncMemoryFlushTest.java
new file mode 100644
index 0000000000..755f9665ca
--- /dev/null
+++ b/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();
+    }
+}
diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/HarnessAgentDynamicHookBuilderTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/HarnessAgentDynamicHookBuilderTest.java
index 12f6183a60..cdb0f33e73 100644
--- a/agentscope-harness/src/test/java/io/agentscope/harness/agent/HarnessAgentDynamicHookBuilderTest.java
+++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/HarnessAgentDynamicHookBuilderTest.java
@@ -49,6 +49,7 @@
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 import org.mockito.ArgumentCaptor;
@@ -76,10 +77,19 @@ class HarnessAgentDynamicHookBuilderTest {
 
     @TempDir Path workspace;
 
+    private HarnessAgent agent;
+
+    @AfterEach
+    void closeAgents() {
+        if (agent != null) {
+            agent.close();
+        }
+    }
+
     @Test
     void defaultBuild_registersDynamicSkillAndSubagentMiddlewares() throws Exception {
         Files.createDirectories(workspace);
-        HarnessAgent agent =
+        agent =
                 HarnessAgent.builder()
                         .name("t")
                         .model(stubModel("ok"))
@@ -107,7 +117,7 @@ void customSkillRepository_composesWithDynamicMiddleware() throws Exception {
         Files.createDirectories(workspace);
         AgentSkillRepository emptyRepo = new EmptySkillRepository();
 
-        HarnessAgent agent =
+        agent =
                 HarnessAgent.builder()
                         .name("t")
                         .model(stubModel("ok"))
@@ -125,7 +135,7 @@ void customSkillRepository_composesWithDynamicMiddleware() throws Exception {
     @Test
     void disableDynamicSkills_skipsDynamicSkillMiddleware() throws Exception {
         Files.createDirectories(workspace);
-        HarnessAgent agent =
+        agent =
                 HarnessAgent.builder()
                         .name("t")
                         .model(stubModel("ok"))
@@ -153,7 +163,7 @@ void disableDynamicSkills_freezesRepositoriesIntoStaticMiddleware() throws Excep
                                         "# Frozen skill",
                                         null)));
 
-        HarnessAgent agent =
+        agent =
                 HarnessAgent.builder()
                         .name("t")
                         .model(model)
@@ -222,7 +232,7 @@ void disableDynamicSkills_appliesBuilderAndVisibilityFiltersToPromptAndLoader()
                                 skill("beta", "beta description"),
                                 skill("gamma", "gamma description")));
 
-        HarnessAgent agent =
+        agent =
                 HarnessAgent.builder()
                         .name("t")
                         .model(model)
@@ -265,7 +275,7 @@ void disableDynamicSkills_skillsEnabledFalseLeavesCatalogEmpty() throws Exceptio
         CountingSkillRepository repository =
                 new CountingSkillRepository(List.of(skill("disabled", "must stay hidden")));
 
-        HarnessAgent agent =
+        agent =
                 HarnessAgent.builder()
                         .name("t")
                         .model(model)
@@ -293,7 +303,7 @@ void disableDynamicSkills_keepsWorkspaceLazyResourcesLoadable() throws Exception
                 "---\nname: lazy-resource\ndescription: Loads a lazy reference\n---\n# Body\n");
         Files.writeString(skillDir.resolve("references/guide.md"), "lazy reference body");
 
-        HarnessAgent agent =
+        agent =
                 HarnessAgent.builder()
                         .name("t")
                         .model(stubModel("ok"))
@@ -331,7 +341,7 @@ void getSkillRepositories_exposesComposedListInOrder() throws Exception {
         Files.createDirectories(workspace);
         AgentSkillRepository custom = new EmptySkillRepository();
 
-        HarnessAgent agent =
+        agent =
                 HarnessAgent.builder()
                         .name("t")
                         .model(stubModel("ok"))
@@ -356,7 +366,7 @@ void getSkillRepositories_exposesComposedListInOrder() throws Exception {
     @Test
     void getSkillRepositories_isEmptyWhenNothingComposed() throws Exception {
         Files.createDirectories(workspace);
-        HarnessAgent agent =
+        agent =
                 HarnessAgent.builder()
                         .name("t")
                         .model(stubModel("ok"))
@@ -371,7 +381,7 @@ void getSkillRepositories_isEmptyWhenNothingComposed() throws Exception {
     @Test
     void getSkillRepositories_returnsImmutableList() throws Exception {
         Files.createDirectories(workspace);
-        HarnessAgent agent =
+        agent =
                 HarnessAgent.builder()
                         .name("t")
                         .model(stubModel("ok"))
@@ -394,7 +404,7 @@ void getSkillRepositories_returnsImmutableList() throws Exception {
     @Test
     void disableDynamicSubagents_fallsBackToStaticSubagentsMiddleware() throws Exception {
         Files.createDirectories(workspace);
-        HarnessAgent agent =
+        agent =
                 HarnessAgent.builder()
                         .name("t")
                         .model(stubModel("ok"))
diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/memory/MemoryConfigTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/memory/MemoryConfigTest.java
index 2def8896c0..db486a07e6 100644
--- a/agentscope-harness/src/test/java/io/agentscope/harness/agent/memory/MemoryConfigTest.java
+++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/memory/MemoryConfigTest.java
@@ -16,10 +16,12 @@
 package io.agentscope.harness.agent.memory;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.time.Duration;
 import org.junit.jupiter.api.Test;
@@ -39,6 +41,7 @@ void defaults_matchHistoricalBehaviour() {
         assertEquals(MemoryConfig.DEFAULT_DAILY_FILE_RETENTION_DAYS, cfg.dailyFileRetentionDays());
         assertEquals(MemoryConfig.DEFAULT_SESSION_RETENTION_DAYS, cfg.sessionRetentionDays());
         assertEquals(MemoryConfig.FlushMode.ALWAYS, cfg.flushTrigger().mode());
+        assertFalse(cfg.asyncFlush(), "asyncFlush default preserves synchronous completion");
     }
 
     @Test
@@ -81,6 +84,7 @@ void builder_overridesAreCarried() {
                         .dailyFileRetentionDays(30)
                         .sessionRetentionDays(60)
                         .flushTrigger(MemoryConfig.FlushTrigger.throttled(Duration.ofMinutes(10)))
+                        .asyncFlush(true)
                         .build();
 
         assertEquals("custom flush", cfg.flushPrompt());
@@ -90,6 +94,7 @@ void builder_overridesAreCarried() {
         assertEquals(60, cfg.sessionRetentionDays());
         assertEquals(MemoryConfig.FlushMode.THROTTLED, cfg.flushTrigger().mode());
         assertEquals(Duration.ofMinutes(10), cfg.flushTrigger().minGap());
+        assertTrue(cfg.asyncFlush());
     }
 
     @Test
diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/middleware/MemoryFlushMiddlewareCompletionTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/middleware/MemoryFlushMiddlewareCompletionTest.java
new file mode 100644
index 0000000000..2e57b714d5
--- /dev/null
+++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/middleware/MemoryFlushMiddlewareCompletionTest.java
@@ -0,0 +1,288 @@
+/*
+ * 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.middleware;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+import io.agentscope.core.agent.Agent;
+import io.agentscope.core.agent.RuntimeContext;
+import io.agentscope.core.event.AgentEvent;
+import io.agentscope.core.event.CustomEvent;
+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.core.state.AgentState;
+import io.agentscope.harness.agent.IsolationScope;
+import io.agentscope.harness.agent.coordination.LocalPeriodicGate;
+import io.agentscope.harness.agent.memory.MemoryConfig;
+import io.agentscope.harness.agent.memory.MemoryFlushManager;
+import io.agentscope.harness.agent.workspace.WorkspaceManager;
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Sinks;
+import reactor.test.StepVerifier;
+
+/** Regression coverage for the response-completion delay reported in issue #2821. */
+class MemoryFlushMiddlewareCompletionTest {
+
+    @Test
+    void asyncFlush_completesResponseWithoutWaitingForMemoryFlush() throws InterruptedException {
+        Sinks.Many<ChatResponse> memoryResponse = Sinks.many().unicast().onBackpressureBuffer();
+        CountDownLatch memoryModelStarted = new CountDownLatch(1);
+        CountDownLatch memoryModelFinished = new CountDownLatch(1);
+        Model slowMemoryModel =
+                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 "controllable-memory-model";
+                    }
+                };
+
+        MemoryFlushMiddleware middleware = asyncMiddleware(slowMemoryModel);
+
+        AgentState state = stateWithUserMessage("Remember this");
+        RuntimeContext context = RuntimeContext.builder().agentState(state).build();
+        AgentEvent downstreamEvent = new CustomEvent("downstream-complete");
+
+        StepVerifier.create(
+                        middleware.onAgent(
+                                mock(Agent.class),
+                                context,
+                                null,
+                                ignored -> Flux.just(downstreamEvent)))
+                .expectNext(downstreamEvent)
+                .verifyComplete();
+
+        assertTrue(
+                memoryModelStarted.await(5, TimeUnit.SECONDS),
+                "memory flush should start in the background");
+        assertTrue(
+                memoryResponse.tryEmitComplete().isSuccess(),
+                "test should release the background memory flush");
+        assertTrue(
+                memoryModelFinished.await(5, TimeUnit.SECONDS),
+                "background memory flush should terminate after release");
+    }
+
+    @Test
+    void asyncFlush_failureIsIsolatedFromResponse() throws InterruptedException {
+        CountDownLatch memoryModelStarted = new CountDownLatch(1);
+        CountDownLatch memoryModelFinished = new CountDownLatch(1);
+        Model failingMemoryModel =
+                new Model() {
+                    @Override
+                    public Flux<ChatResponse> stream(
+                            List<Msg> messages, List<ToolSchema> tools, GenerateOptions options) {
+                        memoryModelStarted.countDown();
+                        return Flux.<ChatResponse>error(new IllegalStateException("flush failed"))
+                                .doFinally(ignored -> memoryModelFinished.countDown());
+                    }
+
+                    @Override
+                    public String getModelName() {
+                        return "failing-memory-model";
+                    }
+                };
+
+        RuntimeContext context =
+                RuntimeContext.builder().agentState(stateWithUserMessage("Remember this")).build();
+        AgentEvent downstreamEvent = new CustomEvent("downstream-complete");
+
+        StepVerifier.create(
+                        asyncMiddleware(failingMemoryModel)
+                                .onAgent(
+                                        mock(Agent.class),
+                                        context,
+                                        null,
+                                        ignored -> Flux.just(downstreamEvent)))
+                .expectNext(downstreamEvent)
+                .verifyComplete();
+
+        assertTrue(
+                memoryModelStarted.await(5, TimeUnit.SECONDS),
+                "failing memory flush should still start in the background");
+        assertTrue(
+                memoryModelFinished.await(5, TimeUnit.SECONDS),
+                "failing memory flush should be consumed and terminate");
+    }
+
+    @Test
+    void asyncFlush_usesCompletionTimeMessageSnapshot() throws InterruptedException {
+        Sinks.Many<ChatResponse> firstResponse = Sinks.many().unicast().onBackpressureBuffer();
+        CountDownLatch firstStarted = new CountDownLatch(1);
+        CountDownLatch secondStarted = new CountDownLatch(1);
+        CountDownLatch secondFinished = new CountDownLatch(1);
+        AtomicInteger invocations = new AtomicInteger();
+        AtomicReference<List<Msg>> secondFlushInput = new AtomicReference<>();
+        Model queuedMemoryModel =
+                new Model() {
+                    @Override
+                    public Flux<ChatResponse> stream(
+                            List<Msg> messages, List<ToolSchema> tools, GenerateOptions options) {
+                        if (invocations.incrementAndGet() == 1) {
+                            firstStarted.countDown();
+                            return firstResponse.asFlux();
+                        }
+                        secondFlushInput.set(List.copyOf(messages));
+                        secondStarted.countDown();
+                        return Flux.<ChatResponse>empty()
+                                .doFinally(ignored -> secondFinished.countDown());
+                    }
+
+                    @Override
+                    public String getModelName() {
+                        return "queued-memory-model";
+                    }
+                };
+        MemoryFlushMiddleware middleware = asyncMiddleware(queuedMemoryModel);
+
+        try {
+            completeDownstream(middleware, stateWithUserMessage("first call"));
+            assertTrue(
+                    firstStarted.await(5, TimeUnit.SECONDS),
+                    "first flush should occupy the serial scheduler");
+
+            AgentState secondState = stateWithUserMessage("snapshot-before-mutation");
+            completeDownstream(middleware, secondState);
+            secondState.contextMutable().clear();
+            secondState.contextMutable().add(userMessage("mutated-after-completion"));
+        } finally {
+            firstResponse.tryEmitComplete();
+        }
+        assertTrue(
+                secondStarted.await(5, TimeUnit.SECONDS),
+                "queued flush should start after the first flush finishes");
+        assertTrue(
+                secondFinished.await(5, TimeUnit.SECONDS),
+                "queued flush should terminate before the test finishes");
+
+        String flushPrompt =
+                secondFlushInput.get().stream()
+                        .map(Msg::getTextContent)
+                        .reduce("", (left, right) -> left + "\n" + right);
+        assertTrue(flushPrompt.contains("snapshot-before-mutation"));
+        assertFalse(flushPrompt.contains("mutated-after-completion"));
+    }
+
+    @Test
+    void responseStream_waitsForMemoryFlushToComplete_currentBehavior() {
+        Sinks.Many<ChatResponse> memoryResponse = Sinks.many().unicast().onBackpressureBuffer();
+        AtomicBoolean memoryModelStarted = new AtomicBoolean();
+        Model slowMemoryModel =
+                new Model() {
+                    @Override
+                    public Flux<ChatResponse> stream(
+                            List<Msg> messages, List<ToolSchema> tools, GenerateOptions options) {
+                        memoryModelStarted.set(true);
+                        return memoryResponse.asFlux();
+                    }
+
+                    @Override
+                    public String getModelName() {
+                        return "controllable-memory-model";
+                    }
+                };
+
+        MemoryFlushMiddleware middleware =
+                new MemoryFlushMiddleware(
+                        mock(WorkspaceManager.class),
+                        slowMemoryModel,
+                        MemoryFlushManager.DEFAULT_FLUSH_PROMPT,
+                        MemoryConfig.FlushTrigger.always(),
+                        IsolationScope.USER);
+
+        AgentState state = stateWithUserMessage("Remember this");
+        RuntimeContext context = RuntimeContext.builder().agentState(state).build();
+        AgentEvent downstreamEvent = new CustomEvent("downstream-complete");
+
+        Flux<AgentEvent> result =
+                middleware.onAgent(
+                        mock(Agent.class), context, null, ignored -> Flux.just(downstreamEvent));
+
+        StepVerifier.create(result)
+                .expectNext(downstreamEvent)
+                .expectNoEvent(Duration.ofMillis(200))
+                .then(
+                        () -> {
+                            assertTrue(
+                                    memoryModelStarted.get(),
+                                    "memory flush should have started after the downstream event");
+                            assertEquals(
+                                    Sinks.EmitResult.OK,
+                                    memoryResponse.tryEmitComplete(),
+                                    "releasing the memory model should unblock response"
+                                            + " completion");
+                        })
+                .verifyComplete();
+    }
+
+    private MemoryFlushMiddleware asyncMiddleware(Model model) {
+        return new MemoryFlushMiddleware(
+                mock(WorkspaceManager.class),
+                model,
+                MemoryFlushManager.DEFAULT_FLUSH_PROMPT,
+                MemoryConfig.FlushTrigger.always(),
+                IsolationScope.USER,
+                new LocalPeriodicGate(),
+                true);
+    }
+
+    private void completeDownstream(MemoryFlushMiddleware middleware, AgentState state) {
+        RuntimeContext context = RuntimeContext.builder().agentState(state).build();
+        middleware
+                .onAgent(
+                        mock(Agent.class),
+                        context,
+                        null,
+                        ignored -> Flux.just(new CustomEvent("downstream-complete")))
+                .then()
+                .block(Duration.ofSeconds(5));
+    }
+
+    private AgentState stateWithUserMessage(String text) {
+        return AgentState.builder().addMessage(userMessage(text)).build();
+    }
+
+    private Msg userMessage(String text) {
+        return Msg.builder()
+                .role(MsgRole.USER)
+                .content(TextBlock.builder().text(text).build())
+                .build();
+    }
+}
diff --git a/docs/v2/en/docs/harness/memory.md b/docs/v2/en/docs/harness/memory.md
index a9c41fdb8d..09c8524cfd 100644
--- a/docs/v2/en/docs/harness/memory.md
+++ b/docs/v2/en/docs/harness/memory.md
@@ -58,7 +58,7 @@ Flush (path 1) is triggered at three different moments:
 
 All three sites share the **same** `flushPrompt`, so customizing it changes all three.
 
-Both flush and offload are **asynchronous**: they are launched in a fire-and-forget fashion via `doOnComplete` after the response stream has ended, so they never block the current `call()` return. The caller receives the full response first; the flush LLM call and JSONL offload run in the background afterward.
+The per-call flush and transcript offload run after the downstream response events have been emitted, but by default they remain part of the response pipeline, so `call()` completion waits for them. Set `MemoryConfig.asyncFlush(true)` to detach only the per-call flush and run it in a fire-and-forget fashion; transcript offload and the other two flush paths retain their existing completion semantics.
 
 ## Enable compaction
 
@@ -105,7 +105,7 @@ CompactionConfig.builder()
 
 ## Customizing the memory pipeline: `MemoryConfig`
 
-`MemoryConfig` is the single place to configure flush / consolidation prompts, throttling, retention, and the per-call flush trigger. Every field has a default; not calling `.memory(...)` reproduces the historical behaviour bit-for-bit.
+`MemoryConfig` is the single place to configure flush / consolidation prompts, throttling, retention, and the per-call flush trigger and completion mode. Every field has a default; not calling `.memory(...)` reproduces the historical behaviour bit-for-bit.
 
 ### Example 1: throttle per-call flush to save tokens
 
@@ -193,6 +193,29 @@ HarnessAgent.builder()
 
 `model(String)` resolves via `ModelRegistry.resolve()`; you can also pass a `Model` instance. When not set, falls back to the agent's primary model.
 
+### Example 7: return without waiting for per-call flush
+
+By default, the agent response waits for the per-call memory flush to finish. If the memory model can be slow and the application does not require the memory write to finish before returning the response, enable asynchronous flush:
+
+```java
+HarnessAgent.builder()
+    ...
+    .memory(MemoryConfig.builder()
+        .asyncFlush(true)
+        .build())
+    .build();
+```
+
+This option only changes **path 1** (the per-call flush). It captures a message snapshot when the response completes, then performs extraction and persistence in the background. Compaction flushes, transcript persistence, and background consolidation keep their existing completion semantics.
+
+Async flush is fire-and-forget:
+
+- failures are logged and do not fail the completed agent response;
+- a process-wide dedicated scheduler runs one flush at a time and queues up to three more; when saturated, new flushes are rejected and logged instead of growing the backlog without bound;
+- in-flight flushes are not awaited during `HarnessAgent.close()`, so an application that exits immediately may stop before the memory write completes.
+
+Leave `asyncFlush` at its default `false` when every accepted response must guarantee that its per-call memory flush has completed.
+
 ### `MemoryConfig` field reference
 
 | Field | Default | Purpose |
@@ -205,6 +228,7 @@ HarnessAgent.builder()
 | `dailyFileRetentionDays` | `90` | Days before a daily log moves to `memory/archive/` |
 | `sessionRetentionDays` | `180` | Days before a `*.log.jsonl` is pruned |
 | `flushTrigger` | `FlushTrigger.always()` | `ALWAYS` / `NEVER` / `THROTTLED(Duration)` |
+| `asyncFlush` | `false` | Return without waiting for the per-call flush; background failures are logged |
 
 ## Large tool-result offloading
 
diff --git a/docs/v2/zh/docs/harness/memory.md b/docs/v2/zh/docs/harness/memory.md
index 7bfc4d58b0..cbd510359a 100644
--- a/docs/v2/zh/docs/harness/memory.md
+++ b/docs/v2/zh/docs/harness/memory.md
@@ -58,7 +58,7 @@ Flush（路径 1）会在以下三个时机被触发：
 
 这三处用的是 **同一份** `flushPrompt`，定制后三处行为一致。
 
-Flush 和 offload 都是**异步执行**的：它们在响应流结束后通过 `doOnComplete` 以 fire-and-forget 方式启动，不会阻塞当前 `call()` 的返回。换句话说，调用方拿到完整响应之后，flush LLM 调用和 JSONL offload 才在后台开始。
+Per-call flush 和 transcript offload 会在下游响应事件发送完成后执行，但默认情况下它们仍属于响应流程的一部分，因此 `call()` 会等待其完成。设置 `MemoryConfig.asyncFlush(true)` 后，只有 per-call flush 会脱离当前响应流程，以 fire-and-forget 方式在后台执行；transcript offload 以及另外两个 flush 路径仍保持现有的完成语义。
 
 ## 开启压缩
 
@@ -105,7 +105,7 @@ CompactionConfig.builder()
 
 ## 定制 Memory pipeline：`MemoryConfig`
 
-`MemoryConfig` 集中管理 flush / consolidation 两条路径的 prompt、节流、保留时长，以及 per-call flush 的触发策略。所有字段都有默认值，不调 `.memory(...)` 时与历史行为完全一致。
+`MemoryConfig` 集中管理 flush / consolidation 两条路径的 prompt、节流、保留时长，以及 per-call flush 的触发策略和完成模式。所有字段都有默认值，不调 `.memory(...)` 时与历史行为完全一致。
 
 ### 例 1：节流 per-call flush，省 token
 
@@ -193,6 +193,29 @@ HarnessAgent.builder()
 
 `model(String)` 走 `ModelRegistry.resolve()`，也可以传 `Model` 实例。不设则 fallback 到 agent 主模型。
 
+### 例 7：不等待 per-call flush 就返回
+
+默认情况下，agent 响应会等待当次记忆 flush 完成。如果记忆模型较慢，而业务不要求记忆写入在返回响应前完成，可以开启异步 flush：
+
+```java
+HarnessAgent.builder()
+    ...
+    .memory(MemoryConfig.builder()
+        .asyncFlush(true)
+        .build())
+    .build();
+```
+
+该选项只改变**路径 1**（per-call flush）。响应完成时会先复制当前消息快照，再在后台执行抽取和持久化。压缩内嵌的 flush、transcript 持久化和后台 consolidation 仍保持原有完成语义。
+
+异步 flush 是 fire-and-forget：
+
+- 失败只记日志，不会让已完成的 agent 响应失败；
+- 进程级专用调度器每次运行一个 flush，最多再排队三个；队列饱和时，新 flush 会被拒绝并记录日志，避免待执行任务无限增长；
+- `HarnessAgent.close()` 不会等待进行中的 flush，因此应用立即退出时，记忆写入可能尚未完成。
+
+如果每个已返回响应都必须保证当次记忆 flush 已完成，请保持 `asyncFlush` 的默认值 `false`。
+
 ### `MemoryConfig` 字段速查
 
 | 字段 | 默认 | 作用 |
@@ -205,6 +228,7 @@ HarnessAgent.builder()
 | `dailyFileRetentionDays` | `90` | 多少天后把日流水账归档到 `memory/archive/` |
 | `sessionRetentionDays` | `180` | 多少天后清掉 `*.log.jsonl` |
 | `flushTrigger` | `FlushTrigger.always()` | `ALWAYS` / `NEVER` / `THROTTLED(Duration)` |
+| `asyncFlush` | `false` | 不等待 per-call flush 就返回；后台失败只记日志 |
 
 ## 大工具结果卸载