SqlFlowSdk.Management.AspNetCore 1.1.1

SqlFlow

SqlFlow is a simple durable execution workflow system, for PostgreSQL and SQL Server. It handles scheduling and retries, without needing any other services to run in addition to PostgreSQL or SQL Server.

The SQL Script for creating the SqlFlow Database Schema is available here:

• https://github.com/bytefish/SqlFlow/blob/main/sql/

It took a large deal of inspiration from Absurd, but it features a much simpler database model.

SqlFlow comes with a Management API and a Control Panel to understand you systems health and quickly find out how many tasks are being processed, which tasks are slow, which tasks currently await events, why tasks are failing and search for specific tasks:

<a href="https://raw.githubusercontent.com/bytefish/SqlFlow/main/docs/control-panel-event-blockades.jpg"> <img src="https://raw.githubusercontent.com/bytefish/SqlFlow/main/docs/control-panel-event-blockades.jpg" alt="Screenshot of Event Blockades within the SqlFlow System" width="100%" /> </a>

There are two complete application examples for using the SDKs in Java and .NET:

• Java SDK: Building a Durable AI Agent
• .NET SDK: Building a Durable AI Agent

Getting Started with the Database

You'll start by creating the ssf database schema and tables:

• PostgreSQL: sql/ssf-postgres.sql
• SQL Server: sql/ssf-sqlserver.sql

Getting Started with the .NET SDK

To include SqlFlowSdk in your project, install the NuGet package using the .NET CLI:

dotnet add package SqlFlowSdk

Also add the SDK Implementation for the Database Management System to use:

dotnet add package SqlFlowSdk.Postgres
dotnet add package SqlFlowSdk.SqlServer

Getting Started with the Java SDK

The SqlFlow SDK for Java is available in Maven Central repository:

<dependencies>
    
    <dependency>
        <groupId>de.bytefish.sqlflow</groupId>
        <artifactId>sqlflow-core</artifactId>
        <version>1.0.1</version>
    </dependency>

    
    <dependency>
        <groupId>de.bytefish.sqlflow</groupId>
        <artifactId>sqlflow-postgres</artifactId>
        <version>1.0.1</version>
    </dependency>

    
    <dependency>
        <groupId>de.bytefish.sqlflow</groupId>
        <artifactId>sqlflow-sqlserver</artifactId>
        <version>1.0.1</version>
    </dependency>
    
</dependencies>

Use either the sqlflow-postgres or sqlflow-sqlserver module.

Control Panel Extensions API in .NET

As of now the Control Panel Endpoints are only available in .NET, they might be ported to other SDKs.

If you want to add the Management Endpoints for the Control Panel, you need to add:

dotnet add package SqlFlowSdk.Management

Also add the Management API Implementation for the Database Management System to use:

dotnet add package SqlFlowSdk.Management.Postgres
dotnet add package SqlFlowSdk.Management.SqlServer

Java SDK: Building a Durable AI Agent

What we are going to build

The classic examples for durable execution are usually e-commerce checkouts or payment processing scenarios. But there's another rapidly growing use case developers are dealing with: Autonomous AI Agents. Building AI agents that interact with external APIs, write code, or execute complex workflows introduces challenges.

1. LLM API calls are inherently slow, prone to timeouts or rate limits. And they are also quite expensive, right? If a server crashes or restarts while waiting for a 30-second AI generation, standard async and await state is lost forever.
2. You don't want an AI to push code to production or execute financial transactions without a human looking at it. Agents need to pause their execution, ask a human for permission and resume only when approved. This is sometimes hours or days later.

Traditional approaches require you to build complex state machines, database polling loops, or heavy external infrastructure. With SqlFlow, we can write our agent as standard, sequential C# code. The framework will automatically checkpoint the state to Postgres, sleep without blocking server threads, and wake up exactly where it left off.

Building an Agent Job

To demonstrate how durable execution with SqlFlow works, we are going to build an autonomous AI agent that fixes bugs. The workflow is quickly laid out as:

1. The agent receives a GitHub issue ID and fetches the stack trace.
2. It generates a potential code fix using a Large Language Model (LLM).
3. It pauses and asks a human for approval.
4. If the human rejects the fix and provides feedback, the agent tries again (up to 3 times).
5. If approved, it creates a Pull Request. If it fails 3 times, it escalates to a senior developer.

So first, let's define the data models that represent our inputs, states and final output:

public record AgentTask(
        @JsonProperty("issue_id") String issueId
) {}

public record Issue(
        @JsonProperty("stack_trace") String stackTrace
) {}

public record Solution(
        @JsonProperty("patched_code") String patchedCode
) {}

public record HumanApproval(
        @JsonProperty("approved") boolean approved,
        @JsonProperty("reason") String reason
) {}

public record AgentResult(
        @JsonProperty("success") boolean success,
        @JsonProperty("pull_request_url") String pullRequestUrl,
        @JsonProperty("reason") String reason
) {}

The LLM Service

Next, we need a service to handle the AI code generation. In the real world, calling an LLM is a slow (and expensive) and the HTTP requests might fail or time out. We are wrapping these expensive calls with SqlFlow, so we don't lose all our state, if the server crashes.

For this demonstration, we are simulating ab LLM API call with some delay and return a hardcoded "code fixes" based on a reviewer's feedback:

// Licensed under the MIT license. See LICENSE file in the project root for full license information.

public interface LlmService {
    Solution generateFix(String log, String lastFeedback);
}

@Service
public class DefaultLlmService implements LlmService {
    private static final Logger logger = LoggerFactory.getLogger(DefaultLlmService.class);

    @Override
    public Solution generateFix(String log, String lastFeedback) {
        logger.info("Agent is thinking: 'Learned from feedback: {}'", lastFeedback);

        // Simulate expensive LLM call
        ServiceUtils.simulateDelay(2500);

        String code = lastFeedback.contains("error handling")
                ? "// AI: Improved Logging & Error-Handling added\nif(data == null) throw new IllegalArgumentException();"
                : "// AI: Simple Fix for the NullReferenceException\nif(data == null) return;";

        logger.info("LLM has generated a potential fix: \n{}", code);

        return new Solution(code);
    }
}

The agent needs to interact with the outside world. The GitHub service handles fetching the initial issue details and creating the final Pull Request. Whenever the LLM has generated has generated a solution, a human review is requested. If the LLM has been using more than a maximum amounts, the issue is escalated to a lead developer.

// Licensed under the MIT license. See LICENSE file in the project root for full license information.

public interface GitHubService {
    Issue getIssueDetails(String id);

    String createPullRequest(String id, String code);

    void requestHumanReview(String issueId, Solution proposedFix, String correlationId);

    void escalateToSenior(String id, String reason);
}

public class DefaultGitHubService implements GitHubService {
    private static final Logger logger = LoggerFactory.getLogger(DefaultGitHubService.class);

    @Override
    public Issue getIssueDetails(String issueId) {
        logger.info("GitHub: Gets Ticket #{} details from the Repository...", issueId);
        ServiceUtils.simulateDelay(800);
        return new Issue("NullReferenceException at PaymentGateway.java:42");
    }

    @Override
    public String createPullRequest(String issueId, String code) {
        logger.info("GitHub: PR for Issue #{} has been created...", issueId);
        ServiceUtils.simulateDelay(1200);
        return "https://github.com/company/repo/pull/" + (int) (Math.random() * 9000 + 1000);
    }

    @Override
    public void escalateToSenior(String id, String reason) {
        logger.error("ESCALATION to Senior Developer: Issue #{} - Reason: {}", id, reason);
        ServiceUtils.simulateDelay(500);
    }

    @Override
    public void requestHumanReview(String issueId, Solution proposedFix, String correlationId) {
        logger.info("ACTION REQUIRED: Solution for Issue #{} with Correlation-ID {} has been created: {}...",
                issueId, correlationId, proposedFix.patchedCode());
        ServiceUtils.simulateDelay(1200);
    }
}

The Autonomous Agent Job

We define our logic inside an IJob. The magic is in the ctx.Step method: every time a step completes, its result is automatically checkpointed to the Postgres database. If the process crashes or is restarted, the framework replays the job. It skips the already completed steps and loads their results directly from the database.

And then instead of blocking a thread with Task.Delay or an infinite polling loop, we use ctx.AwaitEvent to wait for human interaction. This instructs the engine to safely suspend the workflow state to the database and free up the worker until an external system fires the specific event being awaited.

// Licensed under the MIT license. See LICENSE file in the project root for full license information.

@Component
public class AutonomousAgentJob implements Job<AgentTask, AgentResult> {

    private static final Logger logger = LoggerFactory.getLogger(AutonomousAgentJob.class);

    private final LlmService llmService;
    private final GitHubService gitHubService;
    private final LocalNotificationService localNotificationService;

    public AutonomousAgentJob(LlmService llmService, GitHubService gitHubService, LocalNotificationService localNotificationService) {
        this.llmService = llmService;
        this.gitHubService = gitHubService;
        this.localNotificationService = localNotificationService;
    }

    @Override
    public AgentResult execute(TaskContext ctx, AgentTask task) throws Exception {
        logger.info("Agent starts researching ticket {}", task.issueId());

        Issue bugReport = ctx.step("fetch-issue-context", Issue.class, () ->
                gitHubService.getIssueDetails(task.issueId())
        );

        boolean isApproved = false;
        int attempt = 0;
        String lastFeedback = "Initial Attempt";

        while (!isApproved && attempt < 3) {
            attempt++;
            String correlationId = ctx.getTaskId() + "-attempt-" + attempt;

            logger.info("Attempt {}/3: Generating a fix based on: {}", attempt, lastFeedback);

            final String currentFeedback = lastFeedback; // Für Lambda effectively final machen

            Solution proposedFix = ctx.step("generate-code-fix-" + attempt, Solution.class, () ->
                    llmService.generateFix(bugReport.stackTrace(), currentFeedback)
            );

            ctx.step("notify-reviewer-" + attempt, () -> {
                gitHubService.requestHumanReview(task.issueId(), proposedFix, correlationId);

                localNotificationService.notifyReviewer(task.issueId(), correlationId);
            });

            logger.info("Review for {} has been requested. Agent goes idle and waits for the code review...", correlationId);

            Optional<JsonNode> reviewOpt = ctx.awaitEvent(
                    "agent-approval:" + task.issueId() + ":" + correlationId,
                    "wait-for-human-review-" + attempt,
                    null,
                    JsonNode.class
            );

            if (reviewOpt.isPresent()) {
                JsonNode review = reviewOpt.get();
                isApproved = review.has("approved") && review.get("approved").asBoolean();
                lastFeedback = review.has("reason") ? review.get("reason").asText() : "No feedback has been given";

                if (!isApproved) {
                    logger.warn("Attempt {} has been rejected: {}", attempt, lastFeedback);
                }
            }
        }

        if (isApproved) {
            logger.info("Fix approved. Creating Pull Request...");

            String prUrl = ctx.step("create-pull-request", String.class, () ->
                    gitHubService.createPullRequest(task.issueId(), "apply-fix")
            );

            logger.info("Mission accomplished, the PR has been created: {}", prUrl);
            return new AgentResult(true, prUrl, null);

        } else {
            logger.error("Maximum number of attempts reached. Escalates ticket {} to a human.", task.issueId());

            ctx.step("notify-senior-developer", () -> {
                gitHubService.escalateToSenior(task.issueId(), "Agent didn't find a solution after 3 attempts.");
            });

            return new AgentResult(false, null, "Escalated to human supervisor after 3 failures.");
        }
    }
}

Interacting with the System: Providing a Controller with HTTP Endpoints

Now there are two HTTP endpoints for interacting with the Job:

• The /agent/start endpoint kicks off the process asynchronously and returns immediately.
• The /agent/review/... endpoint acts as our callback webhook.
  • When the human reviewer approves or rejects a fix, this endpoint emits the event back into the queue, which then wakes up the sleeping job right where it left off.

Let's define an AgentController for it.

@RestController
@RequestMapping("/agent")
public class AgentController {

    private final ISqlFlow sqlFlow;

    public AgentController(ISqlFlow sqlFlow) {
        this.sqlFlow = sqlFlow;
    }

    @PostMapping("/start")
    public Map<String, String> startAgent(
            @RequestBody AgentTask task) {

        SpawnResult result = sqlFlow.spawn(new SpawnOptions("ai-agent-queue", null, null, null), "solve-bug", task);

        return Map.of(
                "runId", result.runId(),
                "taskId", result.taskId(),
                "status", "Agent dispatched to fix Issue #" + task.issueId());
    }

    @PostMapping("/review/{issueId}/{correlationId}")
    public Map<String, String> review(
            @PathVariable("issueId") String issueId,
            @PathVariable("correlationId") String correlationId,
            @RequestBody HumanApproval approval) {

        String eventName = "agent-approval:" + issueId + ":" + correlationId;

        sqlFlow.emitEvent(
                new EmitEventOptions("ai-agent-queue"),
                eventName,
                approval);

        String message = approval.approved()
                ? "Fix for " + correlationId + " approved. Agent is now completing its work."
                : "Fix for " + correlationId + " rejected. Agent tries again with feedback: '" + approval.reason() + "'";

        return Map.of("message", message);
    }
}

Putting It All Together: Dependency Injection

We are using Spring Boot, so we are using a class annotated with @Configuration to configure our dependencies. We are using TestContainers to spin up a new Postgres instance.

@Configuration
public class SqlFlowConfiguration {

    @Bean(destroyMethod = "stop")
    public PostgreSQLContainer<?> postgresContainer() {
        PostgreSQLContainer<?> postgres =
                new PostgreSQLContainer<>("postgres:18")
                        .withInitScript("ssf-postgres.sql");

        postgres.start();
        return postgres;
    }

    @Bean
    public DataSource dataSource(
            PostgreSQLContainer<?> postgresContainer) {

        HikariConfig config = new HikariConfig();
        config.setJdbcUrl(postgresContainer.getJdbcUrl());
        config.setUsername(postgresContainer.getUsername());
        config.setPassword(postgresContainer.getPassword());
        config.setMaximumPoolSize(10);

        return new HikariDataSource(config);
    }

    @Bean
    public ObjectMapper objectMapper() {
        return new ObjectMapper()
                .registerModule(new JavaTimeModule());
    }

    @Bean
    public JobFactory springJobFactory(ApplicationContext context) {
        return context::getBean;
    }

    @Bean
    public ISqlFlow sqlFlow(
            DataSource dataSource,
            ObjectMapper mapper,
            JobFactory jobFactory) {

        PostgresFlowDatabase db = new PostgresFlowDatabase(dataSource, mapper);

        ISqlFlow client = new SqlFlow(db, mapper);

        client.createQueue("ai-agent-queue");
        client.useJob(
                jobFactory,
                mapper,
                "solve-bug",
                3,
                AutonomousAgentJob.class,
                AgentTask.class);

        return client;
    }

    @Bean
    public SqlFlowWorker sqlFlowWorker(ISqlFlow client) {
        SqlFlowWorker worker = new SqlFlowWorker(
                WorkerOptions.builder()
                        .workerId("spring-worker-1")
                        .queue("ai-agent-queue")
                        .pollInterval(1.0)
                        .concurrency(1)
                        .build(),
                client);

        Thread.ofVirtual().start(worker);
        return worker;
    }
}

Spring Boot now auto-wires everything, and what's left is to define the application entry point.

@SpringBootApplication
public class AiSampleApplication {

    public static void main(String[] args) {
        SpringApplication.run(AiSampleApplication.class, args);
    }
}

An Example Session with the AI Agent Job

Getting the Tooling right

It's not stone age. I want to use tooling to fire my HTTP requests. There's somewhat of a standard established for tooling, which is the http format for HTTP requests.

And while it's easy to use *.http files with Visual Studio, IntelliJ doesn't come with a UI in it's Community Edition. But do not fear, you don't have to fight curl. JetBrains offers a CLI called ijhttp we can use.

We start by downloading it off the JetBrains pages:

curl.exe -f -L -o ijhttp.zip "https://jb.gg/ijhttp/latest"

And extract it to a folder Tools in the User Profile:

Expand-Archive .\ijhttp.zip -DestinationPath "$env:USERPROFILE\Tools\ijhttp"

We can then add ijhttp to the search Path in Windows:

$folder = "$env:USERPROFILE\Tools\ijhttp\ijhttp"
$userPath = [Environment]::GetEnvironmentVariable("Path", "User")

[Environment]::SetEnvironmentVariable("Path", "$userPath;$folder", "User")

The *.http File with the Requests

@baseUrl = https://localhost:5000
@issueId = 12345
@delayMs = 30000

### Start the Agent Job
# @name startAgent
POST {{baseUrl}}/agent/start
Content-Type: application/json

{
  "issue_id": "{{issueId}}"
}

> {%
    client.test("Agent was started", function () {
        client.assert(response.status === 200, "Expected HTTP 200");
        client.assert(response.body.taskId, "Response does not contain taskId");
    });

    client.global.set("taskId", response.body.taskId);
    client.log("Stored taskId: " + response.body.taskId);
%}

### Reject the first attempt after a delay
< {%
    await sleep(Number(request.variables.get("delayMs")));
%}
POST {{baseUrl}}/agent/review/{{issueId}}/{{taskId}}-attempt-1
Content-Type: application/json

{
  "approved": false,
  "reason": "This is way too simple, add a better error handling strategy!"
}

> {%
    client.test("First review was submitted", function () {
        client.assert(response.status === 200, "Expected HTTP 200");
    });
%}

### Approve the second attempt after another delay
< {%
    await sleep(Number(request.variables.get("delayMs")));
%}
POST {{baseUrl}}/agent/review/{{issueId}}/{{taskId}}-attempt-2
Content-Type: application/json

{
  "approved": true,
  "reason": "Now, this looks good!"
}

> {%
    client.test("Second review was submitted", function () {
        client.assert(response.status === 200, "Expected HTTP 200");
    });
%}

Analyzing the Log Output

After starting the Backend we can see the Postgres container being booted:

2026-08-16T10:37:08.743+02:00  INFO 24396 --- [           main] tc.postgres:18                           : Creating container for image: postgres:18
2026-08-16T10:37:08.807+02:00  INFO 24396 --- [           main] tc.postgres:18                           : Container postgres:18 is starting: b98a7b3788ed1e6cf7400ea0cc0d446c03f9aa36b61de449f4fd5e9dd7e8e82d
2026-08-16T10:37:09.877+02:00  INFO 24396 --- [           main] tc.postgres:18                           : Container postgres:18 started in PT1.1334854S
2026-08-16T10:37:09.878+02:00  INFO 24396 --- [           main] tc.postgres:18                           : Container is started (JDBC URL: jdbc:postgresql://localhost:44757/test?loggerLevel=OFF)
2026-08-16T10:37:09.884+02:00  INFO 24396 --- [           main] org.testcontainers.ext.ScriptUtils       : Executing database script from ssf-postgres.sql
2026-08-16T10:37:10.001+02:00  INFO 24396 --- [           main] org.testcontainers.ext.ScriptUtils       : Executed database script from ssf-postgres.sql in 116 ms.

And the Worker for the ai-agent-queue being created:

2026-08-16T10:37:10.141+02:00  INFO 24396 --- [    virtual-103] d.b.sqlflow.core.workers.SqlFlowWorker   : SqlFlow Worker [spring-worker-1] started for queue 'ai-agent-queue'

The Backend is ready to perform. So let's give it something to eat.

We'll switch to the IntelliJ Terminal:

• File → Settings → Tools → Terminal

We'll then run out *.http script using ijhttp -L VERBOSE agent-requests.http.

The first request for fixing an issue 12345 is sent:

PS sqlflow-example\requests> ijhttp -L VERBOSE agent-requests.http
┌─────────────────────────────────────────────────────────────────────────────┐
│                      Running IntelliJ HTTP Client with                      │
├────────────────────────┬────────────────────────────────────────────────────┤
│         Files          │ agent-requests.http                                │
├────────────────────────┼────────────────────────────────────────────────────┤
│   Public Environment   │                                                    │
├────────────────────────┼────────────────────────────────────────────────────┤
│  Private Environment   │                                                    │
└────────────────────────┴────────────────────────────────────────────────────┘
Request 'startAgent' POST http://localhost:5000/agent/start
= request =>
POST http://localhost:5000/agent/start
Content-Type: application/json
Content-Length: 25
User-Agent: IntelliJ HTTP Client/CLI 2026.1
Accept-Encoding: br, deflate, gzip, x-gzip
Accept: */*

{
  "issue_id": "12345"
}

###

<= response =
HTTP/1.1 200
Content-Type: application/json
Content-Length: 144
Date: Sun, 16 Aug 2026 08:44:33 GMT

{"taskId":"d51e000b-b1cc-4ac2-a30a-5521ae3f1c15","runId":"6b565cc1-872c-4cd1-b4b9-a84c5ba2e4b5","status":"Agent dispatched to fix Issue #12345"}

Response code: 200; Time: 448ms (448 ms); Content length: 144 bytes (144 B)

In the Backend we can see our fictional agent doing its fictional work:

2026-08-16T10:45:36.259+02:00  INFO 24396 --- [onPool-worker-3] d.b.s.e.workflows.AutonomousAgentJob     : Agent starts researching ticket 12345
2026-08-16T10:45:36.259+02:00  INFO 24396 --- [onPool-worker-3] d.b.s.e.workflows.AutonomousAgentJob     : Attempt 1/3: Generating a fix based on: Initial Attempt
2026-08-16T10:45:36.259+02:00  INFO 24396 --- [onPool-worker-3] d.b.s.e.workflows.AutonomousAgentJob     : Review for d51e000b-b1cc-4ac2-a30a-5521ae3f1c15-attempt-1 has been requested. Agent goes idle and waits for the code review...

We can see it goes idle and requests a human review. But the ficional fix looks way too simple, so we'll reject it:

Request 'Reject the first attempt after a delay' POST http://localhost:5000/agent/review/12345/d51e000b-b1cc-4ac2-a30a-5521ae3f1c15-attempt-1
= request =>
POST http://localhost:5000/agent/review/12345/d51e000b-b1cc-4ac2-a30a-5521ae3f1c15-attempt-1
Content-Type: application/json
Content-Length: 100
User-Agent: IntelliJ HTTP Client/CLI 2026.1
Accept-Encoding: br, deflate, gzip, x-gzip
Accept: */*

{
  "approved": false,
  "reason": "This is way too simple, add a better error handling strategy!"
}

###

<= response =
HTTP/1.1 200
Content-Type: application/json
Content-Length: 175
Date: Sun, 16 Aug 2026 08:45:05 GMT

{"message":"Fix for d51e000b-b1cc-4ac2-a30a-5521ae3f1c15-attempt-1 rejected. Agent tries again with feedback: 'This is way too simple, add a better error handling strategy!'"}

Response code: 200; Time: 26ms (26 ms); Content length: 175 bytes (175 B)

We can see the Backend receiving the request and the agent is generating another fix, based on our feedback:

2026-08-16T10:45:36.259+02:00  WARN 24396 --- [onPool-worker-3] d.b.s.e.workflows.AutonomousAgentJob     : Attempt 1 has been rejected: This is way too simple, add a better error handling strategy!
2026-08-16T10:45:36.259+02:00  INFO 24396 --- [onPool-worker-3] d.b.s.e.workflows.AutonomousAgentJob     : Attempt 2/3: Generating a fix based on: This is way too simple, add a better error handling strategy!
2026-08-16T10:45:36.259+02:00  INFO 24396 --- [onPool-worker-3] d.b.s.e.workflows.AutonomousAgentJob     : Review for d51e000b-b1cc-4ac2-a30a-5521ae3f1c15-attempt-2 has been requested. Agent goes idle and waits for the code review...

Let's not spend too many fictional tokens on this and accept the fix:

Request 'Approve the second attempt after another delay' POST http://localhost:5000/agent/review/12345/d51e000b-b1cc-4ac2-a30a-5521ae3f1c15-attempt-2
= request =>
POST http://localhost:5000/agent/review/12345/d51e000b-b1cc-4ac2-a30a-5521ae3f1c15-attempt-2
Content-Type: application/json
Content-Length: 59
User-Agent: IntelliJ HTTP Client/CLI 2026.1
Accept-Encoding: br, deflate, gzip, x-gzip
Accept: */*

{
  "approved": true,
  "reason": "Now, this looks good!"
}

###

<= response =
HTTP/1.1 200
Content-Type: application/json
Content-Length: 112
Date: Sun, 16 Aug 2026 08:45:35 GMT

{"message":"Fix for d51e000b-b1cc-4ac2-a30a-5521ae3f1c15-attempt-2 approved. Agent is now completing its work."}

Response code: 200; Time: 19ms (19 ms); Content length: 112 bytes (112 B)

In the logs we can see a happy agent completing the mission and creating a PR:

2026-08-16T10:45:36.259+02:00  INFO 24396 --- [onPool-worker-3] d.b.s.e.workflows.AutonomousAgentJob     : Fix approved. Creating Pull Request...
2026-08-16T10:45:36.261+02:00  INFO 24396 --- [onPool-worker-3] d.b.s.e.s.impl.DefaultGitHubService      : GitHub: PR for Issue #12345 has been created...
2026-08-16T10:45:37.467+02:00  INFO 24396 --- [onPool-worker-3] d.b.s.e.workflows.AutonomousAgentJob     : Mission accomplished, the PR has been created: https://github.com/company/repo/pull/7421

.NET SDK: Building a Durable AI Agent

What we are going to build

The classic examples for durable execution are usually e-commerce checkouts or payment processing scenarios. But there's another rapidly growing use case developers are dealing with: Autonomous AI Agents. Building AI agents that interact with external APIs, write code, or execute complex workflows introduces challenges.

1. LLM API calls are inherently slow, prone to timeouts or rate limits. And they are also quite expensive, right? If a server crashes or restarts while waiting for a 30-second AI generation, standard async and await state is lost forever.
2. You don't want an AI to push code to production or execute financial transactions without a human looking at it. Agents need to pause their execution, ask a human for permission and resume only when approved. This is sometimes hours or days later.

Traditional approaches require you to build complex state machines, database polling loops, or heavy external infrastructure. With SqlFlow, we can write our agent as standard, sequential C# code. The framework will automatically checkpoint the state to Postgres, sleep without blocking server threads, and wake up exactly where it left off.

Building an Agent Job

To demonstrate how durable execution with SqlFlow works, we are going to build an autonomous AI agent that fixes bugs. The workflow is quickly laid out as:

1. The agent receives a GitHub issue ID and fetches the stack trace.
2. It generates a potential code fix using a Large Language Model (LLM).
3. It pauses and asks a human for approval.
4. If the human rejects the fix and provides feedback, the agent tries again (up to 3 times).
5. If approved, it creates a Pull Request. If it fails 3 times, it escalates to a senior developer.

So first, let's define the data models that represent our inputs, states and final output:

public class AgentTask 
{
    [JsonPropertyName("issue_id")]
    public string IssueId { get; set; } = ""; 
}

public class Issue 
{
    [JsonPropertyName("stack_trace")]
    public string StackTrace { get; set; } = ""; 
}

public class Solution 
{
    [JsonPropertyName("patched_code")]
    public string PatchedCode { get; set; } = ""; 
}

public class HumanApproval 
{
    [JsonPropertyName("approved")]
    public bool Approved { get; set; }

    [JsonPropertyName("reason")]
    public string? Reason { get; set; } 
}

public class AgentResult
{
    [JsonPropertyName("success")]
    public bool Success { get; set; }

    [JsonPropertyName("pull_request_url")]
    public string? PullRequestUrl { get; set; }

    [JsonPropertyName("reason")]
    public string? Reason { get; set; }
}

The LLM Service

Next, we need a service to handle the AI code generation. In the real world, calling an LLM is a slow (and expensive) and the HTTP requests might fail or time out. We are wrapping these expensive calls with SqlFlow, so we don't lose all our state, if the server crashes.

For this demonstration, we are simulating ab LLM API call with some delay and return a hardcoded "code fixes" based on a reviewer's feedback:

// Licensed under the MIT license. See LICENSE file in the project root for full license information.

using SqlFlowSdk.AiSample.Models;

namespace SqlFlowSdk.AiSample.Services;

public interface ILlmService
{
    Task<Solution> GenerateFixAsync(string log, string lastFeedback, CancellationToken ct);
}

public class LlmService : ILlmService
{
    private readonly ILogger<LlmService> _logger;
    public LlmService(ILogger<LlmService> logger) => _logger = logger;

    public async Task<Solution> GenerateFixAsync(string log, string lastFeedback, CancellationToken ct)
    {
        _logger.LogInformation("Agent is thinking: 'Learned from feedback: {feedback}'", lastFeedback);

        // Simulate a very expensive LLM call with a delay
        await Task.Delay(2500, ct);

        // Change Code based on human feedback
        string code = lastFeedback.Contains("error handling")
            ? "// AI: Improved Logging & Error-Handling added\nif(data == null) throw new ArgumentNullException();" 
            : "// AI: Simple Fix for the NullReferenceException\nif(data == null) return;";

        _logger.LogInformation("LLM has generated a potential fix: {PatchedCode}", code);

        return new Solution { PatchedCode = code };
    }
}

The agent needs to interact with the outside world. The GitHub service handles fetching the initial issue details and creating the final Pull Request. Whenever the LLM has generated has generated a solution, a human review is requested. If the LLM has been using more than a maximum amounts, the issue is escalated to a lead developer.

// Licensed under the MIT license. See LICENSE file in the project root for full license information.

using SqlFlowSdk.AiSample.Models;

namespace SqlFlowSdk.AiSample.Services;

public interface IGitHubService
{
    Task<Issue> GetIssueDetailsAsync(string id, CancellationToken ct);

    Task<string> CreatePullRequestAsync(string id, string code, CancellationToken ct);

    Task RequestHumanReviewAsync(string issueId, Solution proposedFix, string correlationId, CancellationToken ct);

    Task EscalateToSeniorAsync(string id, string reason, CancellationToken ct);
}

public class GitHubService : IGitHubService
{
    private readonly ILogger<GitHubService> _logger;

    public GitHubService(ILogger<GitHubService> logger)
    {
        _logger = logger;
    }

    public async Task<Issue> GetIssueDetailsAsync(string issueId, CancellationToken ct)
    {
        _logger.LogInformation("GitHub: Gets Ticket #{id} details from the Repository...", issueId);

        await Task.Delay(800, ct);

        return new Issue { StackTrace = "NullReferenceException at PaymentGateway.cs:42" };
    }

    public async Task<string> CreatePullRequestAsync(string issueId, string code, CancellationToken ct)
    {
        _logger.LogInformation("GitHub: PR for Issue #{id} has been created...", issueId);

        await Task.Delay(1200, ct);

        return $"https://github.com/company/repo/pull/{new Random().Next(1000, 9999)}";
    }

    public async Task EscalateToSeniorAsync(string id, string reason, CancellationToken ct)
    {
        _logger.LogCritical("ESCALATION to Senior Developer: Issue #{id} - Grund: {reason}", id, reason);

        await Task.Delay(500, ct);
    }

    public async Task RequestHumanReviewAsync(string issueId, Solution proposedFix, string correlationId, CancellationToken ct)
    {
        _logger.LogInformation("ACTION REQUIRED: Solution for Issue #{id} with Correlation-ID {CorrelationId} has been created: {ProposedFix}...", issueId, correlationId, proposedFix.PatchedCode);

        await Task.Delay(1200, ct);
    }
}

The Autonomous Agent Job

We define our logic inside an IJob. The magic is in the ctx.Step method: every time a step completes, its result is automatically checkpointed to the Postgres database. If the process crashes or is restarted, the framework replays the job. It skips the already completed steps and loads their results directly from the database.

And then instead of blocking a thread with Task.Delay or an infinite polling loop, we use ctx.AwaitEvent to wait for human interaction. This instructs the engine to safely suspend the workflow state to the database and free up the worker until an external system fires the specific event being awaited.

// Licensed under the MIT license. See LICENSE file in the project root for full license information.

using SqlFlowSdk.AiSample.Models;
using SqlFlowSdk.AiSample.Services;
using SqlFlowSdk.Core;
using System.Text.Json.Nodes;

namespace SqlFlowSdk.AiSample;

public class AutonomousAgentJob : IJob<AgentTask, AgentResult>
{
    private readonly ILogger<AutonomousAgentJob> _logger;

    private readonly ILlmService _llmService;
    private readonly IGitHubService _gitHubService;
    private readonly ILocalNotificationService _localNotificationService;

    public AutonomousAgentJob(ILogger<AutonomousAgentJob> logger, ILlmService llmService, IGitHubService gitHubService, ILocalNotificationService localNotificationService)
    {
        _logger = logger;
        _llmService = llmService;
        _gitHubService = gitHubService;
        _localNotificationService = localNotificationService;
    }

    public async Task<AgentResult> ExecuteAsync(TaskContext ctx, AgentTask task)
    {
        _logger.LogInformation("Agent starts researching ticket {IssueId}", task.IssueId);

        // Load the Issue Context first, so the LLM has all relevant information
        var bugReport = await ctx.Step("fetch-issue-context", async () =>
            await _gitHubService.GetIssueDetailsAsync(task.IssueId, ctx.CancellationToken));

        bool isApproved = false;
        int attempt = 0;

        string lastFeedback = "Initial Attempt";

        while (!isApproved && attempt < 3)
        {
            attempt++;

            // Generate CorrelationID for the Event
            string correlationId = $"attempt-{attempt}";

            _logger.LogInformation("Attempt {attempt}/3: Generating a fix based on: {feedback}", attempt, lastFeedback);

            Solution proposedFix = await ctx.Step($"generate-code-fix-{attempt}", async () =>
                await _llmService.GenerateFixAsync(bugReport.StackTrace, lastFeedback, ctx.CancellationToken));

            await ctx.Step($"notify-reviewer-{attempt}", async () => {
                // Notify the reviewer via GitHub service, which could post a link to a GitHub issue or PR for review
                await _gitHubService.RequestHumanReviewAsync(task.IssueId, proposedFix, correlationId, ctx.CancellationToken);
                // Notify the reviewer via local notification service
                await _localNotificationService.NotifyReviewerAsync(task.IssueId, correlationId, ctx.CancellationToken);
            });

            _logger.LogInformation("Review for {CorrelationId} has been requested. Agent goes idle and waits for the code review...", correlationId);

            // Wair for a human decision without blocking a thread
            JsonNode? review = await ctx.AwaitEvent(
                eventName: $"agent-approval:{task.IssueId}:{correlationId}",
                stepName: $"wait-for-human-review-{attempt}"
            );
            
            isApproved = review["approved"]?.GetValue<bool>() ?? false;
            lastFeedback = review["reason"]?.GetValue<string>() ?? "No feedback has been given";

            if (!isApproved)
            {
                _logger.LogWarning("Attempt {attempt} has been rejected: {reason}", attempt, lastFeedback);
            }
        }

        if (isApproved)
        {
            _logger.LogInformation("Fix approved. Creating Pull Request...");

            string prUrl = await ctx.Step("create-pull-request", async () =>
            {
                return await _gitHubService.CreatePullRequestAsync(task.IssueId, "apply-fix", ctx.CancellationToken);
            });

            _logger.LogInformation("Mission accomplished, the PR has been created: {Url}", prUrl);

            return new AgentResult { Success = true, PullRequestUrl = prUrl };
        }
        else
        {
            _logger.LogError("Maximum number of attempts reached. Escalates ticket {IssueId} to a human.", task.IssueId);

            await ctx.Step("notify-senior-developer", async () =>
            {
                await _gitHubService.EscalateToSeniorAsync(task.IssueId, "Agent didn't find a solution after 3 attempts.", ctx.CancellationToken);
            });

            return new AgentResult { Success = false, Reason = "Escalated to human supervisor after 3 failures." };
        }
    }
}

Putting It All Together: Dependency Injection

What's left is registering all dependencies.

In the example I have used TestContainers to spin up a Postgres instance. The Connection String for this instance is then passed to the IServiceCollection#AddSqlFlowSdk(string connectionString) Extension method, that registers and wires up all dependencies for interacting with the Postgres database.

We then configure a background worker that polls a queue for tasks and maps them to our AutonomousAgentJob.

There are two HTTP endpoints for interacting with the Job:

• The /agent/start endpoint kicks off the process asynchronously and returns immediately.
• The /agent/review/... endpoint acts as our callback webhook.
  • When the human reviewer approves or rejects a fix, this endpoint emits the event back into the queue, which then wakes up the sleeping job right where it left off.

It looks like this.

// Licensed under the MIT license. See LICENSE file in the project root for full license information.

// ...

var builder = WebApplication.CreateBuilder(args);

// Start Docker Containers for dependencies
await DockerContainers.StartAllContainersAsync();

string connectionString = $"Host=127.0.0.1;Port=5432;Database=abdurd_db;Username=postgres;Password=password;";

// Add Logging
builder.Services.AddLogging(loggingBuilder => loggingBuilder.AddConsole());

builder.Services.AddSingleton<ILlmService, LlmService>();
builder.Services.AddSingleton<IGitHubService, GitHubService>();
builder.Services.AddSingleton<ILocalNotificationService, LocalNotificationService>();

// Register the SqlFlow SDK
builder.Services.AddSqlFlowSdk(connectionString);

// Configure Workers and Jobs. In this example, we have a queue for AI agents that process tasks related to bug fixing. The
// worker is configured to handle one task at a time and poll for new tasks every second. The job "solve-bug" is defined
// with a maximum of 3 attempts for each task.
builder.Services.AddSqlFlowWorker("ai-agent-queue", worker =>
{
    worker
        .SetConcurrency(1)
        .SetPollInterval(1);

    worker.AddJob<AutonomousAgentJob, AgentTask, AgentResult>("solve-bug", options =>
    {
        options.WithMaxAttempts(3);
    });
});

var app = builder.Build();

// A Webhook triggers the Agent, such as a new JIRA ticket or GitHub issue
app.MapPost("/agent/start", async (ISqlFlow client, [FromBody] AgentTask task, CancellationToken ct) =>
{
    var result = await client.SpawnAsync(new SpawnOptions
    {
        Queue = "ai-agent-queue"
    }, "solve-bug", task, ct);

    return Results.Ok(new { RunId = result.RunId, Status = $"Agent dispatched to fix Isse #{task.IssueId}" });
});

// A Lead-Developer clicks on "Approve" or "Reject", with Feeedback
app.MapPost("/agent/review/{issueId}/{correlationId}", async (
    IEventPublisher publisher,
    string issueId,
    string correlationId,
    [FromBody] HumanApproval approval,
    CancellationToken ct) =>
{
    // Wake up the agent, that is working on the ticket
    await publisher.EmitEventAsync(queue: "ai-agent-queue", eventName: $"agent-approval:{issueId}:{correlationId}", payload: approval, ct);

    string message = approval.Approved
        ? $"Fix for {correlationId} approved. Agent is now completing its work."
        : $"Fix for {correlationId} rejected. Agent tries again with feedback: '{approval.Reason}'";

    return Results.Ok(new { Message = message });
});

app.Run();

An Example Session with the AI Agent Job

After starting the Backend we can see the Postgres container being booted and the SqlFlow Postgres queue being created:

[testcontainers.org 00:00:02.01] Wait for Docker container 5235c226fe2d to complete readiness checks
[testcontainers.org 00:00:03.05] Docker container 5235c226fe2d ready
info: SqlFlowSdk.Workers.SqlFlowGenericWorker[0]
      Create Queue if not exists: 'ai-agent-queue'

In the SqlFlowSdk.AiSample.http we are now executing a Request for fixing an Issue (say 12345):

### Start the Agent Job for an AI fix for the Issue.
# @name startAgent
POST https://localhost:5000/agent/start
Content-Type: application/json
{
    "issue_id": "12345"
}

It returns the following content:

{
  "runId": "019f2db5-d7be-7d89-8fcb-d649e70e698a",
  "taskId": "3ff3cd1c-69ce-4bdb-9a14-6566955cb7cd",
  "status": "Agent dispatched to fix issue #12345"
}

We'll then extract the TaskID, because it's needed to build the CorrelationID for Event CorrelationIDs:

### Extract the Task ID and write it to a variable for later use.
@taskId = {{startAgent.response.body.$.taskId}}

In the Console of the Service, you can see the Agent working and requesting feedback.

info: SqlFlowSdk.AiSample.AutonomousAgentJob[0]
      Agent starts researching ticket 12345
info: SqlFlowSdk.AiSample.Services.GitHubService[0]
      GitHub: Gets Ticket #12345 details from the Repository...
info: SqlFlowSdk.AiSample.AutonomousAgentJob[0]
      Attempt 1/3: Generating a fix based on: Initial Attempt
info: SqlFlowSdk.AiSample.Services.LlmService[0]
      Agent is thinking: 'Learned from feedback: Initial Attempt'
info: SqlFlowSdk.AiSample.Services.LlmService[0]
      LLM has generated a potential fix: // AI: Simple Fix for the NullReferenceException
if(data == null) return;
info: SqlFlowSdk.AiSample.Services.GitHubService[0]
      ACTION REQUIRED: Solution for Issue #12345 with Correlation-ID 3ff3cd1c-69ce-4bdb-9a14-6566955cb7cd-attempt-1 has been created: // AI: Simple Fix for the NullReferenceException
if(data == null) return;...
info: SqlFlowSdk.AiSample.AutonomousAgentJob[0]
      Review for attempt-1 has been requested. Agent goes idle and waits for the code review...

But let's say we don't like the fix and we want it to be rewritten.

We will reject the code and tell it to make it simpler:

### We cannot approve such a simple fix, reject and tell it to improve.
POST https://localhost:5000/agent/review/12345/{{taskId}}-attempt-1
Content-Type: application/json
{
    "approved": false,
    "reason": "This is way too simple, add a better error handling strategy!"
}

We can then see the Agent doing its work:

warn: SqlFlowSdk.AiSample.AutonomousAgentJob[0]
      Attempt 1 has been rejected: This is way too simple, add a better error handling strategy!
info: SqlFlowSdk.AiSample.AutonomousAgentJob[0]
      Attempt 2/3: Generating a fix based on: This is way too simple, add a better error handling strategy!
info: SqlFlowSdk.AiSample.Services.LlmService[0]
      Agent is thinking: 'Learned from feedback: This is way too simple, add a better error handling strategy!'
info: SqlFlowSdk.AiSample.Services.LlmService[0]
      LLM has generated a potential fix: // AI: Improved Logging & Error-Handling added
if(data == null) throw new ArgumentNullException();
info: SqlFlowSdk.AiSample.Services.GitHubService[0]
      ACTION REQUIRED: Solution for Issue #12345 with Correlation-ID 3ff3cd1c-69ce-4bdb-9a14-6566955cb7cd-attempt-2 has been created: // AI: Improved Logging & Error-Handling added
if(data == null) throw new ArgumentNullException();...
info: SqlFlowSdk.AiSample.AutonomousAgentJob[0]
      Review for attempt-2 has been requested. Agent goes idle and waits for the code review...

This looks ok, so let's approve it:

### Send Human Feedback to the Agent
POST https://localhost:5000/agent/review/12345/{{taskId}}-attempt-2
Content-Type: application/json
{
    "approved": true,
    "reason": "Now, this looks good!"
}

And in the Console, we can see the PR finally being created:

info: SqlFlowSdk.AiSample.AutonomousAgentJob[0]
      Fix approved. Creating Pull Request...
info: SqlFlowSdk.AiSample.Services.GitHubService[0]
      GitHub: PR for Issue #12345 has been created...
info: SqlFlowSdk.AiSample.AutonomousAgentJob[0]
      Mission accomplished, the PR has been created: https://github.com/company/repo/pull/4232

And we can now simulate the Agent Loop like this:

### Start the Agent Job for an AI fix for the Issue.
# @name startAgent
POST https://localhost:5000/agent/start
Content-Type: application/json
{
    "issue_id": "12345"
}


### Extract the Task ID and write it to a variable for later use.
@taskId = {{startAgent.response.body.$.taskId}}


### We cannot approve such a simple fix, reject and tell it to improve.
POST https://localhost:5000/agent/review/12345/{{taskId}}-attempt-1
Content-Type: application/json
{
    "approved": false,
    "reason": "This is way too simple, add a better error handling strategy!"
}

### Send Human Feedback to the Agent
POST https://localhost:5000/agent/review/12345/{{taskId}}-attempt-2
Content-Type: application/json
{
    "approved": true,
    "reason": "Now, this looks good!"
}