diff --git a/mateclaw-server/src/main/java/vip/mate/agent/BaseAgent.java b/mateclaw-server/src/main/java/vip/mate/agent/BaseAgent.java index 0c264d44..cd9b0065 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/BaseAgent.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/BaseAgent.java @@ -13,6 +13,7 @@ import org.springframework.util.MimeType; import reactor.core.publisher.Flux; import vip.mate.agent.context.ChatOrigin; import vip.mate.agent.context.ChatOriginHolder; +import vip.mate.agent.context.GoalContinuationContext; import vip.mate.approval.ApprovalPlaceholderUtil; import vip.mate.llm.model.ModelConfigEntity; import vip.mate.llm.routing.MediaCaptionService; @@ -1264,6 +1265,12 @@ public abstract class BaseAgent { * the primary model can't already handle. */ protected CurrentTurnUserMessage buildCurrentUserMessageWithRouting(String conversationId, String userMessageText) { + // Autonomous segments have no new persisted user row. Reconstructing + // from the last user would replace the continuation/recovery instruction. + // History is still loaded normally; queued user turns retain attachment routing. + if (GoalContinuationContext.explicitPrompt()) { + return new CurrentTurnUserMessage(new UserMessage(userMessageText), null); + } // Scheduled-job run (issue #142): the task text is the explicit // userMessageText argument. Never reconstruct it from the conversation // — a shared cron conversation under concurrent runs has no reliable diff --git a/mateclaw-server/src/main/java/vip/mate/agent/context/GoalContinuationContext.java b/mateclaw-server/src/main/java/vip/mate/agent/context/GoalContinuationContext.java index 0d2cb810..970469e4 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/context/GoalContinuationContext.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/context/GoalContinuationContext.java @@ -1,14 +1,20 @@ package vip.mate.agent.context; +import java.util.function.Supplier; + /** Subscription-time marker; callers capture it before asynchronous lifecycle callbacks. */ public final class GoalContinuationContext { - private static final ThreadLocal ACTIVE = new ThreadLocal<>(); + private static final ThreadLocal EXPLICIT_PROMPT = new ThreadLocal<>(); private GoalContinuationContext() {} - public static boolean active() { return Boolean.TRUE.equals(ACTIVE.get()); } - public static T call(java.util.function.Supplier action) { - Boolean previous=ACTIVE.get(); - ACTIVE.set(true); + public static boolean active() { return EXPLICIT_PROMPT.get() != null; } + public static boolean explicitPrompt() { return Boolean.TRUE.equals(EXPLICIT_PROMPT.get()); } + public static T call(Supplier action) { return call(true, action); } + + /** Queued user input keeps normal attachment reconstruction within the same worker. */ + public static T call(boolean explicitPrompt, Supplier action) { + Boolean previous=EXPLICIT_PROMPT.get(); + EXPLICIT_PROMPT.set(explicitPrompt); try { return action.get(); } - finally { if(previous==null) ACTIVE.remove(); else ACTIVE.set(previous); } + finally { if(previous==null) EXPLICIT_PROMPT.remove(); else EXPLICIT_PROMPT.set(previous); } } } diff --git a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalContinuationSupervisor.java b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalContinuationSupervisor.java index 0611f2fb..469eb04e 100644 --- a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalContinuationSupervisor.java +++ b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalContinuationSupervisor.java @@ -92,6 +92,7 @@ public class GoalContinuationSupervisor { private void execute(GoalEntity initial, GoalContinuationStore.Continuation candidate, String token) { LocalDateTime now = LocalDateTime.now(clock); try { + if (closing) return; GoalEntity goal = goals.getById(initial.getId()); if (!eligible(goal)) { settle(initial,token,"paused",now,0,"goal_not_runnable"); return; @@ -112,6 +113,8 @@ public class GoalContinuationSupervisor { case CONTINUE -> { } } GoalSegmentRunner.Result result = runner.run(goal,decision.prompt(),"running".equals(candidate.state())); + // Shutdown cancellation is not user Stop: retain the lease for recovery. + if (closing) return; GoalEntity fresh = goals.getById(goal.getId()); if (fresh.getStatus() == GoalStatus.COMPLETED) { settle(goal,token,"completed",now,0,"goal_completed"); @@ -209,8 +212,10 @@ public class GoalContinuationSupervisor { @PreDestroy public void close() { closing=true; + runner.cancelAll(); if (executor instanceof java.util.concurrent.ExecutorService workers) { - workers.shutdownNow(); + // Cancellation must finish checkpoint persistence without interrupting JDBC I/O. + workers.shutdown(); try { if (!workers.awaitTermination(10,java.util.concurrent.TimeUnit.SECONDS)) { log.warn("Goal workers did not finish shutdown persistence within 10 seconds"); diff --git a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalEvaluationService.java b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalEvaluationService.java index 5aca7f1d..7b3a926f 100644 --- a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalEvaluationService.java +++ b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalEvaluationService.java @@ -131,7 +131,9 @@ public class GoalEvaluationService implements Evaluator { + "\n\n" + format; List messages = new ArrayList<>(2); - messages.add(new SystemMessage(bootstrap ? BOOTSTRAP_SYSTEM_PROMPT : VERDICT_SYSTEM_PROMPT)); + messages.add(new SystemMessage(bootstrap ? BOOTSTRAP_SYSTEM_PROMPT + : Boolean.TRUE.equals(goal.getPersistentExecution()) + ? PERSISTENT_VERDICT_SYSTEM_PROMPT : VERDICT_SYSTEM_PROMPT)); messages.add(new UserMessage(userPrompt)); ChatOptions options = ChatOptions.builder() @@ -219,6 +221,17 @@ public class GoalEvaluationService implements Evaluator { + "'all requirements met'. If a criterion lacks specific evidence, " + "mark it not passed. Output only the requested JSON."; + private static final String PERSISTENT_VERDICT_SYSTEM_PROMPT = + "Judge cumulative progress toward a persistent goal using the latest reply, " + + "conversation evidence and previously verified checklist evidence. " + + "A later step need not repeat completed earlier work. Preserve a prior " + + "passed=true item with nonblank evidence unless new concrete evidence contradicts it. " + + "Revoke it when such contradictory evidence exists, citing that evidence. " + + "An attempted action, a goal description or a claim of completion is not proof. " + + "Newly passed criteria require concrete observable evidence. Return only changed " + + "criterion verdicts; omitted criteria retain their previous state. Keep evidence concise. " + + "Output only the requested JSON."; + private String buildUserPrompt(GoalEntity goal, List existing, List recentMessages, @@ -238,6 +251,12 @@ public class GoalEvaluationService implements Evaluator { sb.append("Current checklist (judge each by id):\n"); for (GoalCriterion c : existing) { sb.append("- ").append(c.id()).append(": ").append(c.text()).append('\n'); + if (Boolean.TRUE.equals(goal.getPersistentExecution())) { + String evidence = safe(c.evidence()); + sb.append(" Previous passed=").append(c.passed()).append("; evidence: ") + .append(evidence.length() > 1000 ? evidence.substring(0, 1000) + " [truncated]" : evidence) + .append('\n'); + } } sb.append('\n'); } @@ -263,6 +282,10 @@ public class GoalEvaluationService implements Evaluator { .append(MAX_BOOTSTRAP_CRITERIA) .append(" criteria. Leave every 'passed' false and 'evidence' empty — " + "this round only defines the checklist."); + } else if (Boolean.TRUE.equals(goal.getPersistentExecution())) { + sb.append("Return only changed criteria with specific evidence. Preserve verified prior work " + + "unless new evidence contradicts it; absence from the latest reply is not a contradiction. " + + "Never treat passed=true without nonblank evidence as verified completion."); } else { sb.append("For every criterion above, return its id with passed=true ONLY when " + "the reply shows concrete evidence; otherwise passed=false with a short " diff --git a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalSegmentRunner.java b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalSegmentRunner.java index bc50196a..7bc9707c 100644 --- a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalSegmentRunner.java +++ b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalSegmentRunner.java @@ -5,6 +5,7 @@ import org.springframework.stereotype.Component; import reactor.core.Disposable; import vip.mate.agent.AgentService; import vip.mate.agent.context.ChatOrigin; +import vip.mate.agent.context.GoalContinuationContext; import vip.mate.agent.runtime.ConversationTurnGate; import vip.mate.approval.ApprovalWorkflowService; import vip.mate.channel.web.AgentStreamAccumulator; @@ -30,10 +31,11 @@ public class GoalSegmentRunner { private final ObjectMapper mapper; private final ConversationTurnGate gate; private final ConcurrentHashMap workers=new ConcurrentHashMap<>(); + private volatile boolean closing; private static final class Worker { final String conversationId; - final Thread thread=Thread.currentThread(); final AtomicBoolean cancelled=new AtomicBoolean(); + final AtomicBoolean interrupted=new AtomicBoolean(); final AtomicReference handle=new AtomicReference<>(); volatile boolean interactive; Worker(String conversationId) { this.conversationId=conversationId; } @@ -64,10 +66,16 @@ public class GoalSegmentRunner { workers.values().stream().filter(w -> Objects.equals(w.conversationId,conversationId)).forEach(this::cancelWorker); } + public void cancelAll() { + closing=true; + workers.values().forEach(this::cancelWorker); + } + private void cancelWorker(Worker worker) { worker.cancelled.set(true); streams.cancelRun(worker.handle.get()); - worker.thread.interrupt(); + // Stream disposal releases the completion latch. Never interrupt this + // worker: it also performs JDBC I/O on shared embedded database channels. } public Result run(GoalEntity goal, String prompt, boolean recovered) { @@ -77,6 +85,11 @@ public class GoalSegmentRunner { Worker worker=new Worker(convId); try { workers.put(goal.getId(),worker); + // Register before checking the shutdown fence so cancellation cannot miss us. + if (closing) { + worker.cancelled.set(true); + return new Result("stopped",false); + } var conv=conversations.findByConversationId(convId); if (conv==null || !Objects.equals(conv.getWorkspaceId(),goal.getWorkspaceId()) || !Objects.equals(conv.getAgentId(),goal.getAgentId()) @@ -121,6 +134,7 @@ public class GoalSegmentRunner { } while (queued!=null); return result; } catch (RuntimeException error) { + if (Thread.interrupted()) worker.interrupted.set(true); // Accepted user input must survive even when this goal cannot continue. ChatStreamTracker.QueuedInput pending; while ((pending=streams.consumeQueuedInput(convId))!=null) { @@ -130,8 +144,20 @@ public class GoalSegmentRunner { "Goal execution interrupted. Queued input was saved; review the execution state before resuming.")); throw error; } finally { - workers.remove(goal.getId(),worker); - permit.close(); + try { + // Cooperative cancellation returns normally, bypassing the error path. + // Accepted input must still survive process exit, including attachments. + if (worker.cancelled.get()) { + ChatStreamTracker.QueuedInput pending; + while ((pending=streams.consumeQueuedInput(convId))!=null) { + if (!pending.persisted()) conversations.saveMessage(convId,"user",pending.message(),pending.contentParts(),"queued"); + } + } + } finally { + workers.remove(goal.getId(),worker); + permit.close(); + if (worker.interrupted.get()) Thread.currentThread().interrupt(); + } } } @@ -153,7 +179,7 @@ public class GoalSegmentRunner { if (worker.cancelled.get() || Thread.currentThread().isInterrupted()) throw new InterruptedException(); conversations.updateStreamStatus(convId,"running"); streams.broadcastObject(convId,"message_start",Map.of("role","assistant","trigger","goal")); - subscription=gate.withPermit(permit,() -> vip.mate.agent.context.GoalContinuationContext.call(() -> + subscription=gate.withPermit(permit,() -> GoalContinuationContext.call(!worker.interactive, () -> reactor.core.publisher.Flux.defer(() -> { if (worker.cancelled.get()) return reactor.core.publisher.Flux.empty(); return agents.chatStructuredStream(goal.getAgentId(),input, @@ -184,10 +210,11 @@ public class GoalSegmentRunner { } return new Result(reason,accumulator.isAwaitingApproval(),evaluationUnavailable.get()); } catch (InterruptedException interrupted) { - Thread.currentThread().interrupt(); + worker.interrupted.set(true); throw new IllegalStateException("Goal worker interrupted; recover from persisted evidence",interrupted); } finally { - if (worker.cancelled.get() || Thread.currentThread().isInterrupted()) streams.cancelRun(handle); + if (Thread.interrupted()) worker.interrupted.set(true); + if (worker.cancelled.get() || worker.interrupted.get()) streams.cancelRun(handle); if (subscription!=null) subscription.dispose(); try { if (!persisted.get()) persist(convId,accumulator,"interrupted"); diff --git a/mateclaw-server/src/test/java/vip/mate/agent/BaseAgentCronIsolationTest.java b/mateclaw-server/src/test/java/vip/mate/agent/BaseAgentCronIsolationTest.java index bdc88a6a..7e56d2a4 100644 --- a/mateclaw-server/src/test/java/vip/mate/agent/BaseAgentCronIsolationTest.java +++ b/mateclaw-server/src/test/java/vip/mate/agent/BaseAgentCronIsolationTest.java @@ -6,12 +6,14 @@ import org.junit.jupiter.api.Test; import org.springframework.ai.chat.messages.Message; import vip.mate.agent.context.ChatOrigin; import vip.mate.agent.context.ChatOriginHolder; +import vip.mate.agent.context.GoalContinuationContext; import vip.mate.workspace.conversation.ConversationService; import vip.mate.workspace.conversation.model.MessageEntity; import java.util.List; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; @@ -98,6 +100,45 @@ class BaseAgentCronIsolationTest { "a normal turn keeps prior history (the trailing current user row is de-duplicated)"); } + @Test + void goalContinuation_keepsHistoryButUsesExplicitInstruction() { + ConversationService conv = mock(ConversationService.class); + List stored = List.of(user("Reply READY"), assistant("READY")); + when(conv.countMessages("goal_conv")).thenReturn((long) stored.size()); + when(conv.listMessages("goal_conv")).thenReturn(stored); + stubRender(conv); + TestAgent agent = newAgent(conv); + ChatOriginHolder.set(ChatOrigin.web("goal_conv", "u1", 1L, null)); + + GoalContinuationContext.call(() -> { + assertEquals(2, agent.history("goal_conv", "Continue with checkpoint 1").size()); + assertEquals("Continue with checkpoint 1", + agent.currentMessage("goal_conv", "Continue with checkpoint 1"), + "the durable continuation must not replay the last persisted user instruction"); + return null; + }); + } + + @Test + void queuedUserTurn_reconstructsStoredContentAndRestoresContinuationScope() { + ConversationService conv = mock(ConversationService.class); + when(conv.listMessages("goal_conv")).thenReturn(List.of(user("Current user with attachment text"))); + stubRender(conv); + TestAgent agent = newAgent(conv); + + GoalContinuationContext.call(() -> { + GoalContinuationContext.call(false, () -> { + assertTrue(GoalContinuationContext.active()); + assertEquals("Current user with attachment text", agent.currentMessage("goal_conv", "fallback")); + return null; + }); + assertEquals("Continue after queued input", agent.currentMessage("goal_conv", "Continue after queued input")); + return null; + }); + assertFalse(GoalContinuationContext.active()); + assertFalse(GoalContinuationContext.explicitPrompt()); + } + // ---------- scaffold ---------- private static TestAgent newAgent(ConversationService conv) { diff --git a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalContinuationSupervisorTest.java b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalContinuationSupervisorTest.java index 3f30a079..d7c59731 100644 --- a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalContinuationSupervisorTest.java +++ b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalContinuationSupervisorTest.java @@ -68,6 +68,16 @@ class GoalContinuationSupervisorTest { verify(store).settle(eq(1L),anyString(),eq("completed"),any(),eq(0),anyString()); } + @Test void shutdownCancellationLeavesLeaseForRestartRecovery() { + when(runner.run(any(),anyString(),anyBoolean())).thenAnswer(inv -> { + supervisor.close(); + return new GoalSegmentRunner.Result("stopped",false); + }); + supervisor.tick(); + verify(runner).cancelAll(); + verify(store,never()).settle(any(),anyString(),anyString(),any(),anyInt(),anyString()); + } + @Test void budgetReachedDuringSegmentIsReportedAsResumableBudgetLimit() { when(runner.run(any(),anyString(),anyBoolean())).thenAnswer(inv -> { goal.setStatus(GoalStatus.PAUSED); diff --git a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalEvaluationServiceTest.java b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalEvaluationServiceTest.java index 0a4cd348..e254b499 100644 --- a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalEvaluationServiceTest.java +++ b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalEvaluationServiceTest.java @@ -5,6 +5,7 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.Mock; +import org.mockito.ArgumentCaptor; import org.mockito.junit.jupiter.MockitoExtension; import org.springframework.ai.chat.messages.AssistantMessage; import org.springframework.ai.chat.messages.UserMessage; @@ -95,6 +96,25 @@ class GoalEvaluationServiceTest { // ==================== Pre-flight guards ==================== + @Test + void persistentVerdictReceivesPriorVerifiedEvidenceOutsideConversationWindow() { + GoalEntity g = goalWithCriteria(); + g.setPersistentExecution(true); + g.setCriteria("[{\"id\":\"C1\",\"text\":\"DNS configured\",\"passed\":true,\"evidence\":\"verified DNS checkpoint\"}," + + "{\"id\":\"C2\",\"text\":\"TLS enabled\",\"passed\":false,\"evidence\":\"\"}]"); + stubChatResponse("{\"criterionVerdicts\":[{\"id\":\"C2\",\"passed\":true,\"evidence\":\"TLS handshake verified\"}],\"summary\":\"done\"}"); + + GoalEvaluationResult result = svc.evaluate(g,List.of(),"TLS handshake verified"); + + ArgumentCaptor request = ArgumentCaptor.forClass(Prompt.class); + verify(chatModel).call(request.capture()); + String prompt = request.getValue().getContents(); + assertTrue(prompt.contains("verified DNS checkpoint"), "persistent evaluation must retain prior evidence after history truncation"); + assertTrue(prompt.contains("passed=true")); + assertTrue(prompt.contains("contradicts")); + assertTrue(result.completed(), "a new verified step can complete previously verified work without repeating it"); + } + @Test void nullGoal_returnsFallback_withoutTouchingProviders() { GoalEvaluationResult r = svc.evaluate(null, List.of(), "anything"); diff --git a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalSegmentRunnerTest.java b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalSegmentRunnerTest.java index e695482c..55771169 100644 --- a/mateclaw-server/src/test/java/vip/mate/goal/service/GoalSegmentRunnerTest.java +++ b/mateclaw-server/src/test/java/vip/mate/goal/service/GoalSegmentRunnerTest.java @@ -5,6 +5,7 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import reactor.core.publisher.Flux; import vip.mate.agent.AgentService; +import vip.mate.agent.context.GoalContinuationContext; import vip.mate.agent.model.AgentEntity; import vip.mate.agent.runtime.ConversationTurnGate; import vip.mate.approval.ApprovalWorkflowService; @@ -12,7 +13,12 @@ import vip.mate.channel.web.ChatStreamTracker; import vip.mate.goal.model.GoalEntity; import vip.mate.workspace.conversation.ConversationService; import vip.mate.workspace.conversation.model.ConversationEntity; +import vip.mate.workspace.conversation.model.MessageContentPart; +import java.util.List; import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import static org.junit.jupiter.api.Assertions.*; import static org.mockito.ArgumentMatchers.*; import static org.mockito.Mockito.*; @@ -64,7 +70,10 @@ class GoalSegmentRunnerTest { var calls=new java.util.concurrent.atomic.AtomicInteger(); when(agents.chatStructuredStream(eq(2L),anyString(),eq("conv"),eq("alice"),isNull(),any())) .thenAnswer(inv -> Flux.defer(() -> { - if(calls.incrementAndGet()==1) streams.enqueueMessage("conv","new user instruction",2L,false); + boolean autonomous = calls.incrementAndGet()==1; + assertTrue(GoalContinuationContext.active()); + assertEquals(autonomous, GoalContinuationContext.explicitPrompt()); + if(autonomous) streams.enqueueMessage("conv","new user instruction",2L,false); return Flux.just(new AgentService.StreamDelta("output",null), AgentService.StreamDelta.event("finish_reason",Map.of("reason","normal"))); })); @@ -103,6 +112,81 @@ class GoalSegmentRunnerTest { assertNotNull(gate.tryAcquire("conv")); } + @Test void cancellationDoesNotInterruptDatabasePersistence() throws Exception { + var saving = new CountDownLatch(1); + var release = new CountDownLatch(1); + var interrupted = new AtomicBoolean(); + when(agents.chatStructuredStream(eq(2L),anyString(),eq("conv"),eq("alice"),isNull(),any())) + .thenReturn(Flux.just(new AgentService.StreamDelta("checkpoint", null))); + when(conversations.saveMessage(eq("conv"),eq("assistant"),anyString(),anyList(),anyString(), + anyInt(),anyInt(),anyInt(),anyInt(),anyInt(),anyString(),anyString(),anyString())) + .thenAnswer(inv -> { + saving.countDown(); + try { release.await(3, TimeUnit.SECONDS); } + catch (InterruptedException error) { interrupted.set(true); } + return null; + }); + Thread worker = Thread.ofVirtual().start(() -> runner.run(goal,"continue",false)); + assertTrue(saving.await(3,TimeUnit.SECONDS)); + runner.cancel(1L); + release.countDown(); + worker.join(3000); + assertFalse(worker.isAlive()); + assertFalse(interrupted.get(), "cancellation must not close embedded database channels by interrupting I/O"); + } + + @Test void shutdownPersistsQueuedInputWithAttachments() throws Exception { + var ready = new CountDownLatch(1); + MessageContentPart attachment = new MessageContentPart(); + attachment.setType("file"); + attachment.setPath("test-evidence.txt"); + var parts = List.of(attachment); + when(agents.chatStructuredStream(eq(2L),anyString(),eq("conv"),eq("alice"),isNull(),any())) + .thenReturn(Flux.never().doOnSubscribe(s -> ready.countDown())); + Thread worker = Thread.ofVirtual().start(() -> runner.run(goal,"continue",false)); + assertTrue(ready.await(3,TimeUnit.SECONDS)); + streams.enqueueMessage("conv","accepted steering",2L,false,parts); + runner.cancelAll(); + worker.join(3000); + assertFalse(worker.isAlive()); + verify(conversations).saveMessage("conv","user","accepted steering",parts,"queued"); + assertFalse(streams.hasQueuedMessage("conv")); + } + + @Test void shutdownRejectsLateWorkerAdmissionWithoutStartingModel() { + when(agents.chatStructuredStream(eq(2L),anyString(),eq("conv"),eq("alice"),isNull(),any())) + .thenReturn(Flux.just(AgentService.StreamDelta.event("finish_reason",Map.of("reason","normal")))); + runner.cancelAll(); + assertEquals("stopped", runner.run(goal,"continue",false).finishReason()); + verify(agents,never()).chatStructuredStream(any(),any(),any(),any(),any(),any()); + } + + @Test void externalInterruptIsRestoredOnlyAfterCheckpointPersistence() throws Exception { + var ready = new CountDownLatch(1); + var persisted = new AtomicBoolean(); + var interruptRestored = new AtomicBoolean(); + when(agents.chatStructuredStream(eq(2L),anyString(),eq("conv"),eq("alice"),isNull(),any())) + .thenReturn(Flux.concat(Flux.just(new AgentService.StreamDelta("partial",null)), + Flux.never().doOnSubscribe(s -> ready.countDown()))); + when(conversations.saveMessage(eq("conv"),eq("assistant"),anyString(),anyList(),anyString(), + anyInt(),anyInt(),anyInt(),anyInt(),anyInt(),anyString(),anyString(),anyString())) + .thenAnswer(inv -> { + assertFalse(Thread.currentThread().isInterrupted()); + persisted.set(true); + return null; + }); + Thread worker = Thread.ofVirtual().start(() -> { + try { runner.run(goal,"continue",false); } + catch (IllegalStateException expected) { interruptRestored.set(Thread.currentThread().isInterrupted()); } + }); + assertTrue(ready.await(3,TimeUnit.SECONDS)); + worker.interrupt(); + worker.join(3000); + assertFalse(worker.isAlive()); + assertTrue(persisted.get()); + assertTrue(interruptRestored.get()); + } + @Test void permanentFailurePersistsAcceptedQueuedInput() { when(agents.chatStructuredStream(eq(2L),anyString(),eq("conv"),eq("alice"),isNull(),any())) .thenReturn(Flux.defer(() -> {