From ecec3fd9c047c32578b2be06c886857bc5a0edcc Mon Sep 17 00:00:00 2001 From: redyyang Date: Mon, 19 Oct 2026 13:41:04 +0800 Subject: [PATCH] feat(planner): add graph-based plan execution with bounded replanning (T40) Add an opt-in PlanExecutor that runs a validated PlanSpec on a fixed StateGraph: ready steps are dispatched as bounded local parallel batches, downstream steps never start before their dependencies complete, and step outcomes are recorded in a JSON PlanRun ledger stored under a new state key. Failures are routed by a PlanFailurePolicy to retry, bounded replan, need-input interruption or stop. Replans are limited by count, deadline and step budget; proposals must pass generated-plan validation, keep completed steps unchanged, never reuse failed or retired step identities, and contain new work, so completed effects are never executed again. Missing, invalid or schema-mismatched results never count as success, and every exit has an explicit terminal status. Refs #143 --- .../agent/planner/PlanExecutionLimits.java | 59 ++ .../ai/graph/agent/planner/PlanExecutor.java | 851 ++++++++++++++++++ .../agent/planner/PlanFailurePolicy.java | 75 ++ .../graph/agent/planner/PlanReasonCodes.java | 67 ++ .../ai/graph/agent/planner/PlanReplanner.java | 69 ++ .../ai/graph/agent/planner/PlanRun.java | 100 ++ .../ai/graph/agent/planner/PlanRunStatus.java | 55 ++ .../graph/agent/planner/PlanStepExecutor.java | 44 + .../graph/agent/planner/PlanStepOutcome.java | 101 +++ .../graph/agent/planner/PlanStepRequest.java | 44 + .../graph/agent/planner/PlanExecutorTest.java | 460 ++++++++++ 11 files changed, 1925 insertions(+) create mode 100644 argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanExecutionLimits.java create mode 100644 argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanExecutor.java create mode 100644 argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanFailurePolicy.java create mode 100644 argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanReasonCodes.java create mode 100644 argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanReplanner.java create mode 100644 argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanRun.java create mode 100644 argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanRunStatus.java create mode 100644 argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanStepExecutor.java create mode 100644 argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanStepOutcome.java create mode 100644 argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanStepRequest.java create mode 100644 argi-agent-framework/src/test/java/io/github/agentic/ai/graph/agent/planner/PlanExecutorTest.java diff --git a/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanExecutionLimits.java b/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanExecutionLimits.java new file mode 100644 index 000000000..59fb683fc --- /dev/null +++ b/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanExecutionLimits.java @@ -0,0 +1,59 @@ +/* + * Copyright 2025-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.github.agentic.ai.graph.agent.planner; + +import java.time.Duration; + +/** + * Bounds of one plan run. Every run is bounded by attempts, replans and the total step + * execution budget; the wall-clock deadline is optional and measured from the start of + * the run, including time spent waiting for external input. + * + * @param maxParallelism maximum ready steps dispatched together in one local batch + * @param maxStepAttempts maximum attempts of one step identity, including the first + * @param maxReplans maximum replanning attempts, valid or not + * @param maxStepExecutions total step dispatches allowed in the run (the budget) + * @param deadline optional wall-clock deadline of the run, or {@code null} + */ +public record PlanExecutionLimits(int maxParallelism, int maxStepAttempts, int maxReplans, int maxStepExecutions, + Duration deadline) { + + public static final int DEFAULT_MAX_PARALLELISM = 4; + + public static final int DEFAULT_MAX_STEP_ATTEMPTS = 2; + + public static final int DEFAULT_MAX_REPLANS = 2; + + public static final int DEFAULT_MAX_STEP_EXECUTIONS = 64; + + public PlanExecutionLimits { + PlanValidator.require(maxParallelism > 0 && maxStepAttempts > 0 && maxReplans >= 0 && maxStepExecutions > 0, + "Plan execution limits must be positive"); + PlanValidator.require(deadline == null || (!deadline.isNegative() && !deadline.isZero()), + "Plan deadline must be positive"); + } + + public static PlanExecutionLimits defaults() { + return new PlanExecutionLimits(DEFAULT_MAX_PARALLELISM, DEFAULT_MAX_STEP_ATTEMPTS, DEFAULT_MAX_REPLANS, + DEFAULT_MAX_STEP_EXECUTIONS, null); + } + + public PlanExecutionLimits withDeadline(Duration deadline) { + return new PlanExecutionLimits(this.maxParallelism, this.maxStepAttempts, this.maxReplans, + this.maxStepExecutions, deadline); + } + +} diff --git a/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanExecutor.java b/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanExecutor.java new file mode 100644 index 000000000..307b581ae --- /dev/null +++ b/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanExecutor.java @@ -0,0 +1,851 @@ +/* + * Copyright 2025-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.github.agentic.ai.graph.agent.planner; + +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.time.Clock; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashSet; +import java.util.HexFormat; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executor; +import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import io.github.agentic.ai.graph.CompileConfig; +import io.github.agentic.ai.graph.CompiledGraph; +import io.github.agentic.ai.graph.KeyStrategy; +import io.github.agentic.ai.graph.KeyStrategyFactoryBuilder; +import io.github.agentic.ai.graph.OverAllState; +import io.github.agentic.ai.graph.RunnableConfig; +import io.github.agentic.ai.graph.StateGraph; +import io.github.agentic.ai.graph.action.AsyncNodeActionWithConfig; +import io.github.agentic.ai.graph.action.InterruptableAction; +import io.github.agentic.ai.graph.action.InterruptionMetadata; +import io.github.agentic.ai.graph.agent.interceptor.ToolOutcome; +import io.github.agentic.ai.graph.exception.GraphStateException; +import com.networknt.schema.Schema; +import tools.jackson.databind.JsonNode; + +import static io.github.agentic.ai.graph.StateGraph.END; +import static io.github.agentic.ai.graph.StateGraph.START; +import static io.github.agentic.ai.graph.action.AsyncEdgeAction.edge_async; +import static io.github.agentic.ai.graph.action.AsyncNodeActionWithConfig.node_async; + +/** + * Opt-in Plan-and-Execute runtime on top of {@link StateGraph}. A validated + * {@link PlanSpec} is executed by a small fixed graph: + * + * 
+ * START -> plan_execute -> (plan_execute | plan_replan | plan_input | END)
+ *
+ * + * Each {@code plan_execute} pass dispatches the ready steps (all dependencies completed) + * of the current revision as one local, bounded parallel batch and records their outcomes + * in the {@link PlanRun} ledger. A downstream step is never dispatched before its + * dependencies completed. Failures are routed by the {@link PlanFailurePolicy} to retry, + * bounded replanning, an input interruption or a terminal stop. + * + *

+ * This is a whole-run, in-process executor: node boundaries checkpoint the ledger through + * the configured saver, and {@code plan_input} uses the standard graph interruption so a + * run suspended with {@link PlanRunStatus#NEED_INPUT} resumes from its checkpoint. + * Durable commit/journal semantics, cross-instance workers and sub-agent protocols remain + * with their dedicated components; step effects stay behind the application's + * {@link PlanStepExecutor} and its existing authorization gates. + *

+ */ +public final class PlanExecutor { + + /** State key holding the JSON-encoded {@link PlanRun}. */ + public static final String RUN_KEY = "argi_plan_run"; + + /** State key holding JSON-encoded input supplied for the awaited step. */ + public static final String INPUT_KEY = "argi_plan_input"; + + public static final String EXECUTE_NODE = "plan_execute"; + + public static final String REPLAN_NODE = "plan_replan"; + + public static final String INPUT_NODE = "plan_input"; + + /** Interruption metadata key carrying the awaited step ID. */ + public static final String INTERRUPT_STEP_ID = "planStepId"; + + /** Interruption metadata key carrying the input request. */ + public static final String INTERRUPT_PROMPT = "planPrompt"; + + /** Interruption metadata key carrying the reason code. */ + public static final String INTERRUPT_REASON = "planReasonCode"; + + private final PlanValidator validator; + + private final PlanStepExecutor stepExecutor; + + private final PlanReplanner replanner; + + private final PlanFailurePolicy failurePolicy; + + private final PlanExecutionLimits limits; + + private final Clock clock; + + private final Executor executor; + + private PlanExecutor(Builder builder) { + this.validator = builder.validator; + this.stepExecutor = builder.stepExecutor; + this.replanner = builder.replanner; + this.failurePolicy = builder.failurePolicy; + this.limits = builder.limits; + this.clock = builder.clock; + this.executor = builder.executor; + } + + public static Builder builder(PlanValidator validator, PlanStepExecutor stepExecutor) { + return new Builder(validator, stepExecutor); + } + + /** + * Create the initial state of a new run. A new run cannot claim progress: the plan + * must pass {@link PlanValidator#validateGenerated(PlanSpec)}. + * @param plan the validated initial revision + * @return the state to pass to the compiled graph + */ + public Map initialState(PlanSpec plan) { + this.validator.validateGenerated(plan); + Ledger ledger = new Ledger(plan, this.clock.millis()); + return Map.of(RUN_KEY, encode(ledger.toRun())); + } + + /** + * State update that supplies input for the awaited step, for use with + * {@link RunnableConfig.Builder#addStateUpdate(Map)} on resume or + * {@link CompiledGraph#updateState(RunnableConfig, Map)}. + * @param stepId the awaited step + * @param jsonInput JSON-encoded input + * @return the state update + */ + public static Map input(String stepId, String jsonInput) { + PlanValidator.text(stepId, "stepId"); + return Map.of(INPUT_KEY, PlanCodec.MAPPER.writeValueAsString(new ProvidedInput(stepId, jsonInput))); + } + + /** Read the run ledger from graph state. */ + public static PlanRun read(OverAllState state) { + Objects.requireNonNull(state, "state"); + return read(state.data()); + } + + /** Read the run ledger from graph state data. */ + public static PlanRun read(Map state) { + Object value = state == null ? null : state.get(RUN_KEY); + PlanValidator.require(value instanceof String, "State does not contain a plan run"); + try { + return PlanCodec.MAPPER.readValue((String) value, PlanRun.class); + } + catch (RuntimeException ex) { + throw new PlanValidationException("Invalid plan run state", ex); + } + } + + /** + * Graph iterations needed by one invocation within the configured limits: each + * execution pass consumes budget, each replan consumes a replanning attempt. + */ + public int recursionLimit() { + return 2 * this.limits.maxStepExecutions() + 2 * this.limits.maxReplans() + 4; + } + + /** Build the uncompiled executor graph. */ + public StateGraph graph() throws GraphStateException { + var keys = new KeyStrategyFactoryBuilder().defaultStrategy(KeyStrategy.REPLACE) + .addStrategy(RUN_KEY, KeyStrategy.REPLACE) + .addStrategy(INPUT_KEY, KeyStrategy.REPLACE) + .build(); + Map routes = Map.of(EXECUTE_NODE, EXECUTE_NODE, REPLAN_NODE, REPLAN_NODE, INPUT_NODE, + INPUT_NODE, END, END); + return new StateGraph("argi-plan-executor", keys) + .addNode(EXECUTE_NODE, node_async((state, config) -> execute(state))) + .addNode(REPLAN_NODE, node_async((state, config) -> replan(state))) + .addNode(INPUT_NODE, new InputNode()) + .addEdge(START, EXECUTE_NODE) + .addConditionalEdges(EXECUTE_NODE, edge_async(PlanExecutor::route), routes) + .addConditionalEdges(REPLAN_NODE, edge_async(PlanExecutor::route), routes) + .addConditionalEdges(INPUT_NODE, edge_async(PlanExecutor::route), routes); + } + + /** + * Compile the executor graph, raising the recursion limit to + * {@link #recursionLimit()} when the given configuration is lower. + */ + public CompiledGraph compile(CompileConfig config) throws GraphStateException { + CompileConfig base = config == null ? CompileConfig.builder().build() : config; + return graph().compile(CompileConfig.builder(base) + .recursionLimit(Math.max(base.recursionLimit(), recursionLimit())) + .build()); + } + + private static String route(OverAllState state) { + PlanRun run = read(state); + if (run.status() == PlanRunStatus.NEED_INPUT) { + return INPUT_NODE; + } + if (run.terminal()) { + return END; + } + return run.failedStepIds().isEmpty() ? EXECUTE_NODE : REPLAN_NODE; + } + + // ---------------------------------------------------------------- execution + + private Map execute(OverAllState state) { + Ledger ledger = new Ledger(read(state)); + if (ledger.status != PlanRunStatus.RUNNING || !ledger.failedStepIds.isEmpty()) { + return Map.of(); + } + if (deadlinePassed(ledger)) { + return write(ledger.terminate(PlanRunStatus.DEADLINE_EXCEEDED, PlanReasonCodes.DEADLINE_EXCEEDED, null)); + } + List ready = ledger.readySteps(); + if (ready.isEmpty()) { + return write(ledger.allCompleted() + ? ledger.terminate(PlanRunStatus.COMPLETED, PlanReasonCodes.PLAN_COMPLETED, null) + : ledger.terminate(PlanRunStatus.FAILED, PlanReasonCodes.NO_READY_STEPS, null)); + } + int budget = this.limits.maxStepExecutions() - ledger.stepExecutions; + if (budget <= 0) { + return write(ledger.terminate(PlanRunStatus.BUDGET_EXHAUSTED, PlanReasonCodes.BUDGET_EXHAUSTED, null)); + } + List batch = ready.subList(0, Math.min(ready.size(), Math.min(budget, this.limits.maxParallelism()))); + + Map> running = new LinkedHashMap<>(); + Map attempts = new LinkedHashMap<>(); + for (PlanStep step : batch) { + ledger.stepExecutions++; + int attempt = ledger.attempts.merge(step.stepId(), 1, Integer::sum); + attempts.put(step.stepId(), attempt); + PlanStepRequest request = new PlanStepRequest(ledger.objective(), ledger.revision(), step, + stepKey(step), attempt, ledger.dependencyResults(step), ledger.inputs.get(step.stepId())); + running.put(step.stepId(), CompletableFuture.supplyAsync(() -> invoke(request), this.executor)); + } + + boolean deadlineHit = false; + Map outcomes = new LinkedHashMap<>(); + for (Map.Entry> entry : running.entrySet()) { + PlanStepOutcome outcome; + try { + Long remaining = remainingMillis(ledger); + outcome = remaining == null ? entry.getValue().get() + : entry.getValue().get(Math.max(0, remaining), TimeUnit.MILLISECONDS); + } + catch (TimeoutException ex) { + entry.getValue().cancel(true); + deadlineHit = true; + outcome = PlanStepOutcome.failed(ToolOutcome.TIMEOUT, "Plan deadline passed while the step was running"); + } + catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + entry.getValue().cancel(true); + outcome = PlanStepOutcome.failed(ToolOutcome.CANCELLED, "Plan execution was interrupted"); + } + catch (ExecutionException ex) { + outcome = PlanStepOutcome.failed(ToolOutcome.ERROR, describe(ex.getCause())); + } + outcomes.put(entry.getKey(), outcome); + } + + Decision decision = null; + for (PlanStep step : batch) { + PlanStepOutcome outcome = outcomes.get(step.stepId()); + if (outcome.effectReference() != null) { + ledger.effectReferences.put(step.stepId(), outcome.effectReference()); + } + Decision next = switch (outcome.kind()) { + case SUCCEEDED -> { + String problem = outputProblem(step, outcome.result()); + if (problem == null) { + ledger.complete(step, outcome.result()); + yield null; + } + yield decide(step, ToolOutcome.ERROR, problem, attempts.get(step.stepId()), + outcome.result() == null ? PlanReasonCodes.STEP_RESULT_MISSING + : PlanReasonCodes.STEP_OUTPUT_INVALID); + } + case NEED_INPUT -> new Decision(PlanFailurePolicy.Action.NEED_INPUT, step, outcome.error(), + PlanReasonCodes.NEED_INPUT); + case FAILED -> decide(step, outcome.outcome(), outcome.error(), attempts.get(step.stepId()), + PlanReasonCodes.STEP_FAILED); + }; + if (next != null && next.action() == PlanFailurePolicy.Action.REPLAN) { + ledger.fail(step, next.detail()); + } + if (next != null && next.action() == PlanFailurePolicy.Action.STOP) { + ledger.fail(step, next.detail()); + } + if (next != null && (decision == null || priority(next.action()) > priority(decision.action()))) { + decision = next; + } + } + + if (deadlineHit) { + ledger.failedStepIds.clear(); + return write(ledger.terminate(PlanRunStatus.DEADLINE_EXCEEDED, PlanReasonCodes.DEADLINE_EXCEEDED, null)); + } + if (decision == null || decision.action() == PlanFailurePolicy.Action.RETRY) { + return write(ledger.allCompleted() + ? ledger.terminate(PlanRunStatus.COMPLETED, PlanReasonCodes.PLAN_COMPLETED, null) : ledger); + } + return write(switch (decision.action()) { + case STOP -> { + ledger.failedStepIds.clear(); + yield ledger.terminate(PlanRunStatus.FAILED, decision.reasonCode(), + decision.step().stepId() + ": " + decision.detail()); + } + case REPLAN -> ledger.mark(PlanRunStatus.RUNNING, decision.reasonCode(), + decision.step().stepId() + ": " + decision.detail()); + case NEED_INPUT -> ledger.await(decision.step(), decision.detail()); + case RETRY -> ledger; + }); + } + + private Decision decide(PlanStep step, String outcome, String error, int attempt, String reasonCode) { + PlanFailurePolicy.Action action = this.failurePolicy + .decide(new PlanFailurePolicy.Failure(step, outcome, error, attempt, this.limits.maxStepAttempts())); + if (action == null) { + action = PlanFailurePolicy.Action.STOP; + } + if (action == PlanFailurePolicy.Action.RETRY && attempt >= this.limits.maxStepAttempts()) { + action = PlanFailurePolicy.Action.REPLAN; + } + return new Decision(action, step, outcome + (error == null ? "" : " - " + error), reasonCode); + } + + /** + * A batch resolves to its most severe continuation: stop, then replan (input + * requests of other steps are asked again in the next revision), then input, then + * retry. Completed steps of the batch are always kept. + */ + private static int priority(PlanFailurePolicy.Action action) { + return switch (action) { + case STOP -> 3; + case REPLAN -> 2; + case NEED_INPUT -> 1; + case RETRY -> 0; + }; + } + + private PlanStepOutcome invoke(PlanStepRequest request) { + try { + PlanStepOutcome outcome = this.stepExecutor.execute(request); + return outcome != null ? outcome + : PlanStepOutcome.failed(ToolOutcome.ERROR, "Step executor returned no outcome"); + } + catch (Exception ex) { + return PlanStepOutcome.failed(ToolOutcome.ERROR, describe(ex)); + } + } + + private static String outputProblem(PlanStep step, String result) { + if (result == null) { + return "Step returned no result"; + } + JsonNode node; + try { + node = PlanCodec.MAPPER.readTree(result); + } + catch (RuntimeException ex) { + return "Step result is not valid JSON"; + } + Schema schema = PlanCodec.SCHEMAS.getSchema(PlanCodec.MAPPER.readTree(step.outputSchema())); + return node == null || !schema.validate(node).isEmpty() ? "Step result does not match its output schema" + : null; + } + + // ---------------------------------------------------------------- replanning + + private Map replan(OverAllState state) { + Ledger ledger = new Ledger(read(state)); + if (ledger.status != PlanRunStatus.RUNNING || ledger.failedStepIds.isEmpty()) { + return Map.of(); + } + if (deadlinePassed(ledger)) { + return write(ledger.terminate(PlanRunStatus.DEADLINE_EXCEEDED, PlanReasonCodes.DEADLINE_EXCEEDED, null)); + } + if (ledger.replans >= this.limits.maxReplans()) { + return write(ledger.terminate(PlanRunStatus.REPLAN_LIMIT_EXCEEDED, PlanReasonCodes.REPLAN_LIMIT_EXCEEDED, + ledger.detail)); + } + if (this.replanner == null) { + return write(ledger.terminate(PlanRunStatus.FAILED, PlanReasonCodes.REPLAN_UNAVAILABLE, ledger.detail)); + } + ledger.replans++; + PlanSpec current = ledger.toRun().plan(); + PlanSpec proposed; + try { + proposed = this.replanner.replan(new PlanReplanner.Request(current, ledger.failedStepIds, + ledger.toRun().results(), current.revision() + 1, ledger.replans)); + } + catch (Exception ex) { + return write(ledger.terminate(PlanRunStatus.FAILED, PlanReasonCodes.REPLAN_FAILED, describe(ex))); + } + if (proposed == null) { + return write(ledger.terminate(PlanRunStatus.REPLAN_REJECTED, PlanReasonCodes.REPLAN_EMPTY, null)); + } + Rejection rejection = check(ledger, current, proposed); + if (rejection != null) { + return write(ledger.terminate(rejection.status(), rejection.reasonCode(), rejection.detail())); + } + return write(ledger.adopt(proposed)); + } + + private Rejection check(Ledger ledger, PlanSpec current, PlanSpec proposed) { + try { + this.validator.validateGenerated(proposed); + } + catch (PlanValidationException ex) { + return new Rejection(PlanRunStatus.INVALID_PLAN, PlanReasonCodes.REPLAN_INVALID, ex.getMessage()); + } + if (proposed.revision() != current.revision() + 1 || !proposed.objective().equals(current.objective())) { + return new Rejection(PlanRunStatus.INVALID_PLAN, PlanReasonCodes.REPLAN_INVALID, + "Proposal must keep the objective and use revision " + (current.revision() + 1)); + } + Map proposedSteps = new LinkedHashMap<>(); + proposed.steps().forEach(step -> proposedSteps.put(step.stepId(), step)); + for (PlanStep step : ledger.steps.values()) { + if (step.status() == PlanStep.Status.COMPLETED) { + PlanStep retained = proposedSteps.get(step.stepId()); + if (retained == null) { + return new Rejection(PlanRunStatus.INVALID_PLAN, PlanReasonCodes.COMPLETED_STEP_DROPPED, + step.stepId()); + } + if (!identity(retained).equals(identity(step))) { + return new Rejection(PlanRunStatus.INVALID_PLAN, PlanReasonCodes.STEP_IDENTITY_REUSED, + step.stepId()); + } + } + } + boolean introducesWork = false; + boolean pendingWork = false; + Set failedDefinitions = new HashSet<>(ledger.failedDefinitions); + ledger.failedStepIds.forEach(id -> failedDefinitions.add(definition(ledger.steps.get(id)))); + for (PlanStep step : proposed.steps()) { + PlanStep existing = ledger.steps.get(step.stepId()); + if (existing != null && existing.status() == PlanStep.Status.COMPLETED) { + continue; + } + pendingWork = true; + if (ledger.retiredStepIds.contains(step.stepId()) || ledger.failedStepIds.contains(step.stepId()) + || (existing != null && !identity(existing).equals(identity(step)))) { + return new Rejection(PlanRunStatus.INVALID_PLAN, PlanReasonCodes.STEP_IDENTITY_REUSED, step.stepId()); + } + if (failedDefinitions.contains(definition(step))) { + return new Rejection(PlanRunStatus.REPLAN_REJECTED, PlanReasonCodes.REPLAN_NO_PROGRESS, + step.stepId()); + } + introducesWork |= existing == null; + } + if (!pendingWork) { + return new Rejection(PlanRunStatus.REPLAN_REJECTED, PlanReasonCodes.REPLAN_EMPTY, null); + } + if (!introducesWork) { + return new Rejection(PlanRunStatus.REPLAN_REJECTED, PlanReasonCodes.REPLAN_NO_PROGRESS, + "Proposal only drops failed work"); + } + return null; + } + + // ---------------------------------------------------------------- helpers + + private boolean deadlinePassed(Ledger ledger) { + Long remaining = remainingMillis(ledger); + return remaining != null && remaining <= 0; + } + + private Long remainingMillis(Ledger ledger) { + if (this.limits.deadline() == null) { + return null; + } + return ledger.startedAtMillis + this.limits.deadline().toMillis() - this.clock.millis(); + } + + private static Map write(Ledger ledger) { + return Map.of(RUN_KEY, encode(ledger.toRun())); + } + + private static String encode(PlanRun run) { + return PlanCodec.MAPPER.writeValueAsString(run); + } + + private static String describe(Throwable error) { + if (error == null) { + return "Unknown failure"; + } + return error.getMessage() == null ? error.getClass().getName() + : error.getClass().getSimpleName() + ": " + error.getMessage(); + } + + /** + * Stable step identity: the ID plus every field that determines what the step does. + */ + static String stepKey(PlanStep step) { + return step.stepId() + "#" + identity(step).substring(0, 16); + } + + private static String identity(PlanStep step) { + return digest(Arrays.asList(step.stepId(), step.objective(), step.dependencies(), step.executionTarget(), + step.inputSchema(), step.outputSchema(), step.completionCriteria())); + } + + /** What a step does, independent of its ID and position in the graph. */ + private static String definition(PlanStep step) { + return digest(Arrays.asList(step.objective(), step.executionTarget(), step.inputSchema(), step.outputSchema(), + step.completionCriteria())); + } + + private static String digest(List fields) { + try { + byte[] bytes = PlanCodec.MAPPER.writeValueAsString(fields).getBytes(StandardCharsets.UTF_8); + return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(bytes)); + } + catch (NoSuchAlgorithmException ex) { + throw new IllegalStateException("SHA-256 is not available", ex); + } + } + + private record Decision(PlanFailurePolicy.Action action, PlanStep step, String detail, String reasonCode) { + + } + + private record Rejection(PlanRunStatus status, String reasonCode, String detail) { + + } + + private record ProvidedInput(String stepId, String input) { + + } + + /** Interrupts before applying until input for the awaited step is present. */ + private final class InputNode implements AsyncNodeActionWithConfig, InterruptableAction { + + @Override + public Optional interrupt(String nodeId, OverAllState state, RunnableConfig config) { + PlanRun run = read(state); + if (run.status() != PlanRunStatus.NEED_INPUT) { + return Optional.empty(); + } + ProvidedInput input = provided(state); + if (input != null && run.awaitingStepId().equals(input.stepId())) { + return Optional.empty(); + } + return Optional.of(InterruptionMetadata.builder(nodeId, state) + .addMetadata(INTERRUPT_STEP_ID, run.awaitingStepId()) + .addMetadata(INTERRUPT_PROMPT, run.awaitingPrompt() == null ? "" : run.awaitingPrompt()) + .addMetadata(INTERRUPT_REASON, run.reasonCode()) + .build()); + } + + @Override + public CompletableFuture> apply(OverAllState state, RunnableConfig config) { + Ledger ledger = new Ledger(read(state)); + ProvidedInput input = provided(state); + if (ledger.status == PlanRunStatus.NEED_INPUT && input != null + && input.stepId().equals(ledger.awaitingStepId)) { + if (deadlinePassed(ledger)) { + ledger.terminate(PlanRunStatus.DEADLINE_EXCEEDED, PlanReasonCodes.DEADLINE_EXCEEDED, null); + } + else if (validJson(input.input())) { + ledger.inputs.put(input.stepId(), input.input()); + ledger.awaitingStepId = null; + ledger.awaitingPrompt = null; + ledger.mark(PlanRunStatus.RUNNING, PlanReasonCodes.INPUT_RECEIVED, null); + } + else { + ledger.mark(PlanRunStatus.NEED_INPUT, PlanReasonCodes.INPUT_INVALID, input.stepId()); + } + } + Map update = new LinkedHashMap<>(write(ledger)); + update.put(INPUT_KEY, OverAllState.MARK_FOR_REMOVAL); + return CompletableFuture.completedFuture(update); + } + + private ProvidedInput provided(OverAllState state) { + Object value = state.data().get(INPUT_KEY); + if (!(value instanceof String json)) { + return null; + } + try { + return PlanCodec.MAPPER.readValue(json, ProvidedInput.class); + } + catch (RuntimeException ex) { + return null; + } + } + + private boolean validJson(String json) { + if (json == null) { + return false; + } + try { + return PlanCodec.MAPPER.readTree(json) != null; + } + catch (RuntimeException ex) { + return false; + } + } + + } + + /** Mutable working copy of a {@link PlanRun}, confined to one node invocation. */ + private static final class Ledger { + + PlanRunStatus status; + + String reasonCode; + + String detail; + + final int schemaVersion; + + long revision; + + final String objective; + + final LinkedHashMap steps = new LinkedHashMap<>(); + + final List revisions; + + int replans; + + int stepExecutions; + + final long startedAtMillis; + + final Map attempts; + + final Map effectReferences; + + final Set retiredStepIds; + + final Set failedDefinitions; + + final List failedStepIds; + + String awaitingStepId; + + String awaitingPrompt; + + final Map inputs; + + Ledger(PlanSpec plan, long startedAtMillis) { + this(new PlanRun(PlanRunStatus.RUNNING, PlanReasonCodes.STARTED, null, plan, List.of(plan.revision()), 0, + 0, startedAtMillis, null, null, null, null, null, null, null, null)); + } + + Ledger(PlanRun run) { + this.status = run.status(); + this.reasonCode = run.reasonCode(); + this.detail = run.detail(); + this.schemaVersion = run.plan().schemaVersion(); + this.revision = run.plan().revision(); + this.objective = run.plan().objective(); + run.plan().steps().forEach(step -> this.steps.put(step.stepId(), step)); + this.revisions = new ArrayList<>(run.revisions()); + this.replans = run.replans(); + this.stepExecutions = run.stepExecutions(); + this.startedAtMillis = run.startedAtMillis(); + this.attempts = new LinkedHashMap<>(run.attempts()); + this.effectReferences = new LinkedHashMap<>(run.effectReferences()); + this.retiredStepIds = new LinkedHashSet<>(run.retiredStepIds()); + this.failedDefinitions = new LinkedHashSet<>(run.failedDefinitions()); + this.failedStepIds = new ArrayList<>(run.failedStepIds()); + this.awaitingStepId = run.awaitingStepId(); + this.awaitingPrompt = run.awaitingPrompt(); + this.inputs = new LinkedHashMap<>(run.inputs()); + } + + String objective() { + return this.objective; + } + + long revision() { + return this.revision; + } + + List readySteps() { + List ready = new ArrayList<>(); + for (PlanStep step : this.steps.values()) { + if (step.status() == PlanStep.Status.PENDING && step.dependencies() + .stream() + .allMatch(id -> this.steps.get(id).status() == PlanStep.Status.COMPLETED)) { + ready.add(step); + } + } + return ready; + } + + boolean allCompleted() { + return this.steps.values().stream().allMatch(step -> step.status() == PlanStep.Status.COMPLETED); + } + + Map dependencyResults(PlanStep step) { + Map results = new LinkedHashMap<>(); + step.dependencies().forEach(id -> results.put(id, this.steps.get(id).result())); + return results; + } + + void complete(PlanStep step, String result) { + this.steps.put(step.stepId(), withOutcome(step, PlanStep.Status.COMPLETED, result, null)); + } + + void fail(PlanStep step, String error) { + this.steps.put(step.stepId(), withOutcome(step, PlanStep.Status.FAILED, null, error)); + if (!this.failedStepIds.contains(step.stepId())) { + this.failedStepIds.add(step.stepId()); + } + } + + Ledger await(PlanStep step, String prompt) { + this.awaitingStepId = step.stepId(); + this.awaitingPrompt = prompt; + return mark(PlanRunStatus.NEED_INPUT, PlanReasonCodes.NEED_INPUT, step.stepId()); + } + + Ledger mark(PlanRunStatus status, String reasonCode, String detail) { + this.status = status; + this.reasonCode = reasonCode; + this.detail = detail; + return this; + } + + Ledger terminate(PlanRunStatus status, String reasonCode, String detail) { + this.awaitingStepId = null; + this.awaitingPrompt = null; + return mark(status, reasonCode, detail); + } + + /** Adopt a checked proposal: completed outcomes come from this ledger only. */ + Ledger adopt(PlanSpec proposed) { + Set kept = new HashSet<>(); + LinkedHashMap next = new LinkedHashMap<>(); + for (PlanStep step : proposed.steps()) { + PlanStep existing = this.steps.get(step.stepId()); + boolean completed = existing != null && existing.status() == PlanStep.Status.COMPLETED; + next.put(step.stepId(), completed ? existing : step); + kept.add(step.stepId()); + } + for (PlanStep step : this.steps.values()) { + if (!kept.contains(step.stepId())) { + this.retiredStepIds.add(step.stepId()); + this.attempts.remove(step.stepId()); + this.inputs.remove(step.stepId()); + } + if (step.status() == PlanStep.Status.FAILED) { + this.failedDefinitions.add(definition(step)); + } + } + this.steps.clear(); + this.steps.putAll(next); + this.revision = proposed.revision(); + this.revisions.add(proposed.revision()); + this.failedStepIds.clear(); + return mark(PlanRunStatus.RUNNING, PlanReasonCodes.REPLANNED, "revision " + proposed.revision()); + } + + PlanRun toRun() { + PlanSpec plan = new PlanSpec(this.schemaVersion, this.revision, this.objective, + new ArrayList<>(this.steps.values())); + return new PlanRun(this.status, this.reasonCode, this.detail, plan, this.revisions, this.replans, + this.stepExecutions, this.startedAtMillis, this.attempts, this.effectReferences, + new ArrayList<>(this.retiredStepIds), new ArrayList<>(this.failedDefinitions), this.failedStepIds, + this.awaitingStepId, this.awaitingPrompt, this.inputs); + } + + private static PlanStep withOutcome(PlanStep step, PlanStep.Status status, String result, String error) { + return new PlanStep(step.stepId(), step.objective(), step.dependencies(), step.executionTarget(), + step.inputSchema(), step.outputSchema(), step.completionCriteria(), status, result, error); + } + + } + + /** Builder of an opt-in plan executor. */ + public static final class Builder { + + private final PlanValidator validator; + + private final PlanStepExecutor stepExecutor; + + private PlanReplanner replanner; + + private PlanFailurePolicy failurePolicy = PlanFailurePolicy.defaults(); + + private PlanExecutionLimits limits = PlanExecutionLimits.defaults(); + + private Clock clock = Clock.systemUTC(); + + private Executor executor = ForkJoinPool.commonPool(); + + private Builder(PlanValidator validator, PlanStepExecutor stepExecutor) { + this.validator = Objects.requireNonNull(validator, "validator"); + this.stepExecutor = Objects.requireNonNull(stepExecutor, "stepExecutor"); + } + + /** Without a replanner, a {@code REPLAN} decision terminates the run as failed. */ + public Builder replanner(PlanReplanner replanner) { + this.replanner = replanner; + return this; + } + + public Builder failurePolicy(PlanFailurePolicy failurePolicy) { + this.failurePolicy = Objects.requireNonNull(failurePolicy, "failurePolicy"); + return this; + } + + public Builder limits(PlanExecutionLimits limits) { + this.limits = Objects.requireNonNull(limits, "limits"); + return this; + } + + public Builder clock(Clock clock) { + this.clock = Objects.requireNonNull(clock, "clock"); + return this; + } + + /** Executor for local parallel batches; use a dedicated one for blocking steps. */ + public Builder executor(Executor executor) { + this.executor = Objects.requireNonNull(executor, "executor"); + return this; + } + + public PlanExecutor build() { + return new PlanExecutor(this); + } + + } + +} diff --git a/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanFailurePolicy.java b/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanFailurePolicy.java new file mode 100644 index 000000000..2212312d6 --- /dev/null +++ b/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanFailurePolicy.java @@ -0,0 +1,75 @@ +/* + * Copyright 2025-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.github.agentic.ai.graph.agent.planner; + +import io.github.agentic.ai.graph.agent.interceptor.ToolOutcome; + +/** + * Decides how a failed step attempt continues. Decisions are bounded by + * {@link PlanExecutionLimits}: a {@link Action#RETRY} beyond the attempt limit becomes a + * {@link Action#REPLAN}, and replanning is limited by count, deadline and budget. + */ +@FunctionalInterface +public interface PlanFailurePolicy { + + Action decide(Failure failure); + + /** + * Default policy: retry transient failures within the attempt limit; stop when the + * effect is unknown (timeout, cancellation, uncertain) or authorization was denied, so + * replanning cannot repeat an external effect or widen authority; replan otherwise. + * @return the default policy + */ + static PlanFailurePolicy defaults() { + return failure -> switch (failure.outcome()) { + case ToolOutcome.TRANSIENT -> failure.attempt() < failure.maxAttempts() ? Action.RETRY : Action.REPLAN; + case ToolOutcome.TIMEOUT, ToolOutcome.CANCELLED, ToolOutcome.UNCERTAIN, ToolOutcome.AUTH_DENIED -> + Action.STOP; + default -> Action.REPLAN; + }; + } + + /** Continuation of a failed step. */ + enum Action { + + /** Run the same step identity again; never used for unknown effects by default. */ + RETRY, + + /** Ask the replanner for a new revision that keeps completed steps. */ + REPLAN, + + /** Suspend the run until external input for the step is supplied. */ + NEED_INPUT, + + /** Terminate the run as failed. */ + STOP + + } + + /** + * A failed attempt. + * + * @param step the failed step + * @param outcome the failure status in the {@link ToolOutcome} vocabulary + * @param error diagnostic text + * @param attempt the 1-based attempt number that failed + * @param maxAttempts the configured attempt limit per step identity + */ + record Failure(PlanStep step, String outcome, String error, int attempt, int maxAttempts) { + + } + +} diff --git a/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanReasonCodes.java b/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanReasonCodes.java new file mode 100644 index 000000000..653af3417 --- /dev/null +++ b/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanReasonCodes.java @@ -0,0 +1,67 @@ +/* + * Copyright 2025-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.github.agentic.ai.graph.agent.planner; + +/** + * Short, stable reason codes recorded in {@link PlanRun#reasonCode()}. They are + * operational labels, never model reasoning. + */ +public final class PlanReasonCodes { + + public static final String STARTED = "PLAN_STARTED"; + + public static final String PLAN_COMPLETED = "PLAN_COMPLETED"; + + public static final String STEP_FAILED = "PLAN_STEP_FAILED"; + + public static final String STEP_RESULT_MISSING = "PLAN_STEP_RESULT_MISSING"; + + public static final String STEP_OUTPUT_INVALID = "PLAN_STEP_OUTPUT_INVALID"; + + public static final String NEED_INPUT = "PLAN_NEED_INPUT"; + + public static final String INPUT_RECEIVED = "PLAN_INPUT_RECEIVED"; + + public static final String INPUT_INVALID = "PLAN_INPUT_INVALID"; + + public static final String NO_READY_STEPS = "PLAN_NO_READY_STEPS"; + + public static final String REPLANNED = "PLAN_REPLANNED"; + + public static final String REPLAN_UNAVAILABLE = "PLAN_REPLAN_UNAVAILABLE"; + + public static final String REPLAN_FAILED = "PLAN_REPLAN_FAILED"; + + public static final String REPLAN_INVALID = "PLAN_REPLAN_INVALID"; + + public static final String REPLAN_EMPTY = "PLAN_REPLAN_EMPTY"; + + public static final String REPLAN_NO_PROGRESS = "PLAN_REPLAN_NO_PROGRESS"; + + public static final String COMPLETED_STEP_DROPPED = "PLAN_COMPLETED_STEP_DROPPED"; + + public static final String STEP_IDENTITY_REUSED = "PLAN_STEP_IDENTITY_REUSED"; + + public static final String REPLAN_LIMIT_EXCEEDED = "PLAN_REPLAN_LIMIT_EXCEEDED"; + + public static final String BUDGET_EXHAUSTED = "PLAN_BUDGET_EXHAUSTED"; + + public static final String DEADLINE_EXCEEDED = "PLAN_DEADLINE_EXCEEDED"; + + private PlanReasonCodes() { + } + +} diff --git a/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanReplanner.java b/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanReplanner.java new file mode 100644 index 000000000..18fd0461b --- /dev/null +++ b/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanReplanner.java @@ -0,0 +1,69 @@ +/* + * Copyright 2025-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.github.agentic.ai.graph.agent.planner; + +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** + * Produces the next plan revision after a step failure, for example through a model call + * or application rules. The proposal is untrusted data and is rejected unless it: + *
    + *
  • passes {@link PlanValidator#validateGenerated(PlanSpec)}: every step is + * {@code PENDING} without result or error, so completed outcomes cannot be claimed;
    • + *
  • uses exactly {@link Request#nextRevision()} and the same objective;
    • + *
  • retains every completed step with an unchanged definition (the executor restores + * their outcomes from its own ledger, so they never run again);
    • + *
  • never reuses the ID of a failed or retired step, nor changes the definition behind + * a retained ID: changed parameters require a new step identity;
    • + *
  • contains new work, and does not repeat a definition that already failed.
    • + *
+ */ +@FunctionalInterface +public interface PlanReplanner { + + /** + * Propose the next revision. Returning {@code null} is an empty decision that + * terminates the run. + * @param request the current progress and failure + * @return the proposed plan + * @throws Exception when no proposal could be produced + */ + PlanSpec replan(Request request) throws Exception; + + /** + * Replanning input. + * + * @param currentPlan the current revision including step statuses, results and errors + * @param failedStepIds the steps whose failure requested replanning + * @param completedResults JSON results of completed steps, keyed by step ID + * @param nextRevision the revision the proposal must carry + * @param replanAttempt the 1-based replanning attempt + */ + record Request(PlanSpec currentPlan, List failedStepIds, Map completedResults, + long nextRevision, int replanAttempt) { + + public Request { + failedStepIds = failedStepIds == null ? List.of() : List.copyOf(failedStepIds); + completedResults = completedResults == null ? Map.of() + : Collections.unmodifiableMap(new LinkedHashMap<>(completedResults)); + } + + } + +} diff --git a/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanRun.java b/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanRun.java new file mode 100644 index 000000000..6bbfed0f7 --- /dev/null +++ b/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanRun.java @@ -0,0 +1,100 @@ +/* + * Copyright 2025-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.github.agentic.ai.graph.agent.planner; + +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import io.github.agentic.ai.graph.agent.action.ActionDecision; +import io.github.agentic.ai.graph.agent.action.ActionType; + +/** + * Authoritative, JSON-serializable progress ledger of a plan run. It is stored as a JSON + * string under {@link PlanExecutor#RUN_KEY}, so it survives every checkpoint serializer + * and never changes existing state keys. Step outcomes in {@link #plan()} are written + * only by the executor, never copied from planner output. + * + * @param status the run status + * @param reasonCode the latest {@link PlanReasonCodes reason code} + * @param detail optional short diagnostic text + * @param plan the current revision with step statuses, results and errors + * @param revisions every adopted revision in order + * @param replans replanning attempts made, valid or not + * @param stepExecutions step dispatches consumed from the budget + * @param startedAtMillis epoch milliseconds at which the run started + * @param attempts attempts per step ID of the current revision + * @param effectReferences effect references reported per step ID, including failed ones + * @param retiredStepIds step IDs that failed or were dropped and can never be reused + * @param failedDefinitions definition fingerprints that already failed + * @param failedStepIds steps awaiting replanning + * @param awaitingStepId the step waiting for input when {@link PlanRunStatus#NEED_INPUT} + * @param awaitingPrompt the input request of the awaited step + * @param inputs supplied JSON inputs per step ID + */ +public record PlanRun(PlanRunStatus status, String reasonCode, String detail, PlanSpec plan, List revisions, + int replans, int stepExecutions, long startedAtMillis, Map attempts, + Map effectReferences, List retiredStepIds, List failedDefinitions, + List failedStepIds, String awaitingStepId, String awaitingPrompt, Map inputs) { + + public PlanRun { + revisions = revisions == null ? List.of() : List.copyOf(revisions); + attempts = copy(attempts); + effectReferences = copy(effectReferences); + retiredStepIds = retiredStepIds == null ? List.of() : List.copyOf(retiredStepIds); + failedDefinitions = failedDefinitions == null ? List.of() : List.copyOf(failedDefinitions); + failedStepIds = failedStepIds == null ? List.of() : List.copyOf(failedStepIds); + inputs = copy(inputs); + } + + /** @return {@code true} when the run can no longer make progress */ + public boolean terminal() { + return this.status.isTerminal(); + } + + /** @return JSON results of completed steps in plan order, keyed by step ID */ + public Map results() { + Map results = new LinkedHashMap<>(); + for (PlanStep step : this.plan.steps()) { + if (step.status() == PlanStep.Status.COMPLETED) { + results.put(step.stepId(), step.result()); + } + } + return Collections.unmodifiableMap(results); + } + + /** + * Express the run state with the shared action vocabulary: completion is a final + * response, a pending replan or input request maps to {@code REPLAN}/{@code NEED_INPUT}, + * every other terminal state is a bounded {@code STOP}. + * @return the decision, or {@code null} while steps are still running + */ + public ActionDecision decision() { + return switch (this.status) { + case COMPLETED -> ActionDecision.of(ActionType.FINAL_RESPONSE, this.reasonCode, this.detail); + case NEED_INPUT -> ActionDecision.of(ActionType.NEED_INPUT, this.reasonCode, this.detail); + case RUNNING -> this.failedStepIds.isEmpty() ? null + : ActionDecision.of(ActionType.REPLAN, this.reasonCode, this.detail); + default -> ActionDecision.of(ActionType.STOP, this.reasonCode, this.detail); + }; + } + + private static Map copy(Map map) { + return map == null ? Map.of() : Collections.unmodifiableMap(new LinkedHashMap<>(map)); + } + +} diff --git a/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanRunStatus.java b/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanRunStatus.java new file mode 100644 index 000000000..bc18c5c83 --- /dev/null +++ b/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanRunStatus.java @@ -0,0 +1,55 @@ +/* + * Copyright 2025-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.github.agentic.ai.graph.agent.planner; + +/** + * Lifecycle of a plan run. Only {@link #COMPLETED} is success, and it requires every step + * of the current revision to be completed with a schema-valid result. + */ +public enum PlanRunStatus { + + /** Ready steps remain or a replan is pending. */ + RUNNING, + + /** Suspended until input for the awaited step is supplied; resumable. */ + NEED_INPUT, + + /** Every step of the current revision completed. */ + COMPLETED, + + /** A step failure stopped the run, or no proposal could be obtained. */ + FAILED, + + /** A proposed revision violated the plan or replanning contract. */ + INVALID_PLAN, + + /** A proposal was empty or repeated work that already failed. */ + REPLAN_REJECTED, + + /** The replanning limit was reached. */ + REPLAN_LIMIT_EXCEEDED, + + /** The step execution budget was exhausted. */ + BUDGET_EXHAUSTED, + + /** The run deadline passed. */ + DEADLINE_EXCEEDED; + + public boolean isTerminal() { + return this != RUNNING && this != NEED_INPUT; + } + +} diff --git a/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanStepExecutor.java b/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanStepExecutor.java new file mode 100644 index 000000000..65a7950d4 --- /dev/null +++ b/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanStepExecutor.java @@ -0,0 +1,44 @@ +/* + * Copyright 2025-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.github.agentic.ai.graph.agent.planner; + +/** + * Application-owned bridge from a validated plan step to an already authorized + * capability. The executor resolves {@link PlanStep#executionTarget()} itself (for + * example through an existing tool, agent or sub-agent call) and keeps the existing + * authorization, validation and interception gates on that path. Planner output is never + * authorization. + * + *

+ * Implementations should use {@link PlanStepRequest#stepKey()} as the idempotency key for + * external effects and must report {@link io.github.agentic.ai.graph.agent.interceptor.ToolOutcome#UNCERTAIN} + * or {@link io.github.agentic.ai.graph.agent.interceptor.ToolOutcome#TIMEOUT} when the + * effect of a failed attempt is unknown, so that the default policy stops instead of + * repeating the effect. Thrown exceptions are recorded as generic failures. + *

+ */ +@FunctionalInterface +public interface PlanStepExecutor { + + /** + * Execute one ready step. A {@code null} outcome is a failure, never a success. + * @param request the step and its trusted execution inputs + * @return the step outcome + * @throws Exception when the capability fails without a structured outcome + */ + PlanStepOutcome execute(PlanStepRequest request) throws Exception; + +} diff --git a/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanStepOutcome.java b/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanStepOutcome.java new file mode 100644 index 000000000..8705e9137 --- /dev/null +++ b/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanStepOutcome.java @@ -0,0 +1,101 @@ +/* + * Copyright 2025-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.github.agentic.ai.graph.agent.planner; + +import java.util.Objects; + +import io.github.agentic.ai.graph.agent.interceptor.ToolCallResponse; +import io.github.agentic.ai.graph.agent.interceptor.ToolOutcome; + +/** + * Structured result of one step attempt. A step is completed only when the outcome is + * {@link Kind#SUCCEEDED} and its result validates against the step output schema; a + * missing result is never treated as success. + * + * @param kind whether the attempt succeeded, failed or needs external input + * @param result JSON-encoded output for a successful attempt + * @param outcome failure status using the {@link ToolOutcome} vocabulary + * @param error diagnostic text for a failure, or the prompt for {@link Kind#NEED_INPUT} + * @param effectReference optional reference (receipt or journal ID) of an external effect + */ +public record PlanStepOutcome(Kind kind, String result, String outcome, String error, String effectReference) { + + public PlanStepOutcome { + Objects.requireNonNull(kind, "kind"); + if (kind == Kind.SUCCEEDED) { + outcome = ToolOutcome.SUCCESS; + } + else if (kind == Kind.FAILED && (outcome == null || outcome.isBlank() || ToolOutcome.SUCCESS.equals(outcome))) { + outcome = ToolOutcome.ERROR; + } + } + + public static PlanStepOutcome succeeded(String result) { + return new PlanStepOutcome(Kind.SUCCEEDED, result, null, null, null); + } + + public static PlanStepOutcome succeeded(String result, String effectReference) { + return new PlanStepOutcome(Kind.SUCCEEDED, result, null, null, effectReference); + } + + public static PlanStepOutcome failed(String outcome, String error) { + return new PlanStepOutcome(Kind.FAILED, null, outcome, error, null); + } + + public static PlanStepOutcome failed(String outcome, String error, String effectReference) { + return new PlanStepOutcome(Kind.FAILED, null, outcome, error, effectReference); + } + + public static PlanStepOutcome needInput(String prompt) { + return new PlanStepOutcome(Kind.NEED_INPUT, null, null, prompt, null); + } + + /** + * Map a tool response produced by the existing tool path, preserving its outcome + * status. The generic {@code error} status is refined by its failure kind so that + * timeouts and cancellations keep their unknown-effect meaning. + * @param response the tool response + * @return the equivalent step outcome + */ + public static PlanStepOutcome fromToolCallResponse(ToolCallResponse response) { + Objects.requireNonNull(response, "response"); + String status = response.getStatus(); + if (ToolOutcome.SUCCESS.equals(status)) { + return succeeded(response.getResult()); + } + if (status == null || ToolOutcome.ERROR.equals(status)) { + String kind = response.getFailureKind().orElse(""); + if (ToolCallResponse.FAILURE_KIND_TIMEOUT.equals(kind)) { + status = ToolOutcome.TIMEOUT; + } + else if (ToolCallResponse.FAILURE_KIND_CANCELLED.equals(kind)) { + status = ToolOutcome.CANCELLED; + } + else { + status = ToolOutcome.ERROR; + } + } + return failed(status, response.getResult()); + } + + /** Attempt result kind. */ + public enum Kind { + + SUCCEEDED, FAILED, NEED_INPUT + + } + +} diff --git a/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanStepRequest.java b/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanStepRequest.java new file mode 100644 index 000000000..a52fd32f5 --- /dev/null +++ b/argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/planner/PlanStepRequest.java @@ -0,0 +1,44 @@ +/* + * Copyright 2025-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.github.agentic.ai.graph.agent.planner; + +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; + +/** + * Input handed to a {@link PlanStepExecutor} for one attempt of a ready step. + * + * @param objective the plan objective + * @param revision the plan revision that scheduled this attempt + * @param step the step definition; its status is always {@code PENDING} + * @param stepKey a stable identity derived from the step ID and definition. It does not + * change across retries of the same step and always changes when the definition changes, + * so it is suitable as an idempotency key for external effects + * @param attempt the 1-based attempt number of this step identity + * @param dependencyResults JSON results of the completed dependencies, keyed by step ID + * @param input JSON input supplied after a {@code NEED_INPUT} interruption, or + * {@code null} + */ +public record PlanStepRequest(String objective, long revision, PlanStep step, String stepKey, int attempt, + Map dependencyResults, String input) { + + public PlanStepRequest { + dependencyResults = dependencyResults == null ? Map.of() + : Collections.unmodifiableMap(new LinkedHashMap<>(dependencyResults)); + } + +} diff --git a/argi-agent-framework/src/test/java/io/github/agentic/ai/graph/agent/planner/PlanExecutorTest.java b/argi-agent-framework/src/test/java/io/github/agentic/ai/graph/agent/planner/PlanExecutorTest.java new file mode 100644 index 000000000..3d4c56f1b --- /dev/null +++ b/argi-agent-framework/src/test/java/io/github/agentic/ai/graph/agent/planner/PlanExecutorTest.java @@ -0,0 +1,460 @@ +/* + * Copyright 2025-2026 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.github.agentic.ai.graph.agent.planner; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneId; +import java.time.ZoneOffset; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; + +import io.github.agentic.ai.graph.CompileConfig; +import io.github.agentic.ai.graph.CompiledGraph; +import io.github.agentic.ai.graph.OverAllState; +import io.github.agentic.ai.graph.RunnableConfig; +import io.github.agentic.ai.graph.agent.action.ActionType; +import io.github.agentic.ai.graph.agent.interceptor.ToolCallResponse; +import io.github.agentic.ai.graph.agent.interceptor.ToolOutcome; +import io.github.agentic.ai.graph.checkpoint.config.SaverConfig; +import io.github.agentic.ai.graph.checkpoint.savers.MemorySaver; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import static io.github.agentic.ai.graph.agent.planner.PlannerFixtures.INPUT; +import static io.github.agentic.ai.graph.agent.planner.PlannerFixtures.OBJECTIVE; +import static io.github.agentic.ai.graph.agent.planner.PlannerFixtures.OUTPUT; +import static io.github.agentic.ai.graph.agent.planner.PlannerFixtures.SEARCH; +import static io.github.agentic.ai.graph.agent.planner.PlannerFixtures.plan; +import static io.github.agentic.ai.graph.agent.planner.PlannerFixtures.step; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class PlanExecutorTest { + + private final ExecutorService pool = Executors.newFixedThreadPool(4); + + private final List calls = Collections.synchronizedList(new ArrayList<>()); + + @AfterEach + void shutdown() { + this.pool.shutdownNow(); + } + + @Test + void serialDependenciesRunInOrderAndPassUpstreamResults() throws Exception { + Map requests = new ConcurrentHashMap<>(); + PlanRun run = run(executor(request -> { + requests.put(request.step().stepId(), request); + return ok(request); + }).build(), plan(step("a"), step("b", "a"), step("c", "b"))); + + assertEquals(PlanRunStatus.COMPLETED, run.status()); + assertEquals(List.of("a", "b", "c"), this.calls); + assertEquals(Map.of("a", "\"a-done\""), requests.get("b").dependencyResults()); + assertEquals(Map.of("a", "\"a-done\"", "b", "\"b-done\"", "c", "\"c-done\""), run.results()); + assertEquals(ActionType.FINAL_RESPONSE, run.decision().type()); + assertEquals(3, run.stepExecutions()); + } + + @Test + void independentStepsRunInParallelAndTheJoinWaitsForBoth() throws Exception { + CountDownLatch bothStarted = new CountDownLatch(2); + PlanRun run = run(executor(request -> { + if (!request.step().stepId().equals("join")) { + bothStarted.countDown(); + assertTrue(bothStarted.await(5, TimeUnit.SECONDS), "branches must overlap"); + } + else { + assertEquals(2, request.dependencyResults().size()); + } + return ok(request); + }).build(), plan(step("left"), step("right"), step("join", "left", "right"))); + + assertEquals(PlanRunStatus.COMPLETED, run.status()); + assertEquals("join", this.calls.get(2)); + } + + @Test + void downstreamStepNeverStartsWhenItsDependencyFails() throws Exception { + PlanRun run = run(executor(request -> PlanStepOutcome.failed(ToolOutcome.UNCERTAIN, "lost ack", "receipt-a")) + .build(), plan(step("a"), step("b", "a"))); + + assertEquals(PlanRunStatus.FAILED, run.status()); + assertEquals(PlanReasonCodes.STEP_FAILED, run.reasonCode()); + assertEquals(List.of("a"), this.calls); + assertEquals(PlanStep.Status.FAILED, run.plan().steps().get(0).status()); + assertEquals(PlanStep.Status.PENDING, run.plan().steps().get(1).status()); + assertEquals("receipt-a", run.effectReferences().get("a")); + assertEquals(ActionType.STOP, run.decision().type()); + } + + @Test + void transientFailureRetriesTheSameStepIdentity() throws Exception { + List keys = Collections.synchronizedList(new ArrayList<>()); + PlanRun run = run(executor(request -> { + keys.add(request.stepKey()); + return request.attempt() == 1 ? PlanStepOutcome.failed(ToolOutcome.TRANSIENT, "busy") : ok(request); + }).build(), plan(step("a"))); + + assertEquals(PlanRunStatus.COMPLETED, run.status()); + assertEquals(2, keys.size()); + assertEquals(keys.get(0), keys.get(1)); + assertEquals(2, run.attempts().get("a")); + } + + @Test + void failureReplansOnceAndNeverRepeatsCompletedEffects() throws Exception { + List replans = new ArrayList<>(); + PlanExecutor executor = executor(request -> request.step().stepId().equals("b") + ? PlanStepOutcome.failed(ToolOutcome.BUSINESS_FAILED, "quota") : ok(request, "receipt-")) + .replanner(request -> { + replans.add(request); + return new PlanSpec(1, request.nextRevision(), OBJECTIVE, + List.of(step("a"), custom("b2", "Perform b with a smaller quota", "a"))); + }) + .build(); + + PlanRun run = run(executor, plan(step("a"), step("b", "a"))); + + assertEquals(PlanRunStatus.COMPLETED, run.status()); + assertEquals(List.of("a", "b", "b2"), this.calls); + assertEquals(List.of(1L, 2L), run.revisions()); + assertEquals(2L, run.plan().revision()); + assertEquals(1, run.replans()); + assertEquals(List.of("b"), run.retiredStepIds()); + assertEquals("receipt-a", run.effectReferences().get("a")); + assertEquals(List.of("b"), replans.get(0).failedStepIds()); + assertEquals(Map.of("a", "\"a-done\""), replans.get(0).completedResults()); + assertEquals(PlanStep.Status.FAILED, replans.get(0).currentPlan().steps().get(1).status()); + assertEquals(Map.of("a", "\"a-done\"", "b2", "\"b2-done\""), run.results()); + } + + @Test + void illegalProposalsTerminateWithInvalidPlan() throws Exception { + assertRejected(request -> new PlanSpec(1, request.nextRevision(), OBJECTIVE, + List.of(step("a"), custom("b", "Perform b differently", "a"))), PlanRunStatus.INVALID_PLAN, + PlanReasonCodes.STEP_IDENTITY_REUSED); + assertRejected(request -> new PlanSpec(1, request.nextRevision(), OBJECTIVE, + List.of(custom("a", "Perform a again"), step("c", "a"))), PlanRunStatus.INVALID_PLAN, + PlanReasonCodes.STEP_IDENTITY_REUSED); + assertRejected(request -> new PlanSpec(1, request.nextRevision(), OBJECTIVE, List.of(step("c"))), + PlanRunStatus.INVALID_PLAN, PlanReasonCodes.COMPLETED_STEP_DROPPED); + assertRejected(request -> new PlanSpec(1, request.nextRevision(), OBJECTIVE, + List.of(claimed("a"), step("c", "a"))), PlanRunStatus.INVALID_PLAN, PlanReasonCodes.REPLAN_INVALID); + assertRejected(request -> new PlanSpec(1, request.nextRevision() + 1, OBJECTIVE, + List.of(step("a"), step("c", "a"))), PlanRunStatus.INVALID_PLAN, PlanReasonCodes.REPLAN_INVALID); + assertRejected(request -> new PlanSpec(1, request.nextRevision(), OBJECTIVE, + List.of(step("a"), step("c", "missing"))), PlanRunStatus.INVALID_PLAN, PlanReasonCodes.REPLAN_INVALID); + } + + @Test + void emptyOrRepeatedDecisionsAreRejected() throws Exception { + assertRejected(request -> null, PlanRunStatus.REPLAN_REJECTED, PlanReasonCodes.REPLAN_EMPTY); + assertRejected(request -> new PlanSpec(1, request.nextRevision(), OBJECTIVE, List.of(step("a"))), + PlanRunStatus.REPLAN_REJECTED, PlanReasonCodes.REPLAN_EMPTY); + assertRejected(request -> new PlanSpec(1, request.nextRevision(), OBJECTIVE, + List.of(step("a"), renamed("b-again", "a"))), PlanRunStatus.REPLAN_REJECTED, + PlanReasonCodes.REPLAN_NO_PROGRESS); + assertRejected(request -> { + throw new IllegalStateException("model unavailable"); + }, PlanRunStatus.FAILED, PlanReasonCodes.REPLAN_FAILED); + } + + @Test + void retainingUntouchedWorkWithoutNewStepsIsNoProgress() throws Exception { + PlanExecutor executor = executor(request -> request.step().stepId().equals("b") + ? PlanStepOutcome.failed(ToolOutcome.BUSINESS_FAILED, "quota") : ok(request)) + .limits(new PlanExecutionLimits(1, 2, 2, 64, null)) + .replanner(request -> new PlanSpec(1, request.nextRevision(), OBJECTIVE, List.of(step("a"), step("c")))) + .build(); + + PlanRun run = run(executor, plan(step("a"), step("b", "a"), step("c"))); + + assertEquals(PlanRunStatus.REPLAN_REJECTED, run.status()); + assertEquals(PlanReasonCodes.REPLAN_NO_PROGRESS, run.reasonCode()); + } + + @Test + void replanningIsBoundedByCount() throws Exception { + AtomicInteger proposals = new AtomicInteger(); + PlanExecutor executor = executor(request -> PlanStepOutcome.failed(ToolOutcome.BUSINESS_FAILED, "no")) + .limits(new PlanExecutionLimits(4, 2, 1, 64, null)) + .replanner(request -> new PlanSpec(1, request.nextRevision(), OBJECTIVE, + List.of(custom("try-" + proposals.incrementAndGet(), "Variant " + proposals.get())))) + .build(); + + PlanRun run = run(executor, plan(step("a"))); + + assertEquals(PlanRunStatus.REPLAN_LIMIT_EXCEEDED, run.status()); + assertEquals(1, proposals.get()); + assertEquals(1, run.replans()); + } + + @Test + void missingInvalidOrAbsentResultsAreNeverSuccess() throws Exception { + for (Function step : List.>of( + request -> PlanStepOutcome.succeeded(null), request -> PlanStepOutcome.succeeded("42"), + request -> PlanStepOutcome.succeeded("not json"), request -> null, request -> { + throw new IllegalStateException("boom"); + })) { + PlanRun run = run(executor(step::apply).build(), plan(step("a"))); + assertEquals(PlanRunStatus.FAILED, run.status()); + assertEquals(PlanReasonCodes.REPLAN_UNAVAILABLE, run.reasonCode()); + assertEquals(PlanStep.Status.FAILED, run.plan().steps().get(0).status()); + assertTrue(run.results().isEmpty()); + } + } + + @Test + void budgetExhaustionStopsBeforeTheNextDispatch() throws Exception { + PlanRun run = run(executor(this::ok).limits(new PlanExecutionLimits(4, 2, 2, 2, null)).build(), + plan(step("a"), step("b", "a"), step("c", "b"))); + + assertEquals(PlanRunStatus.BUDGET_EXHAUSTED, run.status()); + assertEquals(List.of("a", "b"), this.calls); + assertEquals(2, run.results().size()); + } + + @Test + void deadlineStopsTheRun() throws Exception { + MutableClock clock = new MutableClock(); + PlanRun run = run(executor(request -> { + clock.advance(Duration.ofSeconds(11)); + return ok(request); + }).clock(clock).limits(PlanExecutionLimits.defaults().withDeadline(Duration.ofSeconds(10))).build(), + plan(step("a"), step("b", "a"))); + + assertEquals(PlanRunStatus.DEADLINE_EXCEEDED, run.status()); + assertEquals(List.of("a"), this.calls); + assertEquals(Map.of("a", "\"a-done\""), run.results()); + } + + @Test + void stepStillRunningAtTheDeadlineIsCancelled() throws Exception { + CountDownLatch release = new CountDownLatch(1); + PlanRun run = run(executor(request -> { + release.await(5, TimeUnit.SECONDS); + return ok(request); + }).limits(PlanExecutionLimits.defaults().withDeadline(Duration.ofMillis(200))).build(), plan(step("a"))); + release.countDown(); + + assertEquals(PlanRunStatus.DEADLINE_EXCEEDED, run.status()); + assertEquals(PlanStep.Status.FAILED, run.plan().steps().get(0).status()); + assertTrue(run.plan().steps().get(0).error().startsWith(ToolOutcome.TIMEOUT)); + assertTrue(run.results().isEmpty()); + } + + @Test + void missingInputInterruptsAndResumesFromTheCheckpoint() throws Exception { + PlanExecutor executor = executor(request -> { + if (request.step().stepId().equals("approve") && request.input() == null) { + return PlanStepOutcome.needInput("Approve the summary?"); + } + return PlanStepOutcome.succeeded(request.input() != null ? request.input() : done(request)); + }).build(); + CompiledGraph graph = executor + .compile(CompileConfig.builder() + .saverConfig(SaverConfig.builder().register(MemorySaver.builder().build()).build()) + .build()); + RunnableConfig config = RunnableConfig.builder().threadId("plan-input").build(); + + graph.invoke(executor.initialState(plan(step("a"), step("approve", "a"), step("publish", "approve"))), config); + PlanRun waiting = PlanExecutor.read(graph.getState(config).state()); + assertEquals(PlanRunStatus.NEED_INPUT, waiting.status()); + assertEquals("approve", waiting.awaitingStepId()); + assertEquals("Approve the summary?", waiting.awaitingPrompt()); + assertEquals(ActionType.NEED_INPUT, waiting.decision().type()); + assertFalse(waiting.terminal()); + + // Resuming without the awaited input keeps the run suspended. + graph.invoke((Map) null, RunnableConfig.builder(config).resume().build()); + assertEquals(PlanRunStatus.NEED_INPUT, PlanExecutor.read(graph.getState(config).state()).status()); + + Optional resumed = graph.invoke((Map) null, RunnableConfig.builder(config) + .resume() + .addStateUpdate(PlanExecutor.input("approve", "\"approved\"")) + .build()); + + PlanRun run = PlanExecutor.read(resumed.orElseThrow()); + assertEquals(PlanRunStatus.COMPLETED, run.status()); + assertEquals(List.of("a", "approve", "approve", "publish"), this.calls); + assertEquals("\"approved\"", run.results().get("approve")); + assertNull(run.awaitingStepId()); + assertFalse(resumed.get().data().containsKey(PlanExecutor.INPUT_KEY)); + } + + @Test + void newRunsCannotClaimProgress() { + PlanExecutor executor = executor(this::ok).build(); + assertThrows(PlanValidationException.class, () -> executor.initialState(plan(claimed("a")))); + assertThrows(PlanValidationException.class, () -> PlanExecutor.read(Map.of())); + } + + @Test + void defaultPolicyStopsOnUnknownEffectsAndReplansDefiniteFailures() { + PlanFailurePolicy policy = PlanFailurePolicy.defaults(); + PlanStep step = step("a"); + assertEquals(PlanFailurePolicy.Action.RETRY, + policy.decide(new PlanFailurePolicy.Failure(step, ToolOutcome.TRANSIENT, null, 1, 2))); + assertEquals(PlanFailurePolicy.Action.REPLAN, + policy.decide(new PlanFailurePolicy.Failure(step, ToolOutcome.TRANSIENT, null, 2, 2))); + for (String unknown : List.of(ToolOutcome.TIMEOUT, ToolOutcome.CANCELLED, ToolOutcome.UNCERTAIN, + ToolOutcome.AUTH_DENIED)) { + assertEquals(PlanFailurePolicy.Action.STOP, + policy.decide(new PlanFailurePolicy.Failure(step, unknown, null, 1, 2))); + } + for (String definite : List.of(ToolOutcome.ERROR, ToolOutcome.BUSINESS_FAILED, ToolOutcome.VALIDATION_DENIED)) { + assertEquals(PlanFailurePolicy.Action.REPLAN, + policy.decide(new PlanFailurePolicy.Failure(step, definite, null, 1, 2))); + } + } + + @Test + void toolResponsesKeepTheirOutcomeVocabulary() { + assertEquals(PlanStepOutcome.Kind.SUCCEEDED, + PlanStepOutcome.fromToolCallResponse(ToolCallResponse.success("1", "search", "\"x\"")).kind()); + PlanStepOutcome timeout = PlanStepOutcome.fromToolCallResponse(ToolCallResponse.builder() + .toolCallId("1") + .toolName("search") + .content("slow") + .status(ToolOutcome.ERROR) + .metadata(Map.of(ToolCallResponse.FAILURE_KIND_METADATA_KEY, ToolCallResponse.FAILURE_KIND_TIMEOUT)) + .build()); + assertEquals(ToolOutcome.TIMEOUT, timeout.outcome()); + assertEquals(ToolOutcome.BUSINESS_FAILED, PlanStepOutcome.fromToolCallResponse(ToolCallResponse.builder() + .toolCallId("1") + .toolName("search") + .content("rejected") + .status(ToolOutcome.BUSINESS_FAILED) + .build()).outcome()); + assertEquals(ToolOutcome.ERROR, PlanStepOutcome.failed(null, "x").outcome()); + } + + @Test + void runLedgerRoundTripsThroughState() throws Exception { + PlanExecutor executor = executor(this::ok).build(); + Map state = executor.initialState(plan(step("a"))); + assertTrue(state.get(PlanExecutor.RUN_KEY) instanceof String); + PlanRun run = PlanExecutor.read(state); + assertEquals(PlanRunStatus.RUNNING, run.status()); + assertNull(run.decision()); + assertNotNull(PlanExecutor.stepKey(run.plan().steps().get(0))); + assertTrue(executor.compile(null).getMaxIterations() >= executor.recursionLimit()); + } + + // ---------------------------------------------------------------- helpers + + private void assertRejected(PlanReplanner replanner, PlanRunStatus status, String reasonCode) throws Exception { + this.calls.clear(); + PlanExecutor executor = executor(request -> request.step().stepId().equals("b") + ? PlanStepOutcome.failed(ToolOutcome.BUSINESS_FAILED, "quota") : ok(request)) + .replanner(replanner) + .build(); + PlanRun run = run(executor, plan(step("a"), step("b", "a"))); + assertEquals(status, run.status(), run.detail()); + assertEquals(reasonCode, run.reasonCode()); + assertEquals(List.of("a", "b"), this.calls, "rejected proposals must not execute"); + assertEquals(ActionType.STOP, run.decision().type()); + } + + private PlanExecutor.Builder executor(PlanStepExecutor delegate) { + return PlanExecutor.builder(PlannerFixtures.validator(), request -> { + this.calls.add(request.step().stepId()); + return delegate.execute(request); + }).executor(this.pool); + } + + private PlanRun run(PlanExecutor executor, PlanSpec plan) throws Exception { + CompiledGraph graph = executor.compile(CompileConfig.builder().build()); + return PlanExecutor.read(graph.invoke(executor.initialState(plan), RunnableConfig.builder().build()) + .orElseThrow()); + } + + private PlanStepOutcome ok(PlanStepRequest request) { + return PlanStepOutcome.succeeded(done(request)); + } + + private PlanStepOutcome ok(PlanStepRequest request, String receiptPrefix) { + return PlanStepOutcome.succeeded(done(request), receiptPrefix + request.step().stepId()); + } + + private static String done(PlanStepRequest request) { + return "\"" + request.step().stepId() + "-done\""; + } + + private static PlanStep custom(String id, String objective, String... dependencies) { + return new PlanStep(id, objective, List.of(dependencies), SEARCH.executionTarget(), INPUT, OUTPUT, + "Produce a relevant document", PlanStep.Status.PENDING, null, null); + } + + /** Same definition as {@code step("b")} under a new ID. */ + private static PlanStep renamed(String id, String... dependencies) { + return custom(id, "Perform b", dependencies); + } + + private static PlanStep claimed(String id) { + return new PlanStep(id, "Perform " + id, List.of(), SEARCH.executionTarget(), INPUT, OUTPUT, + "Produce a relevant document", PlanStep.Status.COMPLETED, "\"fake\"", null); + } + + private static final class MutableClock extends Clock { + + private final AtomicLong millis = new AtomicLong(1_000_000L); + + void advance(Duration duration) { + this.millis.addAndGet(duration.toMillis()); + } + + @Override + public ZoneId getZone() { + return ZoneOffset.UTC; + } + + @Override + public Clock withZone(ZoneId zone) { + return this; + } + + @Override + public Instant instant() { + return Instant.ofEpochMilli(this.millis.get()); + } + + @Override + public long millis() { + return this.millis.get(); + } + + } + +}