Graph Execution

/**
 * Invokes graph execution with input map and returns final state
 * @param inputs Input data as key-value map
 * @return Optional containing final state if execution completed successfully
 */
Optional<State> invoke(Map<String, Object> inputs);

/**
 * Invokes graph execution with graph input and configuration
 * @param input Graph input (args or resume)
 * @param config Runtime configuration
 * @return Optional containing final state if execution completed successfully
 */
Optional<State> invoke(GraphInput input, RunnableConfig config);

/**
 * Invokes graph execution and returns final NodeOutput with execution details
 * @param input Graph input (args or resume)
 * @param config Runtime configuration
 * @return Optional containing final NodeOutput if execution produced output
 * @since 1.6.1
 */
Optional<NodeOutput<State>> invokeFinal(GraphInput input, RunnableConfig config);

import org.bsc.langgraph4j.*;
import java.util.Map;

// Simple execution
CompiledGraph<MyState> app = workflow.compile();
Optional<MyState> result = app.invoke(Map.of("input", "Hello"));

// Execution with configuration
RunnableConfig config = RunnableConfig.builder()
    .threadId("session-123")
    .build();

Optional<MyState> result = app.invoke(
    new GraphArgs(Map.of("query", "process this")),
    config
);

// Get final output with node information
Optional<NodeOutput<MyState>> finalOutput = app.invokeFinal(
    new GraphArgs(Map.of("data", "input")),
    config
);
if (finalOutput.isPresent()) {
    System.out.println("Last node: " + finalOutput.get().node());
    MyState finalState = finalOutput.get().state();
}

/**
 * Creates streaming execution with input map and default configuration
 * @param inputs Input data as key-value map
 * @return AsyncGenerator stream of NodeOutput
 */
AsyncGenerator<NodeOutput<State>> stream(Map<String, Object> inputs);

/**
 * Creates streaming execution with configuration
 * @param inputs Input data as key-value map
 * @param config Runtime configuration
 * @return AsyncGenerator stream of NodeOutput
 */
AsyncGenerator<NodeOutput<State>> stream(Map<String, Object> inputs, RunnableConfig config);

/**
 * Creates streaming execution with graph input and configuration
 * @param input Graph input (args or resume)
 * @param config Runtime configuration
 * @return AsyncGenerator stream of NodeOutput
 */
AsyncGenerator<NodeOutput<State>> stream(GraphInput input, RunnableConfig config);

import org.bsc.async.AsyncGenerator;

// Basic streaming
AsyncGenerator<NodeOutput<MyState>> stream = app.stream(Map.of("input", "data"));

// Process each output as it arrives
stream.forEachAsync(output -> {
    System.out.println("Node: " + output.node());
    System.out.println("State: " + output.state().data());
    return CompletableFuture.completedFuture(null);
}).get();

// Collect all outputs
List<NodeOutput<MyState>> outputs = stream.stream().toList();

// Stream with configuration for threading
RunnableConfig config = RunnableConfig.builder()
    .threadId("stream-session")
    .build();

AsyncGenerator<NodeOutput<MyState>> configuredStream = app.stream(
    Map.of("query", "streaming query"),
    config
);

/**
 * Creates streaming execution that outputs state snapshots
 * @param input Graph input (args or resume)
 * @param config Runtime configuration
 * @return AsyncGenerator stream of state snapshots
 */
AsyncGenerator<NodeOutput<State>> streamSnapshots(GraphInput input, RunnableConfig config);

/**
 * Creates streaming execution that outputs state snapshots
 * @param inputs Input data as key-value map
 * @param config Runtime configuration
 * @return AsyncGenerator stream of state snapshots
 */
AsyncGenerator<NodeOutput<State>> streamSnapshots(Map<String, Object> inputs, RunnableConfig config);

// Stream state snapshots for debugging
AsyncGenerator<NodeOutput<MyState>> snapshots = app.streamSnapshots(
    Map.of("debug", true),
    config
);

snapshots.forEachAsync(snapshot -> {
    if (snapshot instanceof StateSnapshot) {
        StateSnapshot<MyState> stateSnapshot = (StateSnapshot<MyState>) snapshot;
        System.out.println("Checkpoint ID: " + stateSnapshot.getCheckpointId());
        System.out.println("Node ID: " + stateSnapshot.getNodeId());
        System.out.println("State: " + stateSnapshot.state().data());
    }
    return CompletableFuture.completedFuture(null);
});

/**
 * Get current state snapshot for a thread
 * @param config Runtime configuration with thread ID
 * @return Current state snapshot
 * @throws IllegalStateException if checkpoint saver not configured or no checkpoint found
 */
StateSnapshot<State> getState(RunnableConfig config);

/**
 * Get optional state snapshot for a thread
 * @param config Runtime configuration with thread ID
 * @return Optional containing state snapshot if exists
 * @throws IllegalStateException if checkpoint saver not configured
 */
Optional<StateSnapshot<State>> stateOf(RunnableConfig config);

/**
 * Get last state snapshot for a thread
 * @param config Runtime configuration with thread ID
 * @return Optional containing last state snapshot if any exists
 */
Optional<StateSnapshot<State>> lastStateOf(RunnableConfig config);

/**
 * Update graph state with new values
 * @param config Runtime configuration containing graph state
 * @param values Values to update in state
 * @return Updated runtime configuration
 * @throws Exception when update fails
 */
RunnableConfig updateState(RunnableConfig config, Map<String, Object> values) throws Exception;

/**
 * Update graph state and specify next node for execution
 * @param config Runtime configuration containing graph state
 * @param values Values to update in state
 * @param asNode Node ID to execute next (can be null)
 * @return Updated runtime configuration
 * @throws Exception when update fails
 */
RunnableConfig updateState(RunnableConfig config, Map<String, Object> values, String asNode) throws Exception;

// Get current state
RunnableConfig config = RunnableConfig.builder()
    .threadId("my-thread")
    .build();

try {
    StateSnapshot<MyState> currentState = app.getState(config);
    System.out.println("Current node: " + currentState.getNodeId());
    System.out.println("State data: " + currentState.state().data());
} catch (IllegalStateException e) {
    System.out.println("No checkpoint found");
}

// Update state during execution
RunnableConfig updatedConfig = app.updateState(
    config,
    Map.of("user_input", "new value", "timestamp", System.currentTimeMillis())
);

// Update state and force next node
RunnableConfig configWithNextNode = app.updateState(
    config,
    Map.of("override", true),
    "specific_node"
);

/**
 * Gets execution history for a thread
 * @param config Runtime configuration containing thread ID
 * @return Collection of StateSnapshots ordered by recency (newest first)
 */
Collection<StateSnapshot<State>> getStateHistory(RunnableConfig config);

// Get execution history
Collection<StateSnapshot<MyState>> history = app.getStateHistory(config);

System.out.println("Execution history (" + history.size() + " steps):");
for (StateSnapshot<MyState> snapshot : history) {
    System.out.println("- Node: " + snapshot.getNodeId() +
                      ", Checkpoint: " + snapshot.getCheckpointId());
}

// Find specific checkpoint
Optional<StateSnapshot<MyState>> specificStep = history.stream()
    .filter(snapshot -> snapshot.getNodeId().equals("target_node"))
    .findFirst();

/**
 * Sets maximum number of iterations for graph execution
 * @param maxIterations Maximum iteration count (must be > 0)
 * @throws IllegalArgumentException if maxIterations <= 0
 */
void setMaxIterations(int maxIterations);

// Set iteration limit to prevent infinite loops
app.setMaxIterations(100);

// Execute with limit
try {
    Optional<MyState> result = app.invoke(Map.of("input", "data"));
} catch (IllegalStateException e) {
    if (e.getMessage().contains("Maximum number of iterations")) {
        System.out.println("Graph execution exceeded iteration limit");
    }
}

// Resume from checkpoint using GraphResume input
GraphInput resumeInput = new GraphResume();
RunnableConfig resumeConfig = RunnableConfig.builder()
    .threadId("interrupted-thread")
    .checkPointId("checkpoint-123")
    .build();

Optional<MyState> result = app.invoke(resumeInput, resumeConfig);

// Resume interrupted execution
RunnableConfig interruptedConfig = RunnableConfig.builder()
    .threadId("session-with-interrupt")
    .build();

// Original execution (may be interrupted)
AsyncGenerator<NodeOutput<MyState>> stream = app.stream(
    Map.of("needs_review", true),
    interruptedConfig
);

// Process until interruption
for (NodeOutput<MyState> output : stream.stream().toList()) {
    if (output instanceof InterruptionMetadata) {
        System.out.println("Execution interrupted at: " + output.node());
        break;
    }
}

// Later, resume execution
Optional<MyState> resumedResult = app.invoke(new GraphResume(), interruptedConfig);

enum StreamMode {
    VALUES,    // Stream node execution outputs
    SNAPSHOTS  // Stream state snapshots with checkpoint information
}

// Base input interface
interface GraphInput {}

// Input with arguments
class GraphArgs implements GraphInput {
    GraphArgs(Map<String, Object> value);
    Map<String, Object> value();
}

// Resume from checkpoint
class GraphResume implements GraphInput {}

// Static factory methods
static GraphInput args(Map<String, Object> inputs);

// Node execution output
interface NodeOutput<State extends AgentState> {
    String node();        // Node ID that produced this output
    State state();        // State after node execution
}

// State snapshot with checkpoint information
class StateSnapshot<State extends AgentState> implements NodeOutput<State> {
    String getCheckpointId();
    String getNodeId();
    String getNextNodeId();
    State state();
    RunnableConfig getConfig();
}