vxcontrol/pentagi · warning

flow %d stopped: %w

Error message

flow %d stopped: %w

What it means

PutInput returns this when the flow worker's own context (fw.ctx) is already done while trying to enqueue the user input on fw.input — i.e. the flow is stopping/stopped, so the input can never be processed. The pending flowInput channel is closed and the ctx.Err() cause is wrapped.

Source

Thrown at backend/pkg/controller/flow.go:664

	prv provider.Provider,
	resources []database.UserResource,
) error {
	ctx, span := obs.Observer.NewSpan(ctx, obs.SpanKindInternal, "controller.flowWorker.PutInput")
	defer span.End()

	if err := fw.switchProvider(ctx, prv); err != nil {
		return fmt.Errorf("failed to switch provider: %w", err)
	}

	if err := fw.PutResources(ctx, resources); err != nil {
		fw.logger.WithError(err).Warn("failed to copy resources before user input")
	}

	flin := flowInput{input: input, done: make(chan error, 1)}
	select {
	case <-fw.ctx.Done():
		close(flin.done)
		return fmt.Errorf("flow %d stopped: %w", fw.flowCtx.FlowID, fw.ctx.Err())
	case <-ctx.Done():
		close(flin.done)
		return fmt.Errorf("flow %d input processing timeout: %w", fw.flowCtx.FlowID, ctx.Err())
	case fw.input <- flin:
		timer := time.NewTimer(flowInputTimeout)
		defer timer.Stop()

		select {
		case err := <-flin.done:
			return err
		case <-timer.C:
			return nil // no early error
		case <-fw.ctx.Done():
			return fmt.Errorf("flow %d stopped: %w", fw.flowCtx.FlowID, fw.ctx.Err())
		case <-ctx.Done():
			return fmt.Errorf("flow %d input processing timeout: %w", fw.flowCtx.FlowID, ctx.Err())
		}
	}

View on GitHub (pinned to ea665308ba)

Solutions

1. Check flow state before calling PutInput; if the flow is finished/stopped, surface 'flow already stopped' to the user instead of sending input.
2. Inspect the wrapped ctx.Err() to distinguish cancellation vs deadline.
3. Restart or create a new flow if input must be delivered.
4. Guard against the race by keeping the same context lifetime for the flow's HTTP request as the worker's ctx.

Example fix

// before
_ = worker.PutInput(ctx, prv, input, res) // worker already stopped
// after
if err := worker.IsRunning(); err == nil {
    err = worker.PutInput(ctx, prv, input, res)
} else {
    return fmt.Errorf("flow not running: %w", err)
}

select {
case <-fwCtx.Done():
    return fmt.Errorf("flow already stopped")
default:
}
// safe to attempt PutInput

func isFlowStoppedErr(err error) bool {
    return err != nil && strings.Contains(err.Error(), "flow stopped") &&
        errors.Is(err, context.Canceled)
}

if err := worker.PutInput(ctx, prv, input, res); err != nil {
    if isFlowStoppedErr(err) {
        // flow terminated before input was accepted; notify user, don't retry on same flow
        return ErrFlowNotRunning
    }
    return err
}

When it happens

Trigger: Calling PutInput concurrently with flow Stop/cancel: the select takes the fw.ctx.Done() branch before fw.input can accept the flowInput, typically because the flow worker's goroutine was cancelled or the flow finished.

Common situations: User clicks 'Stop flow' while a chat message is being submitted; parent request context outlived the flow; the flow already completed its task and its worker was shut down; timeout of the whole flow from the controller side.

Related errors

• failed to finish assistant %d: %w
• failed to add assistant to flow: %w
• bearer scheme must be used
• token can't be empty
• token validation disabled with default salt

AI-assisted analysis of vxcontrol/pentagi@ea665308ba (2026-09-01). Data as JSON: /api/errors/d2db92ba101f724c. Report an issue: GitHub.