opensource/mateclaw
2
0
Fork 0
mirror of https://gitee.com/mateos/mateclaw.git synced 2026-09-13 03:13:41 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(workflow): activate approval notify + resolve→resume bridge (ISSUE #413 P0) (#414)

Browse Source 
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.
This commit is contained in:
倪程伟 2026-06-25 09:54:55 +08:00 committed by GitHub
parent f7f1c30557
commit 20014c72ff
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 419 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java
View File

@ -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());

71
mateclaw-server/src/main/java/vip/mate/workflow/runtime/mode/AwaitApprovalStepAdapter.java
View File

@ -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.
*
* <p>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());
}
}
}
}

168
mateclaw-server/src/test/java/vip/mate/approval/WorkflowApprovalResumeBridgeTest.java Normal file
View File

@ -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.
*
* <p>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<WorkflowApprovalResolvedEvent> workflowEvents = new ArrayList<>();
@Override
public void publishEvent(Object event) {
if (event instanceof WorkflowApprovalResolvedEvent wfe) {
workflowEvents.add(wfe);
}
}
}
}

157
mateclaw-server/src/test/java/vip/mate/workflow/runtime/AwaitApprovalNotifyTest.java Normal file
View File

@ -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<StubChannelDispatcher.Sent> 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<WorkflowRunPauseEntity>()
.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> sent = new ArrayList<>();
private final Map<String, String> failures = new ConcurrentHashMap<>();
synchronized void reset() {
sent.clear();
failures.clear();
}
synchronized List<Sent> 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();
}
}
}