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:
+ *
+ * <pre>
+ * START -> plan_execute -> (plan_execute | plan_replan | plan_input | END)
+ * </pre>
+ *
+ * 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.
+ *
+ * <p>
+ * 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.
+ * </p>
+ */
+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<String, Object> 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<String, Object> 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<String, Object> 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<String, String> 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<String, Object> 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<PlanStep> 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<PlanStep> batch = ready.subList(0, Math.min(ready.size(), Math.min(budget, this.limits.maxParallelism())));
+
+		Map<String, CompletableFuture<PlanStepOutcome>> running = new LinkedHashMap<>();
+		Map<String, Integer> 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<String, PlanStepOutcome> outcomes = new LinkedHashMap<>();
+		for (Map.Entry<String, CompletableFuture<PlanStepOutcome>> 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<String, Object> 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<String, PlanStep> 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<String> 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<String, Object> 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<Object> 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<InterruptionMetadata> 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<Map<String, Object>> 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<String, Object> 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<String, PlanStep> steps = new LinkedHashMap<>();
+
+		final List<Long> revisions;
+
+		int replans;
+
+		int stepExecutions;
+
+		final long startedAtMillis;
+
+		final Map<String, Integer> attempts;
+
+		final Map<String, String> effectReferences;
+
+		final Set<String> retiredStepIds;
+
+		final Set<String> failedDefinitions;
+
+		final List<String> failedStepIds;
+
+		String awaitingStepId;
+
+		String awaitingPrompt;
+
+		final Map<String, String> 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<PlanStep> readySteps() {
+			List<PlanStep> 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<String, String> dependencyResults(PlanStep step) {
+			Map<String, String> 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<String> kept = new HashSet<>();
+			LinkedHashMap<String, PlanStep> 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:
+ * <ul>
+ * <li>passes {@link PlanValidator#validateGenerated(PlanSpec)}: every step is
+ * {@code PENDING} without result or error, so completed outcomes cannot be claimed;</li>
+ * <li>uses exactly {@link Request#nextRevision()} and the same objective;</li>
+ * <li>retains every completed step with an unchanged definition (the executor restores
+ * their outcomes from its own ledger, so they never run again);</li>
+ * <li>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;</li>
+ * <li>contains new work, and does not repeat a definition that already failed.</li>
+ * </ul>
+ */
+@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<String> failedStepIds, Map<String, String> 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<Long> revisions,
+		int replans, int stepExecutions, long startedAtMillis, Map<String, Integer> attempts,
+		Map<String, String> effectReferences, List<String> retiredStepIds, List<String> failedDefinitions,
+		List<String> failedStepIds, String awaitingStepId, String awaitingPrompt, Map<String, String> 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<String, String> results() {
+		Map<String, String> 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 <V> Map<String, V> copy(Map<String, V> 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.
+ *
+ * <p>
+ * 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.
+ * </p>
+ */
+@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<String, String> 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<String> calls = Collections.synchronizedList(new ArrayList<>());
+
+	@AfterEach
+	void shutdown() {
+		this.pool.shutdownNow();
+	}
+
+	@Test
+	void serialDependenciesRunInOrderAndPassUpstreamResults() throws Exception {
+		Map<String, PlanStepRequest> 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<String> 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<PlanReplanner.Request> 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<PlanStepRequest, PlanStepOutcome> step : List.<Function<PlanStepRequest, PlanStepOutcome>>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<String, Object>) null, RunnableConfig.builder(config).resume().build());
+		assertEquals(PlanRunStatus.NEED_INPUT, PlanExecutor.read(graph.getState(config).state()).status());
+
+		Optional<OverAllState> resumed = graph.invoke((Map<String, Object>) 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<String, Object> 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();
+		}
+
+	}
+
+}