{"record":{"id":"d2db92ba101f724c","repo":"vxcontrol/pentagi","slug":"flow-d-stopped-w","errorCode":null,"errorMessage":"flow %d stopped: %w","messagePattern":"flow (.+?) stopped: %w","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"backend/pkg/controller/flow.go","lineNumber":664,"sourceCode":"\tprv provider.Provider,\n\tresources []database.UserResource,\n) error {\n\tctx, span := obs.Observer.NewSpan(ctx, obs.SpanKindInternal, \"controller.flowWorker.PutInput\")\n\tdefer span.End()\n\n\tif err := fw.switchProvider(ctx, prv); err != nil {\n\t\treturn fmt.Errorf(\"failed to switch provider: %w\", err)\n\t}\n\n\tif err := fw.PutResources(ctx, resources); err != nil {\n\t\tfw.logger.WithError(err).Warn(\"failed to copy resources before user input\")\n\t}\n\n\tflin := flowInput{input: input, done: make(chan error, 1)}\n\tselect {\n\tcase <-fw.ctx.Done():\n\t\tclose(flin.done)\n\t\treturn fmt.Errorf(\"flow %d stopped: %w\", fw.flowCtx.FlowID, fw.ctx.Err())\n\tcase <-ctx.Done():\n\t\tclose(flin.done)\n\t\treturn fmt.Errorf(\"flow %d input processing timeout: %w\", fw.flowCtx.FlowID, ctx.Err())\n\tcase fw.input <- flin:\n\t\ttimer := time.NewTimer(flowInputTimeout)\n\t\tdefer timer.Stop()\n\n\t\tselect {\n\t\tcase err := <-flin.done:\n\t\t\treturn err\n\t\tcase <-timer.C:\n\t\t\treturn nil // no early error\n\t\tcase <-fw.ctx.Done():\n\t\t\treturn fmt.Errorf(\"flow %d stopped: %w\", fw.flowCtx.FlowID, fw.ctx.Err())\n\t\tcase <-ctx.Done():\n\t\t\treturn fmt.Errorf(\"flow %d input processing timeout: %w\", fw.flowCtx.FlowID, ctx.Err())\n\t\t}\n\t}","sourceCodeStart":646,"sourceCodeEnd":682,"githubUrl":"https://github.com/vxcontrol/pentagi/blob/ea665308baaff015b226f308438a68d929d0f29b/backend/pkg/controller/flow.go#L646-L682","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Check flow state before calling PutInput; if the flow is finished/stopped, surface 'flow already stopped' to the user instead of sending input.","Inspect the wrapped ctx.Err() to distinguish cancellation vs deadline.","Restart or create a new flow if input must be delivered.","Guard against the race by keeping the same context lifetime for the flow's HTTP request as the worker's ctx."],"exampleFix":"// before\n_ = worker.PutInput(ctx, prv, input, res) // worker already stopped\n// after\nif err := worker.IsRunning(); err == nil {\n    err = worker.PutInput(ctx, prv, input, res)\n} else {\n    return fmt.Errorf(\"flow not running: %w\", err)\n}","handlingStrategy":"type-guard","validationCode":"select {\ncase <-fwCtx.Done():\n    return fmt.Errorf(\"flow already stopped\")\ndefault:\n}\n// safe to attempt PutInput","typeGuard":"func isFlowStoppedErr(err error) bool {\n    return err != nil && strings.Contains(err.Error(), \"flow stopped\") &&\n        errors.Is(err, context.Canceled)\n}","tryCatchPattern":"if err := worker.PutInput(ctx, prv, input, res); err != nil {\n    if isFlowStoppedErr(err) {\n        // flow terminated before input was accepted; notify user, don't retry on same flow\n        return ErrFlowNotRunning\n    }\n    return err\n}","preventionTips":["Check flow status (running/finished/stopped) before submitting input.","Keep the request context derived from the same lifetime as the flow worker.","Subscribe to flow status events to disable input UI when the flow is stopping.","Don't call PutInput concurrently with Stop()."],"tags":["go","context-cancelled","concurrency","flow"],"backgroundTag":"context-canceled","analyzedSha":"ea665308baaff015b226f308438a68d929d0f29b","analyzedAt":"2026-09-01T14:16:31.421Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T10:18:20.063Z"}