fix(evidence): seal observation state and rows atomically

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) {
+    private void finish(ExecutionAttempt attempt, ExecutionObservationSink sink, AttemptState overrideState, String summary) {
-        sink.seal();
+        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

mateclaw-server/src/main/java/vip/mate/execution/evidence/service/ExecutionObservationSink.java

@@ -46,6 +46,16 @@ public final class ExecutionObservationSink {
     public synchronized List<EvidenceObservation> observations() { return List.copyOf(observations); }
     public synchronized void seal() { sealed = true; }
 
+    public record Snapshot(AttemptState state, List<EvidenceObservation> 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);

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");
     }