flow: human-in-the-loop pause/resume (durable workflow, stage A)

CHANGELOG.md

@@ -23,6 +23,7 @@ below is kept current between tags and rolled into the next version when it ship
 - **Compacted memory summaries** — agent memory now exposes compacted run summaries for easier inspection and recovery. (`agent/`)
 - **CLI input resume for agent runs** — the CLI can resume agent runs that require additional user input. (`cmd/micro/`, `agent/`)
 - **A2A inbound AP2 mandate verification (opt-in)** — set `Options.AP2PublicKey` (or `a2a.WithPushURLPolicy`'s sibling `a2a.WithAP2PublicKey` for embedded handlers) and the gateway verifies AP2 payment/checkout mandates carried on incoming messages — signature and task/context binding — recording the outcome in each task's `ap2Verifications`, with the x402 settlement rail carried through for the paid path. Off by default; mandates are otherwise carried unverified. (`gateway/a2a/`)
+- **Flow human-in-the-loop pause/resume** — a flow step can suspend a run for external input with `flow.Await(key, prompt)` (or `flow.AwaitStep`): the run checkpoints with status `waiting` and `Execute` returns cleanly. `Flow.Waiting` lists suspended runs with what they await, and `Flow.ResumeWith(ctx, runID, input)` injects the input and continues from the next step. Recovery (`ResumePending`) skips waiting runs since they need input, not a restart. (`flow/`)
 
 ### Changed
 - **Remote agent chat streaming** — `micro chat` now streams replies from remote agents instead of waiting for the full response. (`cmd/micro/`, `agent/`)

flow/otel.go

@@ -92,7 +92,10 @@ func (f *Flow) runStepSpan(ctx context.Context, step Step, in State) (State, int
 			span.SetAttributes(attribute.String(AttrFlowVerificationStatus, "failed"))
 		}
 	}
-	if err != nil {
+	if a, ok := isAwaitInput(err); ok {
+		// A suspend is normal control flow, not a step error.
+		span.SetStatus(codes.Ok, "waiting: "+a.Key)
+	} else if err != nil {
 		span.RecordError(err)
 		span.SetAttributes(attribute.String(AttrFlowErrorKind, string(ai.ClassifyError(err))))
 		span.SetStatus(codes.Error, err.Error())

flow/steps.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"sort"
 	"text/template"
@@ -110,7 +111,8 @@ type Run struct {
 	Flow     string       `json:"flow"`
 	State    State        `json:"state"`
 	Steps    []StepRecord `json:"steps"`
-	Status   string       `json:"status"` // running | done | failed
+	Status   string       `json:"status"` // running | waiting | done | failed
+	Await    *AwaitState  `json:"await,omitempty"`
 	Started  time.Time    `json:"started"`
 	Updated  time.Time    `json:"updated"`
 }
@@ -336,6 +338,54 @@ func LLM(prompt string) StepFunc {
 	}
 }
 
+// AwaitInput is the control signal a step returns (via Await) to suspend a run
+// pending external input. runFrom recognizes it, checkpoints the run as
+// "waiting", and returns cleanly — a suspend is not a failure. ResumeWith
+// injects the input and continues.
+type AwaitInput struct {
+	Key    string // labels what is awaited (e.g. "approval")
+	Prompt string // human-facing description of the input needed
+}
+
+func (e *AwaitInput) Error() string {
+	if e.Prompt != "" {
+		return fmt.Sprintf("flow: awaiting input %q: %s", e.Key, e.Prompt)
+	}
+	return fmt.Sprintf("flow: awaiting input %q", e.Key)
+}
+
+// AwaitState records, on a suspended run, what it is waiting for.
+type AwaitState struct {
+	Step   string `json:"step"`
+	Key    string `json:"key"`
+	Prompt string `json:"prompt,omitempty"`
+}
+
+func isAwaitInput(err error) (*AwaitInput, bool) {
+	var a *AwaitInput
+	if errors.As(err, &a) {
+		return a, true
+	}
+	return nil, false
+}
+
+// Await is a StepFunc that suspends the run pending external input. The run is
+// checkpointed with status "waiting" and returned cleanly; a later call to
+// Flow.ResumeWith(ctx, runID, input) completes this step with the injected
+// input and continues to the next step. key labels what is awaited (surfaced on
+// the run and via Flow.Waiting); prompt describes the input needed.
+func Await(key, prompt string) StepFunc {
+	return func(_ context.Context, in State) (State, error) {
+		return in, &AwaitInput{Key: key, Prompt: prompt}
+	}
+}
+
+// AwaitStep is a convenience for a named await step:
+// Step{Name: name, Run: Await(key, prompt)}.
+func AwaitStep(name, key, prompt string) Step {
+	return Step{Name: name, Run: Await(key, prompt)}
+}
+
 // startRun begins a fresh run of the flow's steps with the given input.
 func (f *Flow) startRun(ctx context.Context, data string) (Run, error) {
 	if err := validateSteps(f.opts.Steps); err != nil {
@@ -421,13 +471,79 @@ func (f *Flow) Pending(ctx context.Context) ([]Run, error) {
 	}
 	var out []Run
 	for _, r := range all {
-		if r.Flow == f.name && r.Status != "done" {
+		// Waiting runs need injected input (ResumeWith), not a restart, so a
+		// recovery loop (ResumePending) should not pick them up.
+		if r.Flow == f.name && r.Status != "done" && r.Status != "waiting" {
 			out = append(out, r)
 		}
 	}
 	return out, nil
 }
 
+// Waiting returns this flow's runs suspended awaiting external input, each with
+// its Await metadata, so a caller can prompt for and inject the needed input
+// with ResumeWith.
+func (f *Flow) Waiting(ctx context.Context) ([]Run, error) {
+	if f.checkpoint == nil {
+		return nil, nil
+	}
+	all, err := f.checkpoint.List(ctx)
+	if err != nil {
+		return nil, err
+	}
+	var out []Run
+	for _, r := range all {
+		if r.Flow == f.name && r.Status == "waiting" {
+			out = append(out, r)
+		}
+	}
+	return out, nil
+}
+
+// ResumeWith completes a suspended (waiting) run: it injects input for the
+// awaited step — the input becomes that step's output state — and continues
+// from the next step. It errors if the run is not waiting for input.
+func (f *Flow) ResumeWith(ctx context.Context, runID, input string) error {
+	ctx, cancel := f.withTimeout(ctx)
+	defer cancel()
+
+	if err := validateSteps(f.opts.Steps); err != nil {
+		return err
+	}
+	if f.checkpoint == nil {
+		return fmt.Errorf("flow %s has no checkpoint configured", f.name)
+	}
+	run, ok, err := f.checkpoint.Load(ctx, runID)
+	if err != nil {
+		return err
+	}
+	if !ok {
+		return fmt.Errorf("run %s not found", runID)
+	}
+	if run.Status != "waiting" {
+		return fmt.Errorf("run %s is not waiting for input (status %q)", runID, run.Status)
+	}
+	steps := f.opts.Steps
+	i := stepIndex(steps, run.State.Stage)
+	if i < 0 {
+		return fmt.Errorf("run %s is waiting at unknown step %q", runID, run.State.Stage)
+	}
+	// The awaited step is satisfied by the injected input; record it done and
+	// advance so runFrom re-enters at the next step.
+	run.Steps[i].Status = "done"
+	run.Steps[i].Result = truncate(input, 200)
+	run.State.Data = []byte(input)
+	if i+1 < len(steps) {
+		run.State.Stage = steps[i+1].Name
+	} else {
+		run.State.Stage = ""
+	}
+	run.Await = nil
+	run.Status = "running"
+	_, err = f.runFrom(ctx, run)
+	return err
+}
+
 // runFrom executes steps from the run's current Stage to the end,
 // checkpointing before and after each step.
 func (f *Flow) runFrom(ctx context.Context, run Run) (Run, error) {
@@ -464,6 +580,19 @@ func (f *Flow) runFrom(ctx context.Context, run Run) (Run, error) {
 		out, attempts, verification, err := f.runStepSpan(ctx, step, run.State)
 		run.Steps[i].Attempts = attempts
 		applyVerificationRecord(&run.Steps[i], verification)
+		if await, ok := isAwaitInput(err); ok {
+			// Suspend the run pending external input — checkpoint and return
+			// cleanly (not a failure). ResumeWith injects the input later.
+			run.Steps[i].Status = "waiting"
+			run.Status = "waiting"
+			run.Await = &AwaitState{Step: step.Name, Key: await.Key, Prompt: await.Prompt}
+			if saveErr := f.save(ctx, run); saveErr != nil {
+				spanErr = saveErr
+				return run, saveErr
+			}
+			f.log.Logf(logger.InfoLevel, "Flow %s run %s waiting for input %q at step %q", f.name, run.ID, await.Key, step.Name)
+			return run, nil
+		}
 		if err != nil {
 			spanErr = err
 			run.Steps[i].Status = "failed"
@@ -537,6 +666,11 @@ func (f *Flow) runStep(ctx context.Context, step Step, in State) (State, int, Ve
 			attemptCtx = ai.WithRunInfo(ctx, info)
 		}
 		out, err := step.Run(attemptCtx, in)
+		// An await signal is control flow, not a failure: suspend immediately
+		// without retrying or grading.
+		if _, ok := isAwaitInput(err); ok {
+			return in, attempt, lastVerification, err
+		}
 		if err == nil && step.Verify != nil {
 			lastVerification, err = step.Verify(attemptCtx, out)
 			if err == nil && !lastVerification.Passed {

flow/steps_test.go

@@ -87,6 +87,93 @@ func TestFlowCheckpointResume(t *testing.T) {
 	}
 }
 
+func TestFlowAwaitAndResumeWith(t *testing.T) {
+	mem := store.NewMemoryStore()
+	var firstCalls int
+	var secondInput string
+
+	steps := []Step{
+		{Name: "first", Run: func(_ context.Context, in State) (State, error) {
+			firstCalls++
+			in.Data = []byte("first-done")
+			return in, nil
+		}},
+		AwaitStep("approval", "approve", "Approve to continue?"),
+		{Name: "second", Run: func(_ context.Context, in State) (State, error) {
+			secondInput = in.String()
+			in.Data = []byte("second-done")
+			return in, nil
+		}},
+	}
+
+	f := New("hitl", WithCheckpoint(StoreCheckpoint(mem, "hitl")), Steps(steps...))
+
+	// Execute suspends at the await step — a clean return, not an error.
+	if err := f.Execute(context.Background(), "start"); err != nil {
+		t.Fatalf("Execute should suspend cleanly, got %v", err)
+	}
+	if firstCalls != 1 {
+		t.Fatalf("first step calls = %d, want 1", firstCalls)
+	}
+
+	// A waiting run is not pending (restart), it needs input.
+	if pend, _ := f.Pending(context.Background()); len(pend) != 0 {
+		t.Errorf("a waiting run must not be pending, got %d", len(pend))
+	}
+	waiting, err := f.Waiting(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(waiting) != 1 {
+		t.Fatalf("waiting runs = %d, want 1", len(waiting))
+	}
+	w := waiting[0]
+	if w.Status != "waiting" || w.Await == nil || w.Await.Key != "approve" ||
+		w.Await.Prompt != "Approve to continue?" || w.Await.Step != "approval" {
+		t.Fatalf("await metadata = %+v (status %q)", w.Await, w.Status)
+	}
+	if w.State.Stage != "approval" {
+		t.Fatalf("waiting stage = %q, want approval", w.State.Stage)
+	}
+
+	// Injecting input completes the awaited step and runs the rest.
+	if err := f.ResumeWith(context.Background(), w.ID, "approved"); err != nil {
+		t.Fatalf("ResumeWith: %v", err)
+	}
+	if firstCalls != 1 {
+		t.Errorf("completed step re-ran on resume; first calls = %d", firstCalls)
+	}
+	if secondInput != "approved" {
+		t.Errorf("second step input = %q, want the injected 'approved'", secondInput)
+	}
+	if wr, _ := f.Waiting(context.Background()); len(wr) != 0 {
+		t.Errorf("no waiting runs after resume, got %d", len(wr))
+	}
+	runs, _ := StoreCheckpoint(mem, "hitl").List(context.Background())
+	if len(runs) != 1 || runs[0].Status != "done" {
+		t.Fatalf("run should be done after resume, got %+v", runs)
+	}
+	if runs[0].Await != nil {
+		t.Errorf("await metadata should be cleared after resume, got %+v", runs[0].Await)
+	}
+}
+
+func TestFlowResumeWithRejectsNonWaiting(t *testing.T) {
+	mem := store.NewMemoryStore()
+	f := New("hitl2", WithCheckpoint(StoreCheckpoint(mem, "hitl2")),
+		Steps(Step{Name: "only", Run: func(_ context.Context, in State) (State, error) { return in, nil }}))
+	if err := f.Execute(context.Background(), "x"); err != nil {
+		t.Fatalf("Execute: %v", err)
+	}
+	runs, _ := StoreCheckpoint(mem, "hitl2").List(context.Background())
+	if len(runs) != 1 {
+		t.Fatalf("runs = %d", len(runs))
+	}
+	if err := f.ResumeWith(context.Background(), runs[0].ID, "input"); err == nil {
+		t.Error("ResumeWith on a completed (non-waiting) run should error")
+	}
+}
+
 func TestFlowStepContextIncludesRunInfo(t *testing.T) {
 	var got ai.RunInfo
 	step := Step{Name: "inspect", Run: func(ctx context.Context, in State) (State, error) {