Skip to content

Commit 2a39306

Browse files
ooplesfranklinicclaude
authored
feat(agentic): Phase 1 — typed durable graph runtime (epic #1544) (#1548)
* feat(agentic): Phase 1 — typed StateGraph runtime core (#1544) Introduces the typed graph runtime (AiDotNet.Agentic.Graph), the LangGraph-surpassing core: - StateGraph<TState> builder: AddNode (sync/async), AddEdge, AddConditionalEdges, SetEntryPoint, Compile() with validation (entry exists, edges resolve, a node can't have both a fixed edge and conditional edges). - CompiledStateGraph<TState>: InvokeAsync (final state) + StreamAsync (per-node GraphStepUpdate); fully typed state threading; cycles bounded by GraphRunOptions.MaxSteps (default 25) -> GraphRecursionException. - GraphSpecialNodes.End sentinel; conditional routers return a node name or End (validated at runtime). - 7 tests (linear, conditional cycle, streaming, recursion-limit, async node, compile validation, dup/reserved names) green on net10.0 + net471. No null-forgiving. Checkpointing/HITL/fan-out/subgraphs are the next slices. Part of epic #1544 (Phase 1). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * feat(agentic): Phase 1 — durable checkpointing, resume & time-travel (#1544) Adds durable execution to the graph runtime (AiDotNet.Agentic.Graph.Checkpointing): - GraphCheckpoint<TState> (threadId, checkpointId, step, nextNode, state, IsComplete); IGraphCheckpointer<TState> (Save/GetLatest/Get/GetHistory); InMemoryGraphCheckpointer<TState> (thread-safe). - CompiledStateGraph: checkpointed InvokeAsync(state, checkpointer, threadId) auto-resumes from the latest checkpoint (saves after every step; cumulative step budget); ResumeFromAsync(threadId, checkpointId) replays from any past checkpoint (time-travel). Shared RunCoreAsync powers plain + checkpointed runs. - 5 tests (fresh+history, resume-skips-earlier, re-invoke-completed, time-travel replay, unknown-checkpoint throws); 12 graph tests total green on net10.0 + net471. No null-forgiving. NEXT: HITL interrupts, channels/reducers+fan-out, DB-backed checkpointers. Part of epic #1544 (Phase 1). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * feat(agentic): Phase 1 — human-in-the-loop interrupts (#1544) Adds interrupt-before HITL to the graph runtime, built on checkpointing: - StateGraph.AddInterruptBefore(node) marks pause points (validated in Compile). - GraphRunResult<TState> (IsComplete/IsInterrupted/State/InterruptedBefore). - CompiledStateGraph.RunAsync(state, checkpointer, threadId, applyOnResume?): fresh run pauses just before the first interrupt node and returns Interrupted; re-invoking the same thread resumes past the pause (optionally applying the human's state edit) until the next interrupt or completion. RunCoreAsync now returns GraphRunResult and honors interrupts; plain/checkpointed InvokeAsync run straight through (no pause). - 4 HITL tests (pause+resume, human edit on resume, interrupt-before-entry, InvokeAsync-ignores-interrupts); 16 graph tests total green on net10.0 + net471. No null-forgiving. NEXT: channels/reducers + dynamic fan-out, DB-backed checkpointers, subgraphs. Part of epic #1544 (Phase 1). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * feat(agentic): Phase 1 — subgraphs, reward-gated edges, deterministic replay (#1544) - StateGraph.AddSubgraph(name, compiledGraph): run a whole graph as one step (composition). - StateGraph.AddRewardGatedEdges(from, reward, threshold, ifMeets, ifBelow): route by a critic/reward score (differentiator); sugar over conditional edges. - CompiledStateGraph.ReplayAsync + GetRecordedStateAsync: deterministically reproduce a recorded run's (node,state) trajectory from checkpoint history without executing nodes. - 3 tests; 19 graph tests total green on net10.0 + net471. No null-forgiving. Part of epic #1544 (Phase 1). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * feat(agentic): Phase 1 — dynamic fan-out / map-reduce nodes (#1544) StateGraph.AddFanOutNode<TItem,TResult>(map, branch, reduce, maxDegreeOfParallelism?): derives items from state, runs branches in parallel (optional SemaphoreSlim throttle), and reduces results back into the typed state — the typed equivalent of LangGraph Send/map-reduce, as a single composable node. 4 tests (map-reduce, throttled, empty, composed); 23 graph tests total green on net10.0 + net471. No null-forgiving. Part of epic #1544 (Phase 1). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * feat(agentic): Phase 1 — durable checkpointers (JSON-file + SQLite) (#1544) - JsonFileGraphCheckpointer<TState> (core, zero new deps): persists all threads' checkpoints to a JSON file for single-process durability across restarts. - SqliteGraphCheckpointer<TState> (opt-in AiDotNet.Storage.Sqlite): durable DB-backed checkpoints (table + JSON state column) for cross-process/cross-machine durability; keeps the SQLite dep out of core. Uses explicit arg checks (core's Guard is internal). - Durability tests prove a fresh checkpointer instance over the same store reads history + resumes. JSON test runs on both TFMs; SQLite test is net10-gated because the e_sqlite3 native lib isn't deployed to the net471 shadow-copy test host (pre-existing, repo-wide; also affects SqliteSqlSyntaxValidatorTests). 25 graph tests green net10.0 / 24 net471. No null-forgiving. Postgres/Redis checkpointers are interface-ready follow-ups (need live servers; not unit-testable in CI). Part of epic #1544 (Phase 1). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * fix(agentic): address PR #1548 review — checkpointer ordering/atomicity/corruption - IGraphCheckpointer.GetHistoryAsync: clarify the contract to chronological/append order (persistent sequence, not logical step) so replay/time-travel is consistent across backends. - InMemoryGraphCheckpointer.GetAsync: return the NEWEST matching checkpoint (LastOrDefault), aligning with the durable backends. - JsonFileGraphCheckpointer.Load: throw a descriptive InvalidOperationException on a corrupt file (with the path) instead of silently returning empty (which would let Save overwrite and lose recoverable data). - JsonFileGraphCheckpointer.Persist: atomic write — temp file + flush-to-disk + atomic replace (File.Move overwrite on net5+, File.Replace/Move fallback on net471), with temp cleanup. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * fix(agentic): address PR #1548 review — MaxSteps validation, Sqlite null-state, test cleanup - GraphRunOptions.MaxSteps: validate > 0 in the setter (ArgumentOutOfRangeException) instead of allowing zero/negative. - SqliteGraphCheckpointer.ReadRow: throw a descriptive InvalidOperationException (with checkpointId/threadId) when the persisted state deserializes to null, rather than passing null into GraphCheckpoint. (Constructor null-checks kept as-is: Guard is internal to core and not accessible from the opt-in AiDotNet.Storage.Sqlite project.) - StateGraphTests: drop the no-op `await Task.CompletedTask;` from two synchronous tests by making them non-async Task methods returning Task.CompletedTask (preserves [Fact(Timeout)], avoids the CS1998 that deleting the await alone would introduce). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: franklinic <franklin@ivorycloud.com> Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
1 parent 29e1bea commit 2a39306

18 files changed

Lines changed: 1974 additions & 0 deletions
Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
namespace AiDotNet.Agentic.Graph.Checkpointing;
2+
3+
/// <summary>
4+
/// An immutable snapshot of a graph run at a point in time: which node runs next and the state as of
5+
/// that point. Saved after each step so a run can be resumed, inspected, or replayed.
6+
/// </summary>
7+
/// <typeparam name="TState">The graph's state type.</typeparam>
8+
/// <remarks>
9+
/// <para>
10+
/// A checkpoint records the <see cref="NextNode"/> (where execution will continue) plus the
11+
/// <see cref="State"/> at that boundary, tagged with a <see cref="ThreadId"/> (the run/conversation) and a
12+
/// monotonically increasing <see cref="Step"/>. The sequence of checkpoints for a thread is its history —
13+
/// the basis for durable resume and time-travel.
14+
/// </para>
15+
/// <para><b>For Beginners:</b> Like a save-game. It remembers exactly where the graph was about to go
16+
/// next and what the data looked like, so you can stop and pick up later (or rewind to an earlier save).
17+
/// </para>
18+
/// </remarks>
19+
public sealed class GraphCheckpoint<TState>
20+
{
21+
/// <summary>
22+
/// Initializes a new checkpoint.
23+
/// </summary>
24+
/// <param name="threadId">The run/thread this checkpoint belongs to.</param>
25+
/// <param name="checkpointId">A unique id for this checkpoint within the thread.</param>
26+
/// <param name="step">The zero-based step index (monotonically increasing within a thread).</param>
27+
/// <param name="nextNode">The node that will execute next (or <see cref="GraphSpecialNodes.End"/> when complete).</param>
28+
/// <param name="state">The state at this boundary.</param>
29+
/// <exception cref="ArgumentNullException">Thrown when a required string is <c>null</c>.</exception>
30+
/// <exception cref="ArgumentException">Thrown when a required string is empty/whitespace.</exception>
31+
/// <exception cref="ArgumentOutOfRangeException">Thrown when <paramref name="step"/> is negative.</exception>
32+
public GraphCheckpoint(string threadId, string checkpointId, int step, string nextNode, TState state)
33+
{
34+
Guard.NotNullOrWhiteSpace(threadId);
35+
Guard.NotNullOrWhiteSpace(checkpointId);
36+
Guard.NotNullOrWhiteSpace(nextNode);
37+
Guard.NonNegative(step);
38+
ThreadId = threadId;
39+
CheckpointId = checkpointId;
40+
Step = step;
41+
NextNode = nextNode;
42+
State = state;
43+
}
44+
45+
/// <summary>Gets the run/thread this checkpoint belongs to.</summary>
46+
public string ThreadId { get; }
47+
48+
/// <summary>Gets the unique id of this checkpoint within the thread.</summary>
49+
public string CheckpointId { get; }
50+
51+
/// <summary>Gets the zero-based, monotonically increasing step index.</summary>
52+
public int Step { get; }
53+
54+
/// <summary>Gets the node that will execute next (or <see cref="GraphSpecialNodes.End"/> when the run is complete).</summary>
55+
public string NextNode { get; }
56+
57+
/// <summary>Gets the state captured at this boundary.</summary>
58+
public TState State { get; }
59+
60+
/// <summary>Gets a value indicating whether this checkpoint represents a completed run.</summary>
61+
public bool IsComplete => NextNode == GraphSpecialNodes.End;
62+
}
Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
namespace AiDotNet.Agentic.Graph.Checkpointing;
2+
3+
/// <summary>
4+
/// Persists and retrieves <see cref="GraphCheckpoint{TState}"/>s, enabling durable resume and time-travel
5+
/// for graph runs. Implementations back onto memory, SQLite, Postgres, Redis, etc.
6+
/// </summary>
7+
/// <typeparam name="TState">The graph's state type.</typeparam>
8+
/// <remarks>
9+
/// <para><b>For Beginners:</b> The save-game store. The graph asks it to save a checkpoint after each
10+
/// step, to fetch the latest checkpoint when resuming, and to list the full history when rewinding.
11+
/// </para>
12+
/// </remarks>
13+
public interface IGraphCheckpointer<TState>
14+
{
15+
/// <summary>
16+
/// Saves (appends) a checkpoint.
17+
/// </summary>
18+
/// <param name="checkpoint">The checkpoint to persist.</param>
19+
/// <param name="cancellationToken">Token used to cancel the operation.</param>
20+
Task SaveAsync(GraphCheckpoint<TState> checkpoint, CancellationToken cancellationToken = default);
21+
22+
/// <summary>
23+
/// Gets the most recent checkpoint for a thread, or <c>null</c> if the thread has none.
24+
/// </summary>
25+
/// <param name="threadId">The run/thread id.</param>
26+
/// <param name="cancellationToken">Token used to cancel the operation.</param>
27+
/// <returns>The latest checkpoint, or <c>null</c>.</returns>
28+
Task<GraphCheckpoint<TState>?> GetLatestAsync(string threadId, CancellationToken cancellationToken = default);
29+
30+
/// <summary>
31+
/// Gets a specific checkpoint by id (used for time-travel / replay from a past point).
32+
/// </summary>
33+
/// <param name="threadId">The run/thread id.</param>
34+
/// <param name="checkpointId">The checkpoint id.</param>
35+
/// <param name="cancellationToken">Token used to cancel the operation.</param>
36+
/// <returns>The checkpoint, or <c>null</c> if not found.</returns>
37+
Task<GraphCheckpoint<TState>?> GetAsync(string threadId, string checkpointId, CancellationToken cancellationToken = default);
38+
39+
/// <summary>
40+
/// Gets the full <see cref="GraphCheckpoint{TState}"/> history for <paramref name="threadId"/> in
41+
/// append (chronological) order — i.e. the order checkpoints were saved, by a persistent monotonic
42+
/// sequence (e.g. the SQLite backend orders by <c>Seq ASC</c>), NOT by logical step number.
43+
/// </summary>
44+
/// <remarks>
45+
/// Replay/time-travel relies on this chronological ordering: logical step values can repeat or branch
46+
/// (loops, re-runs from an earlier checkpoint), so ordering by step would diverge across backends.
47+
/// Implementations MUST return checkpoints oldest-first by save order.
48+
/// </remarks>
49+
/// <param name="threadId">The run/thread id.</param>
50+
/// <param name="cancellationToken">Token used to cancel the operation.</param>
51+
/// <returns>The chronologically-ordered history (possibly empty).</returns>
52+
Task<IReadOnlyList<GraphCheckpoint<TState>>> GetHistoryAsync(string threadId, CancellationToken cancellationToken = default);
53+
}
Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
1+
namespace AiDotNet.Agentic.Graph.Checkpointing;
2+
3+
/// <summary>
4+
/// An in-memory <see cref="IGraphCheckpointer{TState}"/> that keeps each thread's checkpoint history in a
5+
/// dictionary. Suitable for tests, single-process runs, and as the default when no durable store is wired.
6+
/// </summary>
7+
/// <typeparam name="TState">The graph's state type.</typeparam>
8+
/// <remarks>
9+
/// <para>
10+
/// Thread-safe via a per-instance lock. Because it stores the state object by reference, callers that use
11+
/// a mutable reference type for <typeparamref name="TState"/> should treat the state as immutable per step
12+
/// (return a new/copied instance from nodes) to get true snapshots; durable checkpointers serialize and so
13+
/// snapshot inherently.
14+
/// </para>
15+
/// <para><b>For Beginners:</b> Keeps your save-games in memory. Great for tests and short-lived runs; for
16+
/// durability across process restarts, use a database-backed checkpointer (coming in later slices).
17+
/// </para>
18+
/// </remarks>
19+
public sealed class InMemoryGraphCheckpointer<TState> : IGraphCheckpointer<TState>
20+
{
21+
private readonly object _gate = new();
22+
private readonly Dictionary<string, List<GraphCheckpoint<TState>>> _byThread = new(StringComparer.Ordinal);
23+
24+
/// <inheritdoc/>
25+
public Task SaveAsync(GraphCheckpoint<TState> checkpoint, CancellationToken cancellationToken = default)
26+
{
27+
Guard.NotNull(checkpoint);
28+
lock (_gate)
29+
{
30+
if (!_byThread.TryGetValue(checkpoint.ThreadId, out var list))
31+
{
32+
list = new List<GraphCheckpoint<TState>>();
33+
_byThread[checkpoint.ThreadId] = list;
34+
}
35+
36+
list.Add(checkpoint);
37+
}
38+
39+
return Task.CompletedTask;
40+
}
41+
42+
/// <inheritdoc/>
43+
public Task<GraphCheckpoint<TState>?> GetLatestAsync(string threadId, CancellationToken cancellationToken = default)
44+
{
45+
Guard.NotNull(threadId);
46+
lock (_gate)
47+
{
48+
if (_byThread.TryGetValue(threadId, out var list) && list.Count > 0)
49+
{
50+
return Task.FromResult<GraphCheckpoint<TState>?>(list[list.Count - 1]);
51+
}
52+
}
53+
54+
return Task.FromResult<GraphCheckpoint<TState>?>(null);
55+
}
56+
57+
/// <inheritdoc/>
58+
public Task<GraphCheckpoint<TState>?> GetAsync(string threadId, string checkpointId, CancellationToken cancellationToken = default)
59+
{
60+
Guard.NotNull(threadId);
61+
Guard.NotNull(checkpointId);
62+
lock (_gate)
63+
{
64+
if (_byThread.TryGetValue(threadId, out var list))
65+
{
66+
// Return the NEWEST checkpoint with this id (history is append-ordered), matching the durable
67+
// backends which order by descending sequence — so re-saving an id resolves to the latest.
68+
var match = list.LastOrDefault(cp => cp.CheckpointId == checkpointId);
69+
return Task.FromResult<GraphCheckpoint<TState>?>(match);
70+
}
71+
}
72+
73+
return Task.FromResult<GraphCheckpoint<TState>?>(null);
74+
}
75+
76+
/// <inheritdoc/>
77+
public Task<IReadOnlyList<GraphCheckpoint<TState>>> GetHistoryAsync(string threadId, CancellationToken cancellationToken = default)
78+
{
79+
Guard.NotNull(threadId);
80+
lock (_gate)
81+
{
82+
IReadOnlyList<GraphCheckpoint<TState>> history = _byThread.TryGetValue(threadId, out var list)
83+
? list.ToList()
84+
: new List<GraphCheckpoint<TState>>();
85+
return Task.FromResult(history);
86+
}
87+
}
88+
}

0 commit comments

Comments
 (0)