feat(harness): support ordered async memory writes

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

@@ -56,6 +56,7 @@
 import io.agentscope.harness.agent.memory.MemoryConfig;
 import io.agentscope.harness.agent.memory.MemoryConsolidator;
 import io.agentscope.harness.agent.memory.MemoryFlushManager;
+import io.agentscope.harness.agent.memory.MemoryOperationScheduler;
 import io.agentscope.harness.agent.memory.compaction.CompactionConfig;
 import io.agentscope.harness.agent.memory.compaction.ConversationCompactor;
 import io.agentscope.harness.agent.memory.compaction.ToolResultEvictionConfig;
@@ -168,6 +169,7 @@ public class HarnessAgent implements Agent, AutoCloseable {
     private final SkillCurator skillCurator;
     private final SkillAuditLog skillAuditLog;
     private final MemoryConfig memoryConfig;
+    private final MemoryOperationScheduler memoryOperationScheduler;
 
     /** The subagent middleware (either SubagentsMiddleware or DynamicSubagentsMiddleware). */
     private final Object subagentMiddleware;
@@ -199,6 +201,7 @@ private HarnessAgent(
             SkillCurator skillCurator,
             SkillAuditLog skillAuditLog,
             MemoryConfig memoryConfig,
+            MemoryOperationScheduler memoryOperationScheduler,
             Object subagentMiddleware,
             DistributedStore distributedStore,
             WorkspacePathNormalizer pathNormalizer) {
@@ -217,6 +220,7 @@ private HarnessAgent(
         this.skillCurator = skillCurator;
         this.skillAuditLog = skillAuditLog;
         this.memoryConfig = memoryConfig != null ? memoryConfig : MemoryConfig.defaults();
+        this.memoryOperationScheduler = memoryOperationScheduler;
         this.subagentMiddleware = subagentMiddleware;
         this.distributedStore = distributedStore;
         this.pathNormalizer = pathNormalizer;
@@ -378,11 +382,17 @@ public void close() {
             shutdownTaskRepository();
         } finally {
             try {
-                if (ownedWorkspaceIndex != null) {
-                    ownedWorkspaceIndex.close();
+                if (memoryOperationScheduler != null) {
+                    memoryOperationScheduler.close();
                 }
             } finally {
-                delegate.close();
+                try {
+                    if (ownedWorkspaceIndex != null) {
+                        ownedWorkspaceIndex.close();
+                    }
+                } finally {
+                    delegate.close();
+                }
             }
         }
     }
@@ -2180,21 +2190,17 @@ public HarnessAgent build() {
                 inner.middleware(new AtPathExpansionMiddleware(wsManager));
             }
             Model memoryModel = memoryConfig.model() != null ? memoryConfig.model() : model;
+            MemoryOperationScheduler memoryOperationScheduler = null;
             if (memoryModel != null && !disableMemoryHooks) {
                 IsolationScope effectiveIsolationScope = fsIsolationScope;
+                if (memoryConfig.executionMode() == MemoryConfig.ExecutionMode.ASYNC) {
+                    memoryOperationScheduler = new MemoryOperationScheduler();
+                }
 
                 String effectiveFlushPrompt =
                         memoryConfig.flushPrompt() != null
                                 ? memoryConfig.flushPrompt()
                                 : MemoryFlushManager.DEFAULT_FLUSH_PROMPT;
-                inner.middleware(
-                        new MemoryFlushMiddleware(
-                                wsManager,
-                                memoryModel,
-                                effectiveFlushPrompt,
-                                memoryConfig.flushTrigger(),
-                                effectiveIsolationScope));
-
                 String effectiveConsolidationPrompt =
                         memoryConfig.consolidationPrompt() != null
                                 ? memoryConfig.consolidationPrompt()
@@ -2205,14 +2211,28 @@ public HarnessAgent build() {
                                 memoryModel,
                                 effectiveConsolidationPrompt,
                                 memoryConfig.consolidationMaxTokens());
+                // Middleware completion callbacks run from the last registered middleware
+                // outward. Register maintenance first so raw offload + flush is queued before
+                // consolidation for the same isolation key.
                 inner.middleware(
                         new MemoryMaintenanceMiddleware(
                                 wsManager,
                                 consolidator,
                                 memoryConfig.dailyFileRetentionDays(),
                                 memoryConfig.sessionRetentionDays(),
                                 memoryConfig.consolidationMinGap(),
-                                effectiveIsolationScope));
+                                effectiveIsolationScope,
+                                memoryConfig.executionMode(),
+                                memoryOperationScheduler));
+                inner.middleware(
+                        new MemoryFlushMiddleware(
+                                wsManager,
+                                memoryModel,
+                                effectiveFlushPrompt,
+                                memoryConfig.flushTrigger(),
+                                effectiveIsolationScope,
+                                memoryConfig.executionMode(),
+                                memoryOperationScheduler));
             }
             CompactionMiddleware compactionHook = null;
             if (!disableCompaction && compactionConfig != null) {
@@ -2548,6 +2568,7 @@ public HarnessAgent build() {
                     pendingSkillCurator,
                     pendingSkillAuditLog,
                     memoryConfig,
+                    memoryOperationScheduler,
                     capturedSubagentMw,
                     distributedStore,
                     pathNormalizer);

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

@@ -42,11 +42,20 @@
  *       configure via {@code .compaction(CompactionConfig...)} rather than here.</li>
  * </ol>
  *
- * <p>All fields have sensible defaults; {@link #defaults()} returns a config equivalent
- * to the harness's historical behavior so adopting this class is a no-op upgrade.
+ * <p>All fields have sensible defaults; {@link #defaults()} retains the harness's historical
+ * completion semantics. Applications that manage the agent lifecycle can opt into asynchronous
+ * post-call memory work with {@link Builder#executionMode(ExecutionMode)}.
  */
 public final class MemoryConfig {
 
+    /** Determines whether post-call memory work delays completion of the agent call. */
+    public enum ExecutionMode {
+        /** Queue memory work in the background, ordered by the configured isolation key. */
+        ASYNC,
+        /** Wait for memory work before completing the agent call (historical behavior). */
+        BLOCKING
+    }
+
     /** Default {@code consolidationMaxTokens}. */
     public static final int DEFAULT_CONSOLIDATION_MAX_TOKENS = 4_000;
 
@@ -147,6 +156,7 @@ public String toString() {
     private final int dailyFileRetentionDays;
     private final int sessionRetentionDays;
     private final FlushTrigger flushTrigger;
+    private final ExecutionMode executionMode;
 
     private MemoryConfig(Builder b) {
         this.model = b.model;
@@ -157,6 +167,7 @@ private MemoryConfig(Builder b) {
         this.dailyFileRetentionDays = b.dailyFileRetentionDays;
         this.sessionRetentionDays = b.sessionRetentionDays;
         this.flushTrigger = b.flushTrigger;
+        this.executionMode = b.executionMode;
     }
 
     /**
@@ -206,7 +217,11 @@ public FlushTrigger flushTrigger() {
         return flushTrigger;
     }
 
-    /** Returns a config equivalent to the harness's historical defaults. */
+    public ExecutionMode executionMode() {
+        return executionMode;
+    }
+
+    /** Returns the default memory configuration. */
     public static MemoryConfig defaults() {
         return new Builder().build();
     }
@@ -225,6 +240,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 ExecutionMode executionMode = ExecutionMode.BLOCKING;
 
         /**
          * Sets a dedicated model for memory operations (flush + consolidation),
@@ -331,6 +347,18 @@ public Builder flushTrigger(FlushTrigger flushTrigger) {
             return this;
         }
 
+        /**
+         * Controls whether flush and maintenance work runs after the agent call completes or
+         * remains part of its completion path. Must not be null.
+         */
+        public Builder executionMode(ExecutionMode executionMode) {
+            if (executionMode == null) {
+                throw new IllegalArgumentException("executionMode must not be null");
+            }
+            this.executionMode = executionMode;
+            return this;
+        }
+
         public MemoryConfig build() {
             return new MemoryConfig(this);
         }

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

@@ -0,0 +1,122 @@
+/*
+ * 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.memory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+
+/**
+ * Runs post-call memory operations in the background while preserving submission order within
+ * each memory isolation key. Different users or sessions can progress concurrently.
+ *
+ * <p>{@link #close()} stops accepting new work and waits for all accepted operations. This keeps
+ * asynchronous memory writes durable when their owning {@code HarnessAgent} is closed.
+ */
+public final class MemoryOperationScheduler implements AutoCloseable {
+
+    private static final Logger log = LoggerFactory.getLogger(MemoryOperationScheduler.class);
+
+    private final ConcurrentHashMap<String, CompletableFuture<Void>> tails =
+            new ConcurrentHashMap<>();
+    private final Object lifecycleLock = new Object();
+    private boolean accepting = true;
+
+    /**
+     * Queues an operation after all previously submitted work for {@code isolationKey}.
+     *
+     * @return {@code true} when accepted, or {@code false} after this scheduler has closed
+     */
+    public boolean submit(String isolationKey, Supplier<Mono<Void>> operation) {
+        if (operation == null) {
+            throw new IllegalArgumentException("operation must not be null");
+        }
+        String key = isolationKey != null ? isolationKey : "";
+        AtomicReference<CompletableFuture<Void>> submitted = new AtomicReference<>();
+        synchronized (lifecycleLock) {
+            if (!accepting) {
+                return false;
+            }
+            tails.compute(
+                    key,
+                    (ignored, previous) -> {
+                        CompletableFuture<Void> predecessor =
+                                previous == null
+                                        ? CompletableFuture.completedFuture(null)
+                                        : previous.handle((result, error) -> null);
+                        CompletableFuture<Void> next =
+                                predecessor.thenComposeAsync(
+                                        unused -> invoke(operation),
+                                        command -> Schedulers.boundedElastic().schedule(command));
+                        submitted.set(next);
+                        return next;
+                    });
+        }
+
+        CompletableFuture<Void> future = submitted.get();
+        future.whenComplete(
+                (unused, error) -> {
+                    tails.remove(key, future);
+                    if (error != null) {
+                        log.warn(
+                                "Asynchronous memory operation failed for key {}: {}",
+                                key,
+                                unwrap(error).getMessage());
+                    }
+                });
+        return true;
+    }
+
+    private CompletableFuture<Void> invoke(Supplier<Mono<Void>> operation) {
+        try {
+            Mono<Void> task = operation.get();
+            return task != null
+                    ? task.subscribeOn(Schedulers.boundedElastic()).toFuture()
+                    : CompletableFuture.completedFuture(null);
+        } catch (Throwable error) {
+            return CompletableFuture.failedFuture(error);
+        }
+    }
+
+    private static Throwable unwrap(Throwable error) {
+        return error instanceof CompletionException && error.getCause() != null
+                ? error.getCause()
+                : error;
+    }
+
+    @Override
+    public void close() {
+        List<CompletableFuture<Void>> pending;
+        synchronized (lifecycleLock) {
+            if (!accepting && tails.isEmpty()) {
+                return;
+            }
+            accepting = false;
+            pending = new ArrayList<>(tails.values());
+        }
+        CompletableFuture.allOf(pending.toArray(CompletableFuture[]::new))
+                .handle((unused, error) -> null)
+                .join();
+    }
+}