finish updates the debug step with final status and data. A mutex guards the write so concurrent callers (e.g. retried stream wrappers sharing a reuse handle) don't race. Later retries are allowed to overwrite earlier failure results so the step reflects the final outcome, but stale callbacks cann
( ctx context.Context, status Status, response any, usage any, errPayload any, metadata any, )
| 293 | // overwrite earlier failure results so the step reflects the final |
| 294 | // outcome, but stale callbacks cannot regress a terminal state. |
| 295 | func (h *stepHandle) finish( |
| 296 | ctx context.Context, |
| 297 | status Status, |
| 298 | response any, |
| 299 | usage any, |
| 300 | errPayload any, |
| 301 | metadata any, |
| 302 | ) { |
| 303 | if h == nil || h.stepCtx == nil { |
| 304 | return |
| 305 | } |
| 306 | |
| 307 | h.mu.Lock() |
| 308 | defer h.mu.Unlock() |
| 309 | |
| 310 | // Reject stale callbacks that would regress a terminal state. |
| 311 | // Status priority: in_progress < interrupted < error < completed. |
| 312 | // A tardy safety-net writing "interrupted" cannot clobber a step |
| 313 | // that already reached "completed" or "error" from a real retry. |
| 314 | // Equal-priority updates are allowed so that retries ending in the |
| 315 | // same terminal class (e.g. error → error under ReuseStep) can |
| 316 | // still update the step with newer attempt data. |
| 317 | if h.status.IsTerminal() && status.Priority() < h.status.Priority() { |
| 318 | return |
| 319 | } |
| 320 | |
| 321 | h.status = status |
| 322 | h.response = response |
| 323 | h.usage = usage |
| 324 | h.err = errPayload |
| 325 | h.metadata = metadata |
| 326 | if errPayload != nil { |
| 327 | h.hadError = true |
| 328 | } |
| 329 | if h.svc == nil { |
| 330 | return |
| 331 | } |
| 332 | |
| 333 | updateCtx, cancel := stepFinalizeContext(ctx) |
| 334 | defer cancel() |
| 335 | |
| 336 | // When the step completes successfully after a prior failed |
| 337 | // attempt, the error field must be explicitly cleared. A plain |
| 338 | // nil would leave the COALESCE-based SQL untouched, so we send |
| 339 | // jsonClear{} which serializes as a valid JSONB null. Only do |
| 340 | // this when a prior error was actually recorded; otherwise |
| 341 | // clean successes would get a spurious JSONB null that downstream |
| 342 | // aggregation could misread as an error. |
| 343 | errValue := errPayload |
| 344 | if errValue == nil && status == StatusCompleted && h.hadError { |
| 345 | errValue = jsonClear{} |
| 346 | } |
| 347 | |
| 348 | if _, updateErr := h.svc.UpdateStep(updateCtx, UpdateStepParams{ |
| 349 | ID: h.stepCtx.StepID, |
| 350 | ChatID: h.stepCtx.ChatID, |
| 351 | Status: status, |
| 352 | NormalizedResponse: response, |