diff --git a/src/Agentic/Graph/Checkpointing/GraphCheckpoint.cs b/src/Agentic/Graph/Checkpointing/GraphCheckpoint.cs new file mode 100644 index 0000000000..1f8f504ec6 --- /dev/null +++ b/src/Agentic/Graph/Checkpointing/GraphCheckpoint.cs @@ -0,0 +1,62 @@ +namespace AiDotNet.Agentic.Graph.Checkpointing; + +/// +/// An immutable snapshot of a graph run at a point in time: which node runs next and the state as of +/// that point. Saved after each step so a run can be resumed, inspected, or replayed. +/// +/// The graph's state type. +/// +/// +/// A checkpoint records the (where execution will continue) plus the +/// at that boundary, tagged with a (the run/conversation) and a +/// monotonically increasing . The sequence of checkpoints for a thread is its history — +/// the basis for durable resume and time-travel. +/// +/// For Beginners: Like a save-game. It remembers exactly where the graph was about to go +/// next and what the data looked like, so you can stop and pick up later (or rewind to an earlier save). +/// +/// +public sealed class GraphCheckpoint +{ + /// + /// Initializes a new checkpoint. + /// + /// The run/thread this checkpoint belongs to. + /// A unique id for this checkpoint within the thread. + /// The zero-based step index (monotonically increasing within a thread). + /// The node that will execute next (or when complete). + /// The state at this boundary. + /// Thrown when a required string is null. + /// Thrown when a required string is empty/whitespace. + /// Thrown when is negative. + public GraphCheckpoint(string threadId, string checkpointId, int step, string nextNode, TState state) + { + Guard.NotNullOrWhiteSpace(threadId); + Guard.NotNullOrWhiteSpace(checkpointId); + Guard.NotNullOrWhiteSpace(nextNode); + Guard.NonNegative(step); + ThreadId = threadId; + CheckpointId = checkpointId; + Step = step; + NextNode = nextNode; + State = state; + } + + /// Gets the run/thread this checkpoint belongs to. + public string ThreadId { get; } + + /// Gets the unique id of this checkpoint within the thread. + public string CheckpointId { get; } + + /// Gets the zero-based, monotonically increasing step index. + public int Step { get; } + + /// Gets the node that will execute next (or when the run is complete). + public string NextNode { get; } + + /// Gets the state captured at this boundary. + public TState State { get; } + + /// Gets a value indicating whether this checkpoint represents a completed run. + public bool IsComplete => NextNode == GraphSpecialNodes.End; +} diff --git a/src/Agentic/Graph/Checkpointing/IGraphCheckpointer.cs b/src/Agentic/Graph/Checkpointing/IGraphCheckpointer.cs new file mode 100644 index 0000000000..8521f40743 --- /dev/null +++ b/src/Agentic/Graph/Checkpointing/IGraphCheckpointer.cs @@ -0,0 +1,53 @@ +namespace AiDotNet.Agentic.Graph.Checkpointing; + +/// +/// Persists and retrieves s, enabling durable resume and time-travel +/// for graph runs. Implementations back onto memory, SQLite, Postgres, Redis, etc. +/// +/// The graph's state type. +/// +/// For Beginners: The save-game store. The graph asks it to save a checkpoint after each +/// step, to fetch the latest checkpoint when resuming, and to list the full history when rewinding. +/// +/// +public interface IGraphCheckpointer +{ + /// + /// Saves (appends) a checkpoint. + /// + /// The checkpoint to persist. + /// Token used to cancel the operation. + Task SaveAsync(GraphCheckpoint checkpoint, CancellationToken cancellationToken = default); + + /// + /// Gets the most recent checkpoint for a thread, or null if the thread has none. + /// + /// The run/thread id. + /// Token used to cancel the operation. + /// The latest checkpoint, or null. + Task?> GetLatestAsync(string threadId, CancellationToken cancellationToken = default); + + /// + /// Gets a specific checkpoint by id (used for time-travel / replay from a past point). + /// + /// The run/thread id. + /// The checkpoint id. + /// Token used to cancel the operation. + /// The checkpoint, or null if not found. + Task?> GetAsync(string threadId, string checkpointId, CancellationToken cancellationToken = default); + + /// + /// Gets the full history for in + /// append (chronological) order — i.e. the order checkpoints were saved, by a persistent monotonic + /// sequence (e.g. the SQLite backend orders by Seq ASC), NOT by logical step number. + /// + /// + /// Replay/time-travel relies on this chronological ordering: logical step values can repeat or branch + /// (loops, re-runs from an earlier checkpoint), so ordering by step would diverge across backends. + /// Implementations MUST return checkpoints oldest-first by save order. + /// + /// The run/thread id. + /// Token used to cancel the operation. + /// The chronologically-ordered history (possibly empty). + Task>> GetHistoryAsync(string threadId, CancellationToken cancellationToken = default); +} diff --git a/src/Agentic/Graph/Checkpointing/InMemoryGraphCheckpointer.cs b/src/Agentic/Graph/Checkpointing/InMemoryGraphCheckpointer.cs new file mode 100644 index 0000000000..81c17cdaf5 --- /dev/null +++ b/src/Agentic/Graph/Checkpointing/InMemoryGraphCheckpointer.cs @@ -0,0 +1,88 @@ +namespace AiDotNet.Agentic.Graph.Checkpointing; + +/// +/// An in-memory that keeps each thread's checkpoint history in a +/// dictionary. Suitable for tests, single-process runs, and as the default when no durable store is wired. +/// +/// The graph's state type. +/// +/// +/// Thread-safe via a per-instance lock. Because it stores the state object by reference, callers that use +/// a mutable reference type for should treat the state as immutable per step +/// (return a new/copied instance from nodes) to get true snapshots; durable checkpointers serialize and so +/// snapshot inherently. +/// +/// For Beginners: Keeps your save-games in memory. Great for tests and short-lived runs; for +/// durability across process restarts, use a database-backed checkpointer (coming in later slices). +/// +/// +public sealed class InMemoryGraphCheckpointer : IGraphCheckpointer +{ + private readonly object _gate = new(); + private readonly Dictionary>> _byThread = new(StringComparer.Ordinal); + + /// + public Task SaveAsync(GraphCheckpoint checkpoint, CancellationToken cancellationToken = default) + { + Guard.NotNull(checkpoint); + lock (_gate) + { + if (!_byThread.TryGetValue(checkpoint.ThreadId, out var list)) + { + list = new List>(); + _byThread[checkpoint.ThreadId] = list; + } + + list.Add(checkpoint); + } + + return Task.CompletedTask; + } + + /// + public Task?> GetLatestAsync(string threadId, CancellationToken cancellationToken = default) + { + Guard.NotNull(threadId); + lock (_gate) + { + if (_byThread.TryGetValue(threadId, out var list) && list.Count > 0) + { + return Task.FromResult?>(list[list.Count - 1]); + } + } + + return Task.FromResult?>(null); + } + + /// + public Task?> GetAsync(string threadId, string checkpointId, CancellationToken cancellationToken = default) + { + Guard.NotNull(threadId); + Guard.NotNull(checkpointId); + lock (_gate) + { + if (_byThread.TryGetValue(threadId, out var list)) + { + // Return the NEWEST checkpoint with this id (history is append-ordered), matching the durable + // backends which order by descending sequence — so re-saving an id resolves to the latest. + var match = list.LastOrDefault(cp => cp.CheckpointId == checkpointId); + return Task.FromResult?>(match); + } + } + + return Task.FromResult?>(null); + } + + /// + public Task>> GetHistoryAsync(string threadId, CancellationToken cancellationToken = default) + { + Guard.NotNull(threadId); + lock (_gate) + { + IReadOnlyList> history = _byThread.TryGetValue(threadId, out var list) + ? list.ToList() + : new List>(); + return Task.FromResult(history); + } + } +} diff --git a/src/Agentic/Graph/Checkpointing/JsonFileGraphCheckpointer.cs b/src/Agentic/Graph/Checkpointing/JsonFileGraphCheckpointer.cs new file mode 100644 index 0000000000..07905b357a --- /dev/null +++ b/src/Agentic/Graph/Checkpointing/JsonFileGraphCheckpointer.cs @@ -0,0 +1,196 @@ +using System.IO; +using Newtonsoft.Json; + +namespace AiDotNet.Agentic.Graph.Checkpointing; + +/// +/// A durable that persists all threads' checkpoints to a single +/// JSON file, so runs survive process restarts. Zero extra dependencies (uses Newtonsoft.Json). +/// +/// The graph's state type (must be JSON-serializable). +/// +/// +/// Simple and self-contained: every save reads, mutates, and rewrites the file under an in-process lock, +/// so it is correct for single-process durability but not tuned for high-throughput or multi-process +/// concurrent writers. For those, use a database-backed checkpointer (e.g., SQLite in +/// AiDotNet.Storage.Sqlite). +/// +/// For Beginners: Saves your graph's progress to a file on disk, so if the app restarts you +/// can resume right where it left off. +/// +/// +public sealed class JsonFileGraphCheckpointer : IGraphCheckpointer +{ + private readonly object _gate = new(); + private readonly string _path; + + /// + /// Initializes the checkpointer, creating the containing directory if needed. + /// + /// The JSON file used to persist checkpoints. + /// Thrown when is null. + /// Thrown when is empty/whitespace. + public JsonFileGraphCheckpointer(string filePath) + { + Guard.NotNullOrWhiteSpace(filePath); + _path = filePath; + var dir = Path.GetDirectoryName(filePath); + if (!string.IsNullOrEmpty(dir) && !Directory.Exists(dir)) + { + Directory.CreateDirectory(dir); + } + } + + /// + public Task SaveAsync(GraphCheckpoint checkpoint, CancellationToken cancellationToken = default) + { + Guard.NotNull(checkpoint); + lock (_gate) + { + var data = Load(); + if (!data.TryGetValue(checkpoint.ThreadId, out var list)) + { + list = new List>(); + data[checkpoint.ThreadId] = list; + } + + list.Add(checkpoint); + Persist(data); + } + + return Task.CompletedTask; + } + + /// + public Task?> GetLatestAsync(string threadId, CancellationToken cancellationToken = default) + { + Guard.NotNull(threadId); + lock (_gate) + { + var data = Load(); + if (data.TryGetValue(threadId, out var list) && list.Count > 0) + { + return Task.FromResult?>(list[list.Count - 1]); + } + } + + return Task.FromResult?>(null); + } + + /// + public Task?> GetAsync(string threadId, string checkpointId, CancellationToken cancellationToken = default) + { + Guard.NotNull(threadId); + Guard.NotNull(checkpointId); + lock (_gate) + { + var data = Load(); + if (data.TryGetValue(threadId, out var list)) + { + foreach (var cp in list) + { + if (cp.CheckpointId == checkpointId) + { + return Task.FromResult?>(cp); + } + } + } + } + + return Task.FromResult?>(null); + } + + /// + public Task>> GetHistoryAsync(string threadId, CancellationToken cancellationToken = default) + { + Guard.NotNull(threadId); + lock (_gate) + { + var data = Load(); + IReadOnlyList> history = data.TryGetValue(threadId, out var list) + ? list.ToList() + : new List>(); + return Task.FromResult(history); + } + } + + private Dictionary>> Load() + { + if (!File.Exists(_path)) + { + return new Dictionary>>(StringComparer.Ordinal); + } + + var json = File.ReadAllText(_path); + if (string.IsNullOrWhiteSpace(json)) + { + return new Dictionary>>(StringComparer.Ordinal); + } + + try + { + var data = JsonConvert.DeserializeObject>>>(json); + return data ?? new Dictionary>>(StringComparer.Ordinal); + } + catch (JsonException ex) + { + // Surface corruption explicitly rather than silently returning empty — silently dropping the + // history would let the next Save overwrite (and permanently lose) recoverable checkpoint data. + throw new InvalidOperationException( + $"The checkpoint file at '{_path}' is corrupt and could not be parsed. " + + "Move or delete it to start a fresh checkpoint store.", ex); + } + } + + private void Persist(Dictionary>> data) + { + var json = JsonConvert.SerializeObject(data, Formatting.None); + + // Atomic write: serialize to a temp file in the same directory, flush it to disk, then atomically + // replace the target. A crash mid-write therefore leaves the previous (valid) file intact rather than + // a half-written, unparseable one. + var directory = Path.GetDirectoryName(_path); + var tempPath = Path.Combine( + string.IsNullOrEmpty(directory) ? "." : directory, + $".{Path.GetFileName(_path)}.{Guid.NewGuid():N}.tmp"); + + try + { + using (var stream = new FileStream(tempPath, FileMode.Create, FileAccess.Write, FileShare.None)) + using (var writer = new StreamWriter(stream)) + { + writer.Write(json); + writer.Flush(); + stream.Flush(flushToDisk: true); + } + +#if NET5_0_OR_GREATER + File.Move(tempPath, _path, overwrite: true); +#else + if (File.Exists(_path)) + { + // File.Replace is atomic when the destination exists; it also clears the temp file. + File.Replace(tempPath, _path, destinationBackupFileName: null); + } + else + { + File.Move(tempPath, _path); + } +#endif + } + finally + { + if (File.Exists(tempPath)) + { + try + { + File.Delete(tempPath); + } + catch (IOException) + { + // Best-effort cleanup of the temp file; a leftover .tmp is harmless. + } + } + } + } +} diff --git a/src/Agentic/Graph/CompiledStateGraph.cs b/src/Agentic/Graph/CompiledStateGraph.cs new file mode 100644 index 0000000000..09e0214c74 --- /dev/null +++ b/src/Agentic/Graph/CompiledStateGraph.cs @@ -0,0 +1,381 @@ +using System.Runtime.CompilerServices; +using AiDotNet.Agentic.Graph.Checkpointing; + +namespace AiDotNet.Agentic.Graph; + +/// +/// An executable, validated state graph produced by . Runs nodes +/// according to the graph's edges, threading the state from one node to the next until it reaches the end. +/// +/// The graph's state type (threaded through every node). +/// +/// +/// Execution starts at the entry node. After a node runs, the next node is chosen by: its conditional +/// router (if any), else its fixed edge, else the run ends. Cycles are allowed and bounded by +/// . Use for the final state or +/// to observe each step. +/// +/// For Beginners: This is the "compiled program" form of your graph. You give it a starting +/// state and it walks the nodes you wired up, handing each node the latest state, until it reaches the +/// end — then returns the final state. +/// +/// +public sealed class CompiledStateGraph +{ + private readonly IReadOnlyDictionary>> _nodes; + private readonly IReadOnlyDictionary _edges; + private readonly IReadOnlyDictionary> _conditionalEdges; + private readonly HashSet _interruptBefore; + private readonly string _entryPoint; + + internal CompiledStateGraph( + IReadOnlyDictionary>> nodes, + IReadOnlyDictionary edges, + IReadOnlyDictionary> conditionalEdges, + HashSet interruptBefore, + string entryPoint) + { + _nodes = nodes; + _edges = edges; + _conditionalEdges = conditionalEdges; + _interruptBefore = interruptBefore; + _entryPoint = entryPoint; + } + + /// + /// Runs the graph from the supplied initial state and returns the final state. + /// + /// The starting state. + /// Run options (step budget). null uses defaults. + /// Token used to cancel the run. + /// The final state when flow reaches the end node. + /// Thrown when the step budget is exceeded. + public async Task InvokeAsync( + TState initialState, + GraphRunOptions? options = null, + CancellationToken cancellationToken = default) + { + var result = await RunCoreAsync( + initialState, _entryPoint, startStep: 0, ResolveMaxSteps(options), + checkpointer: null, threadId: null, honorInterrupts: false, honorFirstInterrupt: false, cancellationToken).ConfigureAwait(false); + return result.State; + } + + /// + /// Runs the graph with durable checkpointing under a thread id, resuming automatically from the latest + /// saved checkpoint if one exists for that thread (otherwise starting fresh from ). + /// + /// The starting state (used only when the thread has no prior checkpoint). + /// The checkpoint store. + /// The run/thread id under which checkpoints are saved and resumed. + /// Run options (step budget). null uses defaults. The budget is cumulative across resumes. + /// Token used to cancel the run. + /// The final state. + /// Thrown when or is null. + /// Thrown when the step budget is exceeded. + public async Task InvokeAsync( + TState initialState, + IGraphCheckpointer checkpointer, + string threadId, + GraphRunOptions? options = null, + CancellationToken cancellationToken = default) + { + Guard.NotNull(checkpointer); + Guard.NotNullOrWhiteSpace(threadId); + var maxSteps = ResolveMaxSteps(options); + + var latest = await checkpointer.GetLatestAsync(threadId, cancellationToken).ConfigureAwait(false); + string startNode; + TState state; + int startStep; + if (latest is not null) + { + startNode = latest.NextNode; + state = latest.State; + startStep = latest.Step; + } + else + { + startNode = _entryPoint; + state = initialState; + startStep = 0; + await checkpointer.SaveAsync( + new GraphCheckpoint(threadId, $"{threadId}-0", 0, _entryPoint, initialState), cancellationToken).ConfigureAwait(false); + } + + if (startNode == GraphSpecialNodes.End) + { + return state; // thread already complete + } + + var result = await RunCoreAsync( + state, startNode, startStep, maxSteps, checkpointer, threadId, + honorInterrupts: false, honorFirstInterrupt: false, cancellationToken).ConfigureAwait(false); + return result.State; + } + + /// + /// Resumes (replays) a thread from a specific past checkpoint — the basis for time-travel — continuing + /// execution from that checkpoint's next node and state, and saving new checkpoints as it goes. + /// + /// The checkpoint store. + /// The run/thread id. + /// The id of the checkpoint to resume from. + /// Run options (step budget). null uses defaults. + /// Token used to cancel the run. + /// The final state. + /// Thrown when , , or is null. + /// Thrown when the checkpoint does not exist. + public async Task ResumeFromAsync( + IGraphCheckpointer checkpointer, + string threadId, + string checkpointId, + GraphRunOptions? options = null, + CancellationToken cancellationToken = default) + { + Guard.NotNull(checkpointer); + Guard.NotNullOrWhiteSpace(threadId); + Guard.NotNullOrWhiteSpace(checkpointId); + + var checkpoint = await checkpointer.GetAsync(threadId, checkpointId, cancellationToken).ConfigureAwait(false); + if (checkpoint is null) + { + throw new InvalidOperationException($"Checkpoint '{checkpointId}' was not found for thread '{threadId}'."); + } + + if (checkpoint.NextNode == GraphSpecialNodes.End) + { + return checkpoint.State; + } + + var result = await RunCoreAsync( + checkpoint.State, checkpoint.NextNode, checkpoint.Step, ResolveMaxSteps(options), checkpointer, threadId, + honorInterrupts: false, honorFirstInterrupt: false, cancellationToken).ConfigureAwait(false); + return result.State; + } + + /// + /// Runs the graph with human-in-the-loop interrupts under a thread id. On the first call it runs from + /// the entry node and pauses just before the first interrupt node (returning an interrupted result); + /// calling it again on the same thread resumes past that pause and runs until the next interrupt or + /// completion. Optionally applies the human's edit to the state when resuming. + /// + /// The starting state (used only when the thread has no prior checkpoint). + /// The checkpoint store. + /// The run/thread id. + /// When resuming, an optional transform applied to the saved state (the human's input). Ignored on a fresh run. + /// Run options (step budget). null uses defaults. + /// Token used to cancel the run. + /// A indicating completion or an interrupt point. + /// Thrown when or is null. + public async Task> RunAsync( + TState initialState, + IGraphCheckpointer checkpointer, + string threadId, + Func? applyOnResume = null, + GraphRunOptions? options = null, + CancellationToken cancellationToken = default) + { + Guard.NotNull(checkpointer); + Guard.NotNullOrWhiteSpace(threadId); + var maxSteps = ResolveMaxSteps(options); + + var latest = await checkpointer.GetLatestAsync(threadId, cancellationToken).ConfigureAwait(false); + string startNode; + TState state; + int startStep; + bool honorFirstInterrupt; + if (latest is not null) + { + // Resuming: the caller chose to continue, so do not re-pause at the node we stopped before. + startNode = latest.NextNode; + state = applyOnResume is not null ? applyOnResume(latest.State) : latest.State; + startStep = latest.Step; + honorFirstInterrupt = false; + } + else + { + startNode = _entryPoint; + state = initialState; + startStep = 0; + honorFirstInterrupt = true; + await checkpointer.SaveAsync( + new GraphCheckpoint(threadId, $"{threadId}-0", 0, _entryPoint, initialState), cancellationToken).ConfigureAwait(false); + } + + if (startNode == GraphSpecialNodes.End) + { + return GraphRunResult.Complete(state); + } + + return await RunCoreAsync( + state, startNode, startStep, maxSteps, checkpointer, threadId, + honorInterrupts: true, honorFirstInterrupt: honorFirstInterrupt, cancellationToken).ConfigureAwait(false); + } + + private async Task> RunCoreAsync( + TState state, + string startNode, + int startStep, + int maxSteps, + IGraphCheckpointer? checkpointer, + string? threadId, + bool honorInterrupts, + bool honorFirstInterrupt, + CancellationToken cancellationToken) + { + var current = startNode; + var step = startStep; + var firstNode = true; + + while (current != GraphSpecialNodes.End) + { + cancellationToken.ThrowIfCancellationRequested(); + + var considerInterrupt = honorInterrupts && (honorFirstInterrupt || !firstNode); + if (considerInterrupt && _interruptBefore.Contains(current)) + { + // Pause before this node. The latest checkpoint already records NextNode == current, + // so a subsequent resume picks up here. + return GraphRunResult.Interrupted(state, current); + } + + firstNode = false; + + if (++step > maxSteps) + { + throw new GraphRecursionException(maxSteps); + } + + state = await _nodes[current](state, cancellationToken).ConfigureAwait(false); + current = NextNode(current, state); + + if (checkpointer is not null && threadId is not null) + { + await checkpointer.SaveAsync( + new GraphCheckpoint(threadId, $"{threadId}-{step}", step, current, state), cancellationToken).ConfigureAwait(false); + } + } + + return GraphRunResult.Complete(state); + } + + /// + /// Runs the graph and yields an update after each node executes, ending when flow reaches the end node. + /// + /// The starting state. + /// Run options (step budget). null uses defaults. + /// Token used to cancel the run. + /// A stream of ; the final update carries the final state. + /// Thrown when the step budget is exceeded. + public async IAsyncEnumerable> StreamAsync( + TState initialState, + GraphRunOptions? options = null, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + var maxSteps = ResolveMaxSteps(options); + var state = initialState; + var current = _entryPoint; + var steps = 0; + + while (current != GraphSpecialNodes.End) + { + cancellationToken.ThrowIfCancellationRequested(); + if (++steps > maxSteps) + { + throw new GraphRecursionException(maxSteps); + } + + var executed = current; + state = await _nodes[current](state, cancellationToken).ConfigureAwait(false); + yield return new GraphStepUpdate(executed, state); + current = NextNode(current, state); + } + } + + /// + /// Deterministically replays a recorded run from its checkpoint history WITHOUT executing any nodes, + /// yielding the same (node, state) sequence the original run produced. + /// + /// The checkpoint store holding the recorded run. + /// The recorded run/thread id. + /// Token used to cancel the replay. + /// The recorded per-node updates, in order. + /// + /// For Beginners: Re-watch a past run exactly as it happened, reading from the saved + /// checkpoints instead of re-running the (possibly slow or non-deterministic) nodes. Great for + /// debugging and audit. + /// + /// + public async IAsyncEnumerable> ReplayAsync( + IGraphCheckpointer checkpointer, + string threadId, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + Guard.NotNull(checkpointer); + Guard.NotNullOrWhiteSpace(threadId); + + var history = await checkpointer.GetHistoryAsync(threadId, cancellationToken).ConfigureAwait(false); + for (var k = 1; k < history.Count; k++) + { + cancellationToken.ThrowIfCancellationRequested(); + // The node that ran to produce history[k] is the one recorded as "next" by history[k-1]. + var executedNode = history[k - 1].NextNode; + yield return new GraphStepUpdate(executedNode, history[k].State); + } + } + + /// + /// Returns the final recorded state for a thread (read from checkpoints, no execution). + /// + /// The checkpoint store. + /// The run/thread id. + /// Token used to cancel the operation. + /// The latest recorded state. + /// Thrown when the thread has no checkpoints. + public async Task GetRecordedStateAsync( + IGraphCheckpointer checkpointer, + string threadId, + CancellationToken cancellationToken = default) + { + Guard.NotNull(checkpointer); + Guard.NotNullOrWhiteSpace(threadId); + + var latest = await checkpointer.GetLatestAsync(threadId, cancellationToken).ConfigureAwait(false); + if (latest is null) + { + throw new InvalidOperationException($"No checkpoints found for thread '{threadId}'."); + } + + return latest.State; + } + + private static int ResolveMaxSteps(GraphRunOptions? options) + { + var max = options?.MaxSteps ?? new GraphRunOptions().MaxSteps; + return max > 0 ? max : new GraphRunOptions().MaxSteps; + } + + private string NextNode(string current, TState state) + { + if (_conditionalEdges.TryGetValue(current, out var router)) + { + var next = router(state); + if (string.IsNullOrEmpty(next)) + { + throw new InvalidOperationException( + $"The conditional router for node '{current}' returned a null or empty target."); + } + + if (next != GraphSpecialNodes.End && !_nodes.ContainsKey(next)) + { + throw new InvalidOperationException( + $"The conditional router for node '{current}' returned unknown target '{next}'."); + } + + return next; + } + + return _edges.TryGetValue(current, out var target) ? target : GraphSpecialNodes.End; + } +} diff --git a/src/Agentic/Graph/GraphRecursionException.cs b/src/Agentic/Graph/GraphRecursionException.cs new file mode 100644 index 0000000000..3fceeb8a61 --- /dev/null +++ b/src/Agentic/Graph/GraphRecursionException.cs @@ -0,0 +1,30 @@ +namespace AiDotNet.Agentic.Graph; + +/// +/// Thrown when a graph run exceeds its configured maximum number of steps, which usually indicates a +/// cycle that never reaches the end node. +/// +/// +/// For Beginners: Graphs can loop (a node can route back to an earlier node). To stop an +/// accidental infinite loop, every run has a step budget. If the graph keeps going past that budget, +/// this exception is raised so you can fix the routing or raise the limit deliberately. +/// +/// +public sealed class GraphRecursionException : Exception +{ + /// + /// Initializes a new instance of the class. + /// + /// The step budget that was exceeded. + public GraphRecursionException(int maxSteps) + : base($"Graph exceeded its maximum of {maxSteps} steps without reaching the end node. " + + "This usually means a cycle never terminates; check your conditional edges or raise GraphRunOptions.MaxSteps.") + { + MaxSteps = maxSteps; + } + + /// + /// Gets the step budget that was exceeded. + /// + public int MaxSteps { get; } +} diff --git a/src/Agentic/Graph/GraphRunOptions.cs b/src/Agentic/Graph/GraphRunOptions.cs new file mode 100644 index 0000000000..009c8e787d --- /dev/null +++ b/src/Agentic/Graph/GraphRunOptions.cs @@ -0,0 +1,37 @@ +using System; + +namespace AiDotNet.Agentic.Graph; + +/// +/// Per-run settings for executing a . +/// +/// +/// For Beginners: Knobs for a single graph run. The most important one is +/// , which caps how many node executions a run may take before giving up — a +/// safety net against cycles that never end. +/// +/// +public sealed class GraphRunOptions +{ + private int _maxSteps = 25; + + /// + /// Gets or sets the maximum number of node executions allowed in a single run before a + /// is thrown. Default: 25. Must be positive. + /// + /// Thrown when set to a value <= 0. + public int MaxSteps + { + get => _maxSteps; + set + { + if (value <= 0) + { + throw new ArgumentOutOfRangeException( + nameof(value), value, "MaxSteps must be positive (a run needs at least one step)."); + } + + _maxSteps = value; + } + } +} diff --git a/src/Agentic/Graph/GraphRunResult.cs b/src/Agentic/Graph/GraphRunResult.cs new file mode 100644 index 0000000000..3304edce85 --- /dev/null +++ b/src/Agentic/Graph/GraphRunResult.cs @@ -0,0 +1,55 @@ +namespace AiDotNet.Agentic.Graph; + +/// +/// The outcome of a human-in-the-loop graph run: either the run completed, or it paused before an +/// interrupt node awaiting input. Carries the state either way. +/// +/// The graph's state type. +/// +/// +/// When a graph has interrupt points (see ), a run +/// stops just before such a node and returns an interrupted result with +/// set to that node. Call the run again on the same thread to resume past the pause (optionally editing +/// the state with the human's input first). +/// +/// For Beginners: Tells you whether the graph finished or stopped to ask a human. If it +/// stopped, says which step it's waiting to run, and +/// is the data so far so you can review/edit it before continuing. +/// +/// +public sealed class GraphRunResult +{ + private GraphRunResult(bool isComplete, TState state, string? interruptedBefore) + { + IsComplete = isComplete; + State = state; + InterruptedBefore = interruptedBefore; + } + + /// Gets a value indicating whether the run reached the end node. + public bool IsComplete { get; } + + /// Gets a value indicating whether the run paused at an interrupt point. + public bool IsInterrupted => InterruptedBefore is not null; + + /// Gets the state (final when complete, or as of the pause when interrupted). + public TState State { get; } + + /// Gets the node the run paused before, or null when the run completed. + public string? InterruptedBefore { get; } + + /// Creates a completed result. + /// The final state. + /// A completed . + public static GraphRunResult Complete(TState state) => new(true, state, null); + + /// Creates an interrupted result. + /// The state as of the pause. + /// The node the run paused before. + /// An interrupted . + public static GraphRunResult Interrupted(TState state, string node) + { + Guard.NotNullOrWhiteSpace(node); + return new GraphRunResult(false, state, node); + } +} diff --git a/src/Agentic/Graph/GraphSpecialNodes.cs b/src/Agentic/Graph/GraphSpecialNodes.cs new file mode 100644 index 0000000000..54092d23fa --- /dev/null +++ b/src/Agentic/Graph/GraphSpecialNodes.cs @@ -0,0 +1,18 @@ +namespace AiDotNet.Agentic.Graph; + +/// +/// Reserved node names recognized by the graph runtime. +/// +/// +/// For Beginners: A graph finishes when flow reaches the special node. +/// You route to it with AddEdge("lastNode", GraphSpecialNodes.End) (or return it from a +/// conditional router) to say "we're done". +/// +/// +public static class GraphSpecialNodes +{ + /// + /// The terminal node name. When flow reaches this node, the run completes and returns the current state. + /// + public const string End = "__end__"; +} diff --git a/src/Agentic/Graph/GraphStepUpdate.cs b/src/Agentic/Graph/GraphStepUpdate.cs new file mode 100644 index 0000000000..3e094b1414 --- /dev/null +++ b/src/Agentic/Graph/GraphStepUpdate.cs @@ -0,0 +1,38 @@ +namespace AiDotNet.Agentic.Graph; + +/// +/// A streamed update emitted after a single node finishes executing during a graph run: the node's name +/// and the graph state as it stands after that node. +/// +/// The graph's state type. +/// +/// For Beginners: When you stream a graph run, you receive one of these each time a step +/// completes — letting you watch the state evolve node by node (for progress UIs, logging, or +/// debugging). The last update's is the final result. +/// +/// +public sealed class GraphStepUpdate +{ + /// + /// Initializes a new step update. + /// + /// The name of the node that just executed. + /// The graph state after the node executed. + /// Thrown when is null. + public GraphStepUpdate(string nodeName, TState state) + { + Guard.NotNull(nodeName); + NodeName = nodeName; + State = state; + } + + /// + /// Gets the name of the node that just executed. + /// + public string NodeName { get; } + + /// + /// Gets the graph state after the node executed. + /// + public TState State { get; } +} diff --git a/src/Agentic/Graph/StateGraph.cs b/src/Agentic/Graph/StateGraph.cs new file mode 100644 index 0000000000..984d0d61b4 --- /dev/null +++ b/src/Agentic/Graph/StateGraph.cs @@ -0,0 +1,329 @@ +namespace AiDotNet.Agentic.Graph; + +/// +/// A builder for a typed state graph: register nodes (state transformers), wire them with fixed or +/// conditional edges (cycles allowed), set an entry point, then into an executable +/// . +/// +/// +/// The state type threaded through the graph. Each node receives the current state and returns the next +/// state (it may mutate and return the same instance, or return a new one). +/// +/// +/// +/// This is the AiDotNet counterpart to LangGraph's StateGraph, but fully typed: there are no +/// stringly-typed state dictionaries — is your own type, checked by the +/// compiler. Routing is explicit: a node has at most one fixed edge or one conditional router; reaching +/// finishes the run. +/// +/// For Beginners: Think of building a flowchart. adds a box that does +/// some work on your data; draws an arrow to the next box; +/// draws arrows that depend on the data; marks where to start. Arrows can +/// loop back to create cycles. turns the flowchart into something you can run. +/// +/// Example: +/// +/// var graph = new StateGraph<MyState>() +/// .AddNode("plan", (s, ct) => Task.FromResult(s.WithPlan())) +/// .AddNode("act", (s, ct) => Task.FromResult(s.Act())) +/// .AddConditionalEdges("act", s => s.IsDone ? StateGraph<MyState>.End : "act") +/// .AddEdge("plan", "act") +/// .SetEntryPoint("plan") +/// .Compile(); +/// var result = await graph.InvokeAsync(new MyState()); +/// +/// +/// +public sealed class StateGraph +{ + /// + /// The terminal node name; route here to finish the run. Alias of . + /// + public const string End = GraphSpecialNodes.End; + + private readonly Dictionary>> _nodes = new(StringComparer.Ordinal); + private readonly Dictionary _edges = new(StringComparer.Ordinal); + private readonly Dictionary> _conditionalEdges = new(StringComparer.Ordinal); + private readonly HashSet _interruptBefore = new(StringComparer.Ordinal); + private string? _entryPoint; + + /// + /// Adds an asynchronous node that transforms the state. + /// + /// The unique node name (cannot be empty or the reserved end name). + /// The node body: receives the current state and returns the next state. + /// This builder, for chaining. + /// Thrown when or is null. + /// Thrown when the name is empty/whitespace, reserved, or already used. + public StateGraph AddNode(string name, Func> node) + { + Guard.NotNullOrWhiteSpace(name); + Guard.NotNull(node); + if (name == End) + { + throw new ArgumentException($"'{End}' is a reserved node name.", nameof(name)); + } + + if (_nodes.ContainsKey(name)) + { + throw new ArgumentException($"A node named '{name}' has already been added.", nameof(name)); + } + + _nodes[name] = node; + return this; + } + + /// + /// Adds a synchronous node that transforms the state. + /// + /// The unique node name. + /// The node body: receives the current state and returns the next state. + /// This builder, for chaining. + /// Thrown when or is null. + public StateGraph AddNode(string name, Func node) + { + Guard.NotNull(node); + return AddNode(name, (state, _) => Task.FromResult(node(state))); + } + + /// + /// Adds a fixed edge: after runs, flow always moves to . + /// + /// The source node name. + /// The destination node name, or . + /// This builder, for chaining. + /// Thrown when names are empty or already has a fixed edge. + public StateGraph AddEdge(string from, string to) + { + Guard.NotNullOrWhiteSpace(from); + Guard.NotNullOrWhiteSpace(to); + if (_edges.ContainsKey(from)) + { + throw new ArgumentException($"Node '{from}' already has a fixed edge.", nameof(from)); + } + + _edges[from] = to; + return this; + } + + /// + /// Adds conditional edges: after runs, the router inspects the state and + /// returns the next node name (or ). + /// + /// The source node name. + /// Selects the next node from the current state. + /// This builder, for chaining. + /// Thrown when is null. + /// Thrown when is empty or already has conditional edges. + public StateGraph AddConditionalEdges(string from, Func router) + { + Guard.NotNullOrWhiteSpace(from); + Guard.NotNull(router); + if (_conditionalEdges.ContainsKey(from)) + { + throw new ArgumentException($"Node '{from}' already has conditional edges.", nameof(from)); + } + + _conditionalEdges[from] = router; + return this; + } + + /// + /// Adds a node that runs an entire compiled graph as a single step (a subgraph), threading the same + /// state in and out. Enables composition: build small graphs and embed them in larger ones. + /// + /// The node name for the subgraph. + /// The compiled graph to run when this node executes. + /// This builder, for chaining. + /// Thrown when is null. + public StateGraph AddSubgraph(string name, CompiledStateGraph subgraph) + { + Guard.NotNull(subgraph); + return AddNode(name, (state, ct) => subgraph.InvokeAsync(state, null, ct)); + } + + /// + /// Adds reward-gated routing after a node: a scoring function rates the state, and flow goes to + /// when the score is at least , else to + /// . A differentiator for verifier/critic-driven control flow. + /// + /// The source node name. + /// Scores the current state (e.g., a critic/reward model). + /// The minimum score required to take the "meets" branch. + /// The next node when reward(state) >= threshold (may be ). + /// The next node otherwise (may be ). + /// This builder, for chaining. + /// Thrown when is null. + /// Thrown when a node name is empty/whitespace. + public StateGraph AddRewardGatedEdges( + string from, + Func reward, + double threshold, + string ifMeetsThreshold, + string ifBelowThreshold) + { + Guard.NotNull(reward); + Guard.NotNullOrWhiteSpace(ifMeetsThreshold); + Guard.NotNullOrWhiteSpace(ifBelowThreshold); + return AddConditionalEdges(from, state => reward(state) >= threshold ? ifMeetsThreshold : ifBelowThreshold); + } + + /// + /// Adds a dynamic fan-out (map-reduce) node: it derives a set of items from the current state, runs a + /// branch over each item in parallel, then reduces the branch results back into the state. This is the + /// typed equivalent of LangGraph's Send/map-reduce. + /// + /// The per-branch input item type. + /// The per-branch result type. + /// The node name. + /// Derives the items to fan out over from the current state. + /// The work run for each item (in parallel). + /// Merges the ordered branch results back into the state. + /// Optional cap on concurrent branches; null runs all at once. + /// This builder, for chaining. + /// Thrown when , , or is null. + /// Thrown when is less than 1. + public StateGraph AddFanOutNode( + string name, + Func> map, + Func> branch, + Func, TState> reduce, + int? maxDegreeOfParallelism = null) + { + Guard.NotNull(map); + Guard.NotNull(branch); + Guard.NotNull(reduce); + if (maxDegreeOfParallelism is { } dop && dop < 1) + { + throw new ArgumentOutOfRangeException(nameof(maxDegreeOfParallelism), "Max degree of parallelism must be at least 1."); + } + + return AddNode(name, async (state, ct) => + { + var items = (map(state) ?? Enumerable.Empty()).ToList(); + IReadOnlyList results = await RunBranchesAsync(items, branch, maxDegreeOfParallelism, ct).ConfigureAwait(false); + return reduce(state, results); + }); + } + + private static async Task RunBranchesAsync( + IReadOnlyList items, + Func> branch, + int? maxDegreeOfParallelism, + CancellationToken cancellationToken) + { + if (maxDegreeOfParallelism is not { } dop) + { + return await Task.WhenAll(items.Select(item => branch(item, cancellationToken))).ConfigureAwait(false); + } + + using var throttle = new SemaphoreSlim(dop); + async Task RunThrottled(TItem item) + { + await throttle.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + return await branch(item, cancellationToken).ConfigureAwait(false); + } + finally + { + throttle.Release(); + } + } + + return await Task.WhenAll(items.Select(RunThrottled)).ConfigureAwait(false); + } + + /// + /// Marks a node as a human-in-the-loop interrupt point: a run pauses just before this node so a human + /// can review (and optionally edit) the state, then resume. + /// + /// The node to pause before. + /// This builder, for chaining. + /// Thrown when is empty/whitespace. + /// + /// Interrupts are honored by ; + /// the plain InvokeAsync overloads run straight through without pausing. + /// + public StateGraph AddInterruptBefore(string name) + { + Guard.NotNullOrWhiteSpace(name); + _interruptBefore.Add(name); + return this; + } + + /// + /// Sets the node where execution begins. + /// + /// The entry node name. + /// This builder, for chaining. + /// Thrown when is empty/whitespace. + public StateGraph SetEntryPoint(string name) + { + Guard.NotNullOrWhiteSpace(name); + _entryPoint = name; + return this; + } + + /// + /// Validates the graph and produces an executable . + /// + /// The compiled graph. + /// Thrown when the graph is invalid (no/unknown entry point, edges to unknown nodes, or a node with both a fixed edge and conditional edges). + public CompiledStateGraph Compile() + { + var entry = _entryPoint; + if (entry is null || entry.Length == 0) + { + throw new InvalidOperationException("No entry point set. Call SetEntryPoint(...) before Compile()."); + } + + if (!_nodes.ContainsKey(entry)) + { + throw new InvalidOperationException($"Entry point '{entry}' is not a registered node."); + } + + foreach (var edge in _edges) + { + if (!_nodes.ContainsKey(edge.Key)) + { + throw new InvalidOperationException($"Edge source '{edge.Key}' is not a registered node."); + } + + if (edge.Value != End && !_nodes.ContainsKey(edge.Value)) + { + throw new InvalidOperationException($"Edge target '{edge.Value}' (from '{edge.Key}') is not a registered node or End."); + } + } + + foreach (var conditional in _conditionalEdges) + { + if (!_nodes.ContainsKey(conditional.Key)) + { + throw new InvalidOperationException($"Conditional-edge source '{conditional.Key}' is not a registered node."); + } + + if (_edges.ContainsKey(conditional.Key)) + { + throw new InvalidOperationException( + $"Node '{conditional.Key}' has both a fixed edge and conditional edges; use one or the other."); + } + } + + foreach (var node in _interruptBefore) + { + if (!_nodes.ContainsKey(node)) + { + throw new InvalidOperationException($"Interrupt node '{node}' is not a registered node."); + } + } + + // Snapshot so post-compile builder mutations don't affect the compiled graph. + return new CompiledStateGraph( + new Dictionary>>(_nodes, StringComparer.Ordinal), + new Dictionary(_edges, StringComparer.Ordinal), + new Dictionary>(_conditionalEdges, StringComparer.Ordinal), + new HashSet(_interruptBefore, StringComparer.Ordinal), + entry); + } +} diff --git a/src/AiDotNet.Storage.Sqlite/SqliteGraphCheckpointer.cs b/src/AiDotNet.Storage.Sqlite/SqliteGraphCheckpointer.cs new file mode 100644 index 0000000000..93ef9484a0 --- /dev/null +++ b/src/AiDotNet.Storage.Sqlite/SqliteGraphCheckpointer.cs @@ -0,0 +1,174 @@ +using System.Collections.Generic; +using System.Data.Common; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Data.Sqlite; +using Newtonsoft.Json; + +namespace AiDotNet.Agentic.Graph.Checkpointing; + +/// +/// A SQLite-backed that persists graph checkpoints in a database +/// table, giving durable resume and time-travel across process restarts and machines. Ships in the opt-in +/// AiDotNet.Storage.Sqlite package so the core stays free of the SQLite dependency. +/// +/// The graph's state type (serialized to JSON for storage). +/// +/// For Beginners: Saves your graph's progress into a SQLite database file. Point it at a +/// connection string and runs survive restarts — resume or rewind by thread id. +/// +/// +public sealed class SqliteGraphCheckpointer : IGraphCheckpointer +{ + private readonly string _connectionString; + + /// + /// Initializes the checkpointer with a SQLite connection string (e.g. Data Source=graph.db). + /// + /// The SQLite connection string. + /// Thrown when is null. + /// Thrown when is empty/whitespace. + public SqliteGraphCheckpointer(string connectionString) + { + if (string.IsNullOrWhiteSpace(connectionString)) + { + throw new ArgumentException("Connection string cannot be empty.", nameof(connectionString)); + } + + _connectionString = connectionString; + } + + /// + public async Task SaveAsync(GraphCheckpoint checkpoint, CancellationToken cancellationToken = default) + { + if (checkpoint is null) + { + throw new ArgumentNullException(nameof(checkpoint)); + } + + using var connection = await OpenAsync(cancellationToken).ConfigureAwait(false); + using var command = connection.CreateCommand(); + command.CommandText = + "INSERT INTO GraphCheckpoints (ThreadId, CheckpointId, Step, NextNode, StateJson) " + + "VALUES ($threadId, $checkpointId, $step, $nextNode, $stateJson);"; + command.Parameters.AddWithValue("$threadId", checkpoint.ThreadId); + command.Parameters.AddWithValue("$checkpointId", checkpoint.CheckpointId); + command.Parameters.AddWithValue("$step", checkpoint.Step); + command.Parameters.AddWithValue("$nextNode", checkpoint.NextNode); + command.Parameters.AddWithValue("$stateJson", JsonConvert.SerializeObject(checkpoint.State)); + await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + } + + /// + public async Task?> GetLatestAsync(string threadId, CancellationToken cancellationToken = default) + { + if (threadId is null) + { + throw new ArgumentNullException(nameof(threadId)); + } + + using var connection = await OpenAsync(cancellationToken).ConfigureAwait(false); + using var command = connection.CreateCommand(); + command.CommandText = + "SELECT CheckpointId, Step, NextNode, StateJson FROM GraphCheckpoints " + + "WHERE ThreadId = $threadId ORDER BY Seq DESC LIMIT 1;"; + command.Parameters.AddWithValue("$threadId", threadId); + using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); + if (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) + { + return ReadRow(threadId, reader); + } + + return null; + } + + /// + public async Task?> GetAsync(string threadId, string checkpointId, CancellationToken cancellationToken = default) + { + if (threadId is null) + { + throw new ArgumentNullException(nameof(threadId)); + } + + if (checkpointId is null) + { + throw new ArgumentNullException(nameof(checkpointId)); + } + + using var connection = await OpenAsync(cancellationToken).ConfigureAwait(false); + using var command = connection.CreateCommand(); + command.CommandText = + "SELECT CheckpointId, Step, NextNode, StateJson FROM GraphCheckpoints " + + "WHERE ThreadId = $threadId AND CheckpointId = $checkpointId ORDER BY Seq DESC LIMIT 1;"; + command.Parameters.AddWithValue("$threadId", threadId); + command.Parameters.AddWithValue("$checkpointId", checkpointId); + using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); + if (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) + { + return ReadRow(threadId, reader); + } + + return null; + } + + /// + public async Task>> GetHistoryAsync(string threadId, CancellationToken cancellationToken = default) + { + if (threadId is null) + { + throw new ArgumentNullException(nameof(threadId)); + } + + var history = new List>(); + using var connection = await OpenAsync(cancellationToken).ConfigureAwait(false); + using var command = connection.CreateCommand(); + command.CommandText = + "SELECT CheckpointId, Step, NextNode, StateJson FROM GraphCheckpoints " + + "WHERE ThreadId = $threadId ORDER BY Seq ASC;"; + command.Parameters.AddWithValue("$threadId", threadId); + using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); + while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) + { + history.Add(ReadRow(threadId, reader)); + } + + return history; + } + + private static GraphCheckpoint ReadRow(string threadId, DbDataReader reader) + { + var checkpointId = reader.GetString(0); + var step = reader.GetInt32(1); + var nextNode = reader.GetString(2); + var stateJson = reader.GetString(3); + var state = JsonConvert.DeserializeObject(stateJson); + if (state is null) + { + throw new InvalidOperationException( + $"Checkpoint '{checkpointId}' for thread '{threadId}' has a null or unparseable state payload " + + "and cannot be reconstructed."); + } + + return new GraphCheckpoint(threadId, checkpointId, step, nextNode, state); + } + + private async Task OpenAsync(CancellationToken cancellationToken) + { + var connection = new SqliteConnection(_connectionString); + await connection.OpenAsync(cancellationToken).ConfigureAwait(false); + + using var command = connection.CreateCommand(); + command.CommandText = + "CREATE TABLE IF NOT EXISTS GraphCheckpoints (" + + "Seq INTEGER PRIMARY KEY AUTOINCREMENT, " + + "ThreadId TEXT NOT NULL, " + + "CheckpointId TEXT NOT NULL, " + + "Step INTEGER NOT NULL, " + + "NextNode TEXT NOT NULL, " + + "StateJson TEXT NOT NULL);" + + "CREATE INDEX IF NOT EXISTS IX_GraphCheckpoints_Thread ON GraphCheckpoints (ThreadId, Seq);"; + await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false); + + return connection; + } +} diff --git a/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphCheckpointingTests.cs b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphCheckpointingTests.cs new file mode 100644 index 0000000000..2b86a1648f --- /dev/null +++ b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphCheckpointingTests.cs @@ -0,0 +1,86 @@ +using System; +using System.Linq; +using System.Threading.Tasks; +using AiDotNet.Agentic.Graph; +using AiDotNet.Agentic.Graph.Checkpointing; +using Xunit; + +namespace AiDotNetTests.UnitTests.Agentic.Graph +{ + public class GraphCheckpointingTests + { + // a: +1, b: +10, a -> b -> End + private static CompiledStateGraph BuildGraph() => + new StateGraph() + .AddNode("a", s => s + 1) + .AddNode("b", s => s + 10) + .AddEdge("a", "b") + .AddEdge("b", StateGraph.End) + .SetEntryPoint("a") + .Compile(); + + [Fact(Timeout = 60000)] + public async Task CheckpointedRun_Completes_AndRecordsHistory() + { + var graph = BuildGraph(); + var cp = new InMemoryGraphCheckpointer(); + + var result = await graph.InvokeAsync(0, cp, "t1"); + Assert.Equal(11, result); + + var history = await cp.GetHistoryAsync("t1"); + // step0(next=a,0), step1(next=b,1), step2(next=End,11) + Assert.Equal(3, history.Count); + Assert.Equal(new[] { 0, 1, 2 }, history.Select(h => h.Step)); + Assert.Equal(StateGraph.End, history[2].NextNode); + Assert.True(history[2].IsComplete); + Assert.Equal(11, history[2].State); + } + + [Fact(Timeout = 60000)] + public async Task Resume_PicksUpFromLatestCheckpoint_SkippingEarlierNodes() + { + var graph = BuildGraph(); + var cp = new InMemoryGraphCheckpointer(); + + // Simulate that node 'a' already ran (produced 100, next is 'b'). + await cp.SaveAsync(new GraphCheckpoint("t2", "t2-1", 1, "b", 100)); + + var result = await graph.InvokeAsync(initialState: 0, cp, "t2"); + // Resumes at 'b' with state 100 -> 110; 'a' is NOT re-run (would have given 11). + Assert.Equal(110, result); + } + + [Fact(Timeout = 60000)] + public async Task ReinvokingCompletedThread_ReturnsFinalState_WithoutRerunning() + { + var graph = BuildGraph(); + var cp = new InMemoryGraphCheckpointer(); + + await graph.InvokeAsync(0, cp, "t3"); // completes at 11 + var again = await graph.InvokeAsync(999, cp, "t3"); // initialState ignored; thread is complete + Assert.Equal(11, again); + } + + [Fact(Timeout = 60000)] + public async Task TimeTravel_ResumeFromEarlierCheckpoint_Replays() + { + var graph = BuildGraph(); + var cp = new InMemoryGraphCheckpointer(); + await graph.InvokeAsync(0, cp, "t4"); // history: t4-0(a,0), t4-1(b,1), t4-2(End,11) + + // Rewind to the checkpoint after 'a' (next='b', state=1) and replay forward. + var result = await graph.ResumeFromAsync(cp, "t4", "t4-1"); + Assert.Equal(11, result); + } + + [Fact(Timeout = 60000)] + public async Task ResumeFrom_UnknownCheckpoint_Throws() + { + var graph = BuildGraph(); + var cp = new InMemoryGraphCheckpointer(); + await Assert.ThrowsAsync( + () => graph.ResumeFromAsync(cp, "t5", "does-not-exist")); + } + } +} diff --git a/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphCompositionTests.cs b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphCompositionTests.cs new file mode 100644 index 0000000000..c4aff98c9e --- /dev/null +++ b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphCompositionTests.cs @@ -0,0 +1,68 @@ +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using AiDotNet.Agentic.Graph; +using AiDotNet.Agentic.Graph.Checkpointing; +using Xunit; + +namespace AiDotNetTests.UnitTests.Agentic.Graph +{ + public class GraphCompositionTests + { + [Fact(Timeout = 60000)] + public async Task Subgraph_RunsAsSingleStep() + { + var inner = new StateGraph() + .AddNode("plus1", s => s + 1) + .AddEdge("plus1", StateGraph.End) + .SetEntryPoint("plus1") + .Compile(); + + var outer = new StateGraph() + .AddNode("pre", s => s * 2) + .AddSubgraph("sub", inner) + .AddEdge("pre", "sub") + .AddEdge("sub", StateGraph.End) + .SetEntryPoint("pre") + .Compile(); + + Assert.Equal(7, await outer.InvokeAsync(3)); // (3*2) then +1 + } + + [Fact(Timeout = 60000)] + public async Task RewardGatedEdges_LoopUntilThresholdMet() + { + var graph = new StateGraph() + .AddNode("work", s => s + 1) + .AddRewardGatedEdges("work", reward: s => s, threshold: 3, ifMeetsThreshold: StateGraph.End, ifBelowThreshold: "work") + .SetEntryPoint("work") + .Compile(); + + Assert.Equal(3, await graph.InvokeAsync(0)); // loops work until score >= 3 + } + + [Fact(Timeout = 60000)] + public async Task Replay_ReproducesRecordedTrajectory_WithoutExecuting() + { + var graph = new StateGraph() + .AddNode("a", s => s + 1) + .AddNode("b", s => s + 10) + .AddEdge("a", "b") + .AddEdge("b", StateGraph.End) + .SetEntryPoint("a") + .Compile(); + var cp = new InMemoryGraphCheckpointer(); + await graph.InvokeAsync(0, cp, "r1"); + + var replay = new List>(); + await foreach (var u in graph.ReplayAsync(cp, "r1")) + { + replay.Add(u); + } + + Assert.Equal(new[] { "a", "b" }, replay.Select(u => u.NodeName)); + Assert.Equal(new[] { 1, 11 }, replay.Select(u => u.State)); + Assert.Equal(11, await graph.GetRecordedStateAsync(cp, "r1")); + } + } +} diff --git a/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphDurableCheckpointerTests.cs b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphDurableCheckpointerTests.cs new file mode 100644 index 0000000000..dd7fb52215 --- /dev/null +++ b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphDurableCheckpointerTests.cs @@ -0,0 +1,94 @@ +using System.IO; +using System.Threading.Tasks; +using AiDotNet.Agentic.Graph; +using AiDotNet.Agentic.Graph.Checkpointing; +using Xunit; + +namespace AiDotNetTests.UnitTests.Agentic.Graph +{ + public class GraphDurableCheckpointerTests + { + // a: +1, b: +10, a -> b -> End + private static CompiledStateGraph BuildGraph() => + new StateGraph() + .AddNode("a", s => s + 1) + .AddNode("b", s => s + 10) + .AddEdge("a", "b") + .AddEdge("b", StateGraph.End) + .SetEntryPoint("a") + .Compile(); + + // Runs a checkpointed graph via 'writer', then verifies a *fresh* checkpointer instance over the + // same backing store can read the history and resume — proving durability across instances/restarts. + private static async Task AssertDurableAsync(IGraphCheckpointer writer, IGraphCheckpointer freshReader) + { + var graph = BuildGraph(); + + var result = await graph.InvokeAsync(0, writer, "d1"); + Assert.Equal(11, result); + + var history = await freshReader.GetHistoryAsync("d1"); + Assert.Equal(3, history.Count); + Assert.True(history[history.Count - 1].IsComplete); + Assert.Equal(11, history[history.Count - 1].State); + + // Resuming a completed thread through the fresh instance returns the persisted final state. + Assert.Equal(11, await graph.InvokeAsync(999, freshReader, "d1")); + } + + [Fact(Timeout = 60000)] + public async Task JsonFileCheckpointer_PersistsAcrossInstances() + { + var path = Path.GetTempFileName(); + try + { + await AssertDurableAsync( + new JsonFileGraphCheckpointer(path), + new JsonFileGraphCheckpointer(path)); + } + finally + { + TryDelete(path); + } + } + +#if NET5_0_OR_GREATER + // SqliteGraphCheckpointer works on both target frameworks, but the native SQLite library + // (e_sqlite3) is not deployed to the .NET Framework (net471) shadow-copy test host in this + // environment — a pre-existing, repo-wide condition that also affects SqliteSqlSyntaxValidatorTests. + // The cross-TFM durability contract is covered by the JSON-file test above; this adds SQLite coverage + // on the framework where the native dependency is available. + [Fact(Timeout = 60000)] + public async Task SqliteCheckpointer_PersistsAcrossInstances() + { + var path = Path.GetTempFileName(); + var connectionString = $"Data Source={path};Pooling=False"; + try + { + await AssertDurableAsync( + new SqliteGraphCheckpointer(connectionString), + new SqliteGraphCheckpointer(connectionString)); + } + finally + { + TryDelete(path); + } + } +#endif + + private static void TryDelete(string path) + { + try + { + if (File.Exists(path)) + { + File.Delete(path); + } + } + catch (IOException) + { + // Best-effort cleanup; the OS reclaims temp files eventually. + } + } + } +} diff --git a/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphFanOutTests.cs b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphFanOutTests.cs new file mode 100644 index 0000000000..5c62abb9ef --- /dev/null +++ b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphFanOutTests.cs @@ -0,0 +1,63 @@ +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using AiDotNet.Agentic.Graph; +using Xunit; + +namespace AiDotNetTests.UnitTests.Agentic.Graph +{ + public class GraphFanOutTests + { + // Fan out over 1..N, square each branch in parallel, reduce by summing into the state. + private static CompiledStateGraph BuildSquareSumGraph(int? maxParallelism = null) => + new StateGraph() + .AddFanOutNode( + "squareSum", + map: s => Enumerable.Range(1, s), + branch: async (i, ct) => { await Task.Yield(); return i * i; }, + reduce: (s, results) => results.Sum(), + maxDegreeOfParallelism: maxParallelism) + .AddEdge("squareSum", StateGraph.End) + .SetEntryPoint("squareSum") + .Compile(); + + [Fact(Timeout = 60000)] + public async Task FanOut_MapsBranchesAndReduces() + { + var graph = BuildSquareSumGraph(); + Assert.Equal(14, await graph.InvokeAsync(3)); // 1 + 4 + 9 + } + + [Fact(Timeout = 60000)] + public async Task FanOut_RespectsParallelismCap_StillCorrect() + { + var graph = BuildSquareSumGraph(maxParallelism: 2); + Assert.Equal(30, await graph.InvokeAsync(4)); // 1 + 4 + 9 + 16 + } + + [Fact(Timeout = 60000)] + public async Task FanOut_EmptyItemSet_ReducesOverNothing() + { + var graph = BuildSquareSumGraph(); + Assert.Equal(0, await graph.InvokeAsync(0)); // no items -> Sum() == 0 + } + + [Fact(Timeout = 60000)] + public async Task FanOut_ComposesWithOtherNodes() + { + var graph = new StateGraph() + .AddNode("seed", s => s + 2) // 1 -> 3 + .AddFanOutNode( + "squareSum", + map: s => Enumerable.Range(1, s), + branch: (i, ct) => Task.FromResult(i * i), + reduce: (s, results) => results.Sum()) + .AddEdge("seed", "squareSum") + .AddEdge("squareSum", StateGraph.End) + .SetEntryPoint("seed") + .Compile(); + + Assert.Equal(14, await graph.InvokeAsync(1)); // seed: 1->3, then 1+4+9 + } + } +} diff --git a/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphInterruptTests.cs b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphInterruptTests.cs new file mode 100644 index 0000000000..3cd87206ca --- /dev/null +++ b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/GraphInterruptTests.cs @@ -0,0 +1,81 @@ +using System.Threading.Tasks; +using AiDotNet.Agentic.Graph; +using AiDotNet.Agentic.Graph.Checkpointing; +using Xunit; + +namespace AiDotNetTests.UnitTests.Agentic.Graph +{ + public class GraphInterruptTests + { + // draft: +1, approve: *10, draft -> approve -> End, pause before "approve". + private static CompiledStateGraph BuildApprovalGraph() => + new StateGraph() + .AddNode("draft", s => s + 1) + .AddNode("approve", s => s * 10) + .AddEdge("draft", "approve") + .AddEdge("approve", StateGraph.End) + .AddInterruptBefore("approve") + .SetEntryPoint("draft") + .Compile(); + + [Fact(Timeout = 60000)] + public async Task RunAsync_PausesBeforeInterruptNode_ThenResumesToCompletion() + { + var graph = BuildApprovalGraph(); + var cp = new InMemoryGraphCheckpointer(); + + var first = await graph.RunAsync(1, cp, "t1"); + Assert.True(first.IsInterrupted); + Assert.Equal("approve", first.InterruptedBefore); + Assert.Equal(2, first.State); // draft ran (1 -> 2); approve did not + + var second = await graph.RunAsync(0, cp, "t1"); // initialState ignored on resume + Assert.True(second.IsComplete); + Assert.Equal(20, second.State); // approve ran (2 -> 20) + } + + [Fact(Timeout = 60000)] + public async Task RunAsync_AppliesHumanEdit_OnResume() + { + var graph = BuildApprovalGraph(); + var cp = new InMemoryGraphCheckpointer(); + + var first = await graph.RunAsync(1, cp, "t2"); + Assert.True(first.IsInterrupted); + + // Human edits the state (adds 100) before approving. + var second = await graph.RunAsync(0, cp, "t2", applyOnResume: s => s + 100); + Assert.True(second.IsComplete); + Assert.Equal(1020, second.State); // (2 + 100) * 10 + } + + [Fact(Timeout = 60000)] + public async Task RunAsync_CanInterruptBeforeEntryNode() + { + var graph = new StateGraph() + .AddNode("start", s => s + 1) + .AddEdge("start", StateGraph.End) + .AddInterruptBefore("start") + .SetEntryPoint("start") + .Compile(); + var cp = new InMemoryGraphCheckpointer(); + + var first = await graph.RunAsync(5, cp, "t3"); + Assert.True(first.IsInterrupted); + Assert.Equal("start", first.InterruptedBefore); + Assert.Equal(5, first.State); // nothing ran yet + + var second = await graph.RunAsync(0, cp, "t3"); + Assert.True(second.IsComplete); + Assert.Equal(6, second.State); + } + + [Fact(Timeout = 60000)] + public async Task InvokeAsync_RunsStraightThrough_IgnoringInterrupts() + { + var graph = BuildApprovalGraph(); + var result = await graph.InvokeAsync(1); // plain run, no checkpointer + Assert.Equal(20, result); // (1 + 1) * 10 — did not pause + } + } +} diff --git a/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/StateGraphTests.cs b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/StateGraphTests.cs new file mode 100644 index 0000000000..b14dbb783e --- /dev/null +++ b/tests/AiDotNet.Tests/UnitTests/Agentic/Graph/StateGraphTests.cs @@ -0,0 +1,121 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using AiDotNet.Agentic.Graph; +using Xunit; + +namespace AiDotNetTests.UnitTests.Agentic.Graph +{ + public class StateGraphTests + { + [Fact(Timeout = 60000)] + public async Task Linear_RunsNodesInOrder_AndThreadsState() + { + var graph = new StateGraph() + .AddNode("inc", s => s + 1) + .AddNode("double", s => s * 2) + .AddEdge("inc", "double") + .AddEdge("double", StateGraph.End) + .SetEntryPoint("inc") + .Compile(); + + var result = await graph.InvokeAsync(1); + Assert.Equal(4, result); // (1 + 1) * 2 + } + + [Fact(Timeout = 60000)] + public async Task ConditionalEdges_SupportCycles() + { + var graph = new StateGraph() + .AddNode("inc", s => s + 1) + .AddConditionalEdges("inc", s => s < 3 ? "inc" : StateGraph.End) + .SetEntryPoint("inc") + .Compile(); + + var result = await graph.InvokeAsync(0); + Assert.Equal(3, result); // loops inc until value reaches 3 + } + + [Fact(Timeout = 60000)] + public async Task Stream_EmitsUpdatePerNode_WithFinalState() + { + var graph = new StateGraph() + .AddNode("a", s => s + 1) + .AddNode("b", s => s + 10) + .AddEdge("a", "b") + .AddEdge("b", StateGraph.End) + .SetEntryPoint("a") + .Compile(); + + var updates = new List>(); + await foreach (var u in graph.StreamAsync(0)) + { + updates.Add(u); + } + + Assert.Equal(new[] { "a", "b" }, updates.Select(u => u.NodeName)); + Assert.Equal(1, updates[0].State); + Assert.Equal(11, updates[1].State); + } + + [Fact(Timeout = 60000)] + public async Task RecursionLimit_Throws_OnRunawayCycle() + { + var graph = new StateGraph() + .AddNode("loop", s => s + 1) + .AddConditionalEdges("loop", _ => "loop") // never terminates + .SetEntryPoint("loop") + .Compile(); + + var ex = await Assert.ThrowsAsync( + () => graph.InvokeAsync(0, new GraphRunOptions { MaxSteps = 5 })); + Assert.Equal(5, ex.MaxSteps); + } + + [Fact(Timeout = 60000)] + public async Task AsyncNode_IsAwaited() + { + var graph = new StateGraph() + .AddNode("delay", async (s, ct) => { await Task.Yield(); return s + 41; }) + .AddEdge("delay", StateGraph.End) + .SetEntryPoint("delay") + .Compile(); + + Assert.Equal(42, await graph.InvokeAsync(1)); + } + + [Fact(Timeout = 60000)] + public Task Compile_Validates_EntryPointEdgesAndExclusiveRouting() + { + // No entry point. + Assert.Throws(() => + new StateGraph().AddNode("a", s => s).Compile()); + + // Edge to unknown node. + Assert.Throws(() => + new StateGraph().AddNode("a", s => s).AddEdge("a", "ghost").SetEntryPoint("a").Compile()); + + // A node cannot have both a fixed edge and conditional edges. + Assert.Throws(() => + new StateGraph() + .AddNode("a", s => s) + .AddEdge("a", StateGraph.End) + .AddConditionalEdges("a", _ => StateGraph.End) + .SetEntryPoint("a") + .Compile()); + + return Task.CompletedTask; + } + + [Fact(Timeout = 60000)] + public Task AddNode_RejectsDuplicateAndReservedNames() + { + var graph = new StateGraph().AddNode("a", s => s); + Assert.Throws(() => graph.AddNode("a", s => s)); // duplicate + Assert.Throws(() => new StateGraph().AddNode(StateGraph.End, s => s)); // reserved + return Task.CompletedTask; + } + } +}