mirror of
https://github.com/slhaf/Partner.git
synced 2026-05-12 08:43:02 +08:00
refactor(executor): Reconstruct the execution process of ExecutableAction in ActionExecutor, remove ActionRepairer, and adjust the execution timing of ActionCorrector
This commit is contained in:
@@ -0,0 +1,51 @@
|
||||
package work.slhaf.partner.module.modules.action.executor;
|
||||
|
||||
import com.alibaba.fastjson2.JSONObject;
|
||||
import lombok.val;
|
||||
import work.slhaf.partner.api.agent.factory.component.abstracts.AbstractAgentModule;
|
||||
import work.slhaf.partner.api.agent.factory.component.abstracts.ActivateModel;
|
||||
import work.slhaf.partner.api.chat.pojo.Message;
|
||||
import work.slhaf.partner.module.modules.action.executor.entity.CorrectionRecognizerInput;
|
||||
import work.slhaf.partner.module.modules.action.executor.entity.CorrectionRecognizerResult;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 负责在行动链执行过程中判断当前进度是否异常,是否需要引入 corrector 介入。
|
||||
*/
|
||||
public class ActionCorrectionRecognizer extends AbstractAgentModule.Sub<CorrectionRecognizerInput, CorrectionRecognizerResult> implements ActivateModel {
|
||||
@Override
|
||||
public CorrectionRecognizerResult execute(CorrectionRecognizerInput input) {
|
||||
val prompt = buildPrompt(input);
|
||||
return formattedChat(List.of(new Message(Message.Character.USER, prompt)), CorrectionRecognizerResult.class);
|
||||
}
|
||||
|
||||
private String buildPrompt(CorrectionRecognizerInput input) {
|
||||
val prompt = new JSONObject();
|
||||
prompt.put("[行动来源]", input.getSource());
|
||||
prompt.put("[行动倾向]", input.getTendency());
|
||||
prompt.put("[行动描述]", input.getDescription());
|
||||
prompt.put("[行动原因]", input.getReason());
|
||||
prompt.put("[当前阶段]", input.getCurrentStage());
|
||||
prompt.put("[当前阶段位置]", input.getCurrentStageIndex());
|
||||
prompt.put("[是否最后阶段]", input.isLastStage());
|
||||
val stageList = prompt.putArray("[当前行动链阶段列表]");
|
||||
stageList.addAll(input.getOrderedStages());
|
||||
val history = prompt.putArray("[当前阶段已执行情况]");
|
||||
if (input.getHistory() != null) {
|
||||
history.addAll(input.getHistory());
|
||||
}
|
||||
val metaActions = prompt.putArray("[当前阶段行动结果]");
|
||||
metaActions.addAll(input.getCurrentStageMetaActions());
|
||||
val messages = prompt.putArray("[近期对话]");
|
||||
messages.addAll(input.getRecentMessages());
|
||||
val memory = prompt.putArray("[已激活记忆]");
|
||||
memory.addAll(input.getActivatedSlices());
|
||||
return prompt.toJSONString();
|
||||
}
|
||||
|
||||
@Override
|
||||
public String modelKey() {
|
||||
return "action_correction_recognizer";
|
||||
}
|
||||
}
|
||||
@@ -20,6 +20,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
public class ActionExecutor extends AbstractAgentModule.Standalone {
|
||||
private static final int MAX_EXTRACTOR_ATTEMPTS = 3;
|
||||
private final AssemblyHelper assemblyHelper = new AssemblyHelper();
|
||||
@InjectCapability
|
||||
private ActionCapability actionCapability;
|
||||
@@ -30,9 +31,9 @@ public class ActionExecutor extends AbstractAgentModule.Standalone {
|
||||
@InjectModule
|
||||
private ParamsExtractor paramsExtractor;
|
||||
@InjectModule
|
||||
private ActionRepairer actionRepairer;
|
||||
@InjectModule
|
||||
private ActionCorrector actionCorrector;
|
||||
@InjectModule
|
||||
private ActionCorrectionRecognizer actionCorrectionRecognizer;
|
||||
|
||||
private ExecutorService virtualExecutor;
|
||||
private ExecutorService platformExecutor;
|
||||
@@ -156,6 +157,7 @@ public class ActionExecutor extends AbstractAgentModule.Standalone {
|
||||
stageCursor.init();
|
||||
do {
|
||||
val metaActions = actionChain.get(executableAction.getExecutingStage());
|
||||
val recognizerRecord = startRecognizerIfNeeded(executableAction, phaserRecord);
|
||||
val listeningRecord = executeAndListening(metaActions, phaserRecord, source);
|
||||
phaser.awaitAdvance(listeningRecord.phase());
|
||||
// synchronized 同步防止 accepting 循环间、phase guard 判定后发生 stage 推进
|
||||
@@ -166,12 +168,17 @@ public class ActionExecutor extends AbstractAgentModule.Standalone {
|
||||
// 立即尝试推进,本次推进中,如果前方仍有未执行 stage,将执行一次阶段推进
|
||||
stageCursor.requestAdvance();
|
||||
}
|
||||
boolean shouldRunCorrector = hasFailedMetaAction(metaActions);
|
||||
try {
|
||||
// 针对行动链进行修正,修正需要传入执行历史、行动目标等内容
|
||||
// 如果后续运行 corrector 触发频率较高,可考虑增加重试机制
|
||||
val correctorInput = assemblyHelper.buildCorrectorInput(executableAction, source);
|
||||
if (!shouldRunCorrector) {
|
||||
val recognizerResult = resolveRecognizerResult(recognizerRecord);
|
||||
shouldRunCorrector = recognizerResult != null && recognizerResult.isNeedCorrection();
|
||||
}
|
||||
if (shouldRunCorrector) {
|
||||
val correctorInput = assemblyHelper.buildCorrectorInput(executableAction);
|
||||
val correctorResult = actionCorrector.execute(correctorInput);
|
||||
actionCapability.handleInterventions(correctorResult.getMetaInterventionList(), executableAction);
|
||||
}
|
||||
} catch (Exception ignored) {
|
||||
}
|
||||
// 第二次尝试进行阶段推进,本次负责补充上一次在不存在 stage时,但 corrector 执行期间发生了 actionChain 的插入事件
|
||||
@@ -235,43 +242,7 @@ public class ActionExecutor extends AbstractAgentModule.Standalone {
|
||||
return () -> {
|
||||
val actionKey = metaAction.getKey();
|
||||
try {
|
||||
val result = metaAction.getResult();
|
||||
do {
|
||||
val actionData = phaserRecord.executableAction();
|
||||
val executingStage = actionData.getExecutingStage();
|
||||
val historyActionResults = actionData.getHistory().get(executingStage);
|
||||
val additionalContext = actionData.getAdditionalContext().get(executingStage);
|
||||
val extractorInput = assemblyHelper.buildExtractorInput(metaAction, source, historyActionResults, additionalContext);
|
||||
val extractorResult = paramsExtractor.execute(extractorInput);
|
||||
if (extractorResult.isOk()) {
|
||||
metaAction.getParams().putAll(extractorResult.getParams());
|
||||
runnerClient.submit(metaAction);
|
||||
val historyAction = new HistoryAction(actionKey, actionCapability.loadMetaActionInfo(actionKey).getDescription(), metaAction.getResult().getData());
|
||||
actionData.getHistory()
|
||||
.computeIfAbsent(executingStage, integer -> new ArrayList<>())
|
||||
.add(historyAction);
|
||||
} else {
|
||||
val repairerInput = assemblyHelper.buildRepairerInput(historyActionResults, metaAction, source);
|
||||
val repairerResult = actionRepairer.execute(repairerInput);
|
||||
switch (repairerResult.getStatus()) {
|
||||
// 如果本次修复被认为成功,则将补充的信息添加至 additionalContext
|
||||
case RepairerResult.RepairerStatus.OK -> {
|
||||
additionalContext.addAll(repairerResult.getFixedData());
|
||||
result.setStatus(MetaAction.Result.Status.WAITING);
|
||||
}
|
||||
// 此处的修复失败来自系统内部的执行失败:其余方式均不可行时将回退至当前分支
|
||||
case RepairerResult.RepairerStatus.FAILED -> {
|
||||
result.setStatus(MetaAction.Result.Status.FAILED);
|
||||
result.setData("行动执行失败");
|
||||
}
|
||||
// 此处对应已在 repairer 内发起外部请求,故在此处进行阻塞
|
||||
case RepairerResult.RepairerStatus.ACQUIRE -> {
|
||||
phaserRecord.interrupt();
|
||||
result.setStatus(MetaAction.Result.Status.WAITING);
|
||||
}
|
||||
}
|
||||
}
|
||||
} while (result.getStatus().equals(MetaAction.Result.Status.WAITING));
|
||||
executeMetaActionWithRetry(metaAction, phaserRecord, source);
|
||||
} catch (Exception e) {
|
||||
log.error("Action executing failed: {}", actionKey, e);
|
||||
} finally {
|
||||
@@ -280,6 +251,120 @@ public class ActionExecutor extends AbstractAgentModule.Standalone {
|
||||
};
|
||||
}
|
||||
|
||||
private void executeMetaActionWithRetry(MetaAction metaAction, PhaserRecord phaserRecord, String source) {
|
||||
String failureReason = "参数提取失败";
|
||||
val actionData = phaserRecord.executableAction();
|
||||
val actionKey = metaAction.getKey();
|
||||
for (int attempt = 1; attempt <= MAX_EXTRACTOR_ATTEMPTS; attempt++) {
|
||||
val result = metaAction.getResult();
|
||||
result.reset();
|
||||
metaAction.getParams().clear();
|
||||
|
||||
val executingStage = actionData.getExecutingStage();
|
||||
val historyActionResults = actionData.getHistory().get(executingStage);
|
||||
val additionalContext = actionData.getAdditionalContext().get(executingStage);
|
||||
val extractorInput = assemblyHelper.buildExtractorInput(metaAction, source, historyActionResults, additionalContext);
|
||||
ExtractorResult extractorResult;
|
||||
try {
|
||||
extractorResult = paramsExtractor.execute(extractorInput);
|
||||
} catch (Exception e) {
|
||||
failureReason = buildAttemptFailureReason("参数提取异常", e.getLocalizedMessage());
|
||||
continue;
|
||||
}
|
||||
|
||||
if (extractorResult == null || !extractorResult.isOk()) {
|
||||
failureReason = buildAttemptFailureReason("参数提取失败", null);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (extractorResult.getParams() != null) {
|
||||
metaAction.getParams().putAll(extractorResult.getParams());
|
||||
}
|
||||
|
||||
try {
|
||||
runnerClient.submit(metaAction);
|
||||
} catch (Exception e) {
|
||||
failureReason = buildAttemptFailureReason("行动执行异常", e.getLocalizedMessage());
|
||||
continue;
|
||||
}
|
||||
|
||||
if (result.getStatus() == MetaAction.Result.Status.SUCCESS) {
|
||||
val historyAction = new HistoryAction(actionKey, actionCapability.loadMetaActionInfo(actionKey).getDescription(), result.getData());
|
||||
actionData.getHistory()
|
||||
.computeIfAbsent(executingStage, integer -> new ArrayList<>())
|
||||
.add(historyAction);
|
||||
return;
|
||||
}
|
||||
|
||||
failureReason = buildAttemptFailureReason("行动执行失败", result.getData());
|
||||
}
|
||||
metaAction.getResult().setStatus(MetaAction.Result.Status.FAILED);
|
||||
metaAction.getResult().setData(failureReason);
|
||||
}
|
||||
|
||||
private RecognizerTaskRecord startRecognizerIfNeeded(ExecutableAction executableAction, PhaserRecord phaserRecord) {
|
||||
if (!shouldRunCorrectionRecognizer(executableAction)) {
|
||||
return RecognizerTaskRecord.disabled();
|
||||
}
|
||||
val recognizerInput = assemblyHelper.buildRecognizerInput(executableAction);
|
||||
val task = buildRecognizerTask(recognizerInput, phaserRecord.phaser());
|
||||
Future<CorrectionRecognizerResult> future = virtualExecutor.submit(task);
|
||||
return new RecognizerTaskRecord(true, future);
|
||||
}
|
||||
|
||||
private Callable<CorrectionRecognizerResult> buildRecognizerTask(CorrectionRecognizerInput input, Phaser phaser) {
|
||||
phaser.register();
|
||||
return () -> {
|
||||
try {
|
||||
return actionCorrectionRecognizer.execute(input);
|
||||
} finally {
|
||||
phaser.arriveAndDeregister();
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
private CorrectionRecognizerResult resolveRecognizerResult(RecognizerTaskRecord record) {
|
||||
if (record == null || !record.enabled() || record.future() == null) {
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
if (!record.future().isDone()) {
|
||||
return null;
|
||||
}
|
||||
return record.future().get();
|
||||
} catch (Exception e) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private String buildAttemptFailureReason(String prefix, String detail) {
|
||||
if (detail == null || detail.isBlank()) {
|
||||
return prefix;
|
||||
}
|
||||
return prefix + ": " + detail;
|
||||
}
|
||||
|
||||
private boolean hasFailedMetaAction(List<MetaAction> metaActions) {
|
||||
return metaActions.stream().anyMatch(metaAction -> metaAction.getResult().getStatus() == MetaAction.Result.Status.FAILED);
|
||||
}
|
||||
|
||||
private boolean shouldRunCorrectionRecognizer(ExecutableAction executableAction) {
|
||||
val orderedStages = new ArrayList<>(executableAction.getActionChain().keySet());
|
||||
orderedStages.sort(Integer::compareTo);
|
||||
int totalStages = orderedStages.size();
|
||||
if (totalStages < 3) {
|
||||
return false;
|
||||
}
|
||||
int stageIndex = orderedStages.indexOf(executableAction.getExecutingStage());
|
||||
if (stageIndex < 0) {
|
||||
return false;
|
||||
}
|
||||
if (stageIndex == totalStages - 1) {
|
||||
return true;
|
||||
}
|
||||
return stageIndex >= 2 && (stageIndex - 2) % 2 == 0;
|
||||
}
|
||||
|
||||
private void ensureExecutableResult(ExecutableAction executableAction, boolean failed, String failureReason) {
|
||||
if (hasExecutableResult(executableAction)) {
|
||||
return;
|
||||
@@ -346,22 +431,17 @@ public class ActionExecutor extends AbstractAgentModule.Standalone {
|
||||
private record MetaActionsListeningRecord(AtomicBoolean accepting, int phase) {
|
||||
}
|
||||
|
||||
private record RecognizerTaskRecord(boolean enabled, Future<CorrectionRecognizerResult> future) {
|
||||
private static RecognizerTaskRecord disabled() {
|
||||
return new RecognizerTaskRecord(false, null);
|
||||
}
|
||||
}
|
||||
|
||||
@SuppressWarnings("InnerClassMayBeStatic")
|
||||
private class AssemblyHelper {
|
||||
private AssemblyHelper() {
|
||||
}
|
||||
|
||||
private RepairerInput buildRepairerInput(List<HistoryAction> historyActionsResults, MetaAction action, String userId) {
|
||||
RepairerInput input = new RepairerInput();
|
||||
MetaActionInfo metaActionInfo = actionCapability.loadMetaActionInfo(action.getKey());
|
||||
input.setHistoryActionResults(historyActionsResults);
|
||||
input.setParams(metaActionInfo.getParams());
|
||||
input.setRecentMessages(cognitionCapability.getChatMessages());
|
||||
input.setActionDescription(metaActionInfo.getDescription());
|
||||
input.setUserId(userId);
|
||||
return input;
|
||||
}
|
||||
|
||||
private ExtractorInput buildExtractorInput(MetaAction action, String source, List<HistoryAction> historyActionResults,
|
||||
List<String> additionalContext) {
|
||||
ExtractorInput input = new ExtractorInput();
|
||||
@@ -373,7 +453,7 @@ public class ActionExecutor extends AbstractAgentModule.Standalone {
|
||||
return input;
|
||||
}
|
||||
|
||||
private CorrectorInput buildCorrectorInput(ExecutableAction executableAction, String source) {
|
||||
private CorrectorInput buildCorrectorInput(ExecutableAction executableAction) {
|
||||
return CorrectorInput.builder()
|
||||
.tendency(executableAction.getTendency())
|
||||
.source(executableAction.getSource())
|
||||
@@ -385,5 +465,36 @@ public class ActionExecutor extends AbstractAgentModule.Standalone {
|
||||
.activatedSlices(memoryCapability.getActivatedSlices())
|
||||
.build();
|
||||
}
|
||||
|
||||
private CorrectionRecognizerInput buildRecognizerInput(ExecutableAction executableAction) {
|
||||
val orderedStages = new ArrayList<>(executableAction.getActionChain().keySet());
|
||||
orderedStages.sort(Integer::compareTo);
|
||||
int currentStageIndex = orderedStages.indexOf(executableAction.getExecutingStage());
|
||||
List<CorrectionRecognizerMetaActionSnapshot> currentStageMetaActions = executableAction.getActionChain()
|
||||
.getOrDefault(executableAction.getExecutingStage(), List.of())
|
||||
.stream()
|
||||
.map(metaAction -> CorrectionRecognizerMetaActionSnapshot.builder()
|
||||
.key(metaAction.getKey())
|
||||
.name(metaAction.getName())
|
||||
.io(metaAction.getIo())
|
||||
.resultStatus(metaAction.getResult().getStatus().name())
|
||||
.resultData(metaAction.getResult().getData())
|
||||
.build())
|
||||
.toList();
|
||||
return CorrectionRecognizerInput.builder()
|
||||
.tendency(executableAction.getTendency())
|
||||
.source(executableAction.getSource())
|
||||
.reason(executableAction.getReason())
|
||||
.description(executableAction.getDescription())
|
||||
.history(executableAction.getHistory().get(executableAction.getExecutingStage()))
|
||||
.currentStageMetaActions(currentStageMetaActions)
|
||||
.orderedStages(orderedStages)
|
||||
.currentStage(executableAction.getExecutingStage())
|
||||
.currentStageIndex(currentStageIndex)
|
||||
.lastStage(currentStageIndex == orderedStages.size() - 1)
|
||||
.recentMessages(cognitionCapability.getChatMessages())
|
||||
.activatedSlices(memoryCapability.getActivatedSlices())
|
||||
.build();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,27 @@
|
||||
package work.slhaf.partner.module.modules.action.executor.entity;
|
||||
|
||||
import lombok.Builder;
|
||||
import lombok.Data;
|
||||
import work.slhaf.partner.api.chat.pojo.Message;
|
||||
import work.slhaf.partner.core.memory.pojo.ActivatedMemorySlice;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
@Data
|
||||
@Builder
|
||||
public class CorrectionRecognizerInput {
|
||||
private String tendency;
|
||||
private String source;
|
||||
private String reason;
|
||||
private String description;
|
||||
|
||||
private List<HistoryAction> history;
|
||||
private List<CorrectionRecognizerMetaActionSnapshot> currentStageMetaActions;
|
||||
private List<Integer> orderedStages;
|
||||
private int currentStage;
|
||||
private int currentStageIndex;
|
||||
private boolean lastStage;
|
||||
|
||||
private List<Message> recentMessages;
|
||||
private List<ActivatedMemorySlice> activatedSlices;
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
package work.slhaf.partner.module.modules.action.executor.entity;
|
||||
|
||||
import lombok.Builder;
|
||||
import lombok.Data;
|
||||
|
||||
@Data
|
||||
@Builder
|
||||
public class CorrectionRecognizerMetaActionSnapshot {
|
||||
private String key;
|
||||
private String name;
|
||||
private boolean io;
|
||||
private String resultStatus;
|
||||
private String resultData;
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
package work.slhaf.partner.module.modules.action.executor.entity;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
@Data
|
||||
public class CorrectionRecognizerResult {
|
||||
private boolean needCorrection;
|
||||
private String reason;
|
||||
}
|
||||
@@ -1,450 +0,0 @@
|
||||
package work.slhaf.partner.module.modules.action.dispatcher.executor;
|
||||
|
||||
import com.alibaba.fastjson2.JSONObject;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import lombok.val;
|
||||
import org.junit.jupiter.api.*;
|
||||
import org.junit.jupiter.api.extension.ExtendWith;
|
||||
import org.mockito.InjectMocks;
|
||||
import org.mockito.Mock;
|
||||
import org.mockito.junit.jupiter.MockitoExtension;
|
||||
import work.slhaf.partner.core.action.ActionCapability;
|
||||
import work.slhaf.partner.core.action.ActionCore;
|
||||
import work.slhaf.partner.core.action.entity.*;
|
||||
import work.slhaf.partner.core.action.runner.RunnerClient;
|
||||
import work.slhaf.partner.core.cognition.CognitionCapability;
|
||||
import work.slhaf.partner.core.memory.MemoryCapability;
|
||||
import work.slhaf.partner.module.modules.action.executor.ActionCorrector;
|
||||
import work.slhaf.partner.module.modules.action.executor.ActionExecutor;
|
||||
import work.slhaf.partner.module.modules.action.executor.ActionRepairer;
|
||||
import work.slhaf.partner.module.modules.action.executor.ParamsExtractor;
|
||||
import work.slhaf.partner.module.modules.action.executor.entity.CorrectorResult;
|
||||
import work.slhaf.partner.module.modules.action.executor.entity.ExtractorResult;
|
||||
import work.slhaf.partner.module.modules.action.executor.entity.RepairerResult;
|
||||
|
||||
import java.util.*;
|
||||
import java.util.concurrent.*;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.anyString;
|
||||
import static org.mockito.Mockito.*;
|
||||
|
||||
/**
|
||||
* 测试矩阵(与文档一致):
|
||||
* 1) 单行动-单阶段-单MetaAction成功(已覆盖)
|
||||
* 2) 多行动并发执行(未覆盖:需并发稳定性/线程调度控制)
|
||||
* 3) status 非 PREPARE 直接返回(已覆盖)
|
||||
* 4) 多阶段顺序执行(已覆盖)
|
||||
* 5) IO 行动使用虚拟线程池(已覆盖)
|
||||
* 6) extractor 失败 -> repairer OK -> 再成功(已覆盖)
|
||||
* 7) extractor 失败 -> repairer FAILED(已覆盖)
|
||||
* 8) extractor 失败 -> repairer ACQUIRE 阻塞后恢复(已覆盖)
|
||||
* 9) runnerClient.submit 抛异常(未覆盖:需更精细的异常吞吐与线程结束校验)
|
||||
* 10) paramsExtractor.execute 抛异常(未覆盖:与 #9 类似,需更精细的异常吞吐校验)
|
||||
* 11) corrector.execute 抛异常导致资源未清理(已标记已知缺陷,@Disabled)
|
||||
* 12) actionChain 为空导致异常与泄漏(已标记已知缺陷,@Disabled)
|
||||
* 13) metaActions 为空导致 awaitAdvance 阻塞(未覆盖:更适合集成/压测)
|
||||
* 17) result 状态不更新导致循环不退出(未覆盖:更适合集成/压测)
|
||||
* 18) 同 stage 多 metaAction 并发完成顺序不固定(未覆盖:更适合集成/压测)
|
||||
*/
|
||||
@SuppressWarnings("unused")
|
||||
@Slf4j
|
||||
@ExtendWith(MockitoExtension.class)
|
||||
class ActionExecutorTest {
|
||||
|
||||
@Mock
|
||||
ActionCapability actionCapability;
|
||||
@Mock
|
||||
MemoryCapability memoryCapability;
|
||||
@Mock
|
||||
CognitionCapability cognitionCapability;
|
||||
@Mock
|
||||
ParamsExtractor paramsExtractor;
|
||||
@Mock
|
||||
ActionRepairer actionRepairer;
|
||||
@Mock
|
||||
ActionCorrector actionCorrector;
|
||||
@Mock
|
||||
RunnerClient runnerClient;
|
||||
|
||||
@InjectMocks
|
||||
ActionExecutor actionExecutor;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() {
|
||||
lenient().when(cognitionCapability.getChatMessages()).thenReturn(Collections.emptyList());
|
||||
lenient().when(memoryCapability.getActivatedSlices()).thenReturn(Collections.emptyList());
|
||||
lenient().when(actionCapability.putPhaserRecord(any(Phaser.class), any(ExecutableAction.class)))
|
||||
.thenAnswer(inv -> new PhaserRecord(inv.getArgument(0), inv.getArgument(1)));
|
||||
lenient().when(actionCapability.loadMetaActionInfo(anyString())).thenAnswer(inv -> new MetaActionInfo(
|
||||
false,
|
||||
null,
|
||||
Collections.emptyMap(),
|
||||
"desc",
|
||||
Set.of(),
|
||||
Set.of(),
|
||||
Set.of(),
|
||||
false,
|
||||
new JSONObject()
|
||||
));
|
||||
CorrectorResult correctorResult = new CorrectorResult();
|
||||
correctorResult.setMetaInterventionList(Collections.emptyList());
|
||||
lenient().when(actionCorrector.execute(any())).thenReturn(correctorResult);
|
||||
lenient().doNothing().when(actionCapability).handleInterventions(any(), any());
|
||||
}
|
||||
|
||||
// 场景1:B1 -> B3 -> B4 -> B7(成功) -> B10。目的:验证正常主路径与资源清理。
|
||||
@Test
|
||||
void execute_singleAction_singleStage_success() {
|
||||
ExecutorService directExecutor = new DirectExecutorService();
|
||||
stubExecutors(directExecutor, directExecutor);
|
||||
|
||||
ImmediateExecutableAction actionData = buildActionData(singleStageChain(false));
|
||||
|
||||
ExtractorResult extractorResult = new ExtractorResult();
|
||||
extractorResult.setOk(true);
|
||||
when(paramsExtractor.execute(any())).thenReturn(extractorResult);
|
||||
doAnswer(inv -> {
|
||||
MetaAction metaAction = inv.getArgument(0);
|
||||
metaAction.getResult().setStatus(MetaAction.Result.Status.SUCCESS);
|
||||
return null;
|
||||
}).when(runnerClient).submit(any(MetaAction.class));
|
||||
|
||||
actionExecutor.init();
|
||||
actionExecutor.execute(actionData);
|
||||
|
||||
verify(runnerClient, times(1)).submit(any(MetaAction.class));
|
||||
verify(actionCapability, times(1)).removePhaserRecord(any(Phaser.class));
|
||||
assertEquals(ExecutableAction.Status.SUCCESS, actionData.getStatus());
|
||||
assertEquals(1, actionData.getHistory().get(0).size());
|
||||
}
|
||||
|
||||
// 场景3:B1 -> B2。目的:验证非 PREPARE 不执行任何子任务。
|
||||
@Test
|
||||
void execute_statusNotPrepare_shouldSkip() {
|
||||
ExecutorService directExecutor = new DirectExecutorService();
|
||||
stubExecutors(directExecutor, directExecutor);
|
||||
|
||||
ImmediateExecutableAction actionData = buildActionData(singleStageChain(false));
|
||||
actionData.setStatus(ExecutableAction.Status.EXECUTING);
|
||||
|
||||
actionExecutor.init();
|
||||
actionExecutor.execute(actionData);
|
||||
|
||||
verify(actionCapability, never()).putPhaserRecord(any(Phaser.class), any(ExecutableAction.class));
|
||||
verify(runnerClient, never()).submit(any(MetaAction.class));
|
||||
}
|
||||
|
||||
// 场景4:B1 -> B3 -> B4(两轮) -> B7(成功) -> B10。目的:验证多阶段顺序执行。
|
||||
@Test
|
||||
void execute_multiStage_success() {
|
||||
ExecutorService platformExecutor = Executors.newFixedThreadPool(4);
|
||||
ExecutorService virtualExecutor = Executors.newVirtualThreadPerTaskExecutor();
|
||||
stubExecutors(platformExecutor, virtualExecutor);
|
||||
|
||||
Map<Integer, List<MetaAction>> chain = new HashMap<>();
|
||||
chain.put(0, List.of(buildMetaAction("a1", false)));
|
||||
chain.put(1, List.of(buildMetaAction("a2", false)));
|
||||
ImmediateExecutableAction actionData = buildActionData(chain);
|
||||
|
||||
ExtractorResult extractorResult = new ExtractorResult();
|
||||
extractorResult.setOk(true);
|
||||
when(paramsExtractor.execute(any())).thenReturn(extractorResult);
|
||||
doAnswer(inv -> {
|
||||
MetaAction metaAction = inv.getArgument(0);
|
||||
metaAction.getResult().setStatus(MetaAction.Result.Status.SUCCESS);
|
||||
log.info("metaAction result:{}", metaAction.getResult().getStatus());
|
||||
return null;
|
||||
}).when(runnerClient).submit(any(MetaAction.class));
|
||||
|
||||
actionExecutor.init();
|
||||
actionExecutor.execute(actionData);
|
||||
|
||||
verify(runnerClient, timeout(5000).times(2)).submit(any(MetaAction.class));
|
||||
verify(actionCorrector, timeout(5000).times(2)).execute(any());
|
||||
assertEquals(2, actionData.getHistory().size());
|
||||
assertEquals(ExecutableAction.Status.SUCCESS, actionData.getStatus());
|
||||
}
|
||||
|
||||
// 场景5:B4.2。目的:验证 IO 行动使用虚拟线程池。
|
||||
@Test
|
||||
void execute_ioMetaAction_usesVirtualExecutor() {
|
||||
ExecutorService platformExecutor = Executors.newFixedThreadPool(4);
|
||||
ExecutorService virtualExecutor = Executors.newVirtualThreadPerTaskExecutor();
|
||||
stubExecutors(platformExecutor, virtualExecutor);
|
||||
|
||||
ImmediateExecutableAction actionData = buildActionData(singleStageChain(true));
|
||||
|
||||
ExtractorResult extractorResult = new ExtractorResult();
|
||||
extractorResult.setOk(true);
|
||||
lenient().when(paramsExtractor.execute(any())).thenReturn(extractorResult);
|
||||
lenient().doAnswer(inv -> {
|
||||
MetaAction metaAction = inv.getArgument(0);
|
||||
metaAction.getResult().setStatus(MetaAction.Result.Status.SUCCESS);
|
||||
return null;
|
||||
}).when(runnerClient).submit(any(MetaAction.class));
|
||||
|
||||
actionExecutor.init();
|
||||
actionExecutor.execute(actionData);
|
||||
|
||||
verify(actionCapability, times(1)).getExecutor(ActionCore.ExecutorType.VIRTUAL);
|
||||
shutdownExecutor(virtualExecutor);
|
||||
}
|
||||
|
||||
// 场景6:B7.2(失败) -> repairer OK -> B7(成功)。目的:验证修复后成功与上下文追加。
|
||||
@Test
|
||||
void execute_extractorFail_thenRepairOk_thenSuccess() {
|
||||
ExecutorService platformExecutor = Executors.newFixedThreadPool(4);
|
||||
ExecutorService virtualExecutor = Executors.newVirtualThreadPerTaskExecutor();
|
||||
stubExecutors(platformExecutor, virtualExecutor);
|
||||
|
||||
ImmediateExecutableAction actionData = buildActionData(singleStageChain(false));
|
||||
|
||||
ExtractorResult fail = new ExtractorResult();
|
||||
fail.setOk(false);
|
||||
ExtractorResult ok = new ExtractorResult();
|
||||
ok.setOk(true);
|
||||
when(paramsExtractor.execute(any())).thenReturn(fail, ok);
|
||||
|
||||
RepairerResult repairerResult = new RepairerResult();
|
||||
repairerResult.setStatus(RepairerResult.RepairerStatus.OK);
|
||||
repairerResult.setFixedData(List.of("fix-1"));
|
||||
when(actionRepairer.execute(any())).thenReturn(repairerResult);
|
||||
|
||||
doAnswer(inv -> {
|
||||
MetaAction metaAction = inv.getArgument(0);
|
||||
metaAction.getResult().setStatus(MetaAction.Result.Status.SUCCESS);
|
||||
return null;
|
||||
}).when(runnerClient).submit(any(MetaAction.class));
|
||||
|
||||
actionExecutor.init();
|
||||
actionExecutor.execute(actionData);
|
||||
|
||||
try {
|
||||
Thread.sleep(500);
|
||||
} catch (InterruptedException ignored) {
|
||||
}
|
||||
assertEquals(1, actionData.getAdditionalContext().get(0).size());
|
||||
verify(runnerClient, timeout(5000).times(1)).submit(any(MetaAction.class));
|
||||
}
|
||||
|
||||
// 场景7:B7.2(失败) -> repairer FAILED。目的:验证失败分支不提交外部执行。
|
||||
@Test
|
||||
void execute_extractorFail_thenRepairFailed() {
|
||||
ExecutorService platformExecutor = Executors.newFixedThreadPool(4);
|
||||
ExecutorService virtualExecutor = Executors.newVirtualThreadPerTaskExecutor();
|
||||
stubExecutors(platformExecutor, virtualExecutor);
|
||||
|
||||
ImmediateExecutableAction actionData = buildActionData(singleStageChain(false));
|
||||
|
||||
ExtractorResult fail = new ExtractorResult();
|
||||
fail.setOk(false);
|
||||
when(paramsExtractor.execute(any())).thenReturn(fail);
|
||||
|
||||
RepairerResult repairerResult = new RepairerResult();
|
||||
repairerResult.setStatus(RepairerResult.RepairerStatus.FAILED);
|
||||
when(actionRepairer.execute(any())).thenReturn(repairerResult);
|
||||
|
||||
actionExecutor.init();
|
||||
actionExecutor.execute(actionData);
|
||||
|
||||
try {
|
||||
Thread.sleep(500);
|
||||
} catch (InterruptedException ignored) {
|
||||
}
|
||||
MetaAction metaAction = actionData.getActionChain().get(0).getFirst();
|
||||
assertEquals(MetaAction.Result.Status.FAILED, metaAction.getResult().getStatus());
|
||||
verify(runnerClient, never()).submit(any(MetaAction.class));
|
||||
}
|
||||
|
||||
// 场景8:B7.2(ACQUIRE) -> interrupt 阻塞 -> 状态恢复 -> B7(成功)。目的:验证阻塞可恢复且不死锁。
|
||||
@Test
|
||||
@Timeout(3)
|
||||
void execute_extractorFail_thenAcquire_thenResume() throws Exception {
|
||||
ExecutorService platformExecutor = Executors.newCachedThreadPool();
|
||||
ExecutorService virtualExecutor = Executors.newCachedThreadPool();
|
||||
stubExecutors(platformExecutor, virtualExecutor);
|
||||
|
||||
ImmediateExecutableAction actionData = buildActionData(singleStageChain(false));
|
||||
|
||||
ExtractorResult fail = new ExtractorResult();
|
||||
fail.setOk(false);
|
||||
ExtractorResult ok = new ExtractorResult();
|
||||
ok.setOk(true);
|
||||
when(paramsExtractor.execute(any())).thenReturn(fail, ok);
|
||||
|
||||
RepairerResult repairerResult = new RepairerResult();
|
||||
repairerResult.setStatus(RepairerResult.RepairerStatus.ACQUIRE);
|
||||
when(actionRepairer.execute(any())).thenReturn(repairerResult);
|
||||
|
||||
doAnswer(inv -> {
|
||||
MetaAction metaAction = inv.getArgument(0);
|
||||
metaAction.getResult().setStatus(MetaAction.Result.Status.SUCCESS);
|
||||
return null;
|
||||
}).when(runnerClient).submit(any(MetaAction.class));
|
||||
|
||||
CountDownLatch doneLatch = new CountDownLatch(1);
|
||||
doAnswer(inv -> {
|
||||
doneLatch.countDown();
|
||||
return null;
|
||||
}).when(actionCapability).removePhaserRecord(any(Phaser.class));
|
||||
|
||||
ExecutorService resumeExecutor = Executors.newSingleThreadExecutor();
|
||||
resumeExecutor.execute(() -> {
|
||||
while (actionData.getStatus() != ExecutableAction.Status.INTERRUPTED) {
|
||||
try {
|
||||
Thread.sleep(10);
|
||||
} catch (InterruptedException ignored) {
|
||||
}
|
||||
}
|
||||
actionData.setStatus(ExecutableAction.Status.EXECUTING);
|
||||
});
|
||||
|
||||
actionExecutor.init();
|
||||
actionExecutor.execute(actionData);
|
||||
|
||||
assertTrue(doneLatch.await(2, TimeUnit.SECONDS));
|
||||
shutdownExecutor(platformExecutor);
|
||||
shutdownExecutor(virtualExecutor);
|
||||
shutdownExecutor(resumeExecutor);
|
||||
}
|
||||
|
||||
// 场景11:B4.4 异常 -> 资源未清理(已知缺陷)。目的:暴露当前行为。
|
||||
// @Disabled("known-issue: corrector 抛异常时未清理 phaser 记录")
|
||||
// @Tag("known-issue")
|
||||
@Test
|
||||
void execute_correctorThrows_shouldLeakPhaserRecord() {
|
||||
ExecutorService platformExecutor = Executors.newCachedThreadPool();
|
||||
ExecutorService virtualExecutor = Executors.newCachedThreadPool();
|
||||
stubExecutors(platformExecutor, virtualExecutor);
|
||||
|
||||
ImmediateExecutableAction actionData = buildActionData(singleStageChain(false));
|
||||
|
||||
ExtractorResult ok = new ExtractorResult();
|
||||
ok.setOk(true);
|
||||
lenient().when(paramsExtractor.execute(any())).thenReturn(ok);
|
||||
lenient().doAnswer(inv -> {
|
||||
MetaAction metaAction = inv.getArgument(0);
|
||||
metaAction.getResult().setStatus(MetaAction.Result.Status.SUCCESS);
|
||||
return null;
|
||||
}).when(runnerClient).submit(any(MetaAction.class));
|
||||
|
||||
lenient().doThrow(new RuntimeException("boom")).when(actionCorrector).execute(any());
|
||||
|
||||
actionExecutor.init();
|
||||
actionExecutor.execute(actionData);
|
||||
|
||||
try {
|
||||
Thread.sleep(500);
|
||||
} catch (InterruptedException ignored) {
|
||||
}
|
||||
verify(actionCapability).removePhaserRecord(any(Phaser.class));
|
||||
}
|
||||
|
||||
// 场景12:B4.1 actionChain 为空导致异常(已知缺陷)。目的:暴露当前行为。
|
||||
@Disabled("known-issue: actionChain 为空导致 IndexOutOfBounds 与资源未清理")
|
||||
@Tag("known-issue")
|
||||
@Test
|
||||
void execute_emptyActionChain_shouldFail() {
|
||||
ExecutorService platformExecutor = Executors.newCachedThreadPool();
|
||||
ExecutorService virtualExecutor = Executors.newCachedThreadPool();
|
||||
stubExecutors(platformExecutor, virtualExecutor);
|
||||
|
||||
ImmediateExecutableAction actionData = buildActionData(new HashMap<>());
|
||||
|
||||
actionExecutor.init();
|
||||
actionExecutor.execute(actionData);
|
||||
|
||||
try {
|
||||
Thread.sleep(500);
|
||||
} catch (InterruptedException ignored) {
|
||||
}
|
||||
}
|
||||
|
||||
private void stubExecutors(ExecutorService platformExecutor, ExecutorService virtualExecutor) {
|
||||
when(actionCapability.getExecutor(ActionCore.ExecutorType.PLATFORM)).thenReturn(platformExecutor);
|
||||
when(actionCapability.getExecutor(ActionCore.ExecutorType.VIRTUAL)).thenReturn(virtualExecutor);
|
||||
when(actionCapability.runnerClient()).thenReturn(runnerClient);
|
||||
}
|
||||
|
||||
private ImmediateExecutableAction buildActionData(Map<Integer, List<MetaAction>> actionChain) {
|
||||
val immediateActionData = new ImmediateExecutableAction(
|
||||
"tendency",
|
||||
actionChain,
|
||||
"reason",
|
||||
"desc",
|
||||
"source"
|
||||
);
|
||||
immediateActionData.getAdditionalContext().putAll(initAdditionalContext(actionChain));
|
||||
return immediateActionData;
|
||||
}
|
||||
|
||||
private Map<Integer, List<MetaAction>> singleStageChain(boolean io) {
|
||||
Map<Integer, List<MetaAction>> chain = new HashMap<>();
|
||||
chain.put(0, List.of(buildMetaAction("a1", io)));
|
||||
return chain;
|
||||
}
|
||||
|
||||
private MetaAction buildMetaAction(String name, boolean io) {
|
||||
return new MetaAction(
|
||||
name,
|
||||
io,
|
||||
null,
|
||||
MetaAction.Type.ORIGIN,
|
||||
"location"
|
||||
);
|
||||
}
|
||||
|
||||
private Map<Integer, List<String>> initAdditionalContext(Map<Integer, List<MetaAction>> actionChain) {
|
||||
Map<Integer, List<String>> context = new HashMap<>();
|
||||
for (Integer stage : actionChain.keySet()) {
|
||||
context.put(stage, new ArrayList<>());
|
||||
}
|
||||
return context;
|
||||
}
|
||||
|
||||
private void shutdownExecutor(ExecutorService executor) {
|
||||
executor.shutdownNow();
|
||||
try {
|
||||
executor.awaitTermination(1, TimeUnit.SECONDS);
|
||||
} catch (InterruptedException ignored) {
|
||||
}
|
||||
}
|
||||
|
||||
private static final class DirectExecutorService extends AbstractExecutorService {
|
||||
private volatile boolean shutdown;
|
||||
|
||||
@Override
|
||||
public void shutdown() {
|
||||
shutdown = true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<Runnable> shutdownNow() {
|
||||
shutdown = true;
|
||||
return Collections.emptyList();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isShutdown() {
|
||||
return shutdown;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isTerminated() {
|
||||
return shutdown;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean awaitTermination(long timeout, TimeUnit unit) {
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void execute(Runnable command) {
|
||||
command.run();
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user