diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index be663c0061..5136eed9be 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go @@ -568,7 +568,13 @@ func (s *Service) reuseSessionForStep(ctx context.Context, taskID string, curren // and gets delivered to the wrong agent the next time that previous // session is reused (e.g. on the on_turn_complete bounce back). if s.messageQueue != nil { - s.messageQueue.TransferSession(ctx, currentSession.ID, existing.ID) + if err := s.messageQueue.TransferSession(ctx, currentSession.ID, existing.ID); err != nil { + // Fail closed: the workflow switch reuses an existing session, but + // orphaning a queued hand-off prompt on the previous session would + // silently misroute the next prompt. Stop here and surface the + // error so the caller can decide whether to retry. + return nil, fmt.Errorf("transfer queued state to reused session: %w", err) + } } s.completeAndStopSession(ctx, taskID, currentSession) @@ -650,7 +656,17 @@ func (s *Service) createNewSessionForStep(ctx context.Context, taskID string, cu // pending move from the old session to the new one — the queue is keyed by // session ID, and without this the prompt would never reach the new agent. if s.messageQueue != nil { - s.messageQueue.TransferSession(ctx, currentSession.ID, newSession.ID) + if err := s.messageQueue.TransferSession(ctx, currentSession.ID, newSession.ID); err != nil { + s.logger.Error("transfer queue to new session failed; queued prompts on the previous session will not be drained", + zap.String("from_session_id", currentSession.ID), + zap.String("to_session_id", newSession.ID), + zap.Error(err)) + // Continue anyway: the new session is already created and committed + // upstream. Failing closed here would leave the workflow in a + // half-switched state (new session exists but caller thinks it + // failed). The error is surfaced via logs and the orphaned entries + // stay safely in the old session for manual recovery. + } } // Promote the new session to primary so it's loaded when navigating back to this task. diff --git a/apps/backend/internal/orchestrator/messagequeue/service.go b/apps/backend/internal/orchestrator/messagequeue/service.go index bbd185d1e1..4dfbc994ee 100644 --- a/apps/backend/internal/orchestrator/messagequeue/service.go +++ b/apps/backend/internal/orchestrator/messagequeue/service.go @@ -176,19 +176,22 @@ func (s *Service) GetStatus(ctx context.Context, sessionID string) *QueueStatus } // TransferSession moves any queued messages and pending move from one session -// to another. Used by workflow session switches (when a target step has a -// different agent profile and switchSessionForStep creates a new session). -func (s *Service) TransferSession(ctx context.Context, oldSessionID, newSessionID string) { +// to another. Used by workflow session switches. Returns an error so the +// caller can fail closed instead of silently leaving entries orphaned on the +// old session — a transfer that no-ops without a signal would let the workflow +// step move forward while the queue sticks behind. +func (s *Service) TransferSession(ctx context.Context, oldSessionID, newSessionID string) error { if err := s.repo.TransferSession(ctx, oldSessionID, newSessionID); err != nil { s.logger.Error("transfer session failed", zap.String("from_session_id", oldSessionID), zap.String("to_session_id", newSessionID), zap.Error(err)) - return + return err } s.logger.Info("transferred queue between sessions", zap.String("from_session_id", oldSessionID), zap.String("to_session_id", newSessionID)) + return nil } // SetPendingMove records a pending move for a session (replaces any existing one). diff --git a/apps/web/lib/api/domains/queue-api.test.ts b/apps/web/lib/api/domains/queue-api.test.ts index d11bff394c..aa4ec28b7a 100644 --- a/apps/web/lib/api/domains/queue-api.test.ts +++ b/apps/web/lib/api/domains/queue-api.test.ts @@ -11,6 +11,7 @@ describe("rethrowQueueError", () => { }), ).toThrow(QueueFullError); + expect.assertions(5); try { rethrowQueueError({ code: "queue_full", @@ -18,6 +19,7 @@ describe("rethrowQueueError", () => { }); } catch (err) { const qf = err as QueueFullError; + expect(qf).toBeInstanceOf(QueueFullError); expect(qf.queueSize).toBe(9); expect(qf.max).toBe(10); expect(qf.code).toBe("queue_full"); @@ -25,10 +27,12 @@ describe("rethrowQueueError", () => { }); it("defaults missing queue_size / max to 0 when details are sparse", () => { + expect.assertions(3); try { rethrowQueueError({ code: "queue_full" }); } catch (err) { const qf = err as QueueFullError; + expect(qf).toBeInstanceOf(QueueFullError); expect(qf.queueSize).toBe(0); expect(qf.max).toBe(0); } diff --git a/apps/web/lib/api/domains/queue-api.ts b/apps/web/lib/api/domains/queue-api.ts index b49355d116..3b4d516fad 100644 --- a/apps/web/lib/api/domains/queue-api.ts +++ b/apps/web/lib/api/domains/queue-api.ts @@ -33,7 +33,14 @@ type WSError = { }; function asWSError(err: unknown): WSError | undefined { - if (typeof err === "object" && err !== null && "code" in err) { + // Some WS error payloads omit `code` and only carry `message`/`details`; + // narrowing on `code` alone would drop those and stringify the whole object + // as the eventual Error message ("[object Object]"). Real Error instances + // are skipped so they pass through unchanged in `rethrowQueueError`. + if (typeof err !== "object" || err === null || err instanceof Error) { + return undefined; + } + if ("code" in err || "message" in err || "details" in err) { return err as WSError; } return undefined;