From 7736f6b0ab29b720c97bf083a10ce00bca97dfa1 Mon Sep 17 00:00:00 2001 From: matevip Date: Sun, 24 May 2026 23:00:17 +0800 Subject: [PATCH] fix(agent): serialise progress-ledger upsert per conversation --- .../agent/progress/ProgressLedgerService.java | 76 +++++++-- .../ProgressLedgerServiceConcurrencyTest.java | 146 ++++++++++++++++++ 2 files changed, 205 insertions(+), 17 deletions(-) create mode 100644 mateclaw-server/src/test/java/vip/mate/agent/progress/ProgressLedgerServiceConcurrencyTest.java diff --git a/mateclaw-server/src/main/java/vip/mate/agent/progress/ProgressLedgerService.java b/mateclaw-server/src/main/java/vip/mate/agent/progress/ProgressLedgerService.java index 69426c97..ec27c78a 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/progress/ProgressLedgerService.java +++ b/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> 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. + * + *

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 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() .eq(ConversationEntity::getConversationId, conversationId) .select(ConversationEntity::getProgressLedger)); - if (row == null) { - return ProgressLedger.empty(); - } - return parse(row.getProgressLedger()); + return row != null ? row.getProgressLedger() : null; + } + + /** + * 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() + .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); - Map 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); + // Serialise the load-mutate-save sequence per conversation. Without + // this, two parallel @Tool calls on the same conversation race: both + // read the same starting state, each adds its own entry, and the + // last save() drops the other's entry. Observed in production: a + // 12-entry pre-registration collapsed to 8 because four sibling + // tool calls landed in the same window. + Object mutex = upsertLocks.computeIfAbsent(conversationId, k -> new Object()); + synchronized (mutex) { + ProgressLedger ledger = load(conversationId); + Map 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 map) { try { String json = objectMapper.writeValueAsString(map); - conversationMapper.update(null, - new LambdaUpdateWrapper() - .eq(ConversationEntity::getConversationId, conversationId) - .set(ConversationEntity::getProgressLedger, json)); + saveLedgerJson(conversationId, json); } catch (Exception e) { // Surface to caller so the tool can return an error message to // the LLM rather than silently dropping the update. diff --git a/mateclaw-server/src/test/java/vip/mate/agent/progress/ProgressLedgerServiceConcurrencyTest.java b/mateclaw-server/src/test/java/vip/mate/agent/progress/ProgressLedgerServiceConcurrencyTest.java new file mode 100644 index 00000000..53a0cfc8 --- /dev/null +++ b/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. + * + *

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. + * + *

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 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()); + } +}