From 041bdd30f73bdc1f37b344a474a8fc124f742ac9 Mon Sep 17 00:00:00 2001 From: mateaix <7333791@qq.com> Date: Tue, 15 Sep 2026 03:18:27 +0800 Subject: [PATCH] Resume managed Goal approvals with fresh leases and conservative recovery --- .../java/vip/mate/agent/AgentService.java | 14 +++ .../graph/plan/node/StepExecutionNode.java | 2 +- .../service/GoalApprovalReplayStream.java | 118 ++++++++++++++++++ .../goal/service/GoalApprovalRunService.java | 8 ++ .../goal/service/GoalRecoveryService.java | 2 +- .../docs/en/managed-json-acceptance.md | 4 +- .../docs/zh/managed-json-acceptance.md | 4 +- .../GoalJsonAcceptanceIntegrationTest.java | 74 +++++++++++ .../GoalJsonHttpRuntimeIntegrationTest.java | 32 ++++- 9 files changed, 246 insertions(+), 12 deletions(-) create mode 100644 mateclaw-server/src/main/java/vip/mate/goal/service/GoalApprovalReplayStream.java diff --git a/mateclaw-server/src/main/java/vip/mate/agent/AgentService.java b/mateclaw-server/src/main/java/vip/mate/agent/AgentService.java index 7833774b..cfc73126 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/AgentService.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/AgentService.java @@ -89,6 +89,9 @@ public class AgentService { @Autowired(required = false) private vip.mate.agent.runtime.dsh.DshRuntimeService dshRuntimeService; + @Autowired(required = false) + private vip.mate.goal.service.GoalApprovalReplayStream goalApprovalReplay; + /** * Runtime Agent instance cache. Keyed first by agentId, then by a model * key, so a conversation that pins a non-default model gets its own graph @@ -546,6 +549,17 @@ public class AgentService { trackMemoryRecalls(agentId, userMessage, origin); BaseAgent agent = getOrBuildAgentForConversation(agentId, conversationId); ChatOrigin captured = origin != null ? origin : ChatOrigin.EMPTY; + if (goalApprovalReplay != null && goalApprovalReplay.applies(captured)) { + return Flux.using(() -> acquireTurn(conversationId), permit -> + goalApprovalReplay.replay(captured, toolCallPayload, fresh -> { + ChatOriginHolder.set(fresh); + return vip.mate.agent.context.GoalContinuationContext.call(true, () -> + invokeWithLifecycleFlux(agentId, userMessage, conversationId, + (msg, convId) -> agent.chatWithReplayStream(msg, convId, toolCallPayload, + requesterId != null ? requesterId : ""), StreamDelta::content)); + }), vip.mate.agent.runtime.ConversationTurnGate.Permit::close) + .doFinally(signal -> ChatOriginHolder.clear()); + } return Flux.defer(() -> { ChatOriginHolder.set(captured); return withLifecycleFlux(agentId, userMessage, conversationId, diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/node/StepExecutionNode.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/node/StepExecutionNode.java index 16f453f3..2192dc1e 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/node/StepExecutionNode.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/plan/node/StepExecutionNode.java @@ -356,7 +356,7 @@ public class StepExecutionNode implements NodeAction { for (AssistantMessage.ToolCall toolCall : allToolCalls) { if (isPreApprovedToolCall(toolCall.name(), preApprovedPayload)) { String storedArguments = extractArgumentsFromPayload(preApprovedPayload); - events.add(GraphEventPublisher.toolStart(toolCall.name(), toolCall.arguments())); + events.add(GraphEventPublisher.toolStart(toolCall.id(), toolCall.name(), toolCall.arguments())); // RFC-052: pass the directOutputs collector so that an // approved direct tool's full content is captured here // (instead of leaking into the next LLM round). diff --git a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalApprovalReplayStream.java b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalApprovalReplayStream.java new file mode 100644 index 00000000..f66f04d7 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalApprovalReplayStream.java @@ -0,0 +1,118 @@ +package vip.mate.goal.service; + +import org.springframework.stereotype.Service; +import reactor.core.Disposable; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Sinks; +import reactor.core.scheduler.Schedulers; +import vip.mate.agent.AgentService.StreamDelta; +import vip.mate.agent.context.ChatOrigin; +import vip.mate.exception.MateClawException; +import vip.mate.goal.model.SegmentOutcome; + +import java.time.LocalDateTime; +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.function.Function; + +/** Owns the lease and durable checkpoints while the existing channel owns its replay messages. */ +@Service +public class GoalApprovalReplayStream { + private final GoalApprovalRunService runs; + private final GoalRunCoordinator coordinator; + + public GoalApprovalReplayStream(GoalApprovalRunService runs, GoalRunCoordinator coordinator) { + this.runs=runs; this.coordinator=coordinator; + } + + public boolean applies(ChatOrigin origin) { return runs.requiresHandoff(origin); } + + public Flux replay(ChatOrigin origin, String payload, Function> invoke) { + return Flux.using(() -> new Execution(runs.claim(origin, payload)), execution -> + Flux.defer(() -> invoke.apply(execution.claim.origin())) + .doOnSubscribe(subscription -> execution.source = subscription) + .doOnNext(execution::observe) + .takeUntilOther(execution.lost.asMono()) + .doOnComplete(execution::complete), Execution::close); + } + + private final class Execution implements AutoCloseable { + private final GoalApprovalRunService.ReplayRun claim; + private final Sinks.One lost = Sinks.one(); + private final Set inFlight = new HashSet<>(); + private final Disposable renewal; + private volatile org.reactivestreams.Subscription source; + private volatile boolean sourceCompleted; + private boolean unknown = true; + private boolean awaitingApproval; + private boolean evaluationUnavailable; + private String finishReason = "approval_replay_completed"; + + Execution(GoalApprovalRunService.ReplayRun claim) { + this.claim=claim; + // A crash before the forced approved call reports its outcome must not replay it blindly. + checkpoint("uncertain", "approval_replay_started"); + renewal = Schedulers.boundedElastic().schedulePeriodically(() -> { + try { + if (!coordinator.renew(claim.run(), LocalDateTime.now())) { + lost.tryEmitError(new MateClawException(409, "Approved Goal execution lost its owner lease")); + } + } catch (RuntimeException error) { lost.tryEmitError(error); } + }, 20, 20, TimeUnit.SECONDS); + } + + private void checkpoint(String safety, String kind) { + if (!coordinator.checkpoint(claim.run(), safety, kind, null, LocalDateTime.now())) { + throw new MateClawException(409, "Approved Goal execution lost its checkpoint fence"); + } + } + + void observe(StreamDelta delta) { + String event = delta.eventType(); + var data = delta.eventData(); + if ("tool_call_started".equals(event)) { + Object id = data == null ? null : data.get("toolCallId"); + if (id != null && !String.valueOf(id).isBlank()) { inFlight.add(String.valueOf(id)); unknown=false; } + else { inFlight.add(""); unknown=true; } + checkpoint("uncertain", "tool_started"); + } else if ("tool_call_completed".equals(event)) { + Object id = data == null ? null : data.get("toolCallId"); + if (id != null) inFlight.remove(String.valueOf(id)); + // ReAct may deliver these events after a whole action node returns. A later + // tool can already be executing before its start delta arrives, so retain + // uncertainty until the entire replay stream terminates normally. + checkpoint("uncertain", "tool_completed"); + } else if ("tool_approval_requested".equals(event)) { + awaitingApproval=true; + } else if ("goal_evaluated".equals(event) && data != null) { + evaluationUnavailable = Boolean.TRUE.equals(data.get("skipped")) || "fallback".equals(data.get("decision")); + } else if ("finish_reason".equals(event) && data != null && data.get("reason") != null) { + finishReason=String.valueOf(data.get("reason")); + } + } + + void complete() { + sourceCompleted=true; + // On error/cancellation, or an unresolved tool even on normal termination, keep the + // last checkpoint for existing expiry recovery (which pauses uncertain side effects). + if (unknown || !inFlight.isEmpty()) return; + checkpoint("resolved", "approval_replay_finished"); + SegmentOutcome outcome = awaitingApproval ? new SegmentOutcome.AwaitApproval("approval_required") + : "error_fallback".equals(finishReason) ? new SegmentOutcome.Blocked("approval", finishReason) + : evaluationUnavailable ? new SegmentOutcome.Retry("evaluation", "evaluation_unavailable") + : new SegmentOutcome.Continue(finishReason); + if (!coordinator.settle(claim.run(), outcome, LocalDateTime.now())) { + throw new MateClawException(409, "Approved Goal execution lost its settlement fence"); + } + } + + @Override public void close() { + renewal.dispose(); + // Explicitly cancel the producer when lease failure wins the other publisher. + // Merely signalling an error downstream must not leave the graph subscribed. + var subscription = source; + if (!sourceCompleted && subscription != null) subscription.cancel(); + } + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalApprovalRunService.java b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalApprovalRunService.java index a2d54b9b..4d27b5c9 100644 --- a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalApprovalRunService.java +++ b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalApprovalRunService.java @@ -34,6 +34,14 @@ public class GoalApprovalRunService { public record ReplayRun(GoalRunCoordinator.ClaimedRun run, ChatOrigin origin) { } + public boolean requiresHandoff(ChatOrigin origin) { + var link = origin == null ? null : origin.executionAttribution(); + if (link == null || link.goalId() == null || link.approvalId() == null) return false; + var required = jdbc.queryForList("SELECT json_acceptance_required FROM mate_agent_goal WHERE id=? AND deleted=0", + Boolean.class, link.goalId()); + return required.isEmpty() || Boolean.TRUE.equals(required.getFirst()); + } + @Transactional public ReplayRun claim(ChatOrigin requested, String toolCallPayload) { ExecutionAttribution link = requested == null ? null : requested.executionAttribution(); diff --git a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalRecoveryService.java b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalRecoveryService.java index ee714973..fe68c658 100644 --- a/mateclaw-server/src/main/java/vip/mate/goal/service/GoalRecoveryService.java +++ b/mateclaw-server/src/main/java/vip/mate/goal/service/GoalRecoveryService.java @@ -32,7 +32,7 @@ public class GoalRecoveryService { } public RecoveryDecision classify(GoalAttempt attempt) { - if("tool_started".equals(attempt.checkpointType()) && "uncertain".equals(attempt.replaySafety())) { + if("uncertain".equals(attempt.replaySafety())) { return RecoveryDecision.BLOCK_UNCERTAIN_SIDE_EFFECT; } if("message_saved".equals(attempt.checkpointType()) && attempt.assistantMessageId()!=null) { diff --git a/mateclaw-server/src/main/resources/docs/en/managed-json-acceptance.md b/mateclaw-server/src/main/resources/docs/en/managed-json-acceptance.md index 08a24a2e..d5f1b044 100644 --- a/mateclaw-server/src/main/resources/docs/en/managed-json-acceptance.md +++ b/mateclaw-server/src/main/resources/docs/en/managed-json-acceptance.md @@ -67,6 +67,6 @@ JWT requests match the signed userId to the current enabled account ID. Recreati After interactive Web approval, Plan execution restores the original plan and approved call, retaining the requester and managed acceptance requirements. Approval itself does not replace a JSON check or complete the goal. -When a background Goal settles into awaiting approval, its original attempt lease is released. Replaying that persisted identity cannot access managed artifacts or complete the Goal. A fresh attempt can reuse still-eligible evidence, but automatic transfer of a settled approval to a new lease is not yet provided; interactive approval verification does not cover this background path. +When a background Goal settles into awaiting approval, its original attempt lease is released. Replaying that persisted identity cannot access managed artifacts or complete the Goal. The existing replay flow creates an exactly linked fresh attempt and lease for a Goal with selected JSON requirements. That attempt can also reuse still-eligible evidence. A single background approval has been verified through the real ReAct/Plan runtime; repeated approval and arrival during original settlement remain under verification. -V200 records the exact attempt that settled into approval waiting and makes the approval-to-new-attempt association unique. Older waiting rows remain unbound and cannot be inferred into new execution authority. The controlled claim service is verified; automatic replay integration and lifecycle verification are still in progress. +V200 records the exact attempt that settled into approval waiting and makes the approval-to-new-attempt association unique. Older waiting rows remain unbound and cannot be inferred into new execution authority. Replay renews its lease every 20 seconds and settles on normal completion. Cancellation, failure, or lease loss preserves an uncertain checkpoint; expiry recovery pauses for review instead of blindly replaying side effects. Because graph tool events can arrive in batches, an intermediate completion event does not make the whole replay safe to retry. diff --git a/mateclaw-server/src/main/resources/docs/zh/managed-json-acceptance.md b/mateclaw-server/src/main/resources/docs/zh/managed-json-acceptance.md index 88386a4a..0ceb0044 100644 --- a/mateclaw-server/src/main/resources/docs/zh/managed-json-acceptance.md +++ b/mateclaw-server/src/main/resources/docs/zh/managed-json-acceptance.md @@ -69,6 +69,6 @@ JWT请求同时核对签名令牌的userId与当前启用账户ID。同名账户 交互式Web审批后的Plan执行会恢复原计划和已批准调用，并保留请求者及受管验收要求；审批通过本身不能替代JSON检查或完成目标。 -后台 Goal 进入待审批并结算后，原 attempt 的租约已经释放；其持久化审批身份不能再读写托管产物或完成 Goal。新 attempt 可复用仍合格的已有版本，但当前尚未提供已结算审批到新租约的自动接续，不能将交互式审批验证视为这一后台路径已通过。 +后台 Goal 进入待审批并结算后，原 attempt 的租约已经释放；其持久化审批身份不能再读写托管产物或完成 Goal。现有重放流会为已选定JSON要求的Goal建立一个准确关联的新 attempt，使用新租约执行已批准调用；新 attempt 也可复用仍合格的已有版本。单次后台审批已通过真实ReAct/Plan运行时验证，连续再次审批及结算瞬间的竞争仍在验证。 -V200 保存待审批结算对应的准确 attempt，并对审批生成的新 attempt 设置唯一关联。升级前的待审批行保持未绑定，不能推断为任何新的执行权限。受控领取服务已验证，但审批流的自动接入与生命周期验证仍在进行。 +V200 保存待审批结算对应的准确 attempt，并对审批生成的新 attempt 设置唯一关联。升级前的待审批行保持未绑定，不能推断为任何新的执行权限。重放每20秒续租，正常结束后结算；取消、异常或失租保留不确定检查点，到期恢复会暂停并要求核实，不盲目重放工具副作用。图的工具事件可能延后汇总，因此中途完成事件不等于整段执行可安全重试。 diff --git a/mateclaw-server/src/test/java/vip/mate/goal/GoalJsonAcceptanceIntegrationTest.java b/mateclaw-server/src/test/java/vip/mate/goal/GoalJsonAcceptanceIntegrationTest.java index 6bbbebda..649c189a 100644 --- a/mateclaw-server/src/test/java/vip/mate/goal/GoalJsonAcceptanceIntegrationTest.java +++ b/mateclaw-server/src/test/java/vip/mate/goal/GoalJsonAcceptanceIntegrationTest.java @@ -47,6 +47,7 @@ class GoalJsonAcceptanceIntegrationTest { @Autowired private vip.mate.goal.service.GoalAttemptStore attempts; @Autowired private vip.mate.approval.ApprovalWorkflowService approvals; @Autowired private vip.mate.goal.service.GoalApprovalRunService approvalRuns; + @Autowired private vip.mate.goal.service.GoalApprovalReplayStream approvalStream; private String alice; private String bob; @@ -747,6 +748,79 @@ class GoalJsonAcceptanceIntegrationTest { : goals.markRuntimeCompleted(goal.getId(), evaluation, origin); } + @ParameterizedTest @ValueSource(strings = {"normal", "cancel", "error", "unpaired", "renew", "lost"}) + void approvalStreamOwnsLeaseAndConservativelyRecoversInterruptedTools(String kind) throws Exception { + GoalEntity goal = goal(true); + acceptance.configure(goal.getId(), "r", request(0, "summary"), alice); + var original = claimed(goal); + var origin = attemptOrigin(goal, original); + String pending; + vip.mate.agent.context.ChatOriginHolder.set(origin); + try { + pending = approvals.createPending(goal.getConversationId(), alice, "getManagedGoalJsonSlots", "{}", + "offline replay lifecycle fixture", "[]", null, "1"); + } finally { vip.mate.agent.context.ChatOriginHolder.clear(); } + assertTrue(coordinator.settle(original, new SegmentOutcome.AwaitApproval("approval_required"), java.time.LocalDateTime.now())); + assertNotNull(approvals.resolveAndConsume(pending, alice).consumedSnapshot()); + var source = reactor.core.publisher.Sinks.many().unicast().onBackpressureBuffer(); + var current = new java.util.concurrent.atomic.AtomicReference(); + var failure = new java.util.concurrent.atomic.AtomicReference(); + var terminated = new java.util.concurrent.CountDownLatch(1); + var cancelled = new java.util.concurrent.CountDownLatch(1); + var subscription = approvalStream.replay(origin.withApprovalId(pending), "[]", fresh -> { + current.set(fresh); return source.asFlux().doOnCancel(cancelled::countDown); + }).subscribe(delta -> {}, error -> { failure.set(error); terminated.countDown(); }, terminated::countDown); + try { + String attemptId = current.get().executionAttribution().goalAttemptId(); + source.tryEmitNext(vip.mate.agent.AgentService.StreamDelta.event("tool_call_started", java.util.Map.of("toolCallId", "one"))); + source.tryEmitNext(vip.mate.agent.AgentService.StreamDelta.event("tool_call_completed", java.util.Map.of("toolCallId", "one"))); + assertEquals("uncertain", attempts.get(attemptId).replaySafety(), + "Batched graph events do not prove that a later tool has not started"); + long initialLease = jdbc.queryForObject("SELECT lease_until_epoch_second FROM mate_goal_attempt WHERE attempt_id=?", Long.class, attemptId); + if (kind.equals("renew")) { + long deadline = System.nanoTime() + java.util.concurrent.TimeUnit.SECONDS.toNanos(25); + long renewed = initialLease; + while (renewed == initialLease && System.nanoTime() < deadline) { + Thread.sleep(50); + renewed = jdbc.queryForObject("SELECT lease_until_epoch_second FROM mate_goal_attempt WHERE attempt_id=?", Long.class, attemptId); + } + assertTrue(renewed > initialLease, "Actual production 20-second renewal must extend the attempt"); + assertEquals(renewed, jdbc.queryForObject("SELECT lease_until_epoch_second FROM mate_goal_continuation WHERE goal_id=?", Long.class, goal.getId())); + } + if (kind.equals("lost")) { + jdbc.update("UPDATE mate_goal_attempt SET lease_until_epoch_second=0 WHERE attempt_id=?", attemptId); + jdbc.update("UPDATE mate_goal_continuation SET lease_until_epoch_second=0 WHERE goal_id=?", goal.getId()); + assertTrue(terminated.await(25, java.util.concurrent.TimeUnit.SECONDS)); + assertInstanceOf(MateClawException.class, failure.get()); + assertTrue(cancelled.await(2, java.util.concurrent.TimeUnit.SECONDS)); + assertEquals(reactor.core.publisher.Sinks.EmitResult.FAIL_CANCELLED, + source.tryEmitNext(new vip.mate.agent.AgentService.StreamDelta("late", null))); + } else if (kind.equals("cancel")) subscription.dispose(); + else if (kind.equals("error")) source.tryEmitError(new IllegalStateException("offline interrupted tool fixture")); + else { + if (kind.equals("unpaired")) source.tryEmitNext(vip.mate.agent.AgentService.StreamDelta.event( + "tool_call_started", java.util.Map.of("toolCallId", "unfinished"))); + source.tryEmitComplete(); + } + if (kind.equals("normal") || kind.equals("renew")) { + assertTrue(terminated.await(2, java.util.concurrent.TimeUnit.SECONDS)); + assertNull(failure.get()); + assertEquals("succeeded", attempts.get(attemptId).state()); + assertEquals("queued", continuations.get(goal.getId()).state()); + assertEquals(GoalStatus.ACTIVE, goals.getById(goal.getId()).getStatus()); + } else { + assertEquals("running", attempts.get(attemptId).state()); + assertEquals("uncertain", attempts.get(attemptId).replaySafety()); + jdbc.update("UPDATE mate_goal_attempt SET lease_until_epoch_second=0 WHERE attempt_id=?", attemptId); + jdbc.update("UPDATE mate_goal_continuation SET lease_until_epoch_second=0 WHERE goal_id=?", goal.getId()); + assertTrue(recovery.recoverExpired(java.time.Instant.now()) >= 1); + assertEquals("blocked", attempts.get(attemptId).state()); + assertEquals(GoalStatus.PAUSED, goals.getById(goal.getId()).getStatus()); + assertTrue(acceptance.get(goal.getId(), alice).required()); + } + } finally { subscription.dispose(); } + } + @ParameterizedTest @ValueSource(strings = {"valid", "pending", "payload", "paused", "running", "legacy", "archived", "agent", "disabled", "wrong-parent"}) void consumedApprovalCanClaimOnlyItsExactWaitingGoalOnce(String kind) throws Exception { GoalEntity goal = goal(true); diff --git a/mateclaw-server/src/test/java/vip/mate/goal/GoalJsonHttpRuntimeIntegrationTest.java b/mateclaw-server/src/test/java/vip/mate/goal/GoalJsonHttpRuntimeIntegrationTest.java index 157e1d0c..3881d3b6 100644 --- a/mateclaw-server/src/test/java/vip/mate/goal/GoalJsonHttpRuntimeIntegrationTest.java +++ b/mateclaw-server/src/test/java/vip/mate/goal/GoalJsonHttpRuntimeIntegrationTest.java @@ -79,11 +79,11 @@ class GoalJsonHttpRuntimeIntegrationTest { "false,queued,true", "false,reuse,true", "true,reuse,true", "false,recheck,true", "true,recheck,true", "false,supervised,true", "true,supervised,true", "false,supervised-recovered,true", "true,supervised-recovered,true", "false,supervised,false", "true,supervised,false", "false,supervised-recovered,false", "true,supervised-recovered,false", - "false,approval,true", "true,approval,true"}) + "false,approval,true", "true,approval,true", "false,scheduled-approval,true", "true,scheduled-approval,true"}) void authenticatedGoalCompletesThroughHttpOrScheduledProductionRuntime(boolean plan, String entry, boolean accepted) throws Exception { - boolean approval = entry.equals("approval"); + boolean approval = entry.endsWith("approval"); boolean supervised = entry.startsWith("supervised"); - boolean scheduled = entry.equals("scheduled") || entry.equals("recovered") || supervised; + boolean scheduled = entry.equals("scheduled") || entry.equals("scheduled-approval") || entry.equals("recovered") || supervised; boolean reuse = entry.equals("reuse"); boolean recheck = entry.equals("recheck"); boolean queued = entry.equals("queued"); @@ -275,8 +275,17 @@ class GoalJsonHttpRuntimeIntegrationTest { guardRules.insert(rule); guardRegistry.reload(); var guard = guardConfig.getConfig(); guard.setEnabled(true); guardConfig.updateConfig(guard); try { - String waiting = requestBody("POST", "/api/v1/chat/stream", token, - Map.of("agentId", String.valueOf(agentId), "conversationId", conversation, "message", message)); + String waiting; + if (scheduled) { + SegmentOutcome outcome = runner.run(run, message, false); + assertInstanceOf(SegmentOutcome.AwaitApproval.class, outcome); + assertTrue(coordinator.settle(run, outcome, java.time.LocalDateTime.now())); + assertEquals("waiting_approval", continuations.get(goal.getId()).state()); + waiting = outcome.toString(); + } else { + waiting = requestBody("POST", "/api/v1/chat/stream", token, + Map.of("agentId", String.valueOf(agentId), "conversationId", conversation, "message", message)); + } JsonNode pending = request("GET", "/api/v1/chat/" + conversation + "/pending-approvals", token, null).path("data"); assertEquals(1, pending.size(), waiting); String pendingId = pending.get(0).path("pendingId").asText(); @@ -284,13 +293,24 @@ class GoalJsonHttpRuntimeIntegrationTest { assertEquals(GoalStatus.ACTIVE, goals.getById(goal.getId()).getStatus()); assertEquals(0, jdbc.queryForObject("SELECT COUNT(*) FROM mate_goal_json_artifact WHERE goal_id=?", Integer.class, goal.getId())); String persistedOrigin = jdbc.queryForObject("SELECT chat_origin FROM mate_tool_approval WHERE pending_id=?", String.class, pendingId); - assertEquals(userId, approvals.restoreChatOrigin(persistedOrigin).requesterUserId()); + if (scheduled) assertEquals(run.attempt().id(), approvals.restoreChatOrigin(persistedOrigin).executionAttribution().goalAttemptId()); + else assertEquals(userId, approvals.restoreChatOrigin(persistedOrigin).requesterUserId()); Long approvedPlan = plan ? jdbc.queryForObject("SELECT id FROM mate_plan WHERE conversation_id=?", Long.class, conversation) : null; planApprovalReplay.set(plan); String replay = requestBody("POST", "/api/v1/chat/stream", token, Map.of("agentId", String.valueOf(agentId), "conversationId", conversation, "message", "/approve", "pendingApprovalId", pendingId)); assertTrue(replay.contains("Managed JSON fixture completed."), replay); assertEquals("CONSUMED", jdbc.queryForObject("SELECT status FROM mate_tool_approval WHERE pending_id=?", String.class, pendingId)); + if (scheduled) { + String freshAttempt = jdbc.queryForObject("SELECT attempt_id FROM mate_goal_attempt WHERE approval_pending_id=?", String.class, pendingId); + var fresh = attempts.get(freshAttempt); + assertEquals(run.attempt().id(), fresh.parentAttemptId()); + assertNotEquals(run.attempt().leaseToken(), fresh.leaseToken()); + assertEquals("succeeded", fresh.state()); + assertEquals("completed", continuations.get(goal.getId()).state()); + assertFalse(coordinator.renew(run, java.time.LocalDateTime.now())); + assertEquals(freshAttempt, jdbc.queryForObject("SELECT producer_id FROM mate_goal_json_artifact WHERE goal_id=?", String.class, goal.getId())); + } if (plan) { assertEquals(approvedPlan, jdbc.queryForObject("SELECT id FROM mate_plan WHERE conversation_id=?", Long.class, conversation), "Approval replay must finish the original plan without creating a replacement");