com.fasterxml.jackson.core
jackson-databind
diff --git a/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/MissingResponsePolicy.java b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/MissingResponsePolicy.java
new file mode 100644
index 000000000..fd67d1f85
--- /dev/null
+++ b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/MissingResponsePolicy.java
@@ -0,0 +1,27 @@
+/*
+ * 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.agent.durable;
+
+/** Controls whether a model turn with a reserved call but no recorded response may be retried. */
+public enum MissingResponsePolicy {
+
+ /** Stop recovery when the outcome of a reserved model call is unknown. */
+ FAIL_CLOSED,
+
+ /** Permit another model call while the total attempt limit has room. */
+ REASK
+
+}
diff --git a/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/ModelTurnAttemptPolicy.java b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/ModelTurnAttemptPolicy.java
new file mode 100644
index 000000000..0faeedc76
--- /dev/null
+++ b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/ModelTurnAttemptPolicy.java
@@ -0,0 +1,36 @@
+/*
+ * 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.agent.durable;
+
+import static java.util.Objects.requireNonNull;
+
+/** Immutable recovery policy and total model-call limit for one logical turn. */
+public record ModelTurnAttemptPolicy(MissingResponsePolicy missingResponsePolicy, int maxAttempts) {
+
+ public static final int DEFAULT_MAX_ATTEMPTS = 1;
+
+ /** The default permits the initial call and fails closed on recovery. */
+ public static final ModelTurnAttemptPolicy DEFAULT =
+ new ModelTurnAttemptPolicy(MissingResponsePolicy.FAIL_CLOSED, DEFAULT_MAX_ATTEMPTS);
+
+ public ModelTurnAttemptPolicy {
+ requireNonNull(missingResponsePolicy, "missingResponsePolicy cannot be null");
+ if (maxAttempts < 1) {
+ throw new IllegalArgumentException("maxAttempts must be positive");
+ }
+ }
+
+}
diff --git a/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/ModelTurnCodec.java b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/ModelTurnCodec.java
new file mode 100644
index 000000000..69d3bbeaa
--- /dev/null
+++ b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/ModelTurnCodec.java
@@ -0,0 +1,379 @@
+/*
+ * 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.agent.durable;
+
+import io.github.agentic.ai.api.chat.ChatMessage;
+import io.github.agentic.ai.api.chat.ChatRequest;
+import io.github.agentic.ai.api.chat.ChatResponse;
+import io.github.agentic.ai.api.chat.ContentPart;
+import io.github.agentic.ai.api.chat.GenerationOptions;
+import io.github.agentic.ai.api.chat.ToolCall;
+import io.github.agentic.ai.api.chat.ToolDefinition;
+import io.github.agentic.ai.runtime.execution.run.RunEntry;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.StreamReadFeature;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.MapperFeature;
+import com.fasterxml.jackson.databind.ObjectReader;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.fasterxml.jackson.databind.json.JsonMapper;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.HexFormat;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.TreeSet;
+
+/**
+ * Version-tagged JSON payloads for model-turn journal entries.
+ *
+ *
+ * The neutral chat contracts are bound by their declared record components, not by a
+ * hand-written field-by-field codec: the payload and the type stay described in one place,
+ * and a codec bug cannot desynchronise the write and read paths. What is written by hand
+ * is only where the recorded shape must deliberately differ from the type:
+ *
+ * - {@link RequestFingerprint} - a request is reduced to its fingerprint, so the body is
+ * never journaled and adapter-level {@code attributes} are excluded from the hash;
+ * - {@link ToolPlan} - the provider's tool-call id is excluded, because it correlates
+ * protocol messages and is not a task identity;
+ * - {@link AttemptRecord} - a model-call reservation pins its ordinal and request hash;
+ * - {@link Delta} - streamed text is a single scalar.
+ *
+ *
+ *
+ * No Java type information is written: this mapper never enables default typing, so a
+ * payload stays readable by any process and after a class moves. Unknown fields are
+ * ignored on read so a newer writer can add fields without invalidating old records, and
+ * property and map-key ordering is pinned so equal data always produces equal JSON - which
+ * is what makes {@link #requestHash(ChatRequest)} stable.
+ *
+ *
+ * Request bodies are never journaled. Responses and tool arguments are, because replaying
+ * them is the purpose of the journal; applications apply their data-protection policy to
+ * the journal backend.
+ */
+public final class ModelTurnCodec {
+
+ /** Version of every payload schema defined here, carried by {@code RunPayload}. */
+ public static final int SCHEMA_VERSION = 1;
+
+ /** {@code MODEL_CALL}: the request fingerprint, written before the provider call. */
+ public static final String REQUEST_TYPE = "argi.model-request";
+
+ /** {@code MODEL_CALL}: a model-call reservation, committed before invoking the provider. */
+ public static final String ATTEMPT_TYPE = "argi.model-attempt";
+
+ /** {@code MODEL_CALL}: the complete terminal response. */
+ public static final String RESPONSE_TYPE = "argi.model-response";
+
+ /** {@code TASK_CALL}: one planned tool call, committed with its response. */
+ public static final String PLAN_TYPE = "argi.tool-plan";
+
+ /** {@code EVENT}: streamed text received before the terminal response. */
+ public static final String DELTA_TYPE = "argi.model-delta";
+
+ /**
+ * Metadata that belongs to one process run: protocol markers and handles into
+ * in-memory tool-binding snapshots. A recovering invocation re-derives these, so
+ * recording them would only replay stale state - and a snapshot id that changes per
+ * process would change the request fingerprint across a restart.
+ */
+ private static final Set PROCESS_LOCAL_METADATA = Set.of("argiFinalResponse", "argiTextDelta",
+ "argi.toolBindingSnapshot", "argi.toolBindingSelection", "argi.dynamicToolBindings", "argiModelFailure");
+
+ private static final JsonMapper MAPPER = JsonMapper.builder()
+ .enable(StreamReadFeature.STRICT_DUPLICATE_DETECTION)
+ .enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS)
+ .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
+ .enable(MapperFeature.SORT_PROPERTIES_ALPHABETICALLY)
+ .enable(SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS)
+ .build();
+
+ private static final ObjectReader ARGUMENT_READER = MAPPER.readerFor(JsonNode.class)
+ .with(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS);
+
+ private ModelTurnCodec() {
+ }
+
+ /**
+ * Canonical fingerprint of the request a provider actually receives: ordered messages,
+ * disclosed tool definitions and effective generation options.
+ *
+ *
+ * {@link ChatRequest#attributes()} is deliberately excluded. It is a transport escape
+ * hatch whose meaning differs per adapter (Spring AI passes it to advisor parameters,
+ * LangChain4j to listener attributes, Google ADK only checks emptiness), and in the
+ * graph-independent engine it carries the whole mutable agent state. Hashing it would
+ * bind request identity to adapter internals and to state that changes for reasons
+ * unrelated to the request. An application whose adapters derive generation semantics
+ * from attributes must not rely on replay across differing attribute values.
+ */
+ public static String requestHash(ChatRequest request) {
+ Objects.requireNonNull(request, "request cannot be null");
+ RequestFingerprint fingerprint = new RequestFingerprint(sanitize(request.messages()), request.tools(),
+ request.options());
+ return sha256(write(fingerprint));
+ }
+
+ /** The {@link #REQUEST_TYPE} payload: the fingerprint only, never the request body. */
+ public static String encodeRequest(ChatRequest request) {
+ return write(new RequestRecord(requestHash(request)));
+ }
+
+ /** Reads the fingerprint recorded in a {@link #REQUEST_TYPE} payload. */
+ public static String decodeRequestHash(String json) {
+ return read(json, RequestRecord.class).requestHash();
+ }
+
+ static String encodeAttempt(int ordinal, String requestHash) {
+ return write(new AttemptRecord(ordinal, requestHash));
+ }
+
+ static int decodeAttemptOrdinal(String json, String requestHash) {
+ JsonNode record = read(json, JsonNode.class);
+ JsonNode ordinal = record.get("ordinal");
+ JsonNode hash = record.get("requestHash");
+ if (ordinal == null || !ordinal.isInt() || ordinal.intValue() < 1 || hash == null || !hash.isTextual()
+ || !requestHash.equals(hash.textValue())) {
+ throw new ModelTurnJournalException("Invalid model attempt ordinal or request fingerprint");
+ }
+ return ordinal.intValue();
+ }
+
+ /** The {@link #RESPONSE_TYPE} payload. */
+ public static String encodeResponse(ChatResponse response) {
+ return write(sanitize(Objects.requireNonNull(response, "response cannot be null")));
+ }
+
+ public static ChatResponse decodeResponse(String json) {
+ return read(json, ChatResponse.class);
+ }
+
+ /** The {@link #DELTA_TYPE} payload. */
+ public static String encodeDelta(String text) {
+ return write(new Delta(text));
+ }
+
+ /**
+ * The {@link #PLAN_TYPE} payload. The tool-call id assigned by the provider is
+ * deliberately absent: it correlates protocol messages, it is not a task identity.
+ *
+ *
+ * Arguments are stored and hashed in canonical form, so two semantically equal
+ * invocations that differ only in whitespace or object-key order produce the same
+ * record and the same hash. The hash checks input consistency; the logical task is
+ * identified separately by {@link ToolTaskIdentity}.
+ */
+ public static String encodeToolPlan(ToolCall call, ToolDefinition definition, int ordinal) {
+ String arguments = canonicalArguments(call.arguments());
+ return write(new ToolPlan(ordinal, call.name(), definition == null ? null : definitionVersion(definition),
+ sha256(arguments), arguments));
+ }
+
+ /**
+ * Reads an existing plan without resolving tools or validating argument schemas.
+ * Checks the envelope, stored field types, task identity and argument hash only.
+ * The caller supplies the owning turn because entries are scoped to a run by storage,
+ * not by their payload. This method neither verifies a complete model turn nor grants
+ * permission to dispatch; use the recorder to validate the full committed snapshot.
+ * @param turn the owning logical model call
+ * @param entry the stored TASK_CALL entry
+ * @return the typed view of the stored plan, including a possibly absent definition
+ * @throws ModelTurnJournalException if the journal entry is incompatible or inconsistent
+ */
+ public static PlannedToolCall decodeToolPlan(ModelTurnRecorder.Turn turn, RunEntry entry) {
+ Objects.requireNonNull(turn, "turn cannot be null");
+ Objects.requireNonNull(entry, "entry cannot be null");
+ if (entry.schemaVersion() != 1 || entry.kind() != RunEntry.Kind.TASK_CALL
+ || entry.payload().schemaVersion() != SCHEMA_VERSION || !PLAN_TYPE.equals(entry.payload().type())) {
+ throw new ModelTurnJournalException("Incompatible tool plan entry " + entry.id());
+ }
+ JsonNode plan = read(entry.payload().json(), JsonNode.class);
+ JsonNode ordinal = plan.get("ordinal");
+ JsonNode name = plan.get("toolName");
+ JsonNode definition = plan.get("definitionVersion");
+ JsonNode hash = plan.get("argumentsHash");
+ JsonNode arguments = plan.get("arguments");
+ if (ordinal == null || !ordinal.isInt() || ordinal.intValue() < 0
+ || name == null || !name.isTextual() || name.textValue().isBlank()
+ || (definition != null && !definition.isNull() && !definition.isTextual())
+ || hash == null || !hash.isTextual() || arguments == null || !arguments.isTextual()) {
+ throw new ModelTurnJournalException("Invalid tool plan fields in entry " + entry.id());
+ }
+ ToolTaskIdentity identity = new ToolTaskIdentity(turn.runId(), turn.turnId(), ordinal.intValue());
+ if (!identity.entryId().equals(entry.id()) || !sha256(arguments.textValue()).equals(hash.textValue())) {
+ throw new ModelTurnJournalException("Inconsistent tool plan identity or argument hash in entry " + entry.id());
+ }
+ return new PlannedToolCall(identity, name.textValue(),
+ definition == null || definition.isNull() ? null : definition.textValue(),
+ hash.textValue(), arguments.textValue());
+ }
+
+ /**
+ * Canonical form of a tool-argument JSON string: object keys sorted recursively,
+ * compact output.
+ *
+ *
+ * Keys are reordered explicitly rather than through a mapper feature: the mapper's
+ * ordering settings apply to POJO properties and {@code Map} entries, not to an
+ * {@code ObjectNode}, which preserves insertion order. Only objects are reordered.
+ * Array order carries meaning (argument lists are positional) and scalars have no order.
+ * Malformed JSON is returned unchanged rather than rejected: argument validation is
+ * opt-in, so a provider may legitimately produce arguments this codec cannot parse, and
+ * that must not fail a turn that would otherwise dispatch.
+ */
+ public static String canonicalArguments(String arguments) {
+ if (arguments == null || arguments.isBlank()) {
+ return arguments;
+ }
+ try {
+ JsonNode tree = ARGUMENT_READER.readTree(arguments);
+ return tree == null || !tree.isContainerNode() ? arguments : write(sortKeys(tree));
+ }
+ catch (JsonProcessingException malformed) {
+ return arguments;
+ }
+ }
+
+ private static JsonNode sortKeys(JsonNode node) {
+ if (node.isObject()) {
+ ObjectNode sorted = MAPPER.createObjectNode();
+ TreeSet keys = new TreeSet<>();
+ node.fieldNames().forEachRemaining(keys::add);
+ keys.forEach(key -> sorted.set(key, sortKeys(node.get(key))));
+ return sorted;
+ }
+ if (node.isArray()) {
+ ArrayNode values = MAPPER.createArrayNode();
+ node.forEach(child -> values.add(sortKeys(child)));
+ return values;
+ }
+ return node;
+ }
+
+ /** Stable identity of one tool definition; a changed schema must not reuse a task id. */
+ public static String definitionVersion(ToolDefinition definition) {
+ return sha256(write(new DefinitionRecord(definition.name(), definition.description(),
+ definition.inputSchema())));
+ }
+
+ private static List sanitize(List messages) {
+ return messages == null ? List.of() : messages.stream().map(ModelTurnCodec::sanitize).toList();
+ }
+
+ private static ChatResponse sanitize(ChatResponse response) {
+ return new ChatResponse(response.id(), response.model(), sanitize(response.messages()), response.toolCalls(),
+ response.finishReason(), response.usage(), sanitize(response.metadata()));
+ }
+
+ private static ChatMessage sanitize(ChatMessage message) {
+ List content = message.content()
+ .stream()
+ .map(part -> new ContentPart(part.type(), part.text(), part.mimeType(), part.data(),
+ sanitize(part.metadata())))
+ .toList();
+ // Tool arguments are canonicalised here too: an assistant message replaying earlier
+ // calls must fingerprint identically no matter how the provider formatted them.
+ List toolCalls = message.toolCalls()
+ .stream()
+ .map(call -> new ToolCall(call.id(), call.name(), canonicalArguments(call.arguments())))
+ .toList();
+ return new ChatMessage(message.role(), content, toolCalls, message.toolCallId(), message.toolName(),
+ sanitize(message.metadata()));
+ }
+
+ private static Map sanitize(Map metadata) {
+ if (metadata == null || metadata.isEmpty()) {
+ return Map.of();
+ }
+ Map retained = new LinkedHashMap<>();
+ metadata.forEach((key, value) -> {
+ if (key != null && value != null && !PROCESS_LOCAL_METADATA.contains(key)) {
+ retained.put(key, value);
+ }
+ });
+ return retained;
+ }
+
+ private static T read(String json, Class type) {
+ try {
+ return MAPPER.readValue(Objects.requireNonNull(json, "journal payload cannot be null"), type);
+ }
+ catch (JsonProcessingException ex) {
+ throw new ModelTurnJournalException("Cannot decode journal payload as " + type.getSimpleName(), ex);
+ }
+ }
+
+ private static String write(Object payload) {
+ try {
+ return MAPPER.writeValueAsString(payload);
+ }
+ catch (JsonProcessingException ex) {
+ throw new ModelTurnJournalException("Cannot encode journal payload " + payload.getClass().getSimpleName(),
+ ex);
+ }
+ }
+
+ private static String sha256(String value) {
+ try {
+ MessageDigest digest = MessageDigest.getInstance("SHA-256");
+ return HexFormat.of().formatHex(digest.digest(value.getBytes(StandardCharsets.UTF_8)));
+ }
+ catch (NoSuchAlgorithmException ex) {
+ throw new IllegalStateException("SHA-256 is required by the Java platform", ex);
+ }
+ }
+
+ /** Stored shape of {@link #REQUEST_TYPE}: the hash only, never the request body. */
+ private record RequestRecord(String requestHash) {
+ }
+
+ /** Stored shape of {@link #ATTEMPT_TYPE}; a reservation is consumed even if its result is lost. */
+ private record AttemptRecord(int ordinal, String requestHash) {
+ }
+
+ /**
+ * Hashed shape of a request. Excludes {@link ChatRequest#attributes()} on purpose; see
+ * {@link #requestHash(ChatRequest)}.
+ */
+ private record RequestFingerprint(List messages, List tools,
+ GenerationOptions options) {
+ }
+
+ /** Stored shape of {@link #PLAN_TYPE}; excludes the provider's tool-call id. */
+ private record ToolPlan(int ordinal, String toolName, String definitionVersion, String argumentsHash,
+ String arguments) {
+ }
+
+ /** Hashed shape of a tool definition; a changed schema must not reuse a task id. */
+ private record DefinitionRecord(String name, String description, String inputSchema) {
+ }
+
+ /** Stored shape of {@link #DELTA_TYPE}. */
+ private record Delta(String text) {
+ }
+
+}
diff --git a/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/ModelTurnEntryIds.java b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/ModelTurnEntryIds.java
new file mode 100644
index 000000000..2cfe1b03f
--- /dev/null
+++ b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/ModelTurnEntryIds.java
@@ -0,0 +1,113 @@
+/*
+ * 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.agent.durable;
+
+import io.github.agentic.ai.agent.durable.ModelTurnRecorder.Turn;
+
+import java.util.OptionalInt;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Canonical journal entry ids of one model turn. This is the single definition of their
+ * wire format. Entry ids are local to a run; they do not encode its namespace or run id.
+ * Payload versions, journal consistency and execution policy remain separate concerns.
+ */
+public final class ModelTurnEntryIds {
+
+ private static final String PLAN_PREFIX = "#task.";
+
+ private static final String ATTEMPT_PREFIX = "#attempt.";
+
+ private ModelTurnEntryIds() {
+ }
+
+ /** The effective request fingerprint. */
+ public static String request(Turn turn) {
+ return turn.turnId() + "#req";
+ }
+
+ /** The complete terminal response. */
+ public static String response(Turn turn) {
+ return turn.turnId() + "#resp";
+ }
+
+ /** A model-call reservation, numbered from one. */
+ public static String attempt(Turn turn, int ordinal) {
+ if (ordinal < 1) {
+ throw new IllegalArgumentException("attempt ordinal must be positive");
+ }
+ return turn.turnId() + ATTEMPT_PREFIX + ordinal;
+ }
+
+ /** A planned tool call, numbered from zero in response order. */
+ public static String plan(Turn turn, int ordinal) {
+ if (ordinal < 0) {
+ throw new IllegalArgumentException("plan ordinal must be non-negative");
+ }
+ return turn.turnId() + PLAN_PREFIX + ordinal;
+ }
+
+ /** Diagnostic text within a stream attempt; the token is not a call reservation. */
+ public static String delta(Turn turn, String streamAttempt, int seq) {
+ return turn.turnId() + "#progress." + streamAttempt + "#delta." + seq;
+ }
+
+ /**
+ * Reads a plan ordinal of this turn, or empty for an unrelated entry. Matching digit
+ * suffixes with leading zeroes or values outside the int range are corrupt journal ids
+ * and throw rather than disappearing from replay's complete-plan check.
+ */
+ public static OptionalInt planOrdinal(Turn turn, String entryId) {
+ return ordinal(turn, entryId, PLAN_PREFIX, 0);
+ }
+
+ /**
+ * Reads a reservation ordinal of this turn, or empty for an unrelated entry. Matching
+ * digit suffixes must be canonical positive ints; invalid ones throw so a corrupt
+ * reservation cannot silently disappear from the attempt budget.
+ */
+ public static OptionalInt attemptOrdinal(Turn turn, String entryId) {
+ return ordinal(turn, entryId, ATTEMPT_PREFIX, 1);
+ }
+
+ private static OptionalInt ordinal(Turn turn, String entryId, String suffix, int minimum) {
+ requireNonNull(entryId, "entryId cannot be null");
+ String prefix = turn.turnId() + suffix;
+ if (!entryId.startsWith(prefix)) {
+ return OptionalInt.empty();
+ }
+ String digits = entryId.substring(prefix.length());
+ // Turn ids may contain these delimiters: another turn's entry must not leak in.
+ if (digits.isEmpty() || !digits.chars().allMatch(c -> c >= '0' && c <= '9')) {
+ return OptionalInt.empty();
+ }
+ if (digits.length() > 1 && digits.charAt(0) == '0') {
+ throw new ModelTurnJournalException("Non-canonical journal entry ordinal: " + entryId);
+ }
+ try {
+ int value = Integer.parseInt(digits);
+ if (value < minimum) {
+ throw new ModelTurnJournalException("Invalid journal entry ordinal: " + entryId);
+ }
+ return OptionalInt.of(value);
+ }
+ catch (NumberFormatException ex) {
+ throw new ModelTurnJournalException("Journal entry ordinal exceeds the int range: " + entryId, ex);
+ }
+ }
+
+}
diff --git a/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/ModelTurnJournalException.java b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/ModelTurnJournalException.java
new file mode 100644
index 000000000..db08f8372
--- /dev/null
+++ b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/ModelTurnJournalException.java
@@ -0,0 +1,41 @@
+/*
+ * 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.agent.durable;
+
+/**
+ * A model-turn payload could not be encoded or decoded, or recorded data is incomplete,
+ * incompatible or does not describe the run and request being replayed, or the turn's
+ * model-call reservation budget is exhausted. The recorder performs no storage writes;
+ * runtime commits retain the original RunStore exception types.
+ *
+ *
+ * This is not a model failure. An engine that converts caught model exceptions into
+ * user-visible error messages must rethrow this type instead. The runtime must also
+ * propagate storage commit failures before dispatching tools.
+ */
+public class ModelTurnJournalException extends IllegalStateException {
+
+ private static final long serialVersionUID = 1L;
+
+ public ModelTurnJournalException(String message) {
+ super(message);
+ }
+
+ public ModelTurnJournalException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+}
diff --git a/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/ModelTurnRecorder.java b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/ModelTurnRecorder.java
new file mode 100644
index 000000000..870b3e67f
--- /dev/null
+++ b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/ModelTurnRecorder.java
@@ -0,0 +1,383 @@
+/*
+ * 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.agent.durable;
+
+import io.github.agentic.ai.api.chat.ChatRequest;
+import io.github.agentic.ai.api.chat.ChatResponse;
+import io.github.agentic.ai.api.chat.ToolCall;
+import io.github.agentic.ai.api.chat.ToolDefinition;
+import io.github.agentic.ai.runtime.execution.run.RunEntry;
+import io.github.agentic.ai.runtime.execution.run.RunPayload;
+import io.github.agentic.ai.runtime.execution.run.RunSnapshot;
+import io.github.agentic.ai.runtime.execution.run.RunStore;
+import io.github.agentic.ai.runtime.execution.run.RunStoreConflictException;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.function.Predicate;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Prepares journal data for a model turn and validates recorded results on recovery.
+ *
+ *
+ * Stateless: every preparation method returns immutable records and writes nothing. The
+ * runtime owns the store, lease, revision and committed snapshot, and decides whether
+ * journaling is in use at all. It commits the request fingerprint and attempt before
+ * calling the provider, and the response with all its plans and the checkpoint in one
+ * {@link RunStore#commit} before dispatching a tool. Store exceptions propagate unchanged;
+ * they are not model failures.
+ *
+ *
+ * Entry ids are generated and recognized by {@link ModelTurnEntryIds}:
+ *
+ * - {@code #req} - the request fingerprint
+ * - {@code #attempt.} - a model-call reservation, starting at 1
+ * - {@code #progress.#delta.} - streamed text, diagnostic only
+ * - {@code #resp} - the complete terminal response
+ * - {@code #task.} - one planned tool call
+ *
+ *
+ *
+ * A matching request without a response or plan is re-asked only when its
+ * {@link ModelTurnAttemptPolicy} uses {@link MissingResponsePolicy#REASK}, and within its
+ * per-turn attempt limit. By default recovery fails closed and the total limit is one call.
+ * Every invocation requires a committed reservation from
+ * {@link #prepareAttempt(Turn, ChatRequest, RunSnapshot)}; restarting never resets the count.
+ * Even a reservation whose call was not issued consumes an attempt. Run-wide, token and cost
+ * budgets remain runtime responsibilities.
+ */
+public final class ModelTurnRecorder {
+
+ /** Total model-call reservations per turn, including the first invocation. */
+ public static final int DEFAULT_MAX_ATTEMPTS = ModelTurnAttemptPolicy.DEFAULT_MAX_ATTEMPTS;
+
+ private static final int ENTRY_SCHEMA_VERSION = 1;
+
+ /** Creates a stateless recorder; the runtime decides whether to use journaling. */
+ public ModelTurnRecorder() {
+ }
+
+ /**
+ * Engine-supplied identity of one logical model call. The engine restores the same
+ * identity on recovery and assigns distinct turn ids to different calls within a run.
+ * @param runId the run to which the prepared records belong
+ * @param turnId the stable logical call identity within that run
+ */
+ public record Turn(String runId, String turnId) {
+
+ public Turn {
+ if (runId == null || runId.isBlank() || turnId == null || turnId.isBlank()) {
+ throw new IllegalArgumentException("runId and turnId must not be blank");
+ }
+ }
+
+ }
+
+ /**
+ * Returns only the response from {@link #replayTurn(Turn, ChatRequest, RunSnapshot)},
+ * with the same journal consistency checks and no storage side effects.
+ */
+ public Optional replay(Turn turn, ChatRequest request, RunSnapshot snapshot) {
+ return replayTurn(turn, request, snapshot).map(RecordedModelTurn::response);
+ }
+
+ /**
+ * Returns the recorded response and ordered tool plans of this turn when the request
+ * is the one that produced them. Both initial execution after a successful response
+ * commit and recovery can consume this same view of the committed snapshot.
+ *
+ *
+ * Reads only the supplied snapshot. A response requires its matching request and exactly
+ * the tool plans described by that response and request. Orphaned or incompatible data
+ * fails closed. These checks validate current contents, not historical commit boundaries
+ * or the meaning of an application-owned checkpoint.
+ *
+ *
+ * The fingerprint covers the request the provider actually received, after every
+ * request-mutating interceptor. A rewrite that differs on the second call - for example
+ * summarisation that asks a model for its summary - therefore fails closed here instead
+ * of replaying a response produced for other input. Applications that need automatic
+ * recovery across such a rewrite must make it deterministic while journaling.
+ *
+ *
+ * This method cannot detect snapshot staleness. After an empty result, the runtime must
+ * prepare and commit an attempt using this snapshot's revision before calling the model.
+ * An empty result does not grant permission to call the model. A conflict
+ * requires reloading and repeating replay, not merely retrying with a newer revision.
+ * @param turn the engine-supplied identity of this logical call
+ * @param request the request about to be issued
+ * @param snapshot the runtime's consistent committed view of the run
+ * @return the response and plans, or empty when the turn has no committed result or plan
+ * @throws ModelTurnJournalException when recorded data is incomplete, incompatible or
+ * does not match the run or request
+ */
+ public Optional replayTurn(Turn turn, ChatRequest request, RunSnapshot snapshot) {
+ requireNonNull(turn, "turn cannot be null");
+ requireNonNull(request, "request cannot be null");
+ requireNonNull(snapshot, "snapshot cannot be null");
+ if (!turn.runId().equals(snapshot.spec().runId())) {
+ throw new ModelTurnJournalException("Run snapshot does not match model turn " + turn.turnId());
+ }
+ RunEntry recorded = snapshot.journal().get(ModelTurnEntryIds.request(turn));
+ RunEntry response = snapshot.journal().get(ModelTurnEntryIds.response(turn));
+ Map plans = recordedEntries(snapshot,
+ id -> ModelTurnEntryIds.planOrdinal(turn, id).isPresent());
+ Map attempts = recordedEntries(snapshot,
+ id -> ModelTurnEntryIds.attemptOrdinal(turn, id).isPresent());
+ if (recorded == null && response == null && plans.isEmpty() && attempts.isEmpty()) {
+ return Optional.empty();
+ }
+ if (recorded == null) {
+ throw new ModelTurnJournalException("Model turn " + turn.turnId() + " is missing its request fingerprint");
+ }
+ validateEntry(recorded, ModelTurnEntryIds.request(turn), RunEntry.Kind.MODEL_CALL, ModelTurnCodec.REQUEST_TYPE);
+ String hash = ModelTurnCodec.decodeRequestHash(recorded.payload().json());
+ if (hash == null || hash.isBlank()) {
+ throw new ModelTurnJournalException("Model turn " + turn.turnId() + " has an invalid request fingerprint");
+ }
+ if (!hash.equals(ModelTurnCodec.requestHash(request))) {
+ throw new ModelTurnJournalException("Model turn " + turn.turnId()
+ + " was recorded for a different request; refusing to replay it. This happens when the"
+ + " request is rewritten non-deterministically, for example by summarisation, or when"
+ + " the engine reuses a turn identity",
+ new RunStoreConflictException(
+ "Journal entry id identifies different data: " + turn.runId() + "/" + turn.turnId()));
+ }
+ reservedAttempts(turn, attempts, hash);
+ if (response == null) {
+ if (!plans.isEmpty()) {
+ throw new ModelTurnJournalException("Model turn " + turn.turnId() + " has tool plans without a response");
+ }
+ return Optional.empty();
+ }
+ validateEntry(response, ModelTurnEntryIds.response(turn), RunEntry.Kind.MODEL_CALL, ModelTurnCodec.RESPONSE_TYPE);
+ ChatResponse decoded = ModelTurnCodec.decodeResponse(response.payload().json());
+ List expected = preparePlans(turn, decoded, request.tools());
+ if (plans.size() != expected.size()) {
+ throw new ModelTurnJournalException("Model turn " + turn.turnId() + " has an incomplete or extra tool plan");
+ }
+ List decodedPlans = new ArrayList<>();
+ for (RunEntry planned : expected) {
+ RunEntry actual = plans.get(planned.id());
+ if (actual == null) {
+ throw new ModelTurnJournalException("Missing tool plan entry " + planned.id());
+ }
+ PlannedToolCall decodedPlan = ModelTurnCodec.decodeToolPlan(turn, actual);
+ if (!actual.equals(planned)) {
+ throw new ModelTurnJournalException("Recorded tool plan does not match the response and request: "
+ + planned.id());
+ }
+ decodedPlans.add(decodedPlan);
+ }
+ return Optional.of(new RecordedModelTurn(decoded, decodedPlans));
+ }
+
+ /**
+ * Prepares only the request fingerprint. This low-level operation does not reserve a
+ * model call; an invocation path must use
+ * {@link #prepareAttempt(Turn, ChatRequest, RunSnapshot)} before calling the provider.
+ */
+ public List prepareRequest(Turn turn, ChatRequest request) {
+ return List.of(new RunEntry(ENTRY_SCHEMA_VERSION, ModelTurnEntryIds.request(turn),
+ RunEntry.Kind.MODEL_CALL, new RunPayload(ModelTurnCodec.SCHEMA_VERSION, ModelTurnCodec.REQUEST_TYPE,
+ ModelTurnCodec.encodeRequest(request))));
+ }
+
+ /** Prepares a model-call reservation using {@link ModelTurnAttemptPolicy#DEFAULT}. */
+ public List prepareAttempt(Turn turn, ChatRequest request, RunSnapshot snapshot) {
+ return prepareAttempt(turn, request, snapshot, ModelTurnAttemptPolicy.DEFAULT);
+ }
+
+ /**
+ * Prepares the next model-call reservation using the default fail-closed recovery policy
+ * and the supplied total attempt limit.
+ * @param turn the logical model call
+ * @param request the effective provider request
+ * @param snapshot the committed view used for replay and the reservation commit
+ * @param maxAttempts positive total invocation limit
+ * @return immutable entries to commit before at most one provider invocation
+ * @throws ModelTurnJournalException if the turn already has a result, its data is
+ * incompatible, recovery re-asking is disabled, or its model-call budget is exhausted
+ * @throws IllegalArgumentException if maxAttempts is not positive
+ */
+ public List prepareAttempt(Turn turn, ChatRequest request, RunSnapshot snapshot, int maxAttempts) {
+ return prepareAttempt(turn, request, snapshot,
+ new ModelTurnAttemptPolicy(MissingResponsePolicy.FAIL_CLOSED, maxAttempts));
+ }
+
+ /**
+ * Prepares the next model-call reservation and, for a new turn, its request fingerprint.
+ * The runtime must commit the entire batch with this snapshot's revision, retaining the
+ * checkpoint, before invoking the provider. Only a successful commit authorizes one call;
+ * preparation alone never consumes or grants an attempt. On conflict, reload and replay.
+ *
+ *
+ * The limit includes the first call. A legacy request without attempt records counts as
+ * one reservation. Reservations are never refunded for a lost response or an interrupted
+ * invocation; changing the limit does not reset the stored count. Streaming delta tokens
+ * are unrelated to this count. {@link MissingResponsePolicy#FAIL_CLOSED} rejects another
+ * reservation after an earlier call is reserved without a response.
+ * {@link MissingResponsePolicy#REASK} permits it while the total limit has room. This limit
+ * does not replace runtime-owned run budgets.
+ * @param turn the logical model call
+ * @param request the effective provider request
+ * @param snapshot the committed view used for replay and the reservation commit
+ * @param policy recovery policy and positive total invocation limit
+ * @return immutable entries to commit before at most one provider invocation
+ * @throws ModelTurnJournalException if the turn already has a result, its data is
+ * incompatible, recovery re-asking is disabled, or its model-call budget is exhausted
+ */
+ public List prepareAttempt(Turn turn, ChatRequest request, RunSnapshot snapshot,
+ ModelTurnAttemptPolicy policy) {
+ requireNonNull(policy, "policy cannot be null");
+ if (replay(turn, request, snapshot).isPresent()) {
+ throw new ModelTurnJournalException("Model turn " + turn.turnId() + " already has a recorded response");
+ }
+ String hash = ModelTurnCodec.requestHash(request);
+ boolean hasRequest = snapshot.journal().containsKey(ModelTurnEntryIds.request(turn));
+ int used = hasRequest ? reservedAttempts(turn,
+ recordedEntries(snapshot, id -> ModelTurnEntryIds.attemptOrdinal(turn, id).isPresent()), hash) : 0;
+ if (used > 0 && policy.missingResponsePolicy() == MissingResponsePolicy.FAIL_CLOSED) {
+ throw new ModelTurnJournalException("Recovery re-asking is disabled for model turn " + turn.turnId());
+ }
+ if (used >= policy.maxAttempts()) {
+ throw new ModelTurnJournalException(
+ "Model-call budget exhausted for turn " + turn.turnId()
+ + " (maxAttempts=" + policy.maxAttempts() + ")");
+ }
+ List entries = new ArrayList<>();
+ if (!hasRequest) {
+ entries.addAll(prepareRequest(turn, request));
+ }
+ int ordinal = used + 1;
+ entries.add(new RunEntry(ENTRY_SCHEMA_VERSION, ModelTurnEntryIds.attempt(turn, ordinal),
+ RunEntry.Kind.MODEL_CALL, new RunPayload(ModelTurnCodec.SCHEMA_VERSION, ModelTurnCodec.ATTEMPT_TYPE,
+ ModelTurnCodec.encodeAttempt(ordinal, hash))));
+ return List.copyOf(entries);
+ }
+
+ /**
+ * Prepares the terminal response and every tool plan without publishing any records.
+ * The runtime must commit this entire batch with the corresponding checkpoint through
+ * {@link RunStore#commit} before dispatching the first tool. Preparing the batch
+ * does not authorize dispatch.
+ * @param turn the turn being prepared
+ * @param response the terminal response the engine is about to use
+ * @param tools the tool definitions that were disclosed to the model, used to pin each
+ * planned task to the schema the model saw
+ * @return an immutable batch containing the response followed by its tool plans
+ */
+ public List prepareResponse(Turn turn, ChatResponse response, Iterable tools) {
+ List entries = new ArrayList<>();
+ entries.add(new RunEntry(ENTRY_SCHEMA_VERSION, ModelTurnEntryIds.response(turn), RunEntry.Kind.MODEL_CALL,
+ new RunPayload(ModelTurnCodec.SCHEMA_VERSION, ModelTurnCodec.RESPONSE_TYPE,
+ ModelTurnCodec.encodeResponse(response))));
+ entries.addAll(preparePlans(turn, response, tools));
+ return List.copyOf(entries);
+ }
+
+ /**
+ * Prepares diagnostic text received while a response is still streaming.
+ *
+ *
+ * These entries are diagnostic: they explain how far a lost stream got. They are never
+ * a source of truth, because a partial response is not a model result and must not
+ * trigger tool execution. The attempt identifier keeps text from an abandoned attempt
+ * from colliding with a later one, and is stable for every chunk of one attempt so the
+ * recorded text stays ordered.
+ *
+ *
+ * The runtime submits these through its single commit coordinator, retaining the
+ * checkpoint, and handles their completion before committing the terminal response.
+ * @param attempt the attempt token obtained from {@link #newAttempt()}
+ * @param seq the chunk ordinal within this attempt
+ * @param text the text delta received
+ */
+ public List prepareDelta(Turn turn, String attempt, int seq, String text) {
+ if (text == null || text.isEmpty()) {
+ return List.of();
+ }
+ return List.of(new RunEntry(ENTRY_SCHEMA_VERSION,
+ ModelTurnEntryIds.delta(turn, attempt, seq), RunEntry.Kind.EVENT,
+ new RunPayload(ModelTurnCodec.SCHEMA_VERSION, ModelTurnCodec.DELTA_TYPE,
+ ModelTurnCodec.encodeDelta(text))));
+ }
+
+ /**
+ * A fresh diagnostic token for one streaming attempt, not a model-call reservation.
+ * Every subscription that starts a new provider call must use its own token, so an
+ * abandoned attempt cannot overwrite the text of the attempt that replaced it.
+ */
+ public static String newAttempt() {
+ return UUID.randomUUID().toString();
+ }
+
+ private List preparePlans(Turn turn, ChatResponse response, Iterable tools) {
+ List entries = new ArrayList<>();
+ Map byName = new HashMap<>();
+ if (tools != null) {
+ tools.forEach(definition -> byName.put(definition.name(), definition));
+ }
+ int ordinal = 0;
+ for (ToolCall call : response.toolCalls()) {
+ entries.add(new RunEntry(ENTRY_SCHEMA_VERSION, ModelTurnEntryIds.plan(turn, ordinal),
+ RunEntry.Kind.TASK_CALL,
+ new RunPayload(ModelTurnCodec.SCHEMA_VERSION, ModelTurnCodec.PLAN_TYPE,
+ ModelTurnCodec.encodeToolPlan(call, byName.get(call.name()), ordinal))));
+ ordinal++;
+ }
+ return entries;
+ }
+
+ /** Matching entries keyed by id; extra ordinals remain visible to validation. */
+ private Map recordedEntries(RunSnapshot snapshot, Predicate matches) {
+ Map entries = new HashMap<>();
+ snapshot.journal().forEach((id, entry) -> {
+ if (matches.test(id)) {
+ entries.put(id, entry);
+ }
+ });
+ return entries;
+ }
+
+ private int reservedAttempts(Turn turn, Map attempts, String requestHash) {
+ // A request written before attempt journaling still reserves its first invocation.
+ int used = attempts.size() + (attempts.containsKey(ModelTurnEntryIds.attempt(turn, 1)) ? 0 : 1);
+ attempts.forEach((id, entry) -> {
+ validateEntry(entry, id, RunEntry.Kind.MODEL_CALL, ModelTurnCodec.ATTEMPT_TYPE);
+ int ordinal = ModelTurnCodec.decodeAttemptOrdinal(entry.payload().json(), requestHash);
+ if (ordinal > used || !id.equals(ModelTurnEntryIds.attempt(turn, ordinal))) {
+ throw new ModelTurnJournalException("Invalid or non-contiguous model attempt entry " + id);
+ }
+ });
+ return used;
+ }
+
+ private void validateEntry(RunEntry entry, String id, RunEntry.Kind kind, String type) {
+ if (!id.equals(entry.id()) || entry.schemaVersion() != ENTRY_SCHEMA_VERSION || entry.kind() != kind
+ || !type.equals(entry.payload().type())
+ || entry.payload().schemaVersion() != ModelTurnCodec.SCHEMA_VERSION) {
+ throw new ModelTurnJournalException("Incompatible journal entry " + id);
+ }
+ }
+
+}
diff --git a/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/PlannedToolCall.java b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/PlannedToolCall.java
new file mode 100644
index 000000000..802b3332d
--- /dev/null
+++ b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/PlannedToolCall.java
@@ -0,0 +1,43 @@
+/*
+ * 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.agent.durable;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Read-only view of a recorded tool plan, not permission to execute it. The definition
+ * fingerprint describes the disclosed schema, not the tool implementation version, and
+ * may be null when no definition was supplied. Arguments retain their stored form,
+ * including malformed JSON. Tool resolution, argument validation and authorization are
+ * execution-boundary responsibilities. The argument hash checks input consistency; it
+ * does not identify the logical task.
+ *
+ *
+ * The wire field {@code definitionVersion} maps to {@code definitionFingerprint} here.
+ * For protocol correlation, use the identity's ordinal to read the provider call id
+ * from the corresponding response's {@code toolCalls()} list.
+ */
+public record PlannedToolCall(ToolTaskIdentity identity, String toolName, String definitionFingerprint,
+ String argumentsHash, String arguments) {
+
+ public PlannedToolCall {
+ requireNonNull(identity, "identity cannot be null");
+ requireNonNull(toolName, "toolName cannot be null");
+ requireNonNull(argumentsHash, "argumentsHash cannot be null");
+ requireNonNull(arguments, "arguments cannot be null");
+ }
+
+}
diff --git a/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/RecordedModelTurn.java b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/RecordedModelTurn.java
new file mode 100644
index 000000000..549ef3fde
--- /dev/null
+++ b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/RecordedModelTurn.java
@@ -0,0 +1,39 @@
+/*
+ * 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.agent.durable;
+
+import io.github.agentic.ai.api.chat.ChatResponse;
+
+import java.util.Comparator;
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * A recorded response and an immutable view of its plans in ordinal order. The recorder
+ * validates their journal consistency on replay; constructing this data view does not
+ * establish storage commitment, current ownership or permission to execute a tool.
+ */
+public record RecordedModelTurn(ChatResponse response, List plans) {
+
+ public RecordedModelTurn {
+ requireNonNull(response, "response cannot be null");
+ plans = List.copyOf(plans).stream()
+ .sorted(Comparator.comparingInt(plan -> plan.identity().ordinal()))
+ .toList();
+ }
+
+}
diff --git a/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/ToolTaskIdentity.java b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/ToolTaskIdentity.java
new file mode 100644
index 000000000..f6fcd23ae
--- /dev/null
+++ b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/ToolTaskIdentity.java
@@ -0,0 +1,42 @@
+/*
+ * 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.agent.durable;
+
+/**
+ * Stable logical tool-task identity within the run's namespace. The caller owns turn
+ * allocation and recovery. Attempts, provider call ids and argument hashes are not part
+ * of this identity; callers using it outside a run must retain its RunNamespace as well.
+ * @param runId the owning run
+ * @param turnId the caller-supplied logical model call
+ * @param ordinal the zero-based position in the response's tool-call list
+ */
+public record ToolTaskIdentity(String runId, String turnId, int ordinal) {
+
+ public ToolTaskIdentity {
+ if (runId == null || runId.isBlank() || turnId == null || turnId.isBlank()) {
+ throw new IllegalArgumentException("runId and turnId must not be blank");
+ }
+ if (ordinal < 0) {
+ throw new IllegalArgumentException("ordinal must be non-negative");
+ }
+ }
+
+ /** Journal key local to the owning run, not a globally scoped idempotency key. */
+ public String entryId() {
+ return ModelTurnEntryIds.plan(new ModelTurnRecorder.Turn(runId, turnId), ordinal);
+ }
+
+}
diff --git a/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/package-info.java b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/package-info.java
new file mode 100644
index 000000000..a502fefae
--- /dev/null
+++ b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/durable/package-info.java
@@ -0,0 +1,74 @@
+/*
+ * 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.
+ */
+
+/**
+ * Engine-neutral journaling of model turns.
+ *
+ *
+ * A subsequent model call can produce a different assistant message or tool plan. These
+ * types record the response, finish reason and usage, together with a stable identity for
+ * every planned tool call. Preparation is stateless and has no storage
+ * side effects. The runtime persists the data before dispatching tools; recovery then
+ * validates and replays a completed turn's recorded result instead of asking the model again.
+ * {@link io.github.agentic.ai.agent.durable.RecordedModelTurn} exposes that response and
+ * its ordered {@link io.github.agentic.ai.agent.durable.PlannedToolCall} views. Each plan
+ * has a {@link io.github.agentic.ai.agent.durable.ToolTaskIdentity}; its argument hash
+ * checks input consistency and is not a substitute for the logical task identity.
+ * These views preserve the existing journal format. Reading a plan does not resolve
+ * its tool, validate its argument schema or authorize dispatch; an unknown tool or
+ * malformed argument string can still be recorded and read.
+ *
+ *
+ * The contract deliberately knows no graph, session, scheduler or streaming library: it
+ * depends on the neutral chat contracts and on {@code io.github.agentic.ai.runtime.execution.run}.
+ * Graph-backed and graph-independent engines share it, and each supplies stable run
+ * and turn identities. The engine restores the same identity for a recovered logical call
+ * and gives different calls distinct turn ids within a run.
+ *
+ *
+ * The runtime explicitly enables journaling and owns the store, lease and latest committed
+ * snapshot. It commits a prepared request fingerprint and attempt reservation before calling
+ * the model, then commits the full response and plan batch with the corresponding checkpoint
+ * through {@link io.github.agentic.ai.runtime.execution.run.RunStore#commit} before
+ * dispatching the first tool. Planned TASK_CALL entries are journal data, not task lifecycle
+ * updates in {@code RunCommit.tasks}.
+ *
+ *
+ * A turn without a response or plan can reserve another attempt for the same request.
+ * The default limit is two total reservations per turn, including the first invocation;
+ * the caller can supply any positive limit. Each reservation is journaled before its call,
+ * and remains consumed after interruption or a lost response. Legacy request-only records
+ * count as one reservation. Streaming delta tokens do not affect this count, and attempts
+ * do not change task identities. Run-wide, token and cost budgets remain runtime concerns.
+ *
+ *
+ * All writes share one runtime coordinator. Fingerprints, reservations and diagnostic deltas
+ * retain the checkpoint. The runtime settles in-flight delta commits before submitting the
+ * terminal response, may drop unsubmitted deltas, and propagates storage failures by default.
+ * An uncertain commit outcome requires resolving the stored state before continuing from a
+ * cached snapshot; revision conflicts and ownership loss must not be ignored.
+ *
+ *
+ * Replay checks a single supplied snapshot for compatible request/response records and an
+ * exact matching tool plan. It cannot prove that records were committed together, interpret
+ * application checkpoints or detect stale reads. An empty replay must be followed by a
+ * successful attempt reservation using that snapshot's revision before a provider call;
+ * a conflict requires reloading and replaying again. Durable-mode admission, definition and
+ * identity recovery, execution positions and
+ * tool receipts remain runtime responsibilities. An in-memory reference store only models
+ * recovery within one process.
+ */
+package io.github.agentic.ai.agent.durable;
diff --git a/argi-agent/argi-agent-core/src/test/java/io/github/agentic/ai/agent/durable/ModelTurnEntryIdsTest.java b/argi-agent/argi-agent-core/src/test/java/io/github/agentic/ai/agent/durable/ModelTurnEntryIdsTest.java
new file mode 100644
index 000000000..3f50eac34
--- /dev/null
+++ b/argi-agent/argi-agent-core/src/test/java/io/github/agentic/ai/agent/durable/ModelTurnEntryIdsTest.java
@@ -0,0 +1,93 @@
+/*
+ * 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.agent.durable;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.OptionalInt;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class ModelTurnEntryIdsTest {
+
+ private static final ModelTurnRecorder.Turn TURN = new ModelTurnRecorder.Turn("run", "agent:0");
+
+ @Test
+ void preservesTheExistingWireIdsAndTaskIdentity() {
+ assertThat(ModelTurnEntryIds.request(TURN)).isEqualTo("agent:0#req");
+ assertThat(ModelTurnEntryIds.response(TURN)).isEqualTo("agent:0#resp");
+ assertThat(ModelTurnEntryIds.attempt(TURN, 1)).isEqualTo("agent:0#attempt.1");
+ assertThat(ModelTurnEntryIds.plan(TURN, 0)).isEqualTo("agent:0#task.0");
+ assertThat(ModelTurnEntryIds.delta(TURN, "stream-token", 2))
+ .isEqualTo("agent:0#progress.stream-token#delta.2");
+ assertThat(new ToolTaskIdentity(TURN.runId(), TURN.turnId(), 0).entryId())
+ .isEqualTo(ModelTurnEntryIds.plan(TURN, 0));
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = { 1, 2, 10, Integer.MAX_VALUE })
+ void numberedIdsRoundTripWithoutOverflow(int ordinal) {
+ assertThat(ModelTurnEntryIds.planOrdinal(TURN, ModelTurnEntryIds.plan(TURN, ordinal)))
+ .isEqualTo(OptionalInt.of(ordinal));
+ assertThat(ModelTurnEntryIds.attemptOrdinal(TURN, ModelTurnEntryIds.attempt(TURN, ordinal)))
+ .isEqualTo(OptionalInt.of(ordinal));
+ }
+
+ @Test
+ void plansStartAtZeroAndReservationsStartAtOne() {
+ assertThat(ModelTurnEntryIds.planOrdinal(TURN, "agent:0#task.0")).isEqualTo(OptionalInt.of(0));
+ assertThatThrownBy(() -> ModelTurnEntryIds.plan(TURN, -1)).isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> ModelTurnEntryIds.attempt(TURN, 0)).isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> ModelTurnEntryIds.attemptOrdinal(TURN, "agent:0#attempt.0"))
+ .isInstanceOf(ModelTurnJournalException.class);
+ }
+
+ @Test
+ void nestedTurnIdsDoNotLeakIntoTheirParent() {
+ ModelTurnRecorder.Turn nested = new ModelTurnRecorder.Turn("run", "agent:0#task.0#attempt.1");
+ String plan = ModelTurnEntryIds.plan(nested, 2);
+ String attempt = ModelTurnEntryIds.attempt(nested, 3);
+
+ assertThat(ModelTurnEntryIds.planOrdinal(nested, plan)).isEqualTo(OptionalInt.of(2));
+ assertThat(ModelTurnEntryIds.attemptOrdinal(nested, attempt)).isEqualTo(OptionalInt.of(3));
+ assertThat(ModelTurnEntryIds.planOrdinal(TURN, plan)).isEqualTo(OptionalInt.empty());
+ assertThat(ModelTurnEntryIds.attemptOrdinal(TURN, attempt)).isEqualTo(OptionalInt.empty());
+ assertThat(ModelTurnEntryIds.planOrdinal(TURN, ModelTurnEntryIds.request(nested)))
+ .isEqualTo(OptionalInt.empty());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = { "other#task.0", "agent:00#task.0", "agent:0#req", "agent:0#resp",
+ "agent:0#task.", "agent:0#task.x", "agent:0#task.1#resp", "agent:0#attempt.",
+ "agent:0#attempt.x", "agent:0#attempt.1#req" })
+ void unrelatedEntriesAreNotNumberedEntriesOfThisTurn(String entryId) {
+ assertThat(ModelTurnEntryIds.planOrdinal(TURN, entryId)).isEqualTo(OptionalInt.empty());
+ assertThat(ModelTurnEntryIds.attemptOrdinal(TURN, entryId)).isEqualTo(OptionalInt.empty());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = { "00", "01", "2147483648", "9999999999999999999999999" })
+ void corruptNumericSuffixesCannotBeSilentlyDropped(String suffix) {
+ assertThatThrownBy(() -> ModelTurnEntryIds.planOrdinal(TURN, "agent:0#task." + suffix))
+ .isInstanceOf(ModelTurnJournalException.class);
+ assertThatThrownBy(() -> ModelTurnEntryIds.attemptOrdinal(TURN, "agent:0#attempt." + suffix))
+ .isInstanceOf(ModelTurnJournalException.class);
+ }
+
+}
diff --git a/argi-agent/argi-agent-core/src/test/java/io/github/agentic/ai/agent/durable/ModelTurnPlanViewTest.java b/argi-agent/argi-agent-core/src/test/java/io/github/agentic/ai/agent/durable/ModelTurnPlanViewTest.java
new file mode 100644
index 000000000..04e31144e
--- /dev/null
+++ b/argi-agent/argi-agent-core/src/test/java/io/github/agentic/ai/agent/durable/ModelTurnPlanViewTest.java
@@ -0,0 +1,154 @@
+/*
+ * 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.agent.durable;
+
+import io.github.agentic.ai.api.chat.ChatMessage;
+import io.github.agentic.ai.api.chat.ChatResponse;
+import io.github.agentic.ai.api.chat.FinishReason;
+import io.github.agentic.ai.api.chat.ToolCall;
+import io.github.agentic.ai.runtime.execution.run.RunEntry;
+import io.github.agentic.ai.runtime.execution.run.RunPayload;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class ModelTurnPlanViewTest {
+
+ private static final ModelTurnRecorder.Turn TURN = new ModelTurnRecorder.Turn("run", "agent:0");
+
+ private static final String EMPTY_ARGUMENTS_HASH =
+ "44136fa355b3678a1146ad16f7e8649e94fb4fc21fe77e8310c060f61caaff8a";
+
+ /** Version-1 wire fixture, independent of the current writer. */
+ private static final String PLAN_JSON = """
+ {"ordinal":0,"toolName":"unknown","definitionVersion":null,
+ "argumentsHash":"44136fa355b3678a1146ad16f7e8649e94fb4fc21fe77e8310c060f61caaff8a","arguments":"{}"}
+ """;
+
+ @Test
+ void readsTheExistingWireFormatWithoutRewritingIt() {
+ RunEntry entry = planEntry(PLAN_JSON);
+ PlannedToolCall plan = ModelTurnCodec.decodeToolPlan(TURN, entry);
+
+ assertThat(plan).isEqualTo(new PlannedToolCall(new ToolTaskIdentity("run", "agent:0", 0),
+ "unknown", null, EMPTY_ARGUMENTS_HASH, "{}"));
+ assertThat(plan.identity().entryId()).isEqualTo(entry.id());
+ assertThat(planEntry(ModelTurnCodec.encodeToolPlan(new ToolCall("provider", "unknown", "{}"), null, 0)))
+ .isEqualTo(entry);
+ RunEntry withDefinition = planEntry(PLAN_JSON.replace("\"definitionVersion\":null",
+ "\"definitionVersion\":\"schema-fingerprint\""));
+ assertThat(ModelTurnCodec.decodeToolPlan(TURN, withDefinition).definitionFingerprint())
+ .isEqualTo("schema-fingerprint");
+ }
+
+ @Test
+ void readsUnknownToolsAndMalformedArgumentsWithoutBusinessValidation() {
+ String arguments = "{ not valid JSON";
+ RunEntry entry = planEntry(ModelTurnCodec.encodeToolPlan(
+ new ToolCall("provider", "not-disclosed", arguments), null, 0));
+
+ PlannedToolCall plan = ModelTurnCodec.decodeToolPlan(TURN, entry);
+
+ assertThat(plan.toolName()).isEqualTo("not-disclosed");
+ assertThat(plan.definitionFingerprint()).isNull();
+ assertThat(plan.arguments()).isEqualTo(arguments);
+ }
+
+ @Test
+ void rejectsIncompatibleKindsPayloadsAndEntryIdentities() {
+ RunEntry entry = planEntry(PLAN_JSON);
+ List incompatible = List.of(
+ new RunEntry(1, entry.id(), RunEntry.Kind.MODEL_CALL, entry.payload()),
+ new RunEntry(1, entry.id(), entry.kind(), new RunPayload(2, ModelTurnCodec.PLAN_TYPE, PLAN_JSON)),
+ new RunEntry(1, entry.id(), entry.kind(), new RunPayload(1, "other.type", PLAN_JSON)),
+ new RunEntry(1, "agent:0#task.1", entry.kind(), entry.payload()),
+ new RunEntry(1, "other#task.0", entry.kind(), entry.payload()));
+
+ for (RunEntry invalid : incompatible) {
+ assertThatThrownBy(() -> ModelTurnCodec.decodeToolPlan(TURN, invalid))
+ .isInstanceOf(ModelTurnJournalException.class);
+ }
+ }
+
+ @Test
+ void rejectsInvalidStoredFieldsAndInconsistentHashes() {
+ List invalidPayloads = List.of("{}",
+ PLAN_JSON.replace("\"ordinal\":0", "\"ordinal\":-1"),
+ PLAN_JSON.replace("\"ordinal\":0", "\"ordinal\":\"0\""),
+ PLAN_JSON.replace("\"ordinal\":0", "\"ordinal\":0.5"),
+ PLAN_JSON.replace("\"ordinal\":0", "\"ordinal\":2147483648"),
+ PLAN_JSON.replace("\"toolName\":\"unknown\"", "\"toolName\":null"),
+ PLAN_JSON.replace("\"definitionVersion\":null", "\"definitionVersion\":{}"),
+ PLAN_JSON.replace("\"arguments\":\"{}\"", "\"arguments\":{}"),
+ PLAN_JSON.replace(EMPTY_ARGUMENTS_HASH, "different"));
+
+ for (String invalid : invalidPayloads) {
+ assertThatThrownBy(() -> ModelTurnCodec.decodeToolPlan(TURN, planEntry(invalid)))
+ .isInstanceOf(ModelTurnJournalException.class);
+ }
+ }
+
+ @Test
+ void identitiesRemainRunScopedAndPreserveTurnIdsWithDelimiters() {
+ ToolTaskIdentity first = new ToolTaskIdentity("run", "parent#task.0", 0);
+ assertThat(first.entryId()).isEqualTo("parent#task.0#task.0");
+ assertThat(first).isNotEqualTo(new ToolTaskIdentity("other-run", first.turnId(), 0));
+ assertThat(first).isNotEqualTo(new ToolTaskIdentity("run", first.turnId(), 1));
+ assertThatThrownBy(() -> new ToolTaskIdentity("run", "turn", -1))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = { "", " " })
+ void rejectsBlankIdentityComponents(String blank) {
+ assertThatThrownBy(() -> new ToolTaskIdentity(blank, "turn", 0))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> new ToolTaskIdentity("run", blank, 0))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ void recordedTurnCopiesAndOrdersPlansWithoutChangingTheSourceList() {
+ PlannedToolCall first = ModelTurnCodec.decodeToolPlan(TURN, planEntry(PLAN_JSON));
+ PlannedToolCall second = new PlannedToolCall(new ToolTaskIdentity("run", "agent:0", 1),
+ first.toolName(), null, first.argumentsHash(), first.arguments());
+ List source = new ArrayList<>(List.of(second, first));
+ ChatResponse response = new ChatResponse("response", "model", List.of(ChatMessage.assistant("done")),
+ List.of(), FinishReason.STOP, null, Map.of());
+
+ RecordedModelTurn recorded = new RecordedModelTurn(response, source);
+
+ assertThat(source).containsExactly(second, first);
+ source.clear();
+ assertThat(recorded.plans()).containsExactly(first, second);
+ assertThat(recorded.response()).isSameAs(response);
+ assertThatThrownBy(() -> recorded.plans().clear()).isInstanceOf(UnsupportedOperationException.class);
+ }
+
+ private static RunEntry planEntry(String json) {
+ return new RunEntry(1, "agent:0#task.0", RunEntry.Kind.TASK_CALL,
+ new RunPayload(1, ModelTurnCodec.PLAN_TYPE, json));
+ }
+
+}
diff --git a/argi-agent/argi-agent-core/src/test/java/io/github/agentic/ai/agent/durable/ModelTurnRecorderTest.java b/argi-agent/argi-agent-core/src/test/java/io/github/agentic/ai/agent/durable/ModelTurnRecorderTest.java
new file mode 100644
index 000000000..76a0bec92
--- /dev/null
+++ b/argi-agent/argi-agent-core/src/test/java/io/github/agentic/ai/agent/durable/ModelTurnRecorderTest.java
@@ -0,0 +1,722 @@
+/*
+ * 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.agent.durable;
+
+import io.github.agentic.ai.api.chat.ChatMessage;
+import io.github.agentic.ai.api.chat.ChatRequest;
+import io.github.agentic.ai.api.chat.ChatResponse;
+import io.github.agentic.ai.api.chat.FinishReason;
+import io.github.agentic.ai.api.chat.ToolCall;
+import io.github.agentic.ai.api.chat.ToolDefinition;
+import io.github.agentic.ai.runtime.execution.PrincipalReference;
+import io.github.agentic.ai.runtime.execution.TenantReference;
+import io.github.agentic.ai.runtime.execution.TrustedExecutionContext;
+import io.github.agentic.ai.runtime.execution.definition.DefinitionKind;
+import io.github.agentic.ai.runtime.execution.definition.DefinitionReference;
+import io.github.agentic.ai.runtime.execution.run.MemoryRunStore;
+import io.github.agentic.ai.runtime.execution.run.RunCommit;
+import io.github.agentic.ai.runtime.execution.run.RunEntry;
+import io.github.agentic.ai.runtime.execution.run.RunLease;
+import io.github.agentic.ai.runtime.execution.run.RunNamespace;
+import io.github.agentic.ai.runtime.execution.run.RunPayload;
+import io.github.agentic.ai.runtime.execution.run.RunSnapshot;
+import io.github.agentic.ai.runtime.execution.run.RunSpec;
+import io.github.agentic.ai.runtime.execution.run.RunState;
+import io.github.agentic.ai.runtime.execution.run.RunStore;
+import io.github.agentic.ai.runtime.execution.run.RunStoreCapabilities;
+import io.github.agentic.ai.runtime.execution.run.RunStoreConflictException;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** The neutral model-turn contract, exercised without any graph or streaming engine. */
+@Timeout(10)
+class ModelTurnRecorderTest {
+
+ private static final String RUN_ID = "run-1";
+
+ private static final ModelTurnRecorder.Turn TURN = new ModelTurnRecorder.Turn(RUN_ID, "agent:0");
+
+ private static final List TOOLS = List.of(
+ new ToolDefinition("search", "search the web", "{\"type\":\"object\"}"));
+
+ private static final ChatRequest REQUEST = new ChatRequest(List.of(ChatMessage.user("search for argi")),
+ TOOLS, null, Map.of());
+
+ private static final ChatResponse RESPONSE = toolCallingResponse("call-1");
+
+ private static final ModelTurnAttemptPolicy REASK_TWO_ATTEMPTS =
+ new ModelTurnAttemptPolicy(MissingResponsePolicy.REASK, 2);
+
+ private final MemoryRunStore store = new MemoryRunStore();
+
+ private RunStore runtimeStore = store;
+
+ private ModelTurnRecorder recorder = new ModelTurnRecorder();
+
+ private RunSpec run;
+
+ private RunLease lease;
+
+ private RunSnapshot snapshot;
+
+ @BeforeEach
+ void setUp() {
+ TrustedExecutionContext context = TrustedExecutionContext.builder()
+ .tenant(new TenantReference("iam", "tenant"))
+ .principal(new PrincipalReference("iam", "user"))
+ .runId(RUN_ID)
+ .correlationId("trace")
+ .build();
+ run = RunSpec.from(context, "app", "session", "submission",
+ DefinitionReference.of(DefinitionKind.AGENT, "agent", "1"), new RunPayload(1, "test.input", "{}"));
+ store.submit(run, context);
+ lease = store.acquire(run.namespace(), RUN_ID, UUID.randomUUID(), Duration.ofMinutes(1));
+ snapshot = store.commit(run.namespace(), RUN_ID, lease,
+ new RunCommit(0, RunState.RUNNING, new RunPayload(1, "test.checkpoint", "{\"phase\":\"MODEL_READY\"}"),
+ List.of(), List.of(), List.of()));
+ }
+
+ @Test
+ void preparationIsImmutableAndHasNoStorageSideEffects() {
+ RunSnapshot before = snapshot;
+ List> batches = List.of(recorder.prepareRequest(TURN, REQUEST),
+ recorder.prepareAttempt(TURN, REQUEST, snapshot), recorder.prepareResponse(TURN, RESPONSE, TOOLS),
+ recorder.prepareDelta(TURN, ModelTurnRecorder.newAttempt(), 0, "hello"));
+
+ for (List batch : batches) {
+ assertThatThrownBy(batch::clear).isInstanceOf(UnsupportedOperationException.class);
+ }
+ assertThat(recorder.prepareAttempt(TURN, REQUEST, snapshot)).isEqualTo(batches.get(1));
+ assertThat(store.find(run.namespace(), RUN_ID)).contains(before);
+ }
+
+ @Test
+ void toolPlansIgnoreFormattingAndProviderCallIdsWhilePreservingArrayOrder() {
+ String first = ModelTurnCodec.encodeToolPlan(
+ new ToolCall("call-a", "search", "{\"q\":\"x\",\"p\":\"y\"}"), TOOLS.get(0), 0);
+ String second = ModelTurnCodec.encodeToolPlan(
+ new ToolCall("call-b", "search", "{ \"p\" : \"y\", \"q\" : \"x\" }"), TOOLS.get(0), 0);
+
+ assertThat(second).isEqualTo(first);
+ assertThat(ModelTurnCodec.canonicalArguments("{\"tags\":[\"b\",\"a\"]}"))
+ .isEqualTo("{\"tags\":[\"b\",\"a\"]}");
+ }
+
+ @Test
+ void canonicalArgumentsPreserveNestedDecimalValuesAndArrayOrder() {
+ String arguments = "{\"z\":[0.12345678901234567891,{\"b\":0.12345678901234567890,\"a\":1}],\"a\":true}";
+ String expected = "{\"a\":true,\"z\":[0.12345678901234567891,{\"a\":1,\"b\":0.1234567890123456789}]}";
+ String equivalent = "{ \"a\" : true, \"z\" : [0.123456789012345678910,"
+ + "{\"a\":1,\"b\":1.2345678901234567890e-1}] }";
+
+ assertThat(ModelTurnCodec.canonicalArguments(arguments)).isEqualTo(expected);
+ assertThat(ModelTurnCodec.canonicalArguments(equivalent)).isEqualTo(expected);
+ assertThat(ModelTurnCodec.canonicalArguments(expected)).isEqualTo(expected);
+ }
+
+ @Test
+ void highPrecisionToolArgumentsProduceDistinctPlansAndHashesAfterRecovery() {
+ List calls = List.of(new ToolCall("first", "search", "{\"value\":0.12345678901234567890}"),
+ new ToolCall("second", "search", "{\"value\":0.12345678901234567891}"));
+ assertThat(ModelTurnCodec.encodeToolPlan(calls.get(0), TOOLS.get(0), 0))
+ .isNotEqualTo(ModelTurnCodec.encodeToolPlan(calls.get(1), TOOLS.get(0), 0));
+ ChatResponse response = new ChatResponse("response", "model", List.of(ChatMessage.assistant("", calls)),
+ calls, FinishReason.TOOL_CALLS, null, Map.of());
+ commit(null, recorder.prepareAttempt(TURN, REQUEST, snapshot));
+ commit(checkpoint(), recorder.prepareResponse(TURN, response, TOOLS));
+ reload();
+
+ RecordedModelTurn recovered = recorder.replayTurn(TURN, REQUEST, snapshot).orElseThrow();
+
+ assertThat(recovered.plans()).extracting(PlannedToolCall::arguments).containsExactly(
+ "{\"value\":0.1234567890123456789}", "{\"value\":0.12345678901234567891}");
+ assertThat(recovered.plans().get(0).argumentsHash()).isNotEqualTo(recovered.plans().get(1).argumentsHash());
+ assertThat(recovered.response().messages().get(0).toolCalls()).extracting(ToolCall::arguments)
+ .containsExactly("{\"value\":0.1234567890123456789}", "{\"value\":0.12345678901234567891}");
+ assertThat(recovered.response().toolCalls()).isEqualTo(calls);
+ }
+
+ @Test
+ void requestFingerprintsDistinguishHighPrecisionToolArguments() {
+ ChatRequest first = new ChatRequest(List.of(ChatMessage.assistant("",
+ List.of(new ToolCall("call", "search", "{\"value\":0.12345678901234567890}")))),
+ TOOLS, null, Map.of());
+ ChatRequest second = new ChatRequest(List.of(ChatMessage.assistant("",
+ List.of(new ToolCall("call", "search", "{\"value\":0.12345678901234567891}")))),
+ TOOLS, null, Map.of());
+
+ assertThat(ModelTurnCodec.requestHash(first)).isNotEqualTo(ModelTurnCodec.requestHash(second));
+ }
+
+ @Test
+ void referenceDispatchesOnlyAfterTheResponseAllPlansAndCheckpointAreCommitted() {
+ List calls = List.of(new ToolCall("first", "search", "{\"q\":\"first\"}"),
+ new ToolCall("second", "search", "{\"q\":\"second\"}"));
+ ChatResponse response = new ChatResponse("response", "model", List.of(ChatMessage.assistant("", calls)),
+ calls, FinishReason.TOOL_CALLS, null, Map.of());
+ AtomicInteger tools = new AtomicInteger();
+
+ executeReference(() -> {
+ assertThat(snapshot.journal()).containsOnlyKeys("agent:0#req", "agent:0#attempt.1");
+ return response;
+ }, call -> {
+ RunSnapshot committed = store.find(run.namespace(), RUN_ID).orElseThrow();
+ assertThat(committed.checkpoint()).isEqualTo(checkpoint());
+ assertThat(committed.journal()).containsOnlyKeys("agent:0#req", "agent:0#attempt.1",
+ "agent:0#resp", "agent:0#task.0", "agent:0#task.1");
+ tools.incrementAndGet();
+ });
+
+ assertThat(tools.get()).isEqualTo(2);
+ }
+
+ @ParameterizedTest
+ @CsvSource({ "argi.model-request, 0", "argi.model-attempt, 0", "argi.model-response, 1" })
+ void commitFailurePreventsSubsequentSideEffects(String payloadType, int expectedModelCalls) {
+ IllegalStateException unavailable = new IllegalStateException("storage unavailable");
+ runtimeStore = new FailingRunStore(store, payloadType, false, unavailable);
+ RunPayload before = snapshot.checkpoint();
+ AtomicInteger modelCalls = new AtomicInteger();
+ AtomicInteger tools = new AtomicInteger();
+
+ assertThatThrownBy(() -> executeReference(countedModel(modelCalls), call -> tools.incrementAndGet()))
+ .isSameAs(unavailable);
+
+ assertThat(modelCalls.get()).isEqualTo(expectedModelCalls);
+ assertThat(tools.get()).isZero();
+ assertThat(snapshot.checkpoint()).isEqualTo(before);
+ assertThat(store.find(run.namespace(), RUN_ID)).contains(snapshot);
+ }
+
+ @Test
+ void lostAttemptConfirmationDoesNotInvokeTheModelOrRefundTheReservation() {
+ IllegalStateException lost = new IllegalStateException("attempt confirmation lost");
+ runtimeStore = new FailingRunStore(store, ModelTurnCodec.ATTEMPT_TYPE, true, lost);
+ AtomicInteger modelCalls = new AtomicInteger();
+ AtomicInteger tools = new AtomicInteger();
+ RunSnapshot before = snapshot;
+
+ assertThatThrownBy(() -> executeReference(1, countedModel(modelCalls), call -> tools.incrementAndGet()))
+ .isSameAs(lost);
+ assertThat(snapshot).isSameAs(before);
+
+ runtimeStore = store;
+ reload();
+ assertThat(snapshot.journal()).containsOnlyKeys("agent:0#req", "agent:0#attempt.1");
+ assertThatThrownBy(() -> executeReference(1, countedModel(modelCalls), call -> tools.incrementAndGet()))
+ .isInstanceOf(ModelTurnJournalException.class).hasMessageContaining("re-asking is disabled");
+ assertThat(modelCalls.get()).isZero();
+ assertThat(tools.get()).isZero();
+ }
+
+ @Test
+ void lostResponseConfirmationRecoversTheSamePlanWithoutAnotherModelCall() {
+ IllegalStateException lost = new IllegalStateException("response confirmation lost");
+ runtimeStore = new FailingRunStore(store, ModelTurnCodec.RESPONSE_TYPE, true, lost);
+ AtomicInteger modelCalls = new AtomicInteger();
+ AtomicInteger tools = new AtomicInteger();
+
+ assertThatThrownBy(() -> executeReference(countedModel(modelCalls), call -> tools.incrementAndGet()))
+ .isSameAs(lost);
+ assertThat(tools.get()).isZero();
+
+ runtimeStore = store;
+ reload();
+ RunEntry plan = snapshot.journal().get("agent:0#task.0");
+ RunSnapshot committed = snapshot;
+ ChatResponse recovered = executeReference(1, () -> {
+ modelCalls.incrementAndGet();
+ return toolCallingResponse("different-provider-id");
+ }, call -> tools.incrementAndGet());
+
+ assertThat(modelCalls.get()).isEqualTo(1);
+ assertThat(tools.get()).isEqualTo(1);
+ assertThat(recovered.toolCalls()).isEqualTo(RESPONSE.toolCalls());
+ assertThat(snapshot.journal()).containsEntry("agent:0#task.0", plan);
+ assertThat(snapshot).isSameAs(committed);
+ }
+
+ @Test
+ void aStaleSnapshotCannotAuthorizeAnotherModelCall() {
+ RunSnapshot stale = snapshot;
+ commit(null, recorder.prepareAttempt(TURN, REQUEST, snapshot));
+ RunSnapshot committed = snapshot;
+ snapshot = stale;
+ AtomicInteger modelCalls = new AtomicInteger();
+
+ assertThatThrownBy(() -> executeReference(countedModel(modelCalls), call -> {
+ throw new AssertionError("No tool may run after a stale reservation");
+ })).isInstanceOf(RunStoreConflictException.class);
+
+ assertThat(modelCalls.get()).isZero();
+ assertThat(snapshot).isSameAs(stale);
+ assertThat(store.find(run.namespace(), RUN_ID)).contains(committed);
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = { 1, 2, 3 })
+ void committedReservationsSurviveReloadsAndEnforceTheConfiguredLimit(int maxAttempts) {
+ ModelTurnAttemptPolicy policy = new ModelTurnAttemptPolicy(MissingResponsePolicy.REASK, maxAttempts);
+ for (int ordinal = 1; ordinal <= maxAttempts; ordinal++) {
+ commit(null, recorder.prepareAttempt(TURN, REQUEST, snapshot, policy));
+ reload();
+ assertThat(snapshot.journal()).containsKey("agent:0#attempt." + ordinal);
+ }
+ AtomicInteger modelCalls = new AtomicInteger();
+ AtomicInteger tools = new AtomicInteger();
+
+ assertThatThrownBy(() -> executeReference(policy, countedModel(modelCalls),
+ call -> tools.incrementAndGet())).isInstanceOf(ModelTurnJournalException.class)
+ .hasMessageContaining("budget exhausted");
+
+ assertThat(modelCalls.get()).isZero();
+ assertThat(tools.get()).isZero();
+ }
+
+ @Test
+ void aLostResponseFailsClosedByDefaultAfterReload() {
+ IllegalStateException unavailable = new IllegalStateException("response storage unavailable");
+ runtimeStore = new FailingRunStore(store, ModelTurnCodec.RESPONSE_TYPE, false, unavailable);
+ AtomicInteger modelCalls = new AtomicInteger();
+ AtomicInteger tools = new AtomicInteger();
+ Supplier model = () -> toolCallingResponse("provider-call-" + modelCalls.incrementAndGet());
+
+ assertThatThrownBy(() -> executeReference(model, call -> tools.incrementAndGet())).isSameAs(unavailable);
+ assertThat(tools.get()).isZero();
+
+ runtimeStore = store;
+ reload();
+ assertThatThrownBy(() -> executeReference(model, call -> tools.incrementAndGet()))
+ .isInstanceOf(ModelTurnJournalException.class).hasMessageContaining("re-asking is disabled");
+
+ assertThat(modelCalls.get()).isEqualTo(1);
+ assertThat(tools.get()).isZero();
+ assertThat(snapshot.journal()).containsOnlyKeys("agent:0#req", "agent:0#attempt.1");
+ }
+
+ @Test
+ void aLostResponseCanBeReaskedWhenExplicitlyEnabled() {
+ IllegalStateException unavailable = new IllegalStateException("response storage unavailable");
+ runtimeStore = new FailingRunStore(store, ModelTurnCodec.RESPONSE_TYPE, false, unavailable);
+ AtomicInteger modelCalls = new AtomicInteger();
+ AtomicInteger tools = new AtomicInteger();
+ Supplier model = () -> toolCallingResponse("provider-call-" + modelCalls.incrementAndGet());
+ ModelTurnAttemptPolicy policy = REASK_TWO_ATTEMPTS;
+
+ assertThatThrownBy(() -> executeReference(policy, model, call -> tools.incrementAndGet())).isSameAs(unavailable);
+ assertThat(tools.get()).isZero();
+
+ runtimeStore = store;
+ reload();
+ ChatResponse recovered = executeReference(policy, model, call -> tools.incrementAndGet());
+
+ assertThat(modelCalls.get()).isEqualTo(2);
+ assertThat(tools.get()).isEqualTo(1);
+ assertThat(recovered.toolCalls()).extracting(ToolCall::id).containsExactly("provider-call-2");
+ assertThat(snapshot.journal()).containsKeys("agent:0#attempt.1", "agent:0#attempt.2", "agent:0#task.0");
+ }
+
+ @Test
+ void aLegacyRequestCountsAsTheFirstReservation() {
+ commit(null, recorder.prepareRequest(TURN, REQUEST));
+ ModelTurnAttemptPolicy policy = REASK_TWO_ATTEMPTS;
+ List recovery = recorder.prepareAttempt(TURN, REQUEST, snapshot, policy);
+ assertThat(recovery).extracting(RunEntry::id).containsExactly("agent:0#attempt.2");
+
+ commit(null, recovery);
+ reload();
+
+ assertThatThrownBy(() -> recorder.prepareAttempt(TURN, REQUEST, snapshot, policy))
+ .isInstanceOf(ModelTurnJournalException.class).hasMessageContaining("budget exhausted");
+ }
+
+ @Test
+ void attemptLimitAloneDoesNotEnableRecoveryReasking() {
+ commit(null, recorder.prepareAttempt(TURN, REQUEST, snapshot));
+
+ assertThatThrownBy(() -> recorder.prepareAttempt(TURN, REQUEST, snapshot, 5))
+ .isInstanceOf(ModelTurnJournalException.class).hasMessageContaining("re-asking is disabled");
+ }
+
+ @Test
+ void differentTurnsHaveIndependentAttemptBudgets() {
+ commit(null, recorder.prepareAttempt(TURN, REQUEST, snapshot, 1));
+ ModelTurnRecorder.Turn next = new ModelTurnRecorder.Turn(RUN_ID, "next");
+
+ assertThat(recorder.prepareAttempt(next, REQUEST, snapshot, 1)).extracting(RunEntry::id)
+ .containsExactly("next#req", "next#attempt.1");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = { false, true })
+ void aDifferentRequestCannotReplaceAnInterruptedOrCompletedTurn(boolean completed) {
+ commit(null, recorder.prepareAttempt(TURN, REQUEST, snapshot));
+ if (completed) {
+ commit(checkpoint(), recorder.prepareResponse(TURN, RESPONSE, TOOLS));
+ }
+ ChatRequest changed = new ChatRequest(List.of(ChatMessage.user("different request")), TOOLS, null, Map.of());
+
+ assertThatThrownBy(() -> recorder.prepareAttempt(TURN, changed, snapshot))
+ .isInstanceOf(ModelTurnJournalException.class).hasCauseInstanceOf(RunStoreConflictException.class);
+ }
+
+ @Test
+ void replayRejectsIncompleteExtraOrMismatchedPlans() {
+ List complete = completeEntries();
+ RunEntry request = complete.get(0);
+ RunEntry response = complete.get(1);
+ RunEntry plan = complete.get(2);
+ RunEntry extra = new RunEntry(1, "agent:0#task.1", plan.kind(), plan.payload());
+ RunEntry changed = new RunEntry(1, plan.id(), plan.kind(), new RunPayload(1, ModelTurnCodec.PLAN_TYPE,
+ ModelTurnCodec.encodeToolPlan(new ToolCall("call-1", "search", "{\"q\":\"different\"}"),
+ TOOLS.get(0), 0)));
+ List> invalid = List.of(List.of(response, plan), List.of(request, plan),
+ List.of(request, response), List.of(request, response, plan, extra),
+ List.of(request, response, changed));
+
+ for (List entries : invalid) {
+ RunSnapshot recorded = snapshotWithJournal(entries);
+ assertThatThrownBy(() -> recorder.replay(TURN, REQUEST, recorded))
+ .isInstanceOf(ModelTurnJournalException.class);
+ }
+ }
+
+ @Test
+ void replayChecksEntryKindsTypesAndVersions() {
+ List complete = completeEntries();
+ complete.add(recorder.prepareAttempt(TURN, REQUEST, snapshot).get(1));
+ for (int index = 0; index < complete.size(); index++) {
+ RunEntry original = complete.get(index);
+ List incompatible = List.of(
+ new RunEntry(1, original.id(), RunEntry.Kind.EVENT, original.payload()),
+ new RunEntry(1, original.id(), original.kind(),
+ new RunPayload(1, "other.type", original.payload().json())),
+ new RunEntry(1, original.id(), original.kind(),
+ new RunPayload(2, original.payload().type(), original.payload().json())));
+ for (RunEntry replacement : incompatible) {
+ List entries = new ArrayList<>(complete);
+ entries.set(index, replacement);
+ RunSnapshot invalid = snapshotWithJournal(entries);
+ assertThatThrownBy(() -> recorder.replay(TURN, REQUEST, invalid))
+ .isInstanceOf(ModelTurnJournalException.class);
+ }
+ }
+ }
+
+ @Test
+ void inconsistentAttemptRecordsCannotResetTheBudget() {
+ RunEntry request = recorder.prepareRequest(TURN, REQUEST).get(0);
+ String hash = ModelTurnCodec.requestHash(REQUEST);
+ List invalid = List.of(
+ new RunEntry(1, "agent:0#attempt.3", RunEntry.Kind.MODEL_CALL,
+ new RunPayload(1, ModelTurnCodec.ATTEMPT_TYPE, ModelTurnCodec.encodeAttempt(3, hash))),
+ new RunEntry(1, "agent:0#attempt.1", RunEntry.Kind.MODEL_CALL,
+ new RunPayload(1, ModelTurnCodec.ATTEMPT_TYPE, ModelTurnCodec.encodeAttempt(1, "different"))));
+
+ for (RunEntry attempt : invalid) {
+ RunSnapshot recorded = snapshotWithJournal(List.of(request, attempt));
+ assertThatThrownBy(() -> recorder.prepareAttempt(TURN, REQUEST, recorded, 5))
+ .isInstanceOf(ModelTurnJournalException.class);
+ }
+ }
+
+ @Test
+ void theSameTaskWithDifferentArgumentsConflictsWithoutUpdatingTheCheckpoint() {
+ commit(null, recorder.prepareAttempt(TURN, REQUEST, snapshot));
+ commit(checkpoint(), recorder.prepareResponse(TURN, RESPONSE, TOOLS));
+ RunSnapshot before = snapshot;
+ RunEntry changed = new RunEntry(1, "agent:0#task.0", RunEntry.Kind.TASK_CALL,
+ new RunPayload(1, ModelTurnCodec.PLAN_TYPE, ModelTurnCodec.encodeToolPlan(
+ new ToolCall("call-2", "search", "{\"q\":\"different\"}"), TOOLS.get(0), 0)));
+
+ assertThatThrownBy(() -> commit(new RunPayload(1, "test.checkpoint", "{\"changed\":true}"),
+ List.of(changed))).isInstanceOf(RunStoreConflictException.class);
+ assertThat(snapshot).isSameAs(before);
+ assertThat(store.find(run.namespace(), RUN_ID)).contains(before);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = { "0", "01", "2147483648" })
+ void malformedReservationIdsRemainVisibleToBudgetValidation(String suffix) {
+ RunEntry request = recorder.prepareRequest(TURN, REQUEST).get(0);
+ RunEntry attempt = new RunEntry(1, TURN.turnId() + "#attempt." + suffix, RunEntry.Kind.MODEL_CALL,
+ new RunPayload(1, ModelTurnCodec.ATTEMPT_TYPE,
+ ModelTurnCodec.encodeAttempt(1, ModelTurnCodec.requestHash(REQUEST))));
+ RunSnapshot recorded = snapshotWithJournal(List.of(request, attempt));
+
+ assertThatThrownBy(() -> recorder.prepareAttempt(TURN, REQUEST, recorded, 5))
+ .isInstanceOf(ModelTurnJournalException.class);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = { "00", "2147483648" })
+ void malformedExtraPlanIdsRemainVisibleToReplay(String suffix) {
+ List entries = completeEntries();
+ RunEntry plan = entries.get(2);
+ entries.add(new RunEntry(1, TURN.turnId() + "#task." + suffix, plan.kind(), plan.payload()));
+ RunSnapshot recorded = snapshotWithJournal(entries);
+
+ assertThatThrownBy(() -> recorder.replayTurn(TURN, REQUEST, recorded))
+ .isInstanceOf(ModelTurnJournalException.class);
+ }
+
+ @Test
+ void nestedTurnsDoNotAffectReplayOrReservationCounts() {
+ ModelTurnRecorder.Turn nested = new ModelTurnRecorder.Turn(RUN_ID, TURN.turnId() + "#attempt.1#task.0");
+ commit(null, recorder.prepareAttempt(nested, REQUEST, snapshot));
+ commit(checkpoint(), recorder.prepareResponse(nested, RESPONSE, TOOLS));
+
+ assertThat(recorder.replayTurn(TURN, REQUEST, snapshot)).isEmpty();
+ assertThat(recorder.prepareAttempt(TURN, REQUEST, snapshot, 1)).extracting(RunEntry::id)
+ .containsExactly("agent:0#req", "agent:0#attempt.1");
+ assertThat(recorder.replayTurn(nested, REQUEST, snapshot).orElseThrow().plans().get(0).identity())
+ .isEqualTo(new ToolTaskIdentity(RUN_ID, nested.turnId(), 0));
+ }
+
+ @Test
+ void replayTurnExposesTheSameCommittedPlansOnInitialReadAndRecoveryWithoutWriting() {
+ assertThat(recorder.replayTurn(TURN, REQUEST, snapshot)).isEmpty();
+ List calls = List.of(new ToolCall("provider-first", "search", "{\"q\":\"same\"}"),
+ new ToolCall("provider-second", "search", "{\"q\":\"same\"}"));
+ ChatResponse response = new ChatResponse("response", "model", List.of(ChatMessage.assistant("", calls)),
+ calls, FinishReason.TOOL_CALLS, null, Map.of());
+ commit(null, recorder.prepareAttempt(TURN, REQUEST, snapshot));
+ assertThat(recorder.replayTurn(TURN, REQUEST, snapshot)).isEmpty();
+ commit(checkpoint(), recorder.prepareResponse(TURN, response, TOOLS));
+ RunSnapshot before = snapshot;
+
+ RecordedModelTurn initial = recorder.replayTurn(TURN, REQUEST, snapshot).orElseThrow();
+ reload();
+ RecordedModelTurn recovered = recorder.replayTurn(TURN, REQUEST, snapshot).orElseThrow();
+
+ assertThat(recovered).isEqualTo(initial);
+ assertThat(recorder.replay(TURN, REQUEST, snapshot)).contains(recovered.response());
+ assertThat(recovered.plans()).extracting(PlannedToolCall::identity).containsExactly(
+ new ToolTaskIdentity(RUN_ID, TURN.turnId(), 0), new ToolTaskIdentity(RUN_ID, TURN.turnId(), 1));
+ assertThat(recovered.plans().get(0).argumentsHash()).isEqualTo(recovered.plans().get(1).argumentsHash());
+ for (PlannedToolCall plan : recovered.plans()) {
+ assertThat(plan.arguments()).isEqualTo("{\"q\":\"same\"}");
+ assertThat(plan.definitionFingerprint()).isEqualTo(ModelTurnCodec.definitionVersion(TOOLS.get(0)));
+ assertThat(ModelTurnCodec.decodeToolPlan(TURN, snapshot.journal().get(plan.identity().entryId())))
+ .isEqualTo(plan);
+ assertThat(recovered.response().toolCalls().get(plan.identity().ordinal()))
+ .isEqualTo(calls.get(plan.identity().ordinal()));
+ }
+ assertThat(store.find(run.namespace(), RUN_ID)).contains(before);
+ assertThatThrownBy(() -> recovered.plans().clear()).isInstanceOf(UnsupportedOperationException.class);
+ }
+
+ @Test
+ void providerCallIdsAndAttemptReservationsDoNotChangePlanIdentity() {
+ List firstEntries = recorder.prepareResponse(TURN, toolCallingResponse("provider-one"), TOOLS);
+ PlannedToolCall first = ModelTurnCodec.decodeToolPlan(TURN, firstEntries.get(1));
+ commit(null, recorder.prepareAttempt(TURN, REQUEST, snapshot, REASK_TWO_ATTEMPTS));
+ reload();
+ commit(null, recorder.prepareAttempt(TURN, REQUEST, snapshot, REASK_TWO_ATTEMPTS));
+ commit(checkpoint(), recorder.prepareResponse(TURN, toolCallingResponse("provider-two"), TOOLS));
+
+ RecordedModelTurn recorded = recorder.replayTurn(TURN, REQUEST, snapshot).orElseThrow();
+
+ assertThat(recorded.plans()).containsExactly(first);
+ assertThat(recorded.response().toolCalls().get(0).id()).isEqualTo("provider-two");
+ }
+
+ @Test
+ void replayTurnRecordsUnknownToolsAndMalformedArgumentsWithoutAnExecutionGate() {
+ ToolCall call = new ToolCall("provider", "not-disclosed", "{ invalid JSON");
+ ChatResponse response = new ChatResponse("response", "model",
+ List.of(ChatMessage.assistant("", List.of(call))), List.of(call), FinishReason.TOOL_CALLS, null, Map.of());
+ commit(null, recorder.prepareAttempt(TURN, REQUEST, snapshot));
+ commit(checkpoint(), recorder.prepareResponse(TURN, response, REQUEST.tools()));
+
+ RecordedModelTurn recorded = recorder.replayTurn(TURN, REQUEST, snapshot).orElseThrow();
+
+ assertThat(recorded.plans()).hasSize(1);
+ assertThat(recorded.plans().get(0).toolName()).isEqualTo(call.name());
+ assertThat(recorded.plans().get(0).arguments()).isEqualTo(call.arguments());
+ assertThat(recorded.plans().get(0).definitionFingerprint()).isNull();
+ }
+
+ @Test
+ void textOnlyResultsReplayWithoutToolPlans() {
+ ChatResponse text = new ChatResponse("response", "model", List.of(ChatMessage.assistant("done")),
+ List.of(), FinishReason.STOP, null, Map.of());
+ AtomicInteger modelCalls = new AtomicInteger();
+ AtomicInteger tools = new AtomicInteger();
+ Supplier model = () -> {
+ modelCalls.incrementAndGet();
+ return text;
+ };
+
+ assertThat(executeReference(model, call -> tools.incrementAndGet())).isEqualTo(text);
+ assertThat(executeReference(model, call -> tools.incrementAndGet())).isEqualTo(text);
+ assertThat(modelCalls.get()).isEqualTo(1);
+ assertThat(tools.get()).isZero();
+ assertThat(recorder.replayTurn(TURN, REQUEST, snapshot).orElseThrow().plans()).isEmpty();
+ }
+
+ @Test
+ void diagnosticDeltasRetainTheCheckpointUntilTheTerminalResponse() {
+ RunPayload before = snapshot.checkpoint();
+ String attempt = ModelTurnRecorder.newAttempt();
+ AtomicInteger tools = new AtomicInteger();
+
+ executeReference(() -> {
+ commit(null, recorder.prepareDelta(TURN, attempt, 0, "hello"));
+ assertThat(snapshot.checkpoint()).isEqualTo(before);
+ assertThat(snapshot.journal()).doesNotContainKeys("agent:0#resp", "agent:0#task.0");
+ return RESPONSE;
+ }, call -> {
+ assertThat(snapshot.checkpoint()).isEqualTo(checkpoint());
+ tools.incrementAndGet();
+ });
+
+ assertThat(tools.get()).isEqualTo(1);
+ }
+
+ /** In-process reference ordering; production recovery and tool receipts remain runtime concerns. */
+ private ChatResponse executeReference(Supplier model, Consumer dispatch) {
+ return executeReference(ModelTurnAttemptPolicy.DEFAULT, model, dispatch);
+ }
+
+ private ChatResponse executeReference(int maxAttempts, Supplier model, Consumer dispatch) {
+ return executeReference(new ModelTurnAttemptPolicy(MissingResponsePolicy.FAIL_CLOSED, maxAttempts), model,
+ dispatch);
+ }
+
+ private ChatResponse executeReference(ModelTurnAttemptPolicy policy, Supplier model,
+ Consumer dispatch) {
+ ChatResponse response = recorder.replay(TURN, REQUEST, snapshot).orElseGet(() -> {
+ commit(null, recorder.prepareAttempt(TURN, REQUEST, snapshot, policy));
+ ChatResponse result = model.get();
+ commit(checkpoint(), recorder.prepareResponse(TURN, result, REQUEST.tools()));
+ return result;
+ });
+ response.toolCalls().forEach(dispatch);
+ return response;
+ }
+
+ private void commit(RunPayload checkpoint, List entries) {
+ snapshot = runtimeStore.commit(run.namespace(), RUN_ID, lease,
+ new RunCommit(snapshot.revision(), snapshot.state(), checkpoint, List.of(), entries, List.of()));
+ }
+
+ private void reload() {
+ snapshot = store.find(run.namespace(), RUN_ID).orElseThrow();
+ recorder = new ModelTurnRecorder();
+ }
+
+ private List completeEntries() {
+ List entries = new ArrayList<>(recorder.prepareRequest(TURN, REQUEST));
+ entries.addAll(recorder.prepareResponse(TURN, RESPONSE, TOOLS));
+ return entries;
+ }
+
+ private RunSnapshot snapshotWithJournal(List entries) {
+ Map journal = new HashMap<>();
+ entries.forEach(entry -> journal.put(entry.id(), entry));
+ return new RunSnapshot(snapshot.spec(), snapshot.revision(), snapshot.state(), snapshot.checkpoint(),
+ snapshot.tasks(), journal, snapshot.outbox());
+ }
+
+ private static RunPayload checkpoint() {
+ return new RunPayload(1, "test.checkpoint", "{\"phase\":\"TOOLS_PENDING\",\"turnId\":\"agent:0\"}");
+ }
+
+ private static Supplier countedModel(AtomicInteger calls) {
+ return () -> {
+ calls.incrementAndGet();
+ return RESPONSE;
+ };
+ }
+
+ private static ChatResponse toolCallingResponse(String callId) {
+ ToolCall call = new ToolCall(callId, "search", "{\"q\":\"argi\"}");
+ return new ChatResponse("response", "test-model", List.of(ChatMessage.assistant("", List.of(call))),
+ List.of(call), FinishReason.TOOL_CALLS, null, Map.of());
+ }
+
+ /** Simulates definite rejection or a successful write whose acknowledgement is lost. */
+ private record FailingRunStore(RunStore delegate, String payloadType, boolean commitBeforeFailure,
+ RuntimeException failure) implements RunStore {
+
+ @Override
+ public RunStoreCapabilities capabilities() {
+ return delegate.capabilities();
+ }
+
+ @Override
+ public RunSnapshot submit(RunSpec spec, TrustedExecutionContext context) {
+ return delegate.submit(spec, context);
+ }
+
+ @Override
+ public Optional find(RunNamespace namespace, String runId) {
+ return delegate.find(namespace, runId);
+ }
+
+ @Override
+ public Optional activeRun(RunNamespace namespace) {
+ return delegate.activeRun(namespace);
+ }
+
+ @Override
+ public RunLease acquire(RunNamespace namespace, String runId, UUID ownerId, Duration ttl) {
+ return delegate.acquire(namespace, runId, ownerId, ttl);
+ }
+
+ @Override
+ public RunLease renew(RunLease lease, Duration ttl) {
+ return delegate.renew(lease, ttl);
+ }
+
+ @Override
+ public boolean release(RunLease lease) {
+ return delegate.release(lease);
+ }
+
+ @Override
+ public RunSnapshot commit(RunNamespace namespace, String runId, RunLease lease, RunCommit commit) {
+ if (commit.journal().stream().anyMatch(entry -> payloadType.equals(entry.payload().type()))) {
+ if (commitBeforeFailure) {
+ delegate.commit(namespace, runId, lease, commit);
+ }
+ throw failure;
+ }
+ return delegate.commit(namespace, runId, lease, commit);
+ }
+ }
+
+}