perf(session): bulk-load journal segments so replay costs one round-trip

agentscope-core/src/main/java/io/agentscope/core/session/AtomicSessionStorage.java

@@ -15,7 +15,10 @@
  */
 package io.agentscope.core.session;
 
+import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 /** Narrow storage capability. CAS must be atomic across all writers sharing this namespace.
  * Missing is null; read/write failures throw. Implementations must durably store bytes before ACK.
@@ -35,6 +38,20 @@ public byte[] bytes() {
 
     Value read(String path);
 
+    /** Reads several objects in one round-trip when the backend can; entries for paths that do
+     * not exist are simply absent from the result. The default falls back to per-path reads.
+     * Backends whose per-read cost is dominated by connection setup (SQLite over JDBC, for
+     * example) should override this so a journal replay costs one round-trip, not one per
+     * segment. */
+    default Map<String, Value> readAll(Collection<String> paths) {
+        var result = new HashMap<String, Value>();
+        for (String path : paths) {
+            var value = read(path);
+            if (value != null) result.put(path, value);
+        }
+        return result;
+    }
+
     /** Optional read-only discovery of object paths under a prefix. Recovery never uses listing. */
     default List<String> listPaths(String prefix) {
         throw new UnsupportedOperationException("This session storage does not support discovery");

agentscope-core/src/main/java/io/agentscope/core/session/JournalSessionLog.java

@@ -206,7 +206,7 @@ public Head commit(Writer writer, String batchId, long expectedSeq, List<Session
             // Resolve a lost commit ACK even when a later batch has committed.
             if (h.seq() >= seq) {
                 for (String cursor = h.commitId(); cursor != null; ) {
-                    var c = readCommit(cursor);
+                    var c = readCommit(cursor, null);
                     if (c.batchId().equals(batchId)) {
                         if (!c.hash().equals(hash) || c.firstSeq() != expectedSeq + 1)
                             throw new SessionLogException("Batch ID reused with different content");
@@ -271,8 +271,8 @@ public Head commit(Writer writer, String batchId, long expectedSeq, List<Session
         throw new SessionLogException("Commit contention");
     }
 
-    private Commit readCommit(String id) {
-        var value = storage.read(path("commits/" + id + ".json"));
+    private Commit readCommit(String id, Map<String, AtomicSessionStorage.Value> batch) {
+        var value = committedSegment(id, batch);
         if (value == null) throw new SessionLogException("Missing committed segment: " + id);
         var c = parse(value.bytes(), Commit.class);
         if (c.payloadBlobs() != null && !c.payloadBlobs().isEmpty()) {
@@ -317,17 +317,41 @@ private static SessionEvent withPayload(SessionEvent e, String json) {
                 json);
     }
 
-    @Override
+    /** Bulk-load every committed segment under this journal, or {@code null} when the backend
+     * cannot list paths — callers then fall back to per-segment reads. */
+    private Map<String, AtomicSessionStorage.Value> loadCommitSegments() {
+        try {
+            var paths = storage.listPaths(path("commits") + "/");
+            if (paths.isEmpty()) return Map.of();
+            return storage.readAll(paths);
+        } catch (UnsupportedOperationException | SessionLogException ignored) {
+            // The listing also surfaces rows a concurrent writer has not yet referenced
+            // from head (possibly half-written on file backends); a bulk read that trips
+            // over one must degrade rather than fail the scan.
+            return null;
+        }
+    }
+
+    private AtomicSessionStorage.Value committedSegment(
+            String id, Map<String, AtomicSessionStorage.Value> batch) {
+        if (batch != null) return batch.get(path("commits/" + id + ".json"));
+        return storage.read(path("commits/" + id + ".json"));
+    }
+
     public Iterable<SessionEvent> scan(long after, long through) {
         var head = head();
         if (after < 0 || through < after || through > head.seq())
             throw new IllegalArgumentException("Invalid committed prefix");
+        // One round-trip per journal instead of one per segment: bulk-load the committed
+        // segments up front when the backend can list and batch-read, and walk the commit
+        // chain in memory. Backends without discovery fall back to per-segment reads.
+        Map<String, AtomicSessionStorage.Value> loadedSegments = loadCommitSegments();
         var ids = new ArrayDeque<String>();
         var seen = new HashSet<String>();
         long expected = head.seq();
         for (String id = head.commitId(); id != null && expected > after; ) {
             if (!seen.add(id)) throw new SessionLogException("Commit chain cycle");
-            var stored = storage.read(path("commits/" + id + ".json"));
+            var stored = committedSegment(id, loadedSegments);
             if (stored == null) throw new SessionLogException("Missing committed segment");
             var commit = parse(stored.bytes(), Commit.class);
             if (commit.lastSeq() != expected
@@ -360,7 +384,7 @@ public boolean hasNext() {
                             }
                             if (!segments.hasNext())
                                 throw new SessionLogException("Missing committed events");
-                            batch = readCommit(segments.next()).events().iterator();
+                            batch = readCommit(segments.next(), loadedSegments).events().iterator();
                         }
                     }
 

agentscope-harness/src/main/java/io/agentscope/harness/agent/session/LocalSessionStorage.java

@@ -21,6 +21,7 @@
 import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
 import java.nio.file.Path;
 import java.nio.file.StandardCopyOption;
 import java.nio.file.StandardOpenOption;
@@ -75,6 +76,10 @@ public List<String> listPaths(String prefix) {
                     .map(file -> root.relativize(file).toString().replace('\\', '/'))
                     .sorted()
                     .toList();
+        } catch (NoSuchFileException vanished) {
+            // A concurrent delete removed the directory (or an entry) mid-walk; discovery is a
+            // best-effort snapshot, so report what is safely observable as empty.
+            return List.of();
         } catch (IOException error) {
             throw new SessionLogException("Journal discovery failed: " + directory, error);
         }

agentscope-harness/src/main/java/io/agentscope/harness/agent/session/SqliteSessionStorage.java

@@ -25,7 +25,10 @@
 import java.sql.DriverManager;
 import java.sql.SQLException;
 import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import org.sqlite.SQLiteConfig;
 
 /** Windows local journal storage, using atomic SQL writes and synchronous rollback journals. */
@@ -89,6 +92,30 @@ public Value read(String path) {
         }
     }
 
+    @Override
+    public Map<String, Value> readAll(Collection<String> paths) {
+        if (paths.isEmpty() || !Files.exists(database)) return Map.of();
+        // One connection for the whole batch: replay walks hundreds of segments, and per-read
+        // connection setup dominates that traversal on Windows.
+        try (var connection = connect();
+                var statement =
+                        connection.prepareStatement(
+                                "SELECT path, version, payload FROM session_objects WHERE path"
+                                        + " = ?")) {
+            var batch = new HashMap<String, Value>();
+            for (String path : paths) {
+                statement.setString(1, path);
+                try (var result = statement.executeQuery()) {
+                    if (result.next())
+                        batch.put(path, new Value(result.getLong(2), result.getBytes(3)));
+                }
+            }
+            return Map.copyOf(batch);
+        } catch (SQLException | IOException error) {
+            throw new SessionLogException("Journal batch read failed: " + database, error);
+        }
+    }
+
     @Override
     public List<String> listPaths(String prefix) {
         if (!Files.exists(database)) return List.of();

agentscope-service/service-dataplane/src/test/java/io/agentscope/builder/web/toolbus/ToolConfirmationCoordinatorTest.java

@@ -350,8 +350,15 @@ void managedTimeoutWaitsForControlPlaneCancelledDecision() throws Exception {
                         2,
                         false,
                         "timed_out");
+        // RESOLVED is the outcome when this call releases the continuation itself. The ticket's
+        // own poller thread legitimately picks the persisted decision up at any moment, so a
+        // loaded runner can complete the release (and drop the local waiter) between the durable
+        // resolution and the owner lookup, yielding OWNER_UNAVAILABLE. Both codes mean the
+        // decision was accepted; a control-plane retry would observe IDEMPOTENT.
         assertThat(coordinator.resolveManaged(cancelled))
-                .isEqualTo(ToolConfirmationCoordinator.DecisionResult.RESOLVED);
+                .isIn(
+                        ToolConfirmationCoordinator.DecisionResult.RESOLVED,
+                        ToolConfirmationCoordinator.DecisionResult.OWNER_UNAVAILABLE);
         assertThat(waiting.get(1, TimeUnit.SECONDS)).isFalse();
         assertThat(ticket.get().resolutionStatus()).isEqualTo("cancelled");