From 9f89db8262d4b4944c8354a993fc705f997fb1fc Mon Sep 17 00:00:00 2001 From: mateaix <7333791@qq.com> Date: Mon, 14 Sep 2026 00:46:10 +0800 Subject: [PATCH] fix(evidence): seal observation state and rows atomically --- .../service/ExecutionEvidenceRecorder.java | 9 ++-- .../service/ExecutionObservationSink.java | 10 +++++ .../ExecutionEvidenceRecorderTest.java | 41 +++++++++++++++++++ 3 files changed, 56 insertions(+), 4 deletions(-) diff --git a/mateclaw-server/src/main/java/vip/mate/execution/evidence/service/ExecutionEvidenceRecorder.java b/mateclaw-server/src/main/java/vip/mate/execution/evidence/service/ExecutionEvidenceRecorder.java index fa44fec4..2f78a0f7 100644 --- a/mateclaw-server/src/main/java/vip/mate/execution/evidence/service/ExecutionEvidenceRecorder.java +++ b/mateclaw-server/src/main/java/vip/mate/execution/evidence/service/ExecutionEvidenceRecorder.java @@ -81,7 +81,7 @@ public class ExecutionEvidenceRecorder { observedContext = sink.attach(new ToolContext(values)); } String result = callback.call(arguments, observedContext); - finish(attempt, sink, sink.state(), "Tool callback returned"); + finish(attempt, sink, null, "Tool callback returned"); return result; } catch (RuntimeException | Error error) { AttemptState state = error instanceof CancellationException || Thread.currentThread().isInterrupted() @@ -91,8 +91,9 @@ public class ExecutionEvidenceRecorder { } } - private void finish(ExecutionAttempt attempt, ExecutionObservationSink sink, AttemptState state, String summary) { - sink.seal(); + private void finish(ExecutionAttempt attempt, ExecutionObservationSink sink, AttemptState overrideState, String summary) { + var captured = sink.sealAndSnapshot(); + AttemptState state = overrideState != null ? overrideState : captured.state(); if (attempt == null) return; long began = System.nanoTime(); try { @@ -100,7 +101,7 @@ public class ExecutionEvidenceRecorder { failure("owner_lost"); return; } - var observations = new ArrayList<>(sink.observations()); + var observations = new ArrayList<>(captured.observations()); observations.add(new EvidenceObservation("callback", EvidenceKind.TOOL_RETURNED, state == AttemptState.SUCCEEDED ? EvidenceResult.OBSERVED : state == AttemptState.UNKNOWN || state == AttemptState.CANCELLED diff --git a/mateclaw-server/src/main/java/vip/mate/execution/evidence/service/ExecutionObservationSink.java b/mateclaw-server/src/main/java/vip/mate/execution/evidence/service/ExecutionObservationSink.java index f4891273..3413e532 100644 --- a/mateclaw-server/src/main/java/vip/mate/execution/evidence/service/ExecutionObservationSink.java +++ b/mateclaw-server/src/main/java/vip/mate/execution/evidence/service/ExecutionObservationSink.java @@ -46,6 +46,16 @@ public final class ExecutionObservationSink { public synchronized List observations() { return List.copyOf(observations); } public synchronized void seal() { sealed = true; } + public record Snapshot(AttemptState state, List observations) { + public Snapshot { observations = List.copyOf(observations); } + } + + /** State and rows share one linearization point; later observations are ignored. */ + public synchronized Snapshot sealAndSnapshot() { + sealed = true; + return new Snapshot(state, observations); + } + /** Called by the process adapter, never by parsing a tool's returned text. */ public void command(Integer exitCode, boolean timedOut, boolean cancelled, boolean blocked) { command(exitCode, timedOut, cancelled, blocked, null); diff --git a/mateclaw-server/src/test/java/vip/mate/execution/evidence/ExecutionEvidenceRecorderTest.java b/mateclaw-server/src/test/java/vip/mate/execution/evidence/ExecutionEvidenceRecorderTest.java index cbcf41da..0282d1fb 100644 --- a/mateclaw-server/src/test/java/vip/mate/execution/evidence/ExecutionEvidenceRecorderTest.java +++ b/mateclaw-server/src/test/java/vip/mate/execution/evidence/ExecutionEvidenceRecorderTest.java @@ -145,6 +145,47 @@ class ExecutionEvidenceRecorderTest { assertEquals(AttemptState.CANCELLED, cancelled.state()); } + @Test void observationArrivingImmediatelyBeforeSealCannotDisagreeWithStoredState() throws Exception { + var actualSink = new ExecutionObservationSink(false); + when(callback.call(anyString(), any())).thenReturn("callback returned"); + // Deterministically inject the legal interleaving: an observer arrives + // immediately before sealing, after the old recorder read state(). + try (var constructed = mockConstruction(ExecutionObservationSink.class, withSettings().defaultAnswer(call -> { + if (call.getMethod().getName().startsWith("seal")) { + actualSink.command(7, false, false, false); + } + return call.getMethod().invoke(actualSink, call.getArguments()); + }))) { + assertEquals("callback returned", invoke()); + assertEquals(1, constructed.constructed().size()); + verify(store).finish(eq(1L), eq("fence"), eq(AttemptState.FAILED), eq(EffectOutcome.UNCERTAIN), + argThat(rows -> rows.size() == 2 && rows.stream().allMatch(row -> row.result() == EvidenceResult.FAIL))); + } + } + + @Test void sealedSnapshotCannotBeChangedByLateObserversOrItsReader() { + var sink = new ExecutionObservationSink(false); + sink.command(7, false, false, false); + var captured = sink.sealAndSnapshot(); + sink.command(0, false, false, false); + sink.artifact("late", "digest", 1, "text/plain", Instant.now()); + assertEquals(AttemptState.FAILED, captured.state()); + assertEquals(1, captured.observations().size()); + assertEquals(captured, sink.sealAndSnapshot()); + assertThrows(UnsupportedOperationException.class, () -> captured.observations().clear()); + } + + @Test void callbackCancellationOverridesSuccessfulCommandSnapshot() { + when(callback.call(anyString(), any())).thenAnswer(call -> { + ExecutionObservationSink.from(call.getArgument(1)).command(0, false, false, false); + throw new java.util.concurrent.CancellationException("cancelled"); + }); + assertThrows(java.util.concurrent.CancellationException.class, this::invoke); + verify(store).finish(eq(1L), eq("fence"), eq(AttemptState.CANCELLED), eq(EffectOutcome.UNCERTAIN), + argThat(rows -> rows.size() == 2 && rows.getFirst().result() == EvidenceResult.OBSERVED + && rows.getLast().result() == EvidenceResult.UNKNOWN)); + } + private String invoke() { return recorder.invoke(callback, "{}", ChatOrigin.web("conv", "owner", 1L, null).toToolContext(), "invocation", "provider-id"); }