creditflow

The creditflow.example microservice is the showcase for the agentic workflow engine. It implements a mock credit-approval pipeline that exercises most of the workflow features in one cohesive example: parallel fan-out with reducer-driven fan-in, a forEach over a runtime-sized array, a nested subgraph, a goto-driven manual-review loop, and an error transition.

It is also the Foreman’s primary integration-test fixture — the patterns demonstrated here are the same patterns the workflow engine’s tests verify.

creditflow can only be started in the TESTING deployment environment and is not intended for production use.

Depends On

  • foreman.core — the workflow engine that hosts the flow, persists state between steps, and drives transitions.
  • A SQL database for the Foreman’s flow state (in-memory SQLite by default).

Try It

With the examples app running in the TESTING environment:

http://localhost:8080/creditflow.example/demo

The demo page accepts an applicant (name, SSN, address, phone, employers, credit score), kicks off a CreditApproval flow, then renders the result plus a Mermaid diagram of the executed step history. Adjust the inputs to drive different branches:

TweakWhat it exercises
creditScore: 750Happy path — straight through ReviewCredit without revision.
creditScore: 600Review branch — approved after ReviewCredit.
creditScore: 560The goto loop — ReviewCredit → RequestMoreInfo (×2).
ssn ending in 0000 or non-numeric phoneSubgraph rejection — identity verification fails.
Multiple employers (Acme, Globex, Initech)forEach fan-out across employers.

The Workflow Graph

The CreditApproval graph is the centerpiece. It is the same Mermaid source the Foreman emits at runtime from HistoryMermaid, and lives alongside service.go as CREDITAPPROVAL.mmd:

%%{init: {'themeVariables': {'lineColor': '#32a7c1', 'edgeLabelBackground': '#e5f4f3', 'textColor': '#434343', 'titleColor': '#32a7c1', 'clusterTextColor': '#32a7c1'}}}%%
graph LR
    classDef task fill:#32a7c1,color:#f4f2ef,stroke:#32a7c1
    classDef sub fill:#ed2e92,color:#f4f2ef,stroke:#ed2e92
    classDef term fill:#e5f4f3,color:#434343,stroke:#32a7c1

    _title{{"credit-approval"}}:::term --> _start
    _start(( )):::term --> t0
    t0["SubmitCreditApplication"]:::task
    t1["VerifyCredit"]:::task
    t3["IdentityVerification"]:::task
    t4["HandleCreditError"]:::task
    t5["ReviewJoin"]:::task
    t6["ReviewCredit"]:::task
    t7["RequestMoreInfo"]:::task
    t8["Decision"]:::task
    subgraph fo_t0 ["for each in employers"]
        direction LR
        t2["VerifyEmployment"]:::task
    end
    t0 --> t1
    t1 -->|"onError"| t4
    t4 --> t5
    t0 -->|"fan out"| t2
    t0 --> t3
    t1 --> t5
    t2 -->|"fan in"| t5
    t3 --> t5
    t5 -->|"goto"| t7
    t5 --> t6
    t6 -->|"goto"| t7
    t7 --> t6
    t6 --> t8
    t8 --> _end(( )):::term
    style fo_t0 fill:#32a7c1,fill-opacity:0.15,stroke:#434343,stroke-dasharray:4 2
func (svc *Service) CreditApproval(ctx context.Context) (graph *workflow.Graph, err error) {
    graph = workflow.NewGraph("CreditApproval")
    graph.SetEndpoint("SubmitCreditApplication", creditflowapi.SubmitCreditApplication.URL())
    graph.SetEndpoint("VerifyCredit", creditflowapi.VerifyCredit.URL())
    graph.SetEndpoint("VerifyEmployment", creditflowapi.VerifyEmployment.URL())
    // IdentityVerification is a regular task that invokes the IdentityVerification child
    // workflow via flow.Subgraph — there is no static subgraph node.
    graph.SetEndpoint("IdentityVerification", creditflowapi.RunIdentityVerification.URL())
    graph.SetEndpoint("HandleCreditError", creditflowapi.HandleCreditError.URL())
    graph.SetEndpoint("ReviewJoin", creditflowapi.ReviewCredit.URL())
    graph.SetEndpoint("ReviewCredit", creditflowapi.ReviewCredit.URL())
    graph.SetEndpoint("RequestMoreInfo", creditflowapi.RequestMoreInfo.URL())
    graph.SetEndpoint("Decision", creditflowapi.Decision.URL())
    graph.SetFanIn("ReviewJoin")
    // employmentFailures accumulates across the forEach branches; wire the add reducer explicitly.
    graph.SetReducer("employmentFailures", workflow.ReducerAdd)

    // Fan-out from submit
    graph.AddTransition("SubmitCreditApplication", "VerifyCredit")
    graph.AddTransitionOnError("VerifyCredit", "HandleCreditError")
    graph.AddTransition("HandleCreditError", "ReviewJoin")
    graph.AddTransitionForEach("SubmitCreditApplication", "VerifyEmployment", "employers", "employerName")
    graph.AddTransition("SubmitCreditApplication", "IdentityVerification")

    // Fan-in
    graph.AddTransition("VerifyCredit", "ReviewJoin")
    graph.AddTransition("VerifyEmployment", "ReviewJoin")
    graph.AddTransition("IdentityVerification", "ReviewJoin")

    // Goto loop for borderline scores
    graph.AddTransitionGoto("ReviewJoin", "RequestMoreInfo")
    graph.AddTransition("ReviewJoin", "ReviewCredit")
    graph.AddTransitionGoto("ReviewCredit", "RequestMoreInfo")
    graph.AddTransition("RequestMoreInfo", "ReviewCredit")

    graph.AddTransition("ReviewCredit", "Decision")
    graph.AddTransition("Decision", workflow.END)
    return graph, nil
}

A few non-obvious points:

  • AddTransitionForEach("SubmitCreditApplication", "VerifyEmployment", "employers", "employerName") spawns one VerifyEmployment branch per element of the employers array in state, with each branch’s employerName set from its element. The number of parallel branches is determined at runtime — there is no fixed parallelism in the graph definition.
  • SetFanIn("ReviewJoin") marks ReviewJoin as the node where all the parallel branches converge. The Foreman waits for every branch (the three submit fan-outs and every employer iteration) to arrive before running ReviewJoin.
  • ReviewJoin and ReviewCredit are two graph positions sharing the same task URL (creditflowapi.ReviewCredit.URL()). This split is necessary because ReviewJoin is the fan-in nexus for the submit cohort while ReviewCredit hosts the goto loop; the lineage validator needs to close the cohort frame at ReviewJoin without conflicting with re-entry into the loop. The task body is the same; only the graph position differs.
  • graph.SetReducer("employmentFailures", workflow.ReducerAdd) wires the add reducer at fan-in — each VerifyEmployment branch returns its own count, and the Foreman sums them across all branches. The reducer is declared explicitly at graph-build time; the field name carries no hidden semantics.
  • AddTransitionOnError("VerifyCredit", "HandleCreditError") routes a credit-verification failure into the handler instead of failing the whole flow. The handler receives the serialized TracedError as the onErr state field.

The nested IdentityVerification subgraph has the same shape pattern in miniature — fan-out from InitIdentityVerification into three checks (VerifySSN, VerifyAddress, VerifyPhoneNumber), fanned in at IdentityDecision, which is the subgraph’s terminal node. Its source (IDENTITYVERIFICATION.mmd):

%%{init: {'themeVariables': {'lineColor': '#32a7c1', 'edgeLabelBackground': '#e5f4f3', 'textColor': '#434343', 'titleColor': '#32a7c1', 'clusterTextColor': '#32a7c1'}}}%%
graph LR
    classDef task fill:#32a7c1,color:#f4f2ef,stroke:#32a7c1
    classDef sub fill:#ed2e92,color:#f4f2ef,stroke:#ed2e92
    classDef term fill:#e5f4f3,color:#434343,stroke:#32a7c1

    _title{{"identity-verification"}}:::term --> _start
    _start(( )):::term --> t0
    t0["InitIdentityVerification"]:::task
    t1["VerifySSN"]:::task
    t2["VerifyAddress"]:::task
    t3["VerifyPhoneNumber"]:::task
    t4["IdentityDecision"]:::task
    t0 --> t1
    t0 --> t2
    t0 --> t3
    t1 --> t4
    t2 --> t4
    t3 --> t4
    t4 --> _end(( )):::term

Code Walkthrough — Task Handlers

Source: exampleservices/creditflow/service.go.

Workflow tasks are written as ordinary functional handlers that take a *workflow.Flow carrier as their second argument. They read their inputs from named state fields and return their outputs by named return values; the Foreman writes the return values back into state automatically.

SubmitCreditApplication

func (svc *Service) SubmitCreditApplication(ctx context.Context, flow *workflow.Flow, applicant creditflowapi.Applicant) (applicantName string, ssn string, address string, phone string, employers []string, creditScore int, err error) {
    return applicant.ApplicantName, applicant.SSN, applicant.Address, applicant.Phone, applicant.Employers, applicant.CreditScore, nil
}

The opening task. It unpacks the typed Applicant input into individual state fields so downstream tasks can take only the fields they need. The forEach transition that follows iterates over the employers field returned here.

VerifyCredit and HandleCreditError

func (svc *Service) VerifyCredit(ctx context.Context, flow *workflow.Flow, creditScore int) (creditVerified bool, err error) {
    return creditScore >= 550, nil
}

func (svc *Service) HandleCreditError(ctx context.Context, flow *workflow.Flow, onErr *errors.TracedError) (creditVerified bool, err error) {
    svc.LogWarn(ctx, "Credit verification failed, defaulting to not verified", "error", onErr)
    return false, nil
}

The handler receives the prior task’s error via the onErr state field — the Foreman writes the serialized TracedError there before invoking the error-transition target. The handler’s job is to translate the error into application state (here: creditVerified = false) and let the flow continue.

VerifyEmployment (forEach branch)

func (svc *Service) VerifyEmployment(ctx context.Context, flow *workflow.Flow, applicantName string, employerName string) (employmentFailuresOut int, err error) {
    if applicantName == "" || employerName == "" {
        return 1, nil
    }
    return 0, nil
}

One instance runs per element of the employers array. employerName is set per-branch by the forEach transition; applicantName is read from the shared state and is the same in every branch. The output employmentFailuresOut is each branch’s delta (0 or 1); the graph.SetReducer("employmentFailures", workflow.ReducerAdd) wiring tells the Foreman to add them together at fan-in — never the running total.

Identity Subgraph Tasks

func (svc *Service) VerifySSN(ctx context.Context, flow *workflow.Flow, ssn string) (ssnVerified bool, err error) {
    matched, _ := regexp.MatchString(`^\d{3}-\d{2}-\d{4}$`, ssn)
    ssnVerified = matched && !strings.HasSuffix(ssn, "0000")
    return ssnVerified, nil
}
func (svc *Service) VerifyAddress(ctx context.Context, flow *workflow.Flow, address string) (addressVerified bool, err error) {
    addressVerified = address != "" && !strings.Contains(address, "Nowhere")
    return addressVerified, nil
}
func (svc *Service) VerifyPhoneNumber(ctx context.Context, flow *workflow.Flow, phone string) (phoneVerified bool, err error) {
    phoneVerified, _ = regexp.MatchString(`^(\d{3}-\d{3}-\d{4}|\(\d{3}\) \d{3}-\d{4})$`, phone)
    return phoneVerified, nil
}
func (svc *Service) IdentityDecision(ctx context.Context, flow *workflow.Flow, ssnVerified bool, addressVerified bool, phoneVerified bool) (identityVerified bool, err error) {
    identityVerified = ssnVerified && addressVerified && phoneVerified
    return identityVerified, nil
}

The three verify tasks run in parallel after the subgraph’s InitIdentityVerification entry task. IdentityDecision is the subgraph’s fan-in target — it sees all three results and ANDs them.

The parent’s IdentityVerification node is the RunIdentityVerification task, which invokes the child workflow via flow.Subgraph and adopts the result on re-entry — there is no static subgraph node. Only the four fields passed as in cross into the child, and the child’s final state comes back as out:

func (svc *Service) RunIdentityVerification(ctx context.Context, flow *workflow.Flow, applicantName string, ssn string, address string, phone string) (identityVerified bool, err error) {
    var out creditflowapi.IdentityVerificationOut
    yield, err := flow.Subgraph(creditflowapi.IdentityVerification.URL(), creditflowapi.IdentityVerificationIn{
        ApplicantName: applicantName,
        SSN:           ssn,
        Address:       address,
        Phone:         phone,
    }, &out)
    if yield {
        return false, nil // parked, child workflow running
    }
    if err != nil {
        return false, errors.Trace(err)
    }
    return out.IdentityVerified, nil
}

To the parent graph this task behaves like any other: it takes (applicantName, ssn, address, phone) and returns identityVerified.

ReviewCredit and the Goto Loop

func (svc *Service) ReviewCredit(ctx context.Context, flow *workflow.Flow, creditScore int, creditVerified bool, reviewAttempts int) (creditVerifiedOut bool, err error) {
    if creditScore >= 650 {
        return creditVerified, nil
    }
    if creditScore >= 580 {
        return true, nil
    }
    if creditScore >= 550 && reviewAttempts < 2 {
        flow.Goto("RequestMoreInfo")
        return creditVerified, nil
    }
    if creditScore >= 550 {
        return true, nil
    }
    return creditVerified, nil
}

func (svc *Service) RequestMoreInfo(ctx context.Context, flow *workflow.Flow, reviewAttempts int) (reviewAttemptsOut int, err error) {
    return reviewAttempts + 1, nil
}

flow.Goto(...) activates the corresponding AddTransitionGoto(...) declared in the graph — the Foreman takes that transition instead of the normal sequential one. RequestMoreInfo increments the reviewAttempts counter and loops back. After two iterations the score is approved on the third pass; this guards the loop from running unbounded.

Decision

func (svc *Service) Decision(ctx context.Context, flow *workflow.Flow, creditVerified bool, employmentFailures int, identityVerified bool) (approved bool, err error) {
    approved = creditVerified && employmentFailures == 0 && identityVerified
    return approved, nil
}

The terminal task. Combines the three verification results that flowed through ReviewJoin and writes the final approved boolean. The Decision → END transition completes the flow.

Demo

func (svc *Service) Demo(w http.ResponseWriter, r *http.Request) (err error) {
    // ... parse the applicant form ...

    if r.Method == "POST" {
        // Build the Applicant from form fields.
        applicant := creditflowapi.Applicant{ /* ... */ }
        foremanClient := foremanapi.NewClient(svc)
        initialState := creditflowapi.CreditApprovalIn{Applicant: applicant}
        _, result, runErr := svc.runWorkflow(ctx, foremanClient, initialState)
        // ... fill the response data with status, outputs, step history, and a Mermaid diagram ...
    }

    err = svc.WriteResTemplate(w, "demo.html", data)
    return errors.Trace(err)
}

Demo is the only non-task handler. runWorkflow (a helper in the same package) calls foremanapi.NewClient(svc).Create / AwaitAndParse to drive a flow synchronously, then History + HistoryMermaid to render the executed graph for the page. Run is the one-call synchronous shape for a flow that finishes within the request budget; when a flow can outlast a single request — an LLM tool-calling loop, say — launch it with Create and long-poll with Poll, as the bank support demo does.

See Also

  • Agentic workflows — concept overview: tasks, transitions, fan-in, reducers, control signals.
  • State — how the delta-then-merge pattern works for employmentFailures (wired with graph.SetReducer), what forEach branches actually see (including the injected <as>Index / <as>Count fields), and the function-call subgraph boundary where in and out scope what crosses at the IdentityVerification subgraph.
  • Building Agentic Workflows — the step-by-step tutorial; uses this graph shape.
  • Reducers — the merge rules wired explicitly with graph.SetReducer, used at fan-in.
  • Foreman core service — the engine that hosts the flow.
  • Chatbox example — the LLM-engine counterpart to this workflow-engine demo.
  • Flight booking example — human-in-the-loop by parking on a real Interrupt, the contrast to this example’s human-in-the-loop-by-branching.
Backward chatbox embedder Forward