From 20014c72ff8c4d7932fdc7d2829ec66689eda81c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=80=AA=E7=A8=8B=E4=BC=9F?= Date: Thu, 25 Jun 2026 09:54:55 +0800 Subject: [PATCH] =?UTF-8?q?fix(workflow):=20activate=20approval=20notify?= =?UTF-8?q?=20+=20resolve=E2=86=92resume=20bridge=20(ISSUE=20#413=20P0)=20?= =?UTF-8?q?(#414)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two P0 fixes from ISSUE #413 — both address workflow await_approval approvals that silently failed in production: B1 — AwaitApprovalStepAdapter now dispatches the approval notice to every channel in approverChannels that carries a target. Previously approverChannels was write-only metadata: a workflow that declared ["feishu:oc_xxx"] silently dropped the notice and the IM group never learned an approval was waiting. Element format is "channelType" (no push, operator uses admin console) or "channelType:targetId". Each channel failure is logged and skipped — it must not fail the step. B2 — requestWorkflowApproval now registers the wf- approval into the in-memory map via registerRecovered. Previously it only did approvalMapper.insert, so getPending("wf-...") returned null, performResolve short-circuited at the not-pending guard, the WorkflowApprovalResolvedEvent was never published, and ApprovalResumeBridge was dead code. With this fix, resolving a wf- approval walks the full two-phase contract and the bridge fires. Tests: - WorkflowApprovalResumeBridgeTest (3): map registration, event publish on resolve, safe no-op for unregistered wf- ids. - AwaitApprovalNotifyTest (2): targeted channels dispatched, bare "web" skipped, channel failure non-fatal. Regression: ApprovalWorkflowServiceResolveTest (13), AwaitApprovalRuntimeTest (3), DispatchChannelRuntimeTest (3), GcTest (7), RecoveryTest (7) — all green. --- .../approval/ApprovalWorkflowService.java | 23 +++ .../mode/AwaitApprovalStepAdapter.java | 71 ++++++++ .../WorkflowApprovalResumeBridgeTest.java | 168 ++++++++++++++++++ .../runtime/AwaitApprovalNotifyTest.java | 157 ++++++++++++++++ 4 files changed, 419 insertions(+) create mode 100644 mateclaw-server/src/test/java/vip/mate/approval/WorkflowApprovalResumeBridgeTest.java create mode 100644 mateclaw-server/src/test/java/vip/mate/workflow/runtime/AwaitApprovalNotifyTest.java diff --git a/mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java b/mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java index fe40501d..31814fc9 100644 --- a/mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java +++ b/mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java @@ -335,6 +335,29 @@ public class ApprovalWorkflowService implements ApplicationRunner { approvalMapper.insert(entity); log.info("[ApprovalWorkflow] requested workflow approval row id={}, runId={}, workspace={}, kind={}", entity.getId(), runId, workspaceId, kind); + + // ISSUE #413: register the workflow approval into the in-memory map + // so the resolve → resume bridge actually fires. Without this, the + // row only lives in DB and ApprovalService.getPending("wf-...") returns + // null, so performResolve() short-circuits at the "not pending" guard + // and never reaches the WorkflowApprovalResolvedEvent publish in + // Phase 4 — leaving ApprovalResumeBridge as dead code. Mirrors the + // recoverFromDb() snapshot shape exactly. + Instant createdAt = entity.getCreatedAt() != null + ? entity.getCreatedAt().atZone(ZoneId.systemDefault()).toInstant() + : Instant.now(); + PendingApproval snapshot = new PendingApproval( + entity.getPendingId(), + entity.getConversationId(), + /*userId*/ null, + entity.getToolName(), + entity.getToolArguments(), + /*reason*/ entity.getSummary(), + createdAt, + "pending"); + snapshot.setSummary(entity.getSummary()); + approvalService.registerRecovered(snapshot); + return entity.getId(); } catch (Exception e) { log.warn("[ApprovalWorkflow] requestWorkflowApproval failed: {}", e.getMessage()); diff --git a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/mode/AwaitApprovalStepAdapter.java b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/mode/AwaitApprovalStepAdapter.java index 082a6991..cbe1b796 100644 --- a/mateclaw-server/src/main/java/vip/mate/workflow/runtime/mode/AwaitApprovalStepAdapter.java +++ b/mateclaw-server/src/main/java/vip/mate/workflow/runtime/mode/AwaitApprovalStepAdapter.java @@ -10,6 +10,7 @@ import vip.mate.workflow.model.WorkflowRunPauseEntity; import vip.mate.workflow.model.WorkflowRunStepEntity; import vip.mate.workflow.repository.WorkflowRunPauseMapper; import vip.mate.workflow.repository.WorkflowRunStepMapper; +import vip.mate.workflow.runtime.ChannelDispatcher; import vip.mate.workflow.runtime.StepAdapter; import vip.mate.workflow.runtime.StepResult; import vip.mate.workflow.runtime.WorkflowRunContext; @@ -57,6 +58,13 @@ public class AwaitApprovalStepAdapter implements StepAdapter { * adapter falls back to a no-op approval row when null. */ @Autowired(required = false) private ApprovalWorkflowService approvalService; + /** + * ISSUE #413: used to push the approval notice to every channel listed in + * {@code approverChannels}. Optional — null in narrow test contexts (the + * notice is skipped and the pause still resolves via REST / inbox). + */ + @Autowired(required = false) + private ChannelDispatcher channelDispatcher; public AwaitApprovalStepAdapter(WorkflowRunPauseMapper pauseMapper, WorkflowRunStepMapper stepMapper) { @@ -129,7 +137,70 @@ public class AwaitApprovalStepAdapter implements StepAdapter { } } + // ISSUE #413: push the approval notice to the configured channels so + // the approver actually learns an approval is waiting. Before this, + // approverChannels was write-only metadata and a workflow that asked + // for IM notification silently dropped it. + notifyApproverChannels(cfg, context, context.runId(), pauseToken); + return StepResult.paused(pauseToken, "awaiting " + (cfg.approvalKind() == null ? "approval" : cfg.approvalKind())); } + + /** + * ISSUE #413: push the approval notice to every channel in + * {@code approverChannels}. Previously this field was written into the + * approval row's {@code tool_arguments} and never read back, so a workflow + * that declared {@code approverChannels: ["feishu"]} silently dropped the + * notice — the IM group never learned an approval was waiting. + * + *

Element format is {@code "channelType"} (e.g. {@code "web"} — no + * proactive dispatch, operator uses the admin console) or + * {@code "channelType:targetId"} (e.g. {@code "feishu:oc_xxx"} — pushes a + * text notice to that target). The short {@code pauseId} suffix is + * included so the operator can correlate the notice with the inbox entry. + * Each channel failure is logged and skipped — a delivery hiccup on one + * channel must not fail the step (the pause row + REST resume are the + * canonical recovery path). + */ + private void notifyApproverChannels(StepMode.AwaitApproval cfg, WorkflowRunContext context, + long runId, String pauseToken) { + if (channelDispatcher == null || cfg.approverChannels() == null) { + return; + } + String kind = cfg.approvalKind() == null || cfg.approvalKind().isBlank() ? "approval" : cfg.approvalKind().trim(); + String shortToken = pauseToken.substring(0, Math.min(8, pauseToken.length())); + String message = "🔐 工作流审批待处理\n" + + "**类型**: " + kind + "\n" + + (cfg.approvalMessage() != null && !cfg.approvalMessage().isBlank() + ? "**说明**: " + cfg.approvalMessage() + "\n" : "") + + "**runId**: " + runId + "\n" + + "**审批码**: " + shortToken + "\n" + + "请前往管理端审批（工作流 → 运行记录 → 恢复）。"; + + for (String entry : cfg.approverChannels()) { + if (entry == null || entry.isBlank()) continue; + String channelType = entry; + String targetId = null; + int colon = entry.indexOf(':'); + if (colon > 0) { + channelType = entry.substring(0, colon); + targetId = entry.substring(colon + 1); + } + // "web" (and any channel with no target) means "operator handles + // it from the admin console" — no proactive push. + if (targetId == null || targetId.isBlank()) continue; + try { + ChannelDispatcher.DispatchResult result = + channelDispatcher.dispatch(context.workspaceId(), channelType, targetId, message); + if (!result.success()) { + org.slf4j.LoggerFactory.getLogger(AwaitApprovalStepAdapter.class) + .warn("await_approval notify failed for channel '{}': {}", channelType, result.message()); + } + } catch (Exception e) { + org.slf4j.LoggerFactory.getLogger(AwaitApprovalStepAdapter.class) + .warn("await_approval notify threw for channel '{}': {}", channelType, e.getMessage()); + } + } + } } diff --git a/mateclaw-server/src/test/java/vip/mate/approval/WorkflowApprovalResumeBridgeTest.java b/mateclaw-server/src/test/java/vip/mate/approval/WorkflowApprovalResumeBridgeTest.java new file mode 100644 index 00000000..a0c3fb22 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/approval/WorkflowApprovalResumeBridgeTest.java @@ -0,0 +1,168 @@ +package vip.mate.approval; + +import com.baomidou.mybatisplus.core.conditions.Wrapper; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.ibatis.builder.MapperBuilderAssistant; +import org.apache.ibatis.session.Configuration; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.context.ApplicationEventPublisher; +import vip.mate.approval.event.WorkflowApprovalResolvedEvent; +import vip.mate.approval.model.ToolApprovalEntity; +import vip.mate.approval.repository.ToolApprovalMapper; +import vip.mate.workspace.conversation.ConversationService; + +import java.util.ArrayList; +import java.util.List; +import java.util.Set; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.atLeastOnce; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Verifies ISSUE #413 P0-B2: a workflow-scoped approval created via + * {@code requestWorkflowApproval} is registered into the in-memory map, so a + * subsequent {@code resolve("wf-...")} walks the full two-phase contract and + * publishes {@link WorkflowApprovalResolvedEvent} — the event that + * {@code ApprovalResumeBridge} listens for to resume the paused run. + * + *

Before the fix, {@code requestWorkflowApproval} only did + * {@code approvalMapper.insert(entity)} without {@code registerRecovered}, so + * {@code getPending("wf-...")} returned null, {@code performResolve} + * short-circuited at the "not pending" guard, the event was never published, + * and {@code ApprovalResumeBridge} was dead code. + */ +@ExtendWith(MockitoExtension.class) +class WorkflowApprovalResumeBridgeTest { + + @Mock private ToolApprovalMapper approvalMapper; + @Mock private ConversationService conversationService; + + private ApprovalService approvalService; + private CapturingEventPublisher publisher; + private ApprovalWorkflowService workflow; + + @BeforeAll + static void initMyBatisPlusCache() { + TableInfoHelper.initTableInfo( + new MapperBuilderAssistant(new Configuration(), ""), + ToolApprovalEntity.class); + } + + @BeforeEach + void setUp() { + approvalService = new ApprovalService(); + publisher = new CapturingEventPublisher(); + workflow = new ApprovalWorkflowService( + approvalService, approvalMapper, new ObjectMapper(), conversationService); + // events is @Autowired(required = false) with no setter; inject via + // reflection so this unit test (no Spring context) can capture the + // WorkflowApprovalResolvedEvent publish. + try { + var field = ApprovalWorkflowService.class.getDeclaredField("events"); + field.setAccessible(true); + field.set(workflow, publisher); + } catch (Exception e) { + throw new IllegalStateException("failed to inject event publisher", e); + } + } + + @Test + @DisplayName("requestWorkflowApproval registers the wf- approval into the in-memory map") + void requestRegistersIntoMemoryMap() { + // insert returns the row id the adapter writes back as external_approval_id. + when(approvalMapper.insert(any(ToolApprovalEntity.class))).thenAnswer(inv -> { + ((ToolApprovalEntity) inv.getArgument(0)).setId(42L); + return 1; + }); + + Long approvalId = workflow.requestWorkflowApproval( + 1L, 100L, 7L, "manager", "please approve", java.util.List.of("web"), 1800); + + assertThat(approvalId).isEqualTo(42L); + + // The fix: the wf- entry is now in the in-memory map, queryable by the + // synthetic conversation key. Before the fix, getPending("wf-...") + // returned null and the resolve path dead-ended at the "not pending" + // guard, leaving ApprovalResumeBridge as dead code. + PendingApproval pending = approvalService.findPendingByConversation("workflow:run:100"); + assertThat(pending).as("wf- approval must be in the in-memory map after request").isNotNull(); + assertThat(pending.getPendingId()).startsWith("wf-"); + assertThat(pending.getToolName()).isEqualTo("workflow:manager"); + } + + @Test + @DisplayName("resolve('wf-...') publishes WorkflowApprovalResolvedEvent after the fix") + void resolvePublishesWorkflowEvent() { + // Seed via requestWorkflowApproval so the entry is in the map under a + // wf- pendingId (the same path the AwaitApprovalStepAdapter takes). + when(approvalMapper.insert(any(ToolApprovalEntity.class))).thenAnswer(inv -> { + ((ToolApprovalEntity) inv.getArgument(0)).setId(42L); + return 1; + }); + workflow.requestWorkflowApproval( + 1L, 100L, 7L, "manager", "please approve", java.util.List.of("web"), 1800); + + // Recover the generated wf- pendingId via the conversation key. + PendingApproval pending = approvalService.findPendingByConversation("workflow:run:100"); + assertThat(pending).as("wf- approval must be in the in-memory map after request").isNotNull(); + assertThat(pending.getPendingId()).startsWith("wf-"); + String pendingId = pending.getPendingId(); + + // Two-phase resolve: DB UPDATE conditional on PENDING succeeds. + when(approvalMapper.update(isNull(), any(Wrapper.class))).thenReturn(1); + when(conversationService.markPendingApprovalsResolved( + eq("workflow:run:100"), eq(Set.of(pendingId)), any())).thenReturn(0); + // selectOne lookup for the workflow-bridge row id (Phase 4). + ToolApprovalEntity row = new ToolApprovalEntity(); + row.setId(42L); + row.setPendingId(pendingId); + when(approvalMapper.selectOne(any())).thenReturn(row); + + ResolveOutcome outcome = workflow.resolve(pendingId, "operator", "approved"); + + assertThat(outcome.dbSynced()).isTrue(); + assertThat(outcome.decision()).isEqualTo("approved"); + + // The critical assertion: the workflow-resolved event was published. + // ApprovalResumeBridge listens for this to call WorkflowResumer.resume. + assertThat(publisher.workflowEvents).hasSize(1); + WorkflowApprovalResolvedEvent ev = publisher.workflowEvents.get(0); + assertThat(ev.approvalRowId()).isEqualTo(42L); + assertThat(ev.pendingId()).isEqualTo(pendingId); + assertThat(ev.decision()).isEqualTo("approved"); + } + + @Test + @DisplayName("resolve of a wf- approval that was NOT registered is still a safe no-op") + void resolveUnregisteredWorkflowApprovalIsNoop() { + // Simulate the pre-fix state: a wf- pendingId that never entered the map. + ResolveOutcome outcome = workflow.resolve("wf-ghostthatdoesnotexist", "operator", "approved"); + + assertThat(outcome.isAlreadyResolved()).isTrue(); + assertThat(publisher.workflowEvents).isEmpty(); + } + + /** Captures published events so tests can assert on the workflow-bridge event. */ + static class CapturingEventPublisher implements ApplicationEventPublisher { + final List workflowEvents = new ArrayList<>(); + + @Override + public void publishEvent(Object event) { + if (event instanceof WorkflowApprovalResolvedEvent wfe) { + workflowEvents.add(wfe); + } + } + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/workflow/runtime/AwaitApprovalNotifyTest.java b/mateclaw-server/src/test/java/vip/mate/workflow/runtime/AwaitApprovalNotifyTest.java new file mode 100644 index 00000000..e4c2f5fb --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/workflow/runtime/AwaitApprovalNotifyTest.java @@ -0,0 +1,157 @@ +package vip.mate.workflow.runtime; + +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.context.TestConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Import; +import org.springframework.context.annotation.Primary; +import org.springframework.test.context.TestPropertySource; +import vip.mate.MateClawApplication; +import vip.mate.workflow.compiler.WorkflowParser; +import vip.mate.workflow.compiler.ir.WorkflowGraph; +import vip.mate.workflow.model.WorkflowRunPauseEntity; +import vip.mate.workflow.repository.WorkflowRunPauseMapper; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Verifies ISSUE #413 P0-B1: an {@code await_approval} step pushes a notice to + * every channel listed in {@code approverChannels} that carries a target. Before + * the fix, {@code approverChannels} was write-only metadata — a workflow that + * declared {@code ["feishu:oc_xxx"]} silently dropped the notice. + */ +@SpringBootTest( + classes = MateClawApplication.class, + webEnvironment = SpringBootTest.WebEnvironment.NONE +) +@TestPropertySource(properties = { + "spring.datasource.url=jdbc:h2:mem:workflow_notify_${random.uuid};MODE=MySQL;DATABASE_TO_LOWER=TRUE;CASE_INSENSITIVE_IDENTIFIERS=TRUE;DB_CLOSE_DELAY=-1", + "spring.ai.dashscope.api-key=test-key", + "spring.main.web-application-type=none", + "mateclaw.workflow.trigger.async-dispatch=false" +}) +@Import({StubAgentInvokerConfig.class, + AwaitApprovalNotifyTest.StubChannelDispatcherConfig.class}) +class AwaitApprovalNotifyTest { + + @Autowired private WorkflowRunner runner; + @Autowired private WorkflowParser parser; + @Autowired private WorkflowRunPauseMapper pauseMapper; + @Autowired private StubChannelDispatcher stubDispatcher; + + @Test + @DisplayName("approverChannels with target dispatches a notice; bare 'web' does not.") + void dispatchesNoticeToTargetedChannels() { + stubDispatcher.reset(); + + WorkflowGraph graph = parser.parse(""" + { + "steps": [ + {"name":"approve", + "mode":{"type":"await_approval","approvalKind":"manager", + "approverChannels":["feishu:oc_manager_group","web","email:ops@acme.com"], + "approvalMessage":"请经理审批新客户入驻"}} + ] + } + """); + + WorkflowRunResult result = runner.run(graph, + new WorkflowRunRequest(70L, 1L, 99L, "manual", Map.of())); + assertEquals("paused", result.state()); + + // feishu + email both carry a target → two dispatches; "web" has no + // target → skipped (operator uses the admin console). + List sent = stubDispatcher.sentList(); + assertEquals(2, sent.size(), "only channels with an explicit target should be notified"); + assertTrue(sent.stream().anyMatch(s -> "feishu".equals(s.channel()) + && "oc_manager_group".equals(s.target()))); + assertTrue(sent.stream().anyMatch(s -> "email".equals(s.channel()) + && "ops@acme.com".equals(s.target()))); + + // The notice body carries the approval message + runId so the approver + // can correlate it with the inbox entry. + assertTrue(sent.get(0).content().contains("请经理审批新客户入驻")); + assertTrue(sent.get(0).content().contains("runId"), + "notice should mention runId: " + sent.get(0).content()); + + // The pause row still exists — the notice is non-fatal best-effort. + WorkflowRunPauseEntity pause = pauseMapper.selectOne( + new LambdaQueryWrapper() + .eq(WorkflowRunPauseEntity::getRunId, result.runId())); + assertNotNull(pause); + } + + @Test + @DisplayName("A channel dispatch failure does not fail the await_approval step.") + void channelFailureIsNonFatal() { + stubDispatcher.reset(); + stubDispatcher.makeFail("feishu", "rate limited"); + + WorkflowGraph graph = parser.parse(""" + { + "steps": [ + {"name":"approve", + "mode":{"type":"await_approval","approvalKind":"k", + "approverChannels":["feishu:oc_group"]}} + ] + } + """); + + WorkflowRunResult result = runner.run(graph, + new WorkflowRunRequest(71L, 1L, 99L, "manual", Map.of())); + // The step still pauses successfully — a delivery hiccup must not abort + // the run (pause row + REST resume are the canonical recovery path). + assertEquals("paused", result.state()); + } + + @TestConfiguration + static class StubChannelDispatcherConfig { + @Bean + @Primary + StubChannelDispatcher stubChannelDispatcher() { + return new StubChannelDispatcher(); + } + } + + static class StubChannelDispatcher implements ChannelDispatcher { + record Sent(String channel, String target, String content) {} + + private final List sent = new ArrayList<>(); + private final Map failures = new ConcurrentHashMap<>(); + + synchronized void reset() { + sent.clear(); + failures.clear(); + } + + synchronized List sentList() { + return List.copyOf(sent); + } + + void makeFail(String channelType, String message) { + failures.put(channelType, message); + } + + @Override + public synchronized DispatchResult dispatch(long workspaceId, String channelType, + String targetId, String content) { + String forced = failures.get(channelType); + if (forced != null) { + return DispatchResult.fail(forced); + } + sent.add(new Sent(channelType, targetId, content)); + return DispatchResult.ok(); + } + } +}