fix(agent): serialise progress-ledger upsert per conversation

mateclaw-server/src/main/java/vip/mate/agent/progress/ProgressLedgerService.java

@@ -13,6 +13,7 @@ import vip.mate.workspace.conversation.repository.ConversationMapper;
 import java.time.Instant;
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * Loader / writer for the per-conversation progress ledger persisted as a
@@ -36,6 +37,23 @@ public class ProgressLedgerService {
     private static final TypeReference<LinkedHashMap<String, ProgressEntry>> LEDGER_TYPE =
             new TypeReference<>() {};
 
+    /**
+     * Per-conversation mutex for the load-mutate-save sequence inside
+     * {@link #upsert}. Without this guard, a single agent turn that issues
+     * N parallel {@code progress_update} tool calls (observed: 12 calls in
+     * one batch when the model pre-registered every step at task start)
+     * collapses to last-writer-wins, losing every entry but one — defeating
+     * the whole point of the ledger. Different conversations stay
+     * uncontended; only intra-conversation writes serialise.
+     *
+     * <p>Entries are computed on demand and never explicitly removed; even
+     * with thousands of long-running conversations the map stays bounded by
+     * the active conversation set, and any leak is a {@code Object} per
+     * conversation id — small enough to ignore relative to the rest of the
+     * per-conv state already held in memory.
+     */
+    private final ConcurrentHashMap<String, Object> upsertLocks = new ConcurrentHashMap<>();
+
     private final ConversationMapper conversationMapper;
     private final ObjectMapper objectMapper;
 
@@ -47,14 +65,32 @@ public class ProgressLedgerService {
         if (conversationId == null || conversationId.isBlank()) {
             return ProgressLedger.empty();
         }
+        return parse(loadLedgerJson(conversationId));
+    }
+
+    /**
+     * Read the raw JSON column for one conversation, or {@code null} when
+     * the row or column is empty. Protected so concurrency tests can
+     * subclass and back the service with an in-memory map without having
+     * to mock the Mybatis-Plus wrapper internals.
+     */
+    protected String loadLedgerJson(String conversationId) {
         ConversationEntity row = conversationMapper.selectOne(
                 new LambdaQueryWrapper<ConversationEntity>()
                         .eq(ConversationEntity::getConversationId, conversationId)
                         .select(ConversationEntity::getProgressLedger));
-        if (row == null) {
+        return row != null ? row.getProgressLedger() : null;
-            return ProgressLedger.empty();
+    }
-        }
+
-        return parse(row.getProgressLedger());
+    /**
+     * Write the raw JSON column for one conversation. Protected for the
+     * same reason as {@link #loadLedgerJson}.
+     */
+    protected void saveLedgerJson(String conversationId, String json) {
+        conversationMapper.update(null,
+                new LambdaUpdateWrapper<ConversationEntity>()
+                        .eq(ConversationEntity::getConversationId, conversationId)
+                        .set(ConversationEntity::getProgressLedger, json));
     }
 
     /**
@@ -74,15 +110,24 @@ public class ProgressLedgerService {
         if (status == null) {
             throw new IllegalArgumentException("status is required");
         }
-        ProgressLedger ledger = load(conversationId);
+        // Serialise the load-mutate-save sequence per conversation. Without
-        Map<String, ProgressEntry> map = ledger.asMap();
+        // this, two parallel @Tool calls on the same conversation race: both
-        ProgressEntry existing = map.get(key);
+        // read the same starting state, each adds its own entry, and the
-        String effectiveLabel = (label != null && !label.isBlank())
+        // last save() drops the other's entry. Observed in production: a
-                ? label
+        // 12-entry pre-registration collapsed to 8 because four sibling
-                : (existing != null ? existing.getLabel() : key);
+        // tool calls landed in the same window.
-        map.put(key, new ProgressEntry(key, effectiveLabel, status, note, Instant.now()));
+        Object mutex = upsertLocks.computeIfAbsent(conversationId, k -> new Object());
-        persist(conversationId, map);
+        synchronized (mutex) {
-        return new ProgressLedger(map);
+            ProgressLedger ledger = load(conversationId);
+            Map<String, ProgressEntry> map = ledger.asMap();
+            ProgressEntry existing = map.get(key);
+            String effectiveLabel = (label != null && !label.isBlank())
+                    ? label
+                    : (existing != null ? existing.getLabel() : key);
+            map.put(key, new ProgressEntry(key, effectiveLabel, status, note, Instant.now()));
+            persist(conversationId, map);
+            return new ProgressLedger(map);
+        }
     }
 
     private ProgressLedger parse(String json) {
@@ -101,10 +146,7 @@ public class ProgressLedgerService {
     private void persist(String conversationId, Map<String, ProgressEntry> map) {
         try {
             String json = objectMapper.writeValueAsString(map);
-            conversationMapper.update(null,
+            saveLedgerJson(conversationId, json);
-                    new LambdaUpdateWrapper<ConversationEntity>()
-                            .eq(ConversationEntity::getConversationId, conversationId)
-                            .set(ConversationEntity::getProgressLedger, json));
         } catch (Exception e) {
             // Surface to caller so the tool can return an error message to
             // the LLM rather than silently dropping the update.

mateclaw-server/src/test/java/vip/mate/agent/progress/ProgressLedgerServiceConcurrencyTest.java

@@ -0,0 +1,146 @@
+package vip.mate.agent.progress;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Pins the per-conversation mutex inside
+ * {@link ProgressLedgerService#upsert} — the load-mutate-save sequence
+ * must serialise per conversation, otherwise N parallel
+ * {@code progress_update} tool calls on the same conversation collapse
+ * to last-writer-wins and silently drop entries.
+ *
+ * <p>Repro: in round-3 of the LLM-review test, the model pre-registered
+ * 12 entries in a single batch of parallel tool calls; only 7-8 survived
+ * to the DB, the rest were lost, and the agent later re-did completed
+ * work because the snapshot it saw was missing the pending entries.
+ *
+ * <p>Uses an in-memory subclass of the service rather than mocking
+ * Mybatis-Plus: the JSON I/O methods are protected for exactly this
+ * purpose.
+ */
+class ProgressLedgerServiceConcurrencyTest {
+
+    /**
+     * Test double — overrides the two protected DB methods to read/write a
+     * thread-safe in-memory map. The {@code upsert} logic (including the
+     * per-conversation mutex under test) inherits unchanged from the
+     * parent.
+     */
+    private static final class InMemoryProgressLedgerService extends ProgressLedgerService {
+        private final Map<String, String> store = new ConcurrentHashMap<>();
+
+        InMemoryProgressLedgerService() {
+            super(null, new ObjectMapper().registerModule(new JavaTimeModule()));
+        }
+
+        @Override
+        protected String loadLedgerJson(String conversationId) {
+            return store.get(conversationId);
+        }
+
+        @Override
+        protected void saveLedgerJson(String conversationId, String json) {
+            store.put(conversationId, json);
+        }
+    }
+
+    @Test
+    @DisplayName("12 parallel upserts on one conversation all survive — no last-writer-wins drops.")
+    void parallelUpsertsAllSurvive() throws Exception {
+        ProgressLedgerService service = new InMemoryProgressLedgerService();
+        String conv = "conv-race-1";
+
+        int n = 12;
+        ExecutorService pool = Executors.newFixedThreadPool(n);
+        CountDownLatch start = new CountDownLatch(1);
+        CountDownLatch done = new CountDownLatch(n);
+        AtomicInteger failures = new AtomicInteger();
+
+        for (int i = 0; i < n; i++) {
+            final int idx = i;
+            pool.submit(() -> {
+                try {
+                    start.await();
+                    service.upsert(conv, "step_" + idx, "Step " + idx, ProgressStatus.PENDING, null);
+                } catch (Exception e) {
+                    failures.incrementAndGet();
+                } finally {
+                    done.countDown();
+                }
+            });
+        }
+        start.countDown();
+        assertTrue(done.await(10, TimeUnit.SECONDS), "all upsert threads must finish within 10s");
+        pool.shutdown();
+
+        assertEquals(0, failures.get(), "no thread should fail");
+        ProgressLedger finalLedger = service.load(conv);
+        assertEquals(n, finalLedger.size(),
+                "all " + n + " parallel entries must survive; got " + finalLedger.size()
+                        + " — keys=" + finalLedger.asMap().keySet());
+        for (int i = 0; i < n; i++) {
+            assertTrue(finalLedger.asMap().containsKey("step_" + i),
+                    "expected key step_" + i + " in final ledger");
+        }
+    }
+
+    @Test
+    @DisplayName("Parallel upserts on DIFFERENT conversations do not contend.")
+    void differentConversationsAreIndependent() throws Exception {
+        ProgressLedgerService service = new InMemoryProgressLedgerService();
+
+        ExecutorService pool = Executors.newFixedThreadPool(2);
+        CountDownLatch done = new CountDownLatch(2);
+
+        pool.submit(() -> {
+            for (int i = 0; i < 5; i++) {
+                service.upsert("conv-A", "a_" + i, "A " + i, ProgressStatus.DONE, null);
+            }
+            done.countDown();
+        });
+        pool.submit(() -> {
+            for (int i = 0; i < 5; i++) {
+                service.upsert("conv-B", "b_" + i, "B " + i, ProgressStatus.DONE, null);
+            }
+            done.countDown();
+        });
+        assertTrue(done.await(5, TimeUnit.SECONDS));
+        pool.shutdown();
+
+        assertEquals(5, service.load("conv-A").size());
+        assertEquals(5, service.load("conv-B").size());
+    }
+
+    @Test
+    @DisplayName("Sequential updates on the same key advance status in order.")
+    void sequentialStatusTransitions() {
+        ProgressLedgerService service = new InMemoryProgressLedgerService();
+        String conv = "conv-X";
+
+        service.upsert(conv, "step_a", "Step A", ProgressStatus.PENDING, null);
+        service.upsert(conv, "step_a", null, ProgressStatus.IN_PROGRESS, "working");
+        service.upsert(conv, "step_a", null, ProgressStatus.DONE, "finished");
+
+        ProgressLedger ledger = service.load(conv);
+        assertEquals(1, ledger.size());
+        ProgressEntry e = ledger.asMap().get("step_a");
+        assertEquals(ProgressStatus.DONE, e.getStatus());
+        // Label survives the null-label updates by falling back to existing value.
+        assertEquals("Step A", e.getLabel());
+        assertEquals("finished", e.getNote());
+    }
+}