diff --git a/apps/mobile/src/app/(app)/(tabs)/_layout.tsx b/apps/mobile/src/app/(app)/(tabs)/_layout.tsx index ac8c2c7e9c5..5cd172283d5 100644 --- a/apps/mobile/src/app/(app)/(tabs)/_layout.tsx +++ b/apps/mobile/src/app/(app)/(tabs)/_layout.tsx @@ -166,7 +166,10 @@ export default function TabsLayout() { const needsInputRows = activeSessions.map(session => ({ id: session.id, status: session.status, - isAcked: isAttentionAcked(session.id, session.status), + isAcked: isAttentionAcked(session.id, { + raiseId: session.statusUpdatedAt ?? session.status, + status: session.status, + }), })); const needsInputCount = needsInputRows.filter(session => shouldShowNeedsInput({ diff --git a/apps/mobile/src/components/agents/agents-tab-badge.mounted.test.tsx b/apps/mobile/src/components/agents/agents-tab-badge.mounted.test.tsx index d6ca3c3081a..ca45c749dae 100644 --- a/apps/mobile/src/components/agents/agents-tab-badge.mounted.test.tsx +++ b/apps/mobile/src/components/agents/agents-tab-badge.mounted.test.tsx @@ -1,3 +1,4 @@ +/* eslint-disable max-lines -- one cohesive mounted suite: the tab badge, the live header, and the ack store share the mock harness */ import { createElement } from 'react'; import { QueryClientProvider } from '@tanstack/react-query'; import { act, type ReactTestInstance } from '@/test/renderer'; @@ -27,6 +28,27 @@ import { renderWithProviders, waitFor } from '@/test/render-with-providers'; type Mount = Awaited>; const mounts: Mount[] = []; +/** The tray rows a dated raise renders: one waiting permission, one working. */ +function datedRaise(statusUpdatedAt: string): ActiveSession[] { + return [ + { + id: 'personal-0', + connectionId: 'cli', + title: 'Session 0', + status: 'permission', + organizationId: null, + statusUpdatedAt, + }, + { + id: 'personal-1', + connectionId: 'cli', + title: 'Session 1', + status: 'busy', + organizationId: null, + }, + ]; +} + async function mount(queryClient = makeTestQueryClient()) { const result = await renderWithProviders(createElement(CountSurfaces), { queryClient }); mounts.push(result); @@ -136,12 +158,12 @@ describe('Agents needs-input badge and shared live count', () => { const cached = result.queryClient.getQueryData(key()); const fetchCount = fetchSessions.mock.calls.length; act(() => { - ackSessionAttention('personal-0'); + ackSessionAttention('personal-0', status); }); expectCounts(result.renderer, 1, 3); expect(__peekSessionAttentionForTests('personal-0')).toEqual({ raiseId: status }); act(() => { - ackSessionAttention('personal-1'); + ackSessionAttention('personal-1', status); }); expectCounts(result.renderer, undefined, 3); expect(result.queryClient.getQueryData(key())).toBe(cached); @@ -155,11 +177,11 @@ describe('Agents needs-input badge and shared live count', () => { const result = await mount(); await updateSessions(result, sessions(['question'])); act(() => { - ackSessionAttention('personal-0'); + ackSessionAttention('personal-0', 'question'); }); expectCounts(result.renderer, undefined, 1); await updateSessions(result, sessions([status])); - expect(isAttentionAcked('personal-0', 'question')).toBe(false); + expect(isAttentionAcked('personal-0', { raiseId: 'question' })).toBe(false); expectCounts(result.renderer, status === 'permission' ? 1 : undefined, 1); await updateSessions(result, sessions(['question'])); expectCounts(result.renderer, 1, 1); @@ -167,12 +189,15 @@ describe('Agents needs-input badge and shared live count', () => { ); it.each([ - { status: 'question', raiseId: 'question' }, - { status: 'question', raiseId: null }, - { status: 'busy', raiseId: 'question' }, + { status: 'question', raiseId: 'question', suppresses: true }, + // A blob from a build that stored unbound acks names no raise, so it is + // dropped on hydration and suppresses nothing. + { status: 'question', raiseId: null, suppresses: false }, + // The raise it names is gone: the busy row deletes the entry. + { status: 'busy', raiseId: 'question', suppresses: false }, ] as const)( 'reconciles hydrated $raiseId acknowledgments against $status', - async ({ status, raiseId }) => { + async ({ status, raiseId, suppresses }) => { const pending = Promise.withResolvers(); attentionKv.getItem.mockReturnValueOnce(pending.promise); const hydration = __hydrateSessionAttentionForTests(); @@ -193,15 +218,34 @@ describe('Agents needs-input badge and shared live count', () => { ); await hydration; }); - expectCounts(result.renderer, 1, 2); + // The row the dropped ack named is answered only when the ack survived. + const answeredCount = suppresses || status !== 'question' ? 1 : 2; + expectCounts(result.renderer, answeredCount, 2); expect(__peekSessionAttentionForTests('personal-0')).toEqual( - status === 'question' ? { raiseId: 'question' } : undefined + suppresses ? { raiseId: 'question' } : undefined ); await updateSessions(result, sessions(['question', 'permission'])); - expectCounts(result.renderer, status === 'question' ? 1 : 2, 2); + expectCounts(result.renderer, suppresses ? 1 : 2, 2); } ); + it('counts a same-status re-raise after an ack once the row names its raise', async () => { + const result = await mount(); + + await updateSessions(result, datedRaise('2026-09-18T21:00:00.000Z')); + expectCounts(result.renderer, 1, 2); + + // The answer path acks the raise its row names. + act(() => { + ackSessionAttention('personal-0', 'permission', '2026-09-18T21:00:00.000Z'); + }); + expectCounts(result.renderer, undefined, 2); + + // A permission raised later reads as its own raise, not the answered one. + await updateSessions(result, datedRaise('2026-09-18T21:30:00.000Z')); + expectCounts(result.renderer, 1, 2); + }); + it('hides counts while the initial live query loads', async () => { const { renderer, queryClient } = await mount(); expect(queryClient.getQueryState(key())?.fetchStatus).toBe('fetching'); @@ -275,7 +319,7 @@ describe('Agents needs-input badge and shared live count', () => { const result = await mount(queryClient); expectCounts(result.renderer, 2, 3); act(() => { - ackSessionAttention('org-a-0'); + ackSessionAttention('org-a-0', 'question'); }); expectCounts(result.renderer, 1, 3); const pending = Promise.withResolvers<{ sessions: ActiveSession[] }>(); @@ -293,7 +337,7 @@ describe('Agents needs-input badge and shared live count', () => { queryClient.setQueryData(key('org-a'), { sessions: sessions(['busy'], 'org-a') }); }); expectCounts(result.renderer); - expect(isAttentionAcked('org-a-0', 'question')).toBe(true); + expect(isAttentionAcked('org-a-0', { raiseId: 'question' })).toBe(true); fetchSessions.mockResolvedValue({ sessions: [ ...sessions(['question', 'busy'], 'org-b'), @@ -306,7 +350,7 @@ describe('Agents needs-input badge and shared live count', () => { }); await waitFor(() => agentsOptions(result.renderer).tabBarBadge === 1); expectCounts(result.renderer, 1, 2); - expect(isAttentionAcked('org-a-0', 'question')).toBe(true); + expect(isAttentionAcked('org-a-0', { raiseId: 'question' })).toBe(true); expect( queryClient .getQueryCache() diff --git a/apps/mobile/src/components/agents/remote-session-row.tsx b/apps/mobile/src/components/agents/remote-session-row.tsx index ca95a9e543c..5318d7f18ea 100644 --- a/apps/mobile/src/components/agents/remote-session-row.tsx +++ b/apps/mobile/src/components/agents/remote-session-row.tsx @@ -138,12 +138,14 @@ export const RemoteSessionRow = memo(function RemoteSessionRow({ // bumps the shared revision, so the flag re-derives when it ticks. A parent // re-render with an unchanged payload and no ack reuses this result. const needsInput = useMemo( - () => - shouldShowNeedsInput({ + () => { + const raiseId = session.statusUpdatedAt ?? session.status; + return shouldShowNeedsInput({ status: session.status, - raiseId: session.status, - isAcked: isAttentionAcked(session.id, session.status), - }), + raiseId, + isAcked: isAttentionAcked(session.id, { raiseId, status: session.status }), + }); + }, // eslint-disable-next-line react/exhaustive-deps -- the revision is a real input: `isAttentionAcked` reads the ack store, so the flag must re-derive when that store ticks. [session, revision] ); diff --git a/apps/mobile/src/components/agents/session-detail-content.test.ts b/apps/mobile/src/components/agents/session-detail-content.test.ts index b1c81eb0802..329680103db 100644 --- a/apps/mobile/src/components/agents/session-detail-content.test.ts +++ b/apps/mobile/src/components/agents/session-detail-content.test.ts @@ -522,6 +522,9 @@ vi.mock('@/lib/hooks/use-session-model-options', () => { vi.mock('@/lib/hooks/use-remote-model-catalog-retry', () => ({ useRemoteModelCatalogRetry: vi.fn(), })); +vi.mock('@/lib/hooks/use-session-raise-reader', () => ({ + useSessionRaiseReader: () => () => null, +})); vi.mock('@/lib/hooks/use-theme-colors', () => ({ useThemeColors: () => ({}) })); vi.mock('@/lib/persist/drafts', () => ({ agentComposerDraftKey: (id: string) => id })); vi.mock('@/lib/persist/use-draft-load', () => ({ diff --git a/apps/mobile/src/components/agents/session-detail-content.tsx b/apps/mobile/src/components/agents/session-detail-content.tsx index 493146fd222..81116dddc9a 100644 --- a/apps/mobile/src/components/agents/session-detail-content.tsx +++ b/apps/mobile/src/components/agents/session-detail-content.tsx @@ -207,6 +207,7 @@ import { useSessionModelOptions, } from '@/lib/hooks/use-session-model-options'; import { useRemoteModelCatalogRetry } from '@/lib/hooks/use-remote-model-catalog-retry'; +import { useSessionRaiseReader } from '@/lib/hooks/use-session-raise-reader'; import { buildContinueHref, buildContinuePrefillParams, @@ -481,6 +482,7 @@ export function SessionDetailContent({ const analyticsSurface: AnalyticsSurface = fetchedData?.cloudAgentSessionId ? 'cloud-agent' : 'remote-session'; + const readObservedRaise = useSessionRaiseReader(); const { isAnswering, @@ -496,6 +498,7 @@ export function SessionDetailContent({ activeQuestion, activePermission, surface: analyticsSurface, + readObservedRaise, }); const autoApproveState = resolveSessionAutoApproveState({ diff --git a/apps/mobile/src/components/agents/session-detail-queue.test.ts b/apps/mobile/src/components/agents/session-detail-queue.test.ts index abaf7a646b1..a813a100f36 100644 --- a/apps/mobile/src/components/agents/session-detail-queue.test.ts +++ b/apps/mobile/src/components/agents/session-detail-queue.test.ts @@ -320,6 +320,9 @@ vi.mock('@/lib/hooks/use-session-model-options', () => ({ vi.mock('@/lib/hooks/use-remote-model-catalog-retry', () => ({ useRemoteModelCatalogRetry: vi.fn(), })); +vi.mock('@/lib/hooks/use-session-raise-reader', () => ({ + useSessionRaiseReader: () => () => null, +})); vi.mock('@/lib/use-github-repos-refresh', () => ({ useGitHubReposRefresh: () => ({ openGitHubIntegration: vi.fn() }), })); diff --git a/apps/mobile/src/components/agents/session-preview-header.tsx b/apps/mobile/src/components/agents/session-preview-header.tsx index ed547c111bd..5ed313302fe 100644 --- a/apps/mobile/src/components/agents/session-preview-header.tsx +++ b/apps/mobile/src/components/agents/session-preview-header.tsx @@ -51,7 +51,7 @@ export function SessionPreviewHeaderMeta({ ? shouldShowNeedsInput({ status: liveRow.status, raiseId, - isAcked: isAttentionAcked(target.sessionId, raiseId), + isAcked: isAttentionAcked(target.sessionId, { raiseId, status: liveRow.status }), }) : target.needsInput; const liveStatus = liveRow?.status ?? null; diff --git a/apps/mobile/src/components/agents/session-row.mounted.test.tsx b/apps/mobile/src/components/agents/session-row.mounted.test.tsx index a1e8d703420..73ee30d64b1 100644 --- a/apps/mobile/src/components/agents/session-row.mounted.test.tsx +++ b/apps/mobile/src/components/agents/session-row.mounted.test.tsx @@ -13,7 +13,7 @@ import { prefetchSessionTranscript } from '@/lib/agent-session-cache'; import { createKiloAppQueryClient } from '@/lib/query-client'; import { i18n } from '@/i18n'; import { type ActiveSession, type StoredSession } from '@/lib/hooks/use-agent-sessions'; -import { __resetSessionAttentionForTests } from '@/lib/session-attention'; +import { __resetSessionAttentionForTests, ackSessionAttention } from '@/lib/session-attention'; import { RemoteSessionRow } from './remote-session-row'; import { formatScheduledWake } from './session-list-helpers'; import { @@ -294,6 +294,41 @@ describe('StoredSessionRow live speech', () => { } ); + it('brings the badge back for a raise the user has not answered', () => { + const answered = { + ...session, + status: 'permission', + status_updated_at: '2026-09-18T21:00:00.000Z', + }; + const renderer = mount(row({ session: answered, live: true })); + + expect(texts(renderer)).toContain('NEEDS INPUT'); + + // The answer path acks the raise it answered. + act(() => { + ackSessionAttention('stored-1', 'permission', '2026-09-18T21:00:00.000Z'); + }); + expect(texts(renderer)).not.toContain('NEEDS INPUT'); + + // A permission raised after the answered one reaches the row before the + // answered status round-trip lands: it must show again. + act(() => { + renderer.update( + wrapped( + row({ + session: { ...answered, status_updated_at: '2026-09-18T21:30:00.000Z' }, + live: true, + }) + ) + ); + }); + + expect(texts(renderer)).toContain('NEEDS INPUT'); + expect(hosts(renderer, 'Pressable')[0]?.props.accessibilityLabel).toBe( + 'Fix login bug, needs input, feature/live, and CLI' + ); + }); + it('reads SCHEDULED (label only) for a stored scheduled row instead of Idle', () => { // The scheduled branch keys off the status, not the live flag: a stored // history row has no wake time, so it shows the label alone. diff --git a/apps/mobile/src/components/agents/session-row.tsx b/apps/mobile/src/components/agents/session-row.tsx index 98e45853150..6a3d8b276f1 100644 --- a/apps/mobile/src/components/agents/session-row.tsx +++ b/apps/mobile/src/components/agents/session-row.tsx @@ -152,7 +152,7 @@ export function StoredSessionRow({ const needsInput = shouldShowNeedsInput({ status: session.status, raiseId, - isAcked: isAttentionAcked(session.session_id, raiseId), + isAcked: isAttentionAcked(session.session_id, { raiseId, status: session.status }), }); useEffect(() => { reconcileSessionAttention(session.session_id, session.status, session.status_updated_at); diff --git a/apps/mobile/src/components/agents/use-interaction-handlers-scoping.test.ts b/apps/mobile/src/components/agents/use-interaction-handlers-scoping.test.ts index 2f7121421c4..d75c6c54f06 100644 --- a/apps/mobile/src/components/agents/use-interaction-handlers-scoping.test.ts +++ b/apps/mobile/src/components/agents/use-interaction-handlers-scoping.test.ts @@ -32,9 +32,16 @@ type HookDispatcher = { }; type InteractionHandlersArgs = Parameters[0]; +/** The cases here never read a live row, so they may omit the raise reader. */ +type HandlerArgs = Omit; type InteractionHandlersResult = ReturnType; -function renderInteractionHandlers(args: InteractionHandlersArgs) { +function renderInteractionHandlers( + args: Omit & + Partial> +) { + // Read per render: the scoping cases mutate `args` between renders. + const readObservedRaise = args.readObservedRaise ?? (() => null); const reactInternals = React as typeof React & ReactInternals; const hookState: unknown[] = []; let hookIndex = 0; @@ -71,7 +78,7 @@ function renderInteractionHandlers(args: InteractionHandlersArgs) { reactInternals.__CLIENT_INTERNALS_DO_NOT_USE_OR_WARN_USERS_THEY_CANNOT_UPGRADE.H = dispatcher; try { const mountHandlers = useInteractionHandlers; - return mountHandlers(args); + return mountHandlers({ ...args, readObservedRaise }); } finally { reactInternals.__CLIENT_INTERNALS_DO_NOT_USE_OR_WARN_USERS_THEY_CANNOT_UPGRADE.H = previousDispatcher; @@ -92,7 +99,7 @@ describe('useInteractionHandlers request-id scoping', () => { rejectQuestion: vi.fn(), respondToPermission: vi.fn(), }; - const args: InteractionHandlersArgs = { + const args: HandlerArgs = { manager: manager as never, kiloSessionId: 'kilo-session-1', activeQuestion: { requestId: 'q-req-1' }, @@ -125,7 +132,7 @@ describe('useInteractionHandlers request-id scoping', () => { rejectQuestion: vi.fn(), respondToPermission: vi.fn(), }; - const args: InteractionHandlersArgs = { + const args: HandlerArgs = { manager: manager as never, kiloSessionId: 'kilo-session-1', activeQuestion: { requestId: 'q-req-1' }, @@ -168,7 +175,7 @@ describe('useInteractionHandlers request-id scoping', () => { rejectQuestion: vi.fn(), respondToPermission: vi.fn(), }; - const args: InteractionHandlersArgs = { + const args: HandlerArgs = { manager: manager as never, kiloSessionId: 'kilo-session-1', activeQuestion: { requestId: 'q-req-1' }, @@ -207,7 +214,7 @@ describe('useInteractionHandlers request-id scoping', () => { rejectQuestion: vi.fn(), respondToPermission: vi.fn().mockReturnValue(respondPromise), }; - const args: InteractionHandlersArgs = { + const args: HandlerArgs = { manager: manager as never, kiloSessionId: 'kilo-session-1', activeQuestion: null, @@ -240,7 +247,7 @@ describe('useInteractionHandlers request-id scoping', () => { rejectQuestion: vi.fn(), respondToPermission: vi.fn().mockReturnValue(respondPromise), }; - const args: InteractionHandlersArgs = { + const args: HandlerArgs = { manager: manager as never, kiloSessionId: 'kilo-session-1', activeQuestion: null, @@ -283,7 +290,7 @@ describe('useInteractionHandlers request-id scoping', () => { rejectQuestion: vi.fn(), respondToPermission, }; - const args: InteractionHandlersArgs = { + const args: HandlerArgs = { manager: manager as never, kiloSessionId: 'kilo-session-1', activeQuestion: null, diff --git a/apps/mobile/src/components/agents/use-interaction-handlers.test.ts b/apps/mobile/src/components/agents/use-interaction-handlers.test.ts index aaae7835523..280f673a9d5 100644 --- a/apps/mobile/src/components/agents/use-interaction-handlers.test.ts +++ b/apps/mobile/src/components/agents/use-interaction-handlers.test.ts @@ -1,3 +1,4 @@ +/* eslint-disable max-lines -- one cohesive suite: the fake dispatcher, the ack contracts, and the failure paths share the same harness */ import * as React from 'react'; import { beforeEach, describe, expect, it, vi } from 'vitest'; @@ -43,7 +44,12 @@ type HookDispatcher = { type InteractionHandlersArgs = Parameters[0]; type InteractionHandlersResult = ReturnType; -function renderInteractionHandlers(args: InteractionHandlersArgs) { +function renderInteractionHandlers( + args: Omit & + Partial> +) { + // Read per render: the scoping cases mutate `args` between renders. + const readObservedRaise = args.readObservedRaise ?? (() => null); const reactInternals = React as typeof React & ReactInternals; const hookState: unknown[] = []; let hookIndex = 0; @@ -81,7 +87,7 @@ function renderInteractionHandlers(args: InteractionHandlersArgs) { try { // Alias avoids rules-of-hooks lexical false positives under the fake dispatcher. const mountHandlers = useInteractionHandlers; - return mountHandlers(args); + return mountHandlers({ ...args, readObservedRaise }); } finally { reactInternals.__CLIENT_INTERNALS_DO_NOT_USE_OR_WARN_USERS_THEY_CANNOT_UPGRADE.H = previousDispatcher; @@ -115,17 +121,17 @@ describe('useInteractionHandlers attention ack', () => { await render().handleAnswerQuestion([['yes']]); expect(manager.answerQuestion).toHaveBeenCalledWith('q1', [['yes']]); - expect(isAttentionAcked('kilo-session-1', 'R1')).toBe(true); + expect(isAttentionAcked('kilo-session-1', { raiseId: 'question' })).toBe(true); expect( shouldShowNeedsInput({ status: 'question', - raiseId: 'R1', - isAcked: isAttentionAcked('kilo-session-1', 'R1'), + raiseId: 'question', + isAcked: isAttentionAcked('kilo-session-1', { raiseId: 'question' }), }) ).toBe(false); - expect(__peekSessionAttentionForTests('kilo-session-1')).toEqual({ raiseId: null }); + expect(__peekSessionAttentionForTests('kilo-session-1')).toEqual({ raiseId: 'question' }); // Must key by kilo id, never the cloud-agent id. - expect(isAttentionAcked('cloud-agent-id', 'R1')).toBe(false); + expect(isAttentionAcked('cloud-agent-id', { raiseId: 'question' })).toBe(false); expect(captureEvent).toHaveBeenCalledWith('question_answered', { surface: 'remote-session', skipped: false, @@ -133,6 +139,99 @@ describe('useInteractionHandlers attention ack', () => { expect(toastError).not.toHaveBeenCalled(); }); + it('acks the raise the app observed for the session', async () => { + const manager = { + answerQuestion: vi.fn().mockResolvedValue(undefined), + rejectQuestion: vi.fn(), + respondToPermission: vi.fn(), + }; + const { render } = renderInteractionHandlers({ + manager: manager as never, + kiloSessionId: 'kilo-session-1', + activeQuestion: { requestId: 'q1' }, + activePermission: null, + surface: 'remote-session', + readObservedRaise: () => ({ + status: 'question', + statusUpdatedAt: '2026-09-18T21:00:00.000Z', + }), + }); + + await render().handleAnswerQuestion([['yes']]); + + expect(__peekSessionAttentionForTests('kilo-session-1')).toEqual({ + raiseId: '2026-09-18T21:00:00.000Z', + }); + expect(isAttentionAcked('kilo-session-1', { raiseId: '2026-09-18T21:00:00.000Z' })).toBe(true); + // A question raised later is a different raise and keeps its badge. + expect(isAttentionAcked('kilo-session-1', { raiseId: '2026-09-18T21:30:00.000Z' })).toBe(false); + }); + + it.each(['answer', 'skip', 'permission'] as const)( + 'keeps the %s ack on the submitted raise when another arrives in flight', + async operation => { + const gate = Promise.withResolvers(); + const manager = { + answerQuestion: vi.fn().mockReturnValue(gate.promise), + rejectQuestion: vi.fn().mockReturnValue(gate.promise), + respondToPermission: vi.fn().mockReturnValue(gate.promise), + }; + const status = operation === 'permission' ? 'permission' : 'question'; + const firstRaise = '2026-09-18T21:00:00.000Z'; + const nextRaise = '2026-09-18T21:30:00.000Z'; + let observed = { status, statusUpdatedAt: firstRaise }; + const { render } = renderInteractionHandlers({ + manager: manager as never, + kiloSessionId: 'kilo-session-1', + activeQuestion: { requestId: 'q1' }, + activePermission: { requestId: 'p1' }, + surface: 'remote-session', + readObservedRaise: () => observed, + }); + const handlers = render(); + const pending = (async () => { + if (operation === 'answer') { + await handlers.handleAnswerQuestion([['yes']]); + } else if (operation === 'skip') { + await handlers.handleRejectQuestion(); + } else { + await handlers.handleRespondToPermission('once'); + } + })(); + + observed = { status, statusUpdatedAt: nextRaise }; + gate.resolve(undefined); + await pending; + + expect(isAttentionAcked('kilo-session-1', { raiseId: firstRaise, status })).toBe(true); + expect(isAttentionAcked('kilo-session-1', { raiseId: nextRaise, status })).toBe(false); + } + ); + + it('ignores an observed raise of the other kind', async () => { + const manager = { + answerQuestion: vi.fn().mockResolvedValue(undefined), + rejectQuestion: vi.fn(), + respondToPermission: vi.fn(), + }; + const { render } = renderInteractionHandlers({ + manager: manager as never, + kiloSessionId: 'kilo-session-1', + activeQuestion: { requestId: 'q1' }, + activePermission: null, + surface: 'remote-session', + // The row already reads the next raise: it is not the one being answered. + readObservedRaise: () => ({ + status: 'permission', + statusUpdatedAt: '2026-09-18T21:00:00.000Z', + }), + }); + + await render().handleAnswerQuestion([['yes']]); + + expect(__peekSessionAttentionForTests('kilo-session-1')).toEqual({ raiseId: 'question' }); + }); + it('acks after a successful skip/reject', async () => { const manager = { answerQuestion: vi.fn(), @@ -150,7 +249,7 @@ describe('useInteractionHandlers attention ack', () => { await render().handleRejectQuestion(); expect(manager.rejectQuestion).toHaveBeenCalledWith('q1'); - expect(isAttentionAcked('kilo-session-1', 'R1')).toBe(true); + expect(isAttentionAcked('kilo-session-1', { raiseId: 'question' })).toBe(true); expect(captureEvent).toHaveBeenCalledWith('question_answered', { surface: 'cloud-agent', skipped: true, @@ -174,7 +273,7 @@ describe('useInteractionHandlers attention ack', () => { await render().handleRespondToPermission('once'); expect(manager.respondToPermission).toHaveBeenCalledWith('p1', 'once'); - expect(isAttentionAcked('kilo-session-1', 'R1')).toBe(true); + expect(isAttentionAcked('kilo-session-1', { raiseId: 'permission' })).toBe(true); expect(captureEvent).toHaveBeenCalledWith('permission_responded', { surface: 'remote-session', response: 'once', @@ -248,13 +347,13 @@ describe('useInteractionHandlers attention ack', () => { await render().handleAnswerQuestion([['yes']]); - expect(isAttentionAcked('kilo-session-1', 'R1')).toBe(false); + expect(isAttentionAcked('kilo-session-1', { raiseId: 'question' })).toBe(false); expect(__peekSessionAttentionForTests('kilo-session-1')).toBeUndefined(); expect( shouldShowNeedsInput({ status: 'question', - raiseId: 'R1', - isAcked: isAttentionAcked('kilo-session-1', 'R1'), + raiseId: 'question', + isAcked: isAttentionAcked('kilo-session-1', { raiseId: 'question' }), }) ).toBe(true); expect(toastError).toHaveBeenCalledWith('Failed to submit answer. Please try again.'); @@ -277,7 +376,7 @@ describe('useInteractionHandlers attention ack', () => { await render().handleRejectQuestion(); - expect(isAttentionAcked('kilo-session-1', 'R1')).toBe(false); + expect(isAttentionAcked('kilo-session-1', { raiseId: 'question' })).toBe(false); expect(toastError).toHaveBeenCalledWith('Failed to skip question. Please try again.'); expect(captureEvent).not.toHaveBeenCalled(); }); @@ -298,7 +397,7 @@ describe('useInteractionHandlers attention ack', () => { await render().handleRespondToPermission('reject'); - expect(isAttentionAcked('kilo-session-1', 'R1')).toBe(false); + expect(isAttentionAcked('kilo-session-1', { raiseId: 'permission' })).toBe(false); expect(toastError).toHaveBeenCalledWith('Failed to respond to permission. Please try again.'); expect(captureEvent).not.toHaveBeenCalled(); }); @@ -319,7 +418,7 @@ describe('useInteractionHandlers attention ack', () => { await render().handleAnswerQuestion([['ok']]); - expect(isAttentionAcked('session-a', 'R1')).toBe(true); - expect(isAttentionAcked('session-b', 'R1')).toBe(false); + expect(isAttentionAcked('session-a', { raiseId: 'question' })).toBe(true); + expect(isAttentionAcked('session-b', { raiseId: 'question' })).toBe(false); }); }); diff --git a/apps/mobile/src/components/agents/use-interaction-handlers.ts b/apps/mobile/src/components/agents/use-interaction-handlers.ts index 661f876eb9f..063de362cb8 100644 --- a/apps/mobile/src/components/agents/use-interaction-handlers.ts +++ b/apps/mobile/src/components/agents/use-interaction-handlers.ts @@ -13,6 +13,7 @@ import { QUESTION_ANSWERED_EVENT, } from '@/lib/analytics/posthog'; import { ackSessionAttention } from '@/lib/session-attention'; +import { type ObservedSessionRaise } from '@/lib/hooks/use-session-raise-reader'; import { type useSessionManager } from './session-provider'; @@ -26,6 +27,13 @@ type InteractionHandlersArgs = { activeQuestion: { requestId: string; questions?: unknown[] } | null; activePermission: { requestId: string } | null; surface: AnalyticsSurface; + /** + * The raise the app last observed for a session, read before submitting. + * The ack names that raise so it suppresses only the one the user answered; + * null when the session has no live row, and the ack then carries the status + * alone. + */ + readObservedRaise: (sessionId: string) => ObservedSessionRaise | null; }; export function useInteractionHandlers({ @@ -34,7 +42,15 @@ export function useInteractionHandlers({ activeQuestion, activePermission, surface, + readObservedRaise, }: InteractionHandlersArgs) { + const readAnsweredRaiseTime = useCallback( + (status: 'question' | 'permission') => { + const observed = readObservedRaise(kiloSessionId); + return observed?.status === status ? observed.statusUpdatedAt : undefined; + }, + [readObservedRaise, kiloSessionId] + ); const [answeringRequestId, setAnsweringRequestId] = useState(null); const [respondingRequestId, setRespondingRequestId] = useState(null); const [questionSubmissionError, setQuestionSubmissionError] = useState<{ @@ -52,11 +68,12 @@ export function useInteractionHandlers({ return false; } const requestId = activeQuestion.requestId; + const statusUpdatedAt = readAnsweredRaiseTime('question'); setQuestionSubmissionError(null); setAnsweringRequestId(requestId); try { await manager.answerQuestion(requestId, answers); - ackSessionAttention(kiloSessionId); + ackSessionAttention(kiloSessionId, 'question', statusUpdatedAt); captureEvent(QUESTION_ANSWERED_EVENT, { surface, skipped: false }); return true; } catch (error) { @@ -71,7 +88,7 @@ export function useInteractionHandlers({ setAnsweringRequestId(current => (current === requestId ? null : current)); } }, - [manager, kiloSessionId, activeQuestion, surface] + [manager, kiloSessionId, readAnsweredRaiseTime, activeQuestion, surface] ); const handleRejectQuestion = useCallback(async (): Promise => { @@ -79,11 +96,12 @@ export function useInteractionHandlers({ return false; } const requestId = activeQuestion.requestId; + const statusUpdatedAt = readAnsweredRaiseTime('question'); setQuestionSubmissionError(null); setAnsweringRequestId(requestId); try { await manager.rejectQuestion(requestId); - ackSessionAttention(kiloSessionId); + ackSessionAttention(kiloSessionId, 'question', statusUpdatedAt); captureEvent(QUESTION_ANSWERED_EVENT, { surface, skipped: true }); return true; } catch (error) { @@ -95,7 +113,7 @@ export function useInteractionHandlers({ } finally { setAnsweringRequestId(current => (current === requestId ? null : current)); } - }, [manager, kiloSessionId, activeQuestion, surface]); + }, [manager, kiloSessionId, readAnsweredRaiseTime, activeQuestion, surface]); const handleRespondToPermission = useCallback( async (response: 'once' | 'always' | 'reject'): Promise<'ok' | 'retryable' | 'terminal'> => { @@ -105,11 +123,12 @@ export function useInteractionHandlers({ return 'ok'; } const requestId = activePermission.requestId; + const statusUpdatedAt = readAnsweredRaiseTime('permission'); setPermissionSubmissionError(null); setRespondingRequestId(requestId); try { await manager.respondToPermission(requestId, response); - ackSessionAttention(kiloSessionId); + ackSessionAttention(kiloSessionId, 'permission', statusUpdatedAt); captureEvent(PERMISSION_RESPONDED_EVENT, { surface, response }); return 'ok'; } catch (error) { @@ -125,7 +144,7 @@ export function useInteractionHandlers({ setRespondingRequestId(current => (current === requestId ? null : current)); } }, - [manager, kiloSessionId, activePermission, surface] + [manager, kiloSessionId, readAnsweredRaiseTime, activePermission, surface] ); return { diff --git a/apps/mobile/src/components/launcher-surfaces-mount.tsx b/apps/mobile/src/components/launcher-surfaces-mount.tsx index aa96c63392a..0a3f4d23100 100644 --- a/apps/mobile/src/components/launcher-surfaces-mount.tsx +++ b/apps/mobile/src/components/launcher-surfaces-mount.tsx @@ -37,7 +37,10 @@ function selectWaitingSessions(sessions: readonly ActiveSession[]) { shouldShowNeedsInput({ status: session.status, raiseId: session.status, - isAcked: isAttentionAcked(session.id, session.status), + isAcked: isAttentionAcked(session.id, { + raiseId: session.statusUpdatedAt ?? session.status, + status: session.status, + }), }) ) .map(session => ({ diff --git a/apps/mobile/src/lib/active-sessions-live.attention.test.ts b/apps/mobile/src/lib/active-sessions-live.attention.test.ts index 2d9b82464d2..097c9649fdd 100644 --- a/apps/mobile/src/lib/active-sessions-live.attention.test.ts +++ b/apps/mobile/src/lib/active-sessions-live.attention.test.ts @@ -136,11 +136,46 @@ describe('applySessionStatusUpdated', () => { }); expect(result[0]?.statusUpdatedAt).toBe('2026-01-02T00:00:00.000Z'); }); + + it.each([ + ['idle', 'question'], + ['busy', 'permission'], + ['question', 'permission'], + ['permission', 'idle'], + ])('drops the previous status time on an undated %s to %s transition', (status, nextStatus) => { + const current = [makeCached({ id: 'a', status, statusUpdatedAt: '2026-01-01T00:00:00.000Z' })]; + const result = applySessionStatusUpdated(current, { sessionId: 'a', status: nextStatus }); + expect(result[0]?.status).toBe(nextStatus); + expect(result[0]).not.toHaveProperty('statusUpdatedAt'); + }); + + it('keeps the cached time and row for an undated same-status update', () => { + const current = [ + makeCached({ id: 'a', status: 'question', statusUpdatedAt: '2026-01-01T00:00:00.000Z' }), + ]; + const result = applySessionStatusUpdated(current, { sessionId: 'a', status: 'question' }); + expect(result[0]).toBe(current[0]); + expect(result[0]?.statusUpdatedAt).toBe('2026-01-01T00:00:00.000Z'); + }); + + it('keeps the same-status time when an undated update clears a scheduled wake time', () => { + const current = [ + makeCached({ + id: 'a', + status: 'scheduled', + scheduledAt: '2026-01-02T00:00:00.000Z', + statusUpdatedAt: '2026-01-01T00:00:00.000Z', + }), + ]; + const result = applySessionStatusUpdated(current, { sessionId: 'a', status: 'scheduled' }); + expect(result[0]?.scheduledAt).toBeUndefined(); + expect(result[0]?.statusUpdatedAt).toBe('2026-01-01T00:00:00.000Z'); + }); }); describe('statusUpdatedAt through socket merges', () => { it('keeps the status-change time through heartbeat and snapshot merges', () => { - // Socket rows never carry it; losing it blanks the Home newest session. + // Socket rows never carry it; an unchanged status retains its known time. const statusUpdatedAt = '2026-01-01T00:00:00.000Z'; const current = [makeCached({ id: 'a', statusUpdatedAt })]; const afterHeartbeat = mergeHeartbeatForActiveSessions(current, { @@ -153,4 +188,42 @@ describe('statusUpdatedAt through socket merges', () => { expect(afterHeartbeat[0]?.statusUpdatedAt).toBe(statusUpdatedAt); expect(afterSnapshot[0]?.statusUpdatedAt).toBe(statusUpdatedAt); }); + + it.each([ + ['idle', 'question'], + ['busy', 'permission'], + ['idle', 'busy'], + ])('drops stale %s time when heartbeat and snapshot resolve to %s', (status, liveStatus) => { + const current = [makeCached({ id: 'a', status, statusUpdatedAt: '2026-01-01T00:00:00.000Z' })]; + const afterHeartbeat = mergeHeartbeatForActiveSessions(current, { + connectionId: 'c1', + sessions: [{ id: 'a', status: liveStatus, title: 'A' }], + }); + const afterSnapshot = mergeSnapshotForActiveSessions(current, [ + { id: 'a', status: liveStatus, title: 'A', connectionId: 'c1' }, + ]); + for (const result of [afterHeartbeat, afterSnapshot]) { + expect(result[0]?.status).toBe(liveStatus); + expect(result[0]?.statusUpdatedAt).toBeUndefined(); + } + }); + + it.each([ + ['question', 'busy'], + ['permission', 'idle'], + ])('keeps held %s time when heartbeat and snapshot report %s', (status, liveStatus) => { + const statusUpdatedAt = '2026-01-01T00:00:00.000Z'; + const current = [makeCached({ id: 'a', status, statusUpdatedAt })]; + const afterHeartbeat = mergeHeartbeatForActiveSessions(current, { + connectionId: 'c1', + sessions: [{ id: 'a', status: liveStatus, title: 'A' }], + }); + const afterSnapshot = mergeSnapshotForActiveSessions(current, [ + { id: 'a', status: liveStatus, title: 'A', connectionId: 'c1' }, + ]); + for (const result of [afterHeartbeat, afterSnapshot]) { + expect(result[0]?.status).toBe(status); + expect(result[0]?.statusUpdatedAt).toBe(statusUpdatedAt); + } + }); }); diff --git a/apps/mobile/src/lib/active-sessions-live.ts b/apps/mobile/src/lib/active-sessions-live.ts index 92543e2e567..032e6e3f332 100644 --- a/apps/mobile/src/lib/active-sessions-live.ts +++ b/apps/mobile/src/lib/active-sessions-live.ts @@ -198,12 +198,6 @@ type PreservedFields = { * but `lastActivityAt` may be legitimately absent (NULL in DB). */ lastActivityAt: string | undefined; - /** - * Sticky: heartbeat and `sessions.list` rows never carry the status-change - * time. Only a tRPC fetch or a `session.status.updated` event writes it, so - * a merge must keep it or the newest session and its age go blank. - */ - statusUpdatedAt: string | undefined; /** Sticky like the three above: WS payloads never carry an org id. */ organizationId: string | null | undefined; /** @@ -231,7 +225,6 @@ function readEnrichment(current: CachedActiveSession | undefined): PreservedFiel createdAt: current?.createdAt, updatedAt: current?.updatedAt, lastActivityAt: current?.lastActivityAt, - statusUpdatedAt: current?.statusUpdatedAt, organizationId: current?.organizationId, totalCostMicrodollars: current?.totalCostMicrodollars, associatedPr: current?.associatedPr, @@ -246,11 +239,12 @@ function withEnrichmentAndConnectionId( connectionId: string ): CachedActiveSession { const enrichment = readEnrichment(current); + const status = effectiveStatus(row.status, current?.status); return { id: row.id, // Sticky attention: a non-attention WS status must not clear a held // question/permission. "stored" for WS paths is the cached row status. - status: effectiveStatus(row.status, current?.status), + status, // The tray title is DB-authoritative (the router enriches it from // cli_sessions_v2), so once a row has been through a tRPC fetch a heartbeat's // CLI title must not overwrite it — nothing propagates a cloud rename back to @@ -265,6 +259,8 @@ function withEnrichmentAndConnectionId( // only when the WS row omits the field (legacy payloads). capabilities: row.capabilities ?? current?.capabilities, ...enrichment, + // Socket statuses are undated; a changed status cannot inherit the cached time. + statusUpdatedAt: status === current?.status ? current.statusUpdatedAt : undefined, // Wire wins, including omission: a CLI that reports `scheduled` with no // wake time must show scheduled with no time, not a cached timestamp. ...(row.scheduledAt === undefined ? {} : { scheduledAt: row.scheduledAt }), @@ -349,8 +345,8 @@ export function mergeHeartbeatForActiveSessions( /** * Apply an explicit status transition (including leaving attention). * Unknown session ids are ignored — live cache only holds active rows. The - * transition's own time replaces the row's `statusUpdatedAt`, so the newest - * session and its age do not wait for the next tRPC poll. + * transition's own time replaces the row's `statusUpdatedAt`. An undated + * transition preserves the cached time only when the status is unchanged. */ export function applySessionStatusUpdated( current: readonly CachedActiveSession[], @@ -377,6 +373,8 @@ export function applySessionStatusUpdated( } if (statusUpdatedAt !== undefined) { next.statusUpdatedAt = statusUpdatedAt; + } else if (row.status !== update.status) { + delete next.statusUpdatedAt; } return next; }); diff --git a/apps/mobile/src/lib/glanceable/approve-ask.test.ts b/apps/mobile/src/lib/glanceable/approve-ask.test.ts index 11c605baafe..f44859f25cc 100644 --- a/apps/mobile/src/lib/glanceable/approve-ask.test.ts +++ b/apps/mobile/src/lib/glanceable/approve-ask.test.ts @@ -334,7 +334,21 @@ describe('runGlanceableApprove', () => { readWaitingAsk.mockResolvedValue(PERMISSION_ASK); answerPermissionMutate.mockResolvedValue(undefined); await expect(runGlanceableApprove()).resolves.toEqual({ kind: 'approved' }); - expect(isAttentionAcked('ses_1', null)).toBe(true); + expect(isAttentionAcked('ses_1', { raiseId: 'permission' })).toBe(true); + }); + + it('acks the raise its ask recorded, not whatever raise is observed next', async () => { + readWaitingAsk.mockResolvedValue({ + ...PERMISSION_ASK, + statusUpdatedAt: '2026-09-18T21:00:00.000Z', + }); + answerPermissionMutate.mockResolvedValue(undefined); + + await expect(runGlanceableApprove()).resolves.toEqual({ kind: 'approved' }); + + expect(isAttentionAcked('ses_1', { raiseId: '2026-09-18T21:00:00.000Z' })).toBe(true); + // A permission raised after the answered one is not suppressed. + expect(isAttentionAcked('ses_1', { raiseId: '2026-09-18T21:30:00.000Z' })).toBe(false); }); it('acks a raise proven gone, and never acks a retryable failure', async () => { @@ -343,20 +357,20 @@ describe('runGlanceableApprove', () => { readWaitingAsk.mockResolvedValue(PERMISSION_ASK); answerPermissionMutate.mockRejectedValue(withCode('NOT_FOUND')); await expect(runGlanceableApprove()).resolves.toEqual({ kind: 'gone' }); - expect(isAttentionAcked('ses_1', null)).toBe(true); + expect(isAttentionAcked('ses_1', { raiseId: 'permission' })).toBe(true); // Retryable keeps the ask live: the next tap has to answer that same // session, so an ack here would hide the raise the user must retry. __resetSessionAttentionForTests(); answerPermissionMutate.mockRejectedValue(new Error('network down')); await expect(runGlanceableApprove()).resolves.toEqual({ kind: 'retryable' }); - expect(isAttentionAcked('ses_1', null)).toBe(false); + expect(isAttentionAcked('ses_1', { raiseId: 'permission' })).toBe(false); }); it('does not ack an ask it never answered', async () => { readWaitingAsk.mockResolvedValue({ ...PERMISSION_ASK, isCloudAgent: false }); await expect(runGlanceableApprove()).resolves.toEqual({ kind: 'none' }); - expect(isAttentionAcked('ses_1', null)).toBe(false); + expect(isAttentionAcked('ses_1', { raiseId: 'permission' })).toBe(false); }); it.each([ @@ -533,7 +547,7 @@ describe('refreshGlanceableSnapshot', () => { // row would republish the pre-action snapshot — the answered session still // waiting — on the ongoing card and the widget, which is what the ack is // for. The poll still reads the raw row: that is what proves the sync. - ackSessionAttention('ses_1'); + ackSessionAttention('ses_1', 'permission'); const publisher = makePublisher(); const rows = [{ id: 'ses_1', status: 'permission' }]; const fetchRows = vi.fn(async () => rows); diff --git a/apps/mobile/src/lib/glanceable/approve-ask.ts b/apps/mobile/src/lib/glanceable/approve-ask.ts index 7b04d58cb80..39dbe1d521d 100644 --- a/apps/mobile/src/lib/glanceable/approve-ask.ts +++ b/apps/mobile/src/lib/glanceable/approve-ask.ts @@ -211,10 +211,11 @@ export async function runGlanceableApprove( ): Promise { if (options.target !== undefined) { // A session card names its own session and request; the recorded ask is - // the aggregate card's and stays as it is. + // the aggregate card's and stays as it is. A card's Approve answers a + // permission, and the card carries no status time. const result = await answerPendingPermission(options.target, options); if (result.kind === 'approved' || result.kind === 'gone') { - ackSessionAttention(options.target.kiloSessionId); + ackSessionAttention(options.target.kiloSessionId, 'permission'); } return result; } @@ -227,7 +228,7 @@ export async function runGlanceableApprove( recordWaitingAsk(null); } if (result.kind === 'approved' || result.kind === 'gone') { - ackSessionAttention(ask.kiloSessionId); + ackSessionAttention(ask.kiloSessionId, ask.status, ask.statusUpdatedAt); } return result; } diff --git a/apps/mobile/src/lib/glanceable/approve-front-agent.test.ts b/apps/mobile/src/lib/glanceable/approve-front-agent.test.ts index 93a4eb1ba74..2b8d2dd38e8 100644 --- a/apps/mobile/src/lib/glanceable/approve-front-agent.test.ts +++ b/apps/mobile/src/lib/glanceable/approve-front-agent.test.ts @@ -90,7 +90,8 @@ function createHarness(input: HarnessInput) { throw input.destroyRejection; } }); - const ackSessionAttention = vi.fn<(kiloSessionId: string) => void>(); + const ackSessionAttention = + vi.fn<(kiloSessionId: string, status: string, statusUpdatedAt?: string | null) => void>(); let listCall = 0; const listSessions = vi.fn(async (_organizationId: string | null) => { listCall += 1; @@ -169,7 +170,11 @@ describe('approveFrontAgent', () => { expect(registered.manager.switchSession).not.toHaveBeenCalled(); expect(harness.destroyLiveSessionManager).not.toHaveBeenCalled(); expect(harness.headless.manager.switchSession).not.toHaveBeenCalled(); - expect(harness.ackSessionAttention).toHaveBeenCalledWith('session-1'); + expect(harness.ackSessionAttention).toHaveBeenCalledWith( + 'session-1', + 'permission', + '2026-01-01T00:00:00.000Z' + ); // The refresh carries the post-approval rows, so the count drops without // waiting for the background push. expect(refreshedRows(harness)).toEqual([BUSY_ROW]); @@ -187,7 +192,11 @@ describe('approveFrontAgent', () => { expect(harness.headless.manager.switchSession).toHaveBeenCalledWith('session-1'); expect(harness.headless.manager.respondToPermission).toHaveBeenCalledWith('perm-2', 'once'); expect(harness.destroyLiveSessionManager).toHaveBeenCalledTimes(1); - expect(harness.ackSessionAttention).toHaveBeenCalledWith('session-1'); + expect(harness.ackSessionAttention).toHaveBeenCalledWith( + 'session-1', + 'permission', + '2026-01-01T00:00:00.000Z' + ); }); it('returns nothing-to-approve for a question-only front and still refreshes the surfaces', async () => { @@ -349,7 +358,11 @@ describe('approveFrontAgent', () => { const outcome = await approveFrontAgent(harness.deps); expect(outcome).toEqual({ kind: 'approved' }); - expect(harness.ackSessionAttention).toHaveBeenCalledWith('session-1'); + expect(harness.ackSessionAttention).toHaveBeenCalledWith( + 'session-1', + 'permission', + '2026-01-01T00:00:00.000Z' + ); }); it('fails without rejecting when the headless manager factory rejects', async () => { diff --git a/apps/mobile/src/lib/glanceable/approve-front-agent.ts b/apps/mobile/src/lib/glanceable/approve-front-agent.ts index ac720db9b9b..0955bcf43a3 100644 --- a/apps/mobile/src/lib/glanceable/approve-front-agent.ts +++ b/apps/mobile/src/lib/glanceable/approve-front-agent.ts @@ -60,7 +60,7 @@ export type FrontApprovalDeps = { /** Tear down a manager `createLiveSessionManager` made. Never a registered one. */ destroyLiveSessionManager(handle: LiveSessionManagerHandle): void; /** The ack the permission card performs after a successful response. */ - ackSessionAttention(kiloSessionId: string): void; + ackSessionAttention(kiloSessionId: string, status: string, statusUpdatedAt?: string | null): void; /** Publish fresh rows to the glanceable sinks through the push sink loop. */ refreshGlanceableSurfaces(input: { rows: readonly FrontApprovableRow[]; @@ -139,7 +139,7 @@ async function runApproval(deps: FrontApprovalDeps): Promise { const wait = await waitForAsk(deps, handle); if (wait.kind === 'timeout') { @@ -185,8 +185,9 @@ async function answerFrontAsk( } catch (error) { return { kind: 'failed', retryable: deps.classifyFailure(error) === 'retryable' }; } - // The same ack the permission card performs once the wrapper accepted it. - deps.ackSessionAttention(kiloSessionId); + // The same ack the permission card performs once the wrapper accepted it, + // bound to the raise this row was waiting on. + deps.ackSessionAttention(front.id, front.status, front.statusUpdatedAt); return { kind: 'approved' }; } diff --git a/apps/mobile/src/lib/glanceable/attention-rows.test.ts b/apps/mobile/src/lib/glanceable/attention-rows.test.ts index 1efbbdfee6f..122b879645a 100644 --- a/apps/mobile/src/lib/glanceable/attention-rows.test.ts +++ b/apps/mobile/src/lib/glanceable/attention-rows.test.ts @@ -26,7 +26,7 @@ describe('resolveAnsweredRaises', () => { }); it('resolves an answered raise to idle so the surfaces stop reporting Needs input', () => { - ackSessionAttention('s1'); + ackSessionAttention('s1', 'permission'); expect(resolveAnsweredRaises([{ id: 's1', status: 'permission' }])).toEqual([ { id: 's1', status: 'idle' }, @@ -34,7 +34,7 @@ describe('resolveAnsweredRaises', () => { }); it('keeps a different session waiting when only one raise was answered', () => { - ackSessionAttention('s1'); + ackSessionAttention('s1', 'question'); expect( resolveAnsweredRaises([ @@ -48,7 +48,7 @@ describe('resolveAnsweredRaises', () => { }); it('does not reclassify a non-attention row the ack store still holds', () => { - ackSessionAttention('s1'); + ackSessionAttention('s1', 'permission'); reconcileSessionAttention('s1', 'permission', 'R1'); expect(resolveAnsweredRaises([{ id: 's1', status: 'busy' }])).toEqual([ @@ -57,8 +57,7 @@ describe('resolveAnsweredRaises', () => { }); it('counts a re-raise again once reconcile replaces the answered raise', () => { - ackSessionAttention('s1'); - reconcileSessionAttention('s1', 'permission', 'R1'); + ackSessionAttention('s1', 'permission', 'R1'); reconcileSessionAttention('s1', 'permission', 'R2'); expect( @@ -67,8 +66,7 @@ describe('resolveAnsweredRaises', () => { }); it('matches the raise identity the reconciler stored', () => { - ackSessionAttention('s1'); - reconcileSessionAttention('s1', 'permission', 'R1'); + ackSessionAttention('s1', 'permission', 'R1'); expect(isAnsweredAttentionRow({ id: 's1', status: 'permission', statusUpdatedAt: 'R1' })).toBe( true @@ -76,7 +74,7 @@ describe('resolveAnsweredRaises', () => { }); it('resolves a raise the tray reconciler pinned to the raw status while the row carries a timestamp', () => { - ackSessionAttention('s1'); + ackSessionAttention('s1', 'permission'); // The tray rows reconcile with no timestamp (`(tabs)/_layout.tsx`, // `remote-session-row.tsx`), so the ack pins to the raw status while the // same tray row carries the server `statusUpdatedAt`. @@ -90,7 +88,7 @@ describe('resolveAnsweredRaises', () => { }); it('drops an answered raise from the derived counts while a waiting sibling stays', () => { - ackSessionAttention('s1'); + ackSessionAttention('s1', 'permission'); const rows = [ { id: 's1', status: 'permission' }, { id: 's2', status: 'permission' }, @@ -115,7 +113,7 @@ describe('resolveAnsweredRaises', () => { // reads the same tray row carrying its server `statusUpdatedAt`. Both shade // surfaces must agree, so the Active-agents status must not still say // "Needs input" beside the needs-input notification's "Request approved". - ackSessionAttention('s1'); + ackSessionAttention('s1', 'permission'); reconcileSessionAttention('s1', 'permission', null); const rows = [{ id: 's1', status: 'permission', statusUpdatedAt: '2026-09-18T21:00:00.000Z' }]; diff --git a/apps/mobile/src/lib/glanceable/attention-rows.ts b/apps/mobile/src/lib/glanceable/attention-rows.ts index 54a65c5b528..04cf89bd804 100644 --- a/apps/mobile/src/lib/glanceable/attention-rows.ts +++ b/apps/mobile/src/lib/glanceable/attention-rows.ts @@ -22,12 +22,10 @@ import { isAttentionAcked, sessionNeedsInput } from '@/lib/session-attention'; * and `reconcileSessionAttention` drops the ack once the server status leaves * attention (or the raise is replaced), so a re-raise counts again. * - * Raise identity is `statusUpdatedAt ?? status` (see `session-attention`), but - * the two kinds of observer do not always have both fields: the stored-session - * rows reconcile with `status_updated_at`, while the tray rows reconcile with - * no timestamp, so their ack pins to the raw status. A tray row still carries - * the server `statusUpdatedAt`, so the ack must be matched against either - * identity or an answered tray raise keeps counting as "Needs input". + * Raise identity is `statusUpdatedAt ?? status` (see `session-attention`). A + * tray row carries the server `statusUpdatedAt`, so the ack is read against + * that identity and the row's status: an answered raise counts as idle, while a + * re-raise under the same status counts as waiting again. */ /** The row fields the ack resolution reads. */ @@ -43,10 +41,10 @@ export function isAnsweredAttentionRow(row: AttentionRow): boolean { if (!sessionNeedsInput(row.status)) { return false; } - return ( - isAttentionAcked(row.id, row.statusUpdatedAt ?? row.status) || - isAttentionAcked(row.id, row.status) - ); + return isAttentionAcked(row.id, { + raiseId: row.statusUpdatedAt ?? row.status, + status: row.status, + }); } /** Resolve every row to the status the glanceable surfaces should count. */ diff --git a/apps/mobile/src/lib/glanceable/waiting-ask.test.ts b/apps/mobile/src/lib/glanceable/waiting-ask.test.ts index 7614dd3e232..890b93a684e 100644 --- a/apps/mobile/src/lib/glanceable/waiting-ask.test.ts +++ b/apps/mobile/src/lib/glanceable/waiting-ask.test.ts @@ -92,6 +92,20 @@ describe('selectWaitingAsk', () => { }); }); + it("carries the row's status time so the answer can name its raise", () => { + const ask = selectWaitingAsk( + [{ id: 's9', status: 'permission', statusUpdatedAt: at(-60_000) }], + CTX, + NOW + ); + expect(ask).toMatchObject({ kiloSessionId: 's9', statusUpdatedAt: at(-60_000) }); + }); + + it('leaves the status time absent when the row carries none', () => { + const ask = selectWaitingAsk([{ id: 's9', status: 'permission' }], CTX, NOW); + expect(ask?.statusUpdatedAt).toBeUndefined(); + }); + it('breaks a tie on tray order', () => { const ask = selectWaitingAsk( [ @@ -275,7 +289,11 @@ describe('waiting ask store', () => { }); it('survives a JS restart through the SecureStore mirror', async () => { - const ask = askFor({ kiloSessionId: 's7', status: 'question' }); + const ask = askFor({ + kiloSessionId: 's7', + status: 'question', + statusUpdatedAt: at(-60_000), + }); recordWaitingAsk(ask); // A JS restart drops every module-local value; only the mirror remains. diff --git a/apps/mobile/src/lib/glanceable/waiting-ask.ts b/apps/mobile/src/lib/glanceable/waiting-ask.ts index 5444e3d5a3b..34504c7fad8 100644 --- a/apps/mobile/src/lib/glanceable/waiting-ask.ts +++ b/apps/mobile/src/lib/glanceable/waiting-ask.ts @@ -27,6 +27,13 @@ export type WaitingAsk = { kiloSessionId: string; /** 'permission' | 'question' — the tray row's status. */ status: string; + /** + * ISO 8601 status time of the ask's row, the raise's own identity + * (`statusUpdatedAt ?? status`), so the ack an Approve writes names exactly + * the raise it answered. Absent on rows that never carried one, and on a + * record mirrored by a build that did not store it. + */ + statusUpdatedAt?: string; /** True when the row came from the cloud-agent control plane. */ isCloudAgent: boolean; /** @@ -63,6 +70,7 @@ const WAITING_ASK_KEY = 'glanceable-waiting-ask'; const waitingAskSchema = z.object({ kiloSessionId: z.string().min(1), status: z.string(), + statusUpdatedAt: z.string().optional(), isCloudAgent: z.boolean(), scopeKey: z.string().min(1), organizationId: z.string().nullable(), @@ -130,6 +138,7 @@ export function selectWaitingAsk( return { kiloSessionId: chosen.id, status: chosen.status, + statusUpdatedAt: chosen.statusUpdatedAt, isCloudAgent: chosen.connectionId === CLOUD_AGENT_CONNECTION_ID, // The snapshot's own scope key, so a sink can fence the ask against the // surface it belongs to without re-deriving the hash. diff --git a/apps/mobile/src/lib/glanceable/widget-actions.test.ts b/apps/mobile/src/lib/glanceable/widget-actions.test.ts index 4349771b2c4..78b10f637d1 100644 --- a/apps/mobile/src/lib/glanceable/widget-actions.test.ts +++ b/apps/mobile/src/lib/glanceable/widget-actions.test.ts @@ -287,6 +287,26 @@ describe('runWidgetApprove', () => { }); }); + it('acks the raise the press answered, not the next one', async () => { + wireTrpc({ + sessions: [ + { id: 'waiting', status: 'permission', statusUpdatedAt: '2026-09-18T21:00:00.000Z' }, + ], + cloudAgentSessionId: 'workspace_agent_1', + permissions: [{ id: 'perm-1' }], + }); + const { release } = collectSink(); + + await expect(runWidgetApprove(keyFor('waiting', 'perm-1'))).resolves.toEqual({ + kind: 'approved', + }); + + expect(isAttentionAcked('waiting', { raiseId: '2026-09-18T21:00:00.000Z' })).toBe(true); + // A permission raised after the answered one keeps its badge. + expect(isAttentionAcked('waiting', { raiseId: '2026-09-18T21:30:00.000Z' })).toBe(false); + release(); + }); + it('answers nothing when the displayed session is not a cloud-agent session', async () => { const rpc = wireTrpc({ sessions: [{ id: 'waiting', status: 'permission' }], @@ -326,7 +346,7 @@ describe('runWidgetApprove', () => { // makes the redraw count it as answered instead of showing the pre-action // counts. expect(rpc.activeSessions.list.query).toHaveBeenCalledTimes(2); - expect(isAttentionAcked('waiting', 'permission')).toBe(true); + expect(isAttentionAcked('waiting', { raiseId: 'permission' })).toBe(true); expect(snapshots).toHaveLength(1); expect(snapshots[0]).toMatchObject({ needsInput: 0, status: 'happy' }); // The card outlives the JS process, so the republish must start/update it, diff --git a/apps/mobile/src/lib/glanceable/widget-actions.ts b/apps/mobile/src/lib/glanceable/widget-actions.ts index 3d9d571b378..1061b4f1c97 100644 --- a/apps/mobile/src/lib/glanceable/widget-actions.ts +++ b/apps/mobile/src/lib/glanceable/widget-actions.ts @@ -256,7 +256,7 @@ async function approveVisiblePermission( organizationId, }) : trpcClient.cloudAgentNext.answerPermission.mutate(answer)); - ackSessionAttention(row.id); + ackSessionAttention(row.id, row.status, row.statusUpdatedAt); return { kind: 'approved', answeredSessionId: row.id }; } } diff --git a/apps/mobile/src/lib/hooks/use-session-raise-reader.ts b/apps/mobile/src/lib/hooks/use-session-raise-reader.ts new file mode 100644 index 00000000000..6061162e764 --- /dev/null +++ b/apps/mobile/src/lib/hooks/use-session-raise-reader.ts @@ -0,0 +1,49 @@ +import { useQueryClient } from '@tanstack/react-query'; +import { useCallback, useMemo } from 'react'; + +import { + buildActiveSessionsTrayInput, + type CachedActiveSessionsData, +} from '@/lib/active-sessions-live'; +import { useOrganization } from '@/lib/organization-context'; +import { useTRPC } from '@/lib/trpc'; + +/** The raise a live session row is waiting on, as the tray cache holds it. */ +export type ObservedSessionRaise = { + status: string; + statusUpdatedAt?: string; +}; + +/** + * Read the raise the tray last observed before submitting a response. + * + * The answer paths know which request they answered but not when its raise was + * created; the raise id is `statusUpdatedAt ?? status` of the row, and the row + * the app has observed lives in the `activeSessions.list` cache the tab layout, + * the list rows and the live-sync owner all write. This reads that cache once + * per submission — no fetch, no subscription — so an ack can name the exact raise the + * user answered instead of degrading to its status. + * + * Returns null when the session is not in the cache (an offline or already + * finished session), and the caller then acks with the status alone. + */ +export function useSessionRaiseReader(): (sessionId: string) => ObservedSessionRaise | null { + const trpc = useTRPC(); + const queryClient = useQueryClient(); + const { organizationId } = useOrganization(); + const queryKey = useMemo( + () => trpc.activeSessions.list.queryKey(buildActiveSessionsTrayInput(organizationId)), + [trpc, organizationId] + ); + return useCallback( + (sessionId: string): ObservedSessionRaise | null => { + const data = queryClient.getQueryData(queryKey); + const row = data?.sessions.find(session => session.id === sessionId); + if (row === undefined) { + return null; + } + return { status: row.status, statusUpdatedAt: row.statusUpdatedAt }; + }, + [queryClient, queryKey] + ); +} diff --git a/apps/mobile/src/lib/notification-action-interaction.test.ts b/apps/mobile/src/lib/notification-action-interaction.test.ts index 8ddb758556f..dbd4495d5dd 100644 --- a/apps/mobile/src/lib/notification-action-interaction.test.ts +++ b/apps/mobile/src/lib/notification-action-interaction.test.ts @@ -174,7 +174,9 @@ const POLL_INTERVAL_MS = 100; /** Deps shared by every test that drives a fake manager on the fake clock. */ function fakeManagerDeps( fake: FakeManager, - extra?: { ack?: (kiloSessionId: string) => void } + extra?: { + ack?: (kiloSessionId: string, status: string, statusUpdatedAt?: string | null) => void; + } ): NonNullable[0]['deps']> { const clock = createFakeClock(); return { @@ -208,7 +210,8 @@ describe('runNeedsInputInteraction', () => { seeds.store.set(seeds.activePermission, { requestId: 'perm-1' }); }, }); - const ack = vi.fn<(kiloSessionId: string) => void>(); + const ack = + vi.fn<(kiloSessionId: string, status: string, statusUpdatedAt?: string | null) => void>(); const outcome = await runNeedsInputInteraction({ kiloSessionId: SESSION_ID, @@ -220,7 +223,7 @@ describe('runNeedsInputInteraction', () => { expect(fake.switchSession).toHaveBeenCalledWith(SESSION_ID); expect(fake.respondToPermission).toHaveBeenCalledWith('perm-1', 'once'); expect(fake.answerQuestion).not.toHaveBeenCalled(); - expect(ack).toHaveBeenCalledWith(SESSION_ID); + expect(ack).toHaveBeenCalledWith(SESSION_ID, 'permission', null); expect(fake.destroy).toHaveBeenCalledTimes(1); }); @@ -230,7 +233,8 @@ describe('runNeedsInputInteraction', () => { seeds.store.set(seeds.activeQuestion, { requestId: 'question-1' }); }, }); - const ack = vi.fn<(kiloSessionId: string) => void>(); + const ack = + vi.fn<(kiloSessionId: string, status: string, statusUpdatedAt?: string | null) => void>(); const outcome = await runNeedsInputInteraction({ kiloSessionId: SESSION_ID, @@ -242,7 +246,7 @@ describe('runNeedsInputInteraction', () => { expect(outcome).toBe('ok'); expect(fake.answerQuestion).toHaveBeenCalledWith('question-1', [['use the second option']]); expect(fake.respondToPermission).not.toHaveBeenCalled(); - expect(ack).toHaveBeenCalledWith(SESSION_ID); + expect(ack).toHaveBeenCalledWith(SESSION_ID, 'question', null); expect(fake.destroy).toHaveBeenCalledTimes(1); }); @@ -380,7 +384,8 @@ describe('runNeedsInputInteraction', () => { seeds.store.set(seeds.activePermission, { requestId: 'perm-1' }); }, }); - const ack = vi.fn<(kiloSessionId: string) => void>(); + const ack = + vi.fn<(kiloSessionId: string, status: string, statusUpdatedAt?: string | null) => void>(); const outcome = await runNeedsInputInteraction({ kiloSessionId: SESSION_ID, @@ -450,7 +455,8 @@ describe('runNeedsInputInteraction', () => { it('returns retryable when no raise of either kind appears in the budget, so the actions are kept', async () => { const fake = createFakeManager(); - const ack = vi.fn<(kiloSessionId: string) => void>(); + const ack = + vi.fn<(kiloSessionId: string, status: string, statusUpdatedAt?: string | null) => void>(); const outcome = await runNeedsInputInteraction({ kiloSessionId: SESSION_ID, @@ -470,7 +476,8 @@ describe('runNeedsInputInteraction', () => { it('returns retryable for a reply when neither kind is pending at the deadline', async () => { const fake = createFakeManager(); - const ack = vi.fn<(kiloSessionId: string) => void>(); + const ack = + vi.fn<(kiloSessionId: string, status: string, statusUpdatedAt?: string | null) => void>(); const outcome = await runNeedsInputInteraction({ kiloSessionId: SESSION_ID, @@ -533,7 +540,8 @@ describe('runNeedsInputInteraction', () => { }, }); getSessionQuery.mockResolvedValue({ organization_id: 'org-1', status: 'idle' }); - const ack = vi.fn<(kiloSessionId: string) => void>(); + const ack = + vi.fn<(kiloSessionId: string, status: string, statusUpdatedAt?: string | null) => void>(); const outcome = await runNeedsInputInteraction({ kiloSessionId: SESSION_ID, @@ -545,7 +553,7 @@ describe('runNeedsInputInteraction', () => { expect(fake.respondToPermission).not.toHaveBeenCalled(); expect(fake.answerQuestion).not.toHaveBeenCalled(); // The raise is over, so the Agents row and badge stop offering it. - expect(ack).toHaveBeenCalledWith(SESSION_ID); + expect(ack).toHaveBeenCalledWith(SESSION_ID, 'permission', null); expect(fake.destroy).toHaveBeenCalledTimes(1); }); @@ -564,7 +572,8 @@ describe('runNeedsInputInteraction', () => { getSessionQuery .mockResolvedValueOnce({ organization_id: 'org-1', status: 'permission' }) .mockResolvedValue({ organization_id: 'org-1', status: 'permission' }); - const ack = vi.fn<(kiloSessionId: string) => void>(); + const ack = + vi.fn<(kiloSessionId: string, status: string, statusUpdatedAt?: string | null) => void>(); const outcome = await runNeedsInputInteraction({ kiloSessionId: SESSION_ID, @@ -575,7 +584,7 @@ describe('runNeedsInputInteraction', () => { expect(outcome).toBe('ok'); expect(fake.respondToPermission).not.toHaveBeenCalled(); expect(fake.answerQuestion).not.toHaveBeenCalled(); - expect(ack).toHaveBeenCalledWith(SESSION_ID); + expect(ack).toHaveBeenCalledWith(SESSION_ID, 'permission', null); expect(fake.destroy).toHaveBeenCalledTimes(1); }); @@ -589,7 +598,8 @@ describe('runNeedsInputInteraction', () => { }, }); getSessionQuery.mockResolvedValue({ organization_id: 'org-1', status: 'permission' }); - const ack = vi.fn<(kiloSessionId: string) => void>(); + const ack = + vi.fn<(kiloSessionId: string, status: string, statusUpdatedAt?: string | null) => void>(); const outcome = await runNeedsInputInteraction({ kiloSessionId: SESSION_ID, @@ -615,7 +625,8 @@ describe('runNeedsInputInteraction', () => { .mockRejectedValue( Object.assign(new Error('network down'), { data: { code: 'INTERNAL_SERVER_ERROR' } }) ); - const ack = vi.fn<(kiloSessionId: string) => void>(); + const ack = + vi.fn<(kiloSessionId: string, status: string, statusUpdatedAt?: string | null) => void>(); const outcome = await runNeedsInputInteraction({ kiloSessionId: SESSION_ID, @@ -640,7 +651,8 @@ describe('runNeedsInputInteraction', () => { seeds.store.set(seeds.isReadOnly, true); }, }); - const ack = vi.fn<(kiloSessionId: string) => void>(); + const ack = + vi.fn<(kiloSessionId: string, status: string, statusUpdatedAt?: string | null) => void>(); const outcome = await runNeedsInputInteraction({ kiloSessionId: SESSION_ID, @@ -662,7 +674,8 @@ describe('runNeedsInputInteraction', () => { seeds.store.set(seeds.agentStatus, { type: 'error', message: 'transport failed' }); }, }); - const ack = vi.fn<(kiloSessionId: string) => void>(); + const ack = + vi.fn<(kiloSessionId: string, status: string, statusUpdatedAt?: string | null) => void>(); const outcome = await runNeedsInputInteraction({ kiloSessionId: SESSION_ID, @@ -682,7 +695,8 @@ describe('runNeedsInputInteraction', () => { seeds.store.set(seeds.sessionType, null); }, }); - const ack = vi.fn<(kiloSessionId: string) => void>(); + const ack = + vi.fn<(kiloSessionId: string, status: string, statusUpdatedAt?: string | null) => void>(); const outcome = await runNeedsInputInteraction({ kiloSessionId: SESSION_ID, @@ -699,7 +713,8 @@ describe('runNeedsInputInteraction', () => { // `connecting` is the never-opened socket: the budget expiring here is // exactly the cold headless start the raise can still be live behind. const fake = createFakeManager(); - const ack = vi.fn<(kiloSessionId: string) => void>(); + const ack = + vi.fn<(kiloSessionId: string, status: string, statusUpdatedAt?: string | null) => void>(); const outcome = await runNeedsInputInteraction({ kiloSessionId: SESSION_ID, @@ -712,7 +727,8 @@ describe('runNeedsInputInteraction', () => { }); it('acks a session the user can no longer see', async () => { - const ack = vi.fn<(kiloSessionId: string) => void>(); + const ack = + vi.fn<(kiloSessionId: string, status: string, statusUpdatedAt?: string | null) => void>(); const outcome = await runNeedsInputInteraction({ kiloSessionId: SESSION_ID, @@ -726,7 +742,31 @@ describe('runNeedsInputInteraction', () => { }); expect(outcome).toBe('unavailable'); - expect(ack).toHaveBeenCalledWith(SESSION_ID); + expect(ack).toHaveBeenCalledWith(SESSION_ID, 'permission', null); + }); + + it('acks the raise its stored row reports, not the next one', async () => { + const fake = createFakeManager({ + onSwitch: seeds => { + seeds.store.set(seeds.activePermission, { requestId: 'perm-1' }); + }, + }); + getSessionQuery.mockResolvedValue({ + organization_id: 'org-1', + status: 'permission', + status_updated_at: '2026-09-18 21:00:00+00', + }); + const ack = + vi.fn<(kiloSessionId: string, status: string, statusUpdatedAt?: string | null) => void>(); + + const outcome = await runNeedsInputInteraction({ + kiloSessionId: SESSION_ID, + action: 'approve', + deps: fakeManagerDeps(fake, { ack }), + }); + + expect(outcome).toBe('ok'); + expect(ack).toHaveBeenCalledWith(SESSION_ID, 'permission', '2026-09-18 21:00:00+00'); }); it('answers a raise that lands with the snapshot replay just after the budget', async () => { @@ -740,7 +780,8 @@ describe('runNeedsInputInteraction', () => { }); const clock = createFakeClock(); let seeded = false; - const ack = vi.fn<(kiloSessionId: string) => void>(); + const ack = + vi.fn<(kiloSessionId: string, status: string, statusUpdatedAt?: string | null) => void>(); const outcome = await runNeedsInputInteraction({ kiloSessionId: SESSION_ID, @@ -764,7 +805,7 @@ describe('runNeedsInputInteraction', () => { expect(outcome).toBe('ok'); expect(fake.respondToPermission).toHaveBeenCalledWith('perm-late', 'once'); - expect(ack).toHaveBeenCalledWith(SESSION_ID); + expect(ack).toHaveBeenCalledWith(SESSION_ID, 'permission', null); }); it('returns retryable when the raise does not match the action, so the raise keeps its actions', async () => { @@ -810,7 +851,8 @@ describe('runNeedsInputInteraction', () => { const fake = createFakeManager(); const clock = createFakeClock(); let seeded = false; - const ack = vi.fn<(kiloSessionId: string) => void>(); + const ack = + vi.fn<(kiloSessionId: string, status: string, statusUpdatedAt?: string | null) => void>(); const outcome = await runNeedsInputInteraction({ kiloSessionId: SESSION_ID, diff --git a/apps/mobile/src/lib/notification-action-interaction.ts b/apps/mobile/src/lib/notification-action-interaction.ts index e7284595df6..5c55b7997f2 100644 --- a/apps/mobile/src/lib/notification-action-interaction.ts +++ b/apps/mobile/src/lib/notification-action-interaction.ts @@ -107,7 +107,18 @@ type ManagerFactoryOptions = { * The session row this module needs: existence, the manager's org scope, and * the status that decides a proven-absent raise's outcome. */ -type SessionRow = { organization_id?: string | null; status?: string | null }; +/** + * The raise a notification action answered: its status, plus the stored row's + * own status time when that row still reported the answered status. + */ +type AnsweredRaise = { status: string; statusUpdatedAt: string | null }; + +type SessionRow = { + organization_id?: string | null; + status?: string | null; + /** When the stored row's status last changed, the raise's own identity. */ + status_updated_at?: string | null; +}; export type NeedsInputInteractionDeps = { /** @@ -126,7 +137,7 @@ export type NeedsInputInteractionDeps = { /** Reads the authenticated user id that scopes the transcript cache. */ getUserId?: () => Promise; /** Records the session-attention ack after a successful interaction. */ - ack?: (kiloSessionId: string) => void; + ack?: (kiloSessionId: string, status: string, statusUpdatedAt?: string | null) => void; /** Clock the wait for the raise is bounded by. */ now?: () => number; sleep?: (ms: number) => Promise; @@ -181,15 +192,27 @@ export async function runNeedsInputInteraction({ const pollIntervalMs = deps?.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS; const missingRaiseSettleMs = deps?.missingRaiseSettleMs ?? DEFAULT_MISSING_RAISE_SETTLE_MS; + // The raise the notification offered. A card's Approve answers a permission + // and a Reply answers a question, and the stored row's own status time names + // the raise when the row still reads that status. + const answeredStatus = action === 'approve' ? 'permission' : 'question'; let organizationId: string | undefined = undefined; + let answeredRaise: AnsweredRaise = { status: answeredStatus, statusUpdatedAt: null }; try { const session = await getSession(kiloSessionId); organizationId = session.organization_id ?? undefined; + answeredRaise = { + status: answeredStatus, + // A row already reporting a different status has moved on from this + // raise: its time names the next one, so the ack keeps the status alone. + statusUpdatedAt: + session.status === answeredStatus ? (session.status_updated_at ?? null) : null, + }; } catch (error) { // A session the user can no longer see is stale, not retryable; every // other read failure (deadline, 5xx, offline) may succeed on a later tap. if (readTrpcErrorField(error, 'code') === 'NOT_FOUND') { - ack(kiloSessionId); + ack(kiloSessionId, answeredRaise.status, answeredRaise.statusUpdatedAt); return 'unavailable'; } return 'retryable'; @@ -235,7 +258,7 @@ export async function runNeedsInputInteraction({ // that can no longer do anything — the same clear a successful // answer performs. ackGone: () => { - ack(kiloSessionId); + ack(kiloSessionId, answeredRaise.status, answeredRaise.statusUpdatedAt); }, }); if (resolution.kind === 'answered') { @@ -249,7 +272,7 @@ export async function runNeedsInputInteraction({ await (action === 'approve' ? manager.respondToPermission(requestId, 'once') : manager.answerQuestion(requestId, [[replyText]])); - ack(kiloSessionId); + ack(kiloSessionId, answeredRaise.status, answeredRaise.statusUpdatedAt); return 'ok'; } finally { manager.destroy(); diff --git a/apps/mobile/src/lib/session-attention.test.ts b/apps/mobile/src/lib/session-attention.test.ts index 1e4c4252ba9..8da51b4271b 100644 --- a/apps/mobile/src/lib/session-attention.test.ts +++ b/apps/mobile/src/lib/session-attention.test.ts @@ -126,66 +126,167 @@ describe('shouldShowNeedsInput', () => { }); describe('ack store state machine', () => { - it('pending ack hides the indicator immediately for any raiseId', () => { - ackSessionAttention('s1'); - expect(isAttentionAcked('s1', 'R1')).toBe(true); - expect(isAttentionAcked('s1', 'R2')).toBe(true); - expect(isAttentionAcked('s1', null)).toBe(true); + it('an ack hides only the raise it names', () => { + ackSessionAttention('s1', 'question', 'R1'); + + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(true); + expect(isAttentionAcked('s1', { raiseId: 'R2' })).toBe(false); + expect(__peekSessionAttentionForTests('s1')).toEqual({ raiseId: 'R1' }); }); - it('reconcile resolves a pending entry to the observed raise', () => { - ackSessionAttention('s1'); - reconcileSessionAttention('s1', 'question', 'R1'); - expect(isAttentionAcked('s1', 'R1')).toBe(true); - expect(isAttentionAcked('s1', 'R2')).toBe(false); + it('an ack with no timestamp hides the raise it was answered from', () => { + ackSessionAttention('s1', 'question'); + + // The caller named the status the user answered, nothing more. + expect(isAttentionAcked('s1', { raiseId: 'question' })).toBe(true); + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(false); }); - it('a resolved ack hides its own raise but not a new one', () => { - ackSessionAttention('s1'); - reconcileSessionAttention('s1', 'question', 'R1'); - expect(isAttentionAcked('s1', 'R1')).toBe(true); - // a new status_updated_at means a new raise — should show again - expect(isAttentionAcked('s1', 'R2')).toBe(false); + it('a different raise observed after an ack keeps its own badge', () => { + // The user answered a question; a permission the user has not seen arrives + // before the answer's status round-trip lands. + ackSessionAttention('s1', 'question', 'R1'); + + reconcileSessionAttention('s1', 'permission', 'R2'); + + expect(isAttentionAcked('s1', { raiseId: 'R2' })).toBe(false); + expect( + shouldShowNeedsInput({ + status: 'permission', + raiseId: 'R2', + isAcked: isAttentionAcked('s1', { raiseId: 'R2' }), + }) + ).toBe(true); + expect(__peekSessionAttentionEntryForTests('s1')).toEqual({ + raiseId: 'R2', + status: 'permission', + ackedAt: null, + expiresAt: null, + }); }); - it('a later successful answer re-pends a resolved ack and hides the new raise', () => { - ackSessionAttention('s1'); - reconcileSessionAttention('s1', 'question', 'R1'); - // new raise R2 arrives and is not acked - expect(isAttentionAcked('s1', 'R2')).toBe(false); - // user answers again → ack overwrites with pending - ackSessionAttention('s1'); - expect(isAttentionAcked('s1', 'R2')).toBe(true); + it('a re-raise of the acked status after an ack keeps its own badge', () => { + ackSessionAttention('s1', 'permission', 'R1'); + + reconcileSessionAttention('s1', 'permission', 'R2'); + + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(false); + expect(isAttentionAcked('s1', { raiseId: 'R2' })).toBe(false); + }); + + it('a status-only ack learns its raise on the first observation, then a later raise shows', () => { + ackSessionAttention('s1', 'permission'); + + // The raise the user answered, observed for the first time. + reconcileSessionAttention('s1', 'permission', 'R1'); + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(true); + + // A new permission under the same status. + reconcileSessionAttention('s1', 'permission', 'R2'); + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(false); + expect(isAttentionAcked('s1', { raiseId: 'R2' })).toBe(false); + }); + + it('a same-status re-raise shows on a reader that knows the row time', () => { + ackSessionAttention('s1', 'permission', 'R1'); + + // The Agents rows read the row they render: its status and its status time. + expect(isAttentionAcked('s1', { raiseId: 'R1', status: 'permission' })).toBe(true); + expect(isAttentionAcked('s1', { raiseId: 'R2', status: 'permission' })).toBe(false); + // A row the surface cannot date still reads the acked status. + expect(isAttentionAcked('s1', { raiseId: 'permission' })).toBe(true); + }); + + it('reads a status-only ack as answered for every raise of that status', () => { + ackSessionAttention('s1', 'permission'); + + // Nothing the ack holds can tell two permissions apart, so it answers both; + // a raise of the other kind is a different raise. + expect(isAttentionAcked('s1', { raiseId: 'R2', status: 'permission' })).toBe(true); + expect(isAttentionAcked('s1', { raiseId: 'R2', status: 'question' })).toBe(false); + }); + + it('does not reuse a raise that was already answered', () => { + ackSessionAttention('s1', 'question', 'R1'); + + // A question the store has not observed yet; the answer names only the + // status, and the acked raise is not the one the user answered. + ackSessionAttention('s1', 'question'); + + expect(__peekSessionAttentionForTests('s1')).toEqual({ raiseId: 'question' }); + }); + + it('an observation without a timestamp keeps the acked status hidden', () => { + ackSessionAttention('s1', 'permission', 'R1'); + + // The tray badge reads a row without its status time. + reconcileSessionAttention('s1', 'permission', null); + + expect(__peekSessionAttentionForTests('s1')).toEqual({ raiseId: 'R1' }); + expect(isAttentionAcked('s1', { raiseId: 'permission' })).toBe(true); + }); + + it('reads a stored row PostgreSQL time and a live row ISO time as one raise', () => { + ackSessionAttention('s1', 'permission', '2026-10-12T09:45:12.945Z'); + + reconcileSessionAttention('s1', 'permission', '2026-10-12 09:45:12.945+00'); + + expect(isAttentionAcked('s1', { raiseId: '2026-10-12T09:45:12.945Z' })).toBe(true); + expect(__peekSessionAttentionForTests('s1')).toEqual({ raiseId: '2026-10-12T09:45:12.945Z' }); + }); + + it('a later successful answer acks the raise the user answered', () => { + ackSessionAttention('s1', 'question', 'R1'); + // A new raise arrives and is not acked. + reconcileSessionAttention('s1', 'question', 'R2'); + expect(isAttentionAcked('s1', { raiseId: 'R2' })).toBe(false); + + ackSessionAttention('s1', 'question', 'R2'); + + expect(isAttentionAcked('s1', { raiseId: 'R2' })).toBe(true); + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(false); + }); + + it('an ack with no timestamp keeps the raise the store already tracks for its status', () => { + // R2 is the current raise the store observed; the in-app answer names only + // the status, so the ack keeps the raised identity the store holds. + ackSessionAttention('s1', 'question', 'R1'); + reconcileSessionAttention('s1', 'question', 'R2'); + + ackSessionAttention('s1', 'question'); + + expect(__peekSessionAttentionForTests('s1')).toEqual({ raiseId: 'R2' }); + reconcileSessionAttention('s1', 'question', 'R3'); + expect(isAttentionAcked('s1', { raiseId: 'R3' })).toBe(false); }); it('without an ack, opening alone leaves the badge visible for any raise', () => { // DEF-4: no on-mount ack — viewing a blocked session must not hide NEEDS INPUT - expect(isAttentionAcked('s1', 'R1')).toBe(false); + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(false); expect( shouldShowNeedsInput({ status: 'question', raiseId: 'R1', - isAcked: isAttentionAcked('s1', 'R1'), + isAcked: isAttentionAcked('s1', { raiseId: 'R1' }), }) ).toBe(true); }); it('acking one session never suppresses a raise on a different session', () => { - ackSessionAttention('s1'); - expect(isAttentionAcked('s1', 'R1')).toBe(true); - expect(isAttentionAcked('s2', 'R1')).toBe(false); + ackSessionAttention('s1', 'question', 'R1'); + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(true); + expect(isAttentionAcked('s2', { raiseId: 'R1' })).toBe(false); expect( shouldShowNeedsInput({ status: 'question', raiseId: 'R1', - isAcked: isAttentionAcked('s2', 'R1'), + isAcked: isAttentionAcked('s2', { raiseId: 'R1' }), }) ).toBe(true); }); it('reconcile deletes the entry on a non-attention status, so the next raise shows', () => { - ackSessionAttention('s1'); - reconcileSessionAttention('s1', 'question', 'R1'); + ackSessionAttention('s1', 'question', 'R1'); expect(__peekSessionAttentionForTests('s1')).toEqual({ raiseId: 'R1' }); // status drops to busy — entry should be cleared @@ -194,12 +295,12 @@ describe('ack store state machine', () => { // next question raise (no timestamp — remote active-only row) // is NOT acked and therefore visible - expect(isAttentionAcked('s1', 'question')).toBe(false); + expect(isAttentionAcked('s1', { raiseId: 'question' })).toBe(false); }); it('timestamp-less remote question → busy → question cycle re-raises visibly', () => { - ackSessionAttention('s1'); - // resolve to status string (no statusUpdatedAt) + ackSessionAttention('s1', 'question'); + // The same status with no timestamp is the raise the user answered. reconcileSessionAttention('s1', 'question', null); expect(__peekSessionAttentionForTests('s1')).toEqual({ raiseId: 'question' }); @@ -208,18 +309,17 @@ describe('ack store state machine', () => { expect(__peekSessionAttentionForTests('s1')).toBeUndefined(); // fresh question raise — not acked, visible - expect(isAttentionAcked('s1', 'question')).toBe(false); + expect(isAttentionAcked('s1', { raiseId: 'question' })).toBe(false); }); - it('answer-then-clear sequence: pending entry resolves via reconcile, then next raise is not absorbed', () => { - // user answers (ack → pending), then status still reports the same raise - ackSessionAttention('s1'); - // reconcile resolves the pending entry to R1 + it('answer-then-clear sequence: the ack names its raise, so the next raise is not absorbed', () => { + ackSessionAttention('s1', 'question', 'R1'); + // Server status still reports the same raise. reconcileSessionAttention('s1', 'question', 'R1'); - expect(isAttentionAcked('s1', 'R1')).toBe(true); + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(true); - // new raise R2 — pending is already resolved to R1, so R2 is NOT acked - expect(isAttentionAcked('s1', 'R2')).toBe(false); + // new raise R2 — the ack names R1, so R2 is NOT acked + expect(isAttentionAcked('s1', { raiseId: 'R2' })).toBe(false); }); it('reconcile with attention status and no entry is a no-op', () => { @@ -231,7 +331,7 @@ describe('ack store state machine', () => { expect(getRevisionSnapshot()).toBe(before); expect(listener).not.toHaveBeenCalled(); - expect(isAttentionAcked('s1', 'R1')).toBe(false); + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(false); unsubscribe(); }); @@ -250,27 +350,25 @@ describe('ack store state machine', () => { }); it('reconcile with attention status and a resolved entry is a no-op for the same raise', () => { - ackSessionAttention('s1'); - reconcileSessionAttention('s1', 'question', 'R1'); - // now entry.raiseId === 'R1' + ackSessionAttention('s1', 'question', 'R1'); const before = getRevisionSnapshot(); const listener = vi.fn<() => void>(); const unsubscribe = subscribe(listener); - // same raiseId → still resolved, no change + // same raiseId → no change reconcileSessionAttention('s1', 'question', 'R1'); expect(getRevisionSnapshot()).toBe(before); expect(listener).not.toHaveBeenCalled(); - expect(isAttentionAcked('s1', 'R1')).toBe(true); + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(true); unsubscribe(); }); }); describe('entry shape and re-raise', () => { - it('resolves a pending entry with the observed raise and ack metadata', () => { - ackSessionAttention('s1'); + it('an ack without a timestamp learns its raise and keeps its ack metadata', () => { + ackSessionAttention('s1', 'question'); reconcileSessionAttention('s1', 'question', 'R1'); const entry = __peekSessionAttentionEntryForTests('s1'); expect(entry).toMatchObject({ raiseId: 'R1', status: 'question' }); @@ -279,14 +377,13 @@ describe('entry shape and re-raise', () => { }); it('replaces the raise and clears the ack on a same-session re-raise', () => { - ackSessionAttention('s1'); - reconcileSessionAttention('s1', 'question', 'R1'); - expect(isAttentionAcked('s1', 'R1')).toBe(true); + ackSessionAttention('s1', 'question', 'R1'); + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(true); // a new status_updated_at is a new raise: the badge returns reconcileSessionAttention('s1', 'question', 'R2'); - expect(isAttentionAcked('s1', 'R1')).toBe(false); - expect(isAttentionAcked('s1', 'R2')).toBe(false); + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(false); + expect(isAttentionAcked('s1', { raiseId: 'R2' })).toBe(false); expect(__peekSessionAttentionEntryForTests('s1')).toEqual({ raiseId: 'R2', status: 'question', @@ -295,32 +392,31 @@ describe('entry shape and re-raise', () => { }); }); - it('acking a re-raised entry re-pends it and hides the new raise', () => { - ackSessionAttention('s1'); - reconcileSessionAttention('s1', 'question', 'R1'); + it('acking a re-raised entry acks the new raise, not the old one', () => { + ackSessionAttention('s1', 'question', 'R1'); // re-raise reconcileSessionAttention('s1', 'question', 'R2'); - expect(isAttentionAcked('s1', 'R2')).toBe(false); + expect(isAttentionAcked('s1', { raiseId: 'R2' })).toBe(false); // user answers the new raise - ackSessionAttention('s1'); - expect(isAttentionAcked('s1', 'R2')).toBe(true); - expect(__peekSessionAttentionForTests('s1')).toEqual({ raiseId: null }); + ackSessionAttention('s1', 'question', 'R2'); + expect(isAttentionAcked('s1', { raiseId: 'R2' })).toBe(true); + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(false); + expect(__peekSessionAttentionForTests('s1')).toEqual({ raiseId: 'R2' }); }); }); describe('durable persistence', () => { it('round-trips acks across a simulated restart', async () => { - ackSessionAttention('s1'); - reconcileSessionAttention('s1', 'question', 'R1'); + ackSessionAttention('s1', 'question', 'R1'); await __flushSessionAttentionWritesForTests(); // Simulated restart: clear the in-memory store, then re-hydrate from KV. __resetSessionAttentionForTests(); await __hydrateSessionAttentionForTests(); - expect(isAttentionAcked('s1', 'R1')).toBe(true); - expect(isAttentionAcked('s1', 'R2')).toBe(false); + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(true); + expect(isAttentionAcked('s1', { raiseId: 'R2' })).toBe(false); expect(__peekSessionAttentionEntryForTests('s1')).toEqual({ raiseId: 'R1', status: 'question', @@ -329,21 +425,40 @@ describe('durable persistence', () => { }); }); - it('restores a pending ack as pending across a restart', async () => { - ackSessionAttention('s1'); + it('a different raise observed after a restart keeps its badge', async () => { + ackSessionAttention('s1', 'question', 'R1'); await __flushSessionAttentionWritesForTests(); __resetSessionAttentionForTests(); await __hydrateSessionAttentionForTests(); + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(true); - // A pending ack hides any raise after restart. - expect(isAttentionAcked('s1', 'R1')).toBe(true); - expect(isAttentionAcked('s1', 'R2')).toBe(true); + // A raise that arrived while the app was dead: the restored ack names R1, + // so R2 is not suppressed. + reconcileSessionAttention('s1', 'question', 'R2'); + + expect(isAttentionAcked('s1', { raiseId: 'R2' })).toBe(false); + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(false); + expect(__peekSessionAttentionForTests('s1')).toEqual({ raiseId: 'R2' }); + }); + + it('drops a restored ack that names no raise', async () => { + const now = Date.now(); + seedAttentionKv([ + { sessionId: 's1', raiseId: null, status: null, ackedAt: now, expiresAt: now + DAY_MS }, + ]); + + __resetSessionAttentionForTests(); + await __hydrateSessionAttentionForTests(); + + // An unbound ack suppresses nothing: the raise stays visible. + expect(__peekSessionAttentionForTests('s1')).toBeUndefined(); + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(false); + expect(isAttentionAcked('s1', { raiseId: 'question' })).toBe(false); }); it('persists a deleted entry as gone across a restart', async () => { - ackSessionAttention('s1'); - reconcileSessionAttention('s1', 'question', 'R1'); + ackSessionAttention('s1', 'question', 'R1'); // delete reconcileSessionAttention('s1', 'busy', null); await __flushSessionAttentionWritesForTests(); @@ -374,7 +489,7 @@ describe('durable persistence', () => { const hydration = __hydrateSessionAttentionForTests(); // A write for a different session lands while hydration is still reading. - ackSessionAttention('s2'); + ackSessionAttention('s2', 'question'); // Release the stale persisted read. readGate.resolve(persisted); @@ -390,11 +505,11 @@ describe('durable persistence', () => { it('keeps in-memory behavior when a KV write fails and retries on the next bump', async () => { kvMock.setItem.mockRejectedValueOnce(new Error('disk full')); - ackSessionAttention('s1'); + ackSessionAttention('s1', 'question'); // In-memory store is authoritative: the badge hides immediately. - expect(isAttentionAcked('s1', 'R1')).toBe(true); + expect(isAttentionAcked('s1', { raiseId: 'question' })).toBe(true); await __flushSessionAttentionWritesForTests(); - expect(__peekSessionAttentionForTests('s1')).toEqual({ raiseId: null }); + expect(__peekSessionAttentionForTests('s1')).toEqual({ raiseId: 'question' }); // The next bump retries the write. reconcileSessionAttention('s1', 'question', 'R1'); @@ -419,8 +534,8 @@ describe('durable persistence', () => { expect(__peekSessionAttentionForTests('s1')).toBeUndefined(); - ackSessionAttention('s1'); - expect(isAttentionAcked('s1', 'R1')).toBe(true); + ackSessionAttention('s1', 'question', 'R1'); + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(true); }); }); @@ -454,8 +569,7 @@ describe('expiry', () => { vi.useFakeTimers(); try { vi.setSystemTime(new Date('2026-01-01T00:00:00Z')); - ackSessionAttention('s1'); - reconcileSessionAttention('s1', 'question', 'R1'); + ackSessionAttention('s1', 'question', 'R1'); expect(__peekSessionAttentionForTests('s1')).toEqual({ raiseId: 'R1' }); // 9 days later the ack has expired. @@ -485,24 +599,24 @@ describe('hydration gating', () => { // Hydration is in flight: the store is still empty, so the badge derives // from server status (no stale ack suppression). - expect(isAttentionAcked('s1', 'R1')).toBe(false); + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(false); expect( shouldShowNeedsInput({ status: 'question', raiseId: 'R1', - isAcked: isAttentionAcked('s1', 'R1'), + isAcked: isAttentionAcked('s1', { raiseId: 'R1' }), }) ).toBe(true); await hydration; // After hydration the restored ack suppresses its raise. - expect(isAttentionAcked('s1', 'R1')).toBe(true); + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(true); expect( shouldShowNeedsInput({ status: 'question', raiseId: 'R1', - isAcked: isAttentionAcked('s1', 'R1'), + isAcked: isAttentionAcked('s1', { raiseId: 'R1' }), }) ).toBe(false); }); @@ -527,16 +641,16 @@ describe('hydration gating', () => { const hydration = __hydrateSessionAttentionForTests(); // The fresh ack lands while hydration is still reading. - ackSessionAttention('s1'); - expect(__peekSessionAttentionForTests('s1')).toEqual({ raiseId: null }); + ackSessionAttention('s1', 'question'); + expect(__peekSessionAttentionForTests('s1')).toEqual({ raiseId: 'question' }); // Release the stale persisted read. readGate.resolve(persisted); await hydration; - // The fresh ack survives: still pending, not the persisted resolved entry. - expect(__peekSessionAttentionForTests('s1')).toEqual({ raiseId: null }); - expect(isAttentionAcked('s1', 'R1')).toBe(true); + // The fresh ack survives: still its own raise, not the persisted entry. + expect(__peekSessionAttentionForTests('s1')).toEqual({ raiseId: 'question' }); + expect(isAttentionAcked('s1', { raiseId: 'question' })).toBe(true); }); }); @@ -546,7 +660,7 @@ describe('revision snapshot and listener notification', () => { const listener = vi.fn<() => void>(); const unsubscribe = subscribe(listener); - ackSessionAttention('s1'); + ackSessionAttention('s1', 'question', 'R1'); expect(getRevisionSnapshot()).toBe(before + 1); expect(listener).toHaveBeenCalledTimes(1); @@ -554,43 +668,61 @@ describe('revision snapshot and listener notification', () => { unsubscribe(); }); - it('does not bump revision or notify when re-acking an already-pending entry', () => { - ackSessionAttention('s1'); + it('does not bump revision or notify when re-acking the same raise', () => { + ackSessionAttention('s1', 'question', 'R1'); const after = getRevisionSnapshot(); const listener = vi.fn<() => void>(); const unsubscribe = subscribe(listener); - // Repeated successful answers while still pending is a no-op. - ackSessionAttention('s1'); - ackSessionAttention('s1'); + // Repeated successful answers for the same raise are a no-op. + ackSessionAttention('s1', 'question', 'R1'); + ackSessionAttention('s1', 'question', 'R1'); expect(getRevisionSnapshot()).toBe(after); expect(listener).not.toHaveBeenCalled(); - expect(isAttentionAcked('s1', 'R1')).toBe(true); + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(true); unsubscribe(); }); - it('bumps revision when re-acking a resolved entry (re-pends it)', () => { - ackSessionAttention('s1'); - reconcileSessionAttention('s1', 'question', 'R1'); - // entry is now resolved to R1 + it('bumps revision when acking a different raise than the one already acked', () => { + ackSessionAttention('s1', 'question', 'R1'); const after = getRevisionSnapshot(); const listener = vi.fn<() => void>(); const unsubscribe = subscribe(listener); - ackSessionAttention('s1'); + ackSessionAttention('s1', 'question', 'R2'); expect(getRevisionSnapshot()).toBe(after + 1); expect(listener).toHaveBeenCalledTimes(1); - // re-pended: a new raise is once again absorbed - expect(isAttentionAcked('s1', 'R2')).toBe(true); + // The ack moved to R2; R1 is no longer suppressed. + expect(isAttentionAcked('s1', { raiseId: 'R2' })).toBe(true); + expect(isAttentionAcked('s1', { raiseId: 'R1' })).toBe(false); unsubscribe(); }); + it.each([true, false])( + 'keeps the newer raise with acked=%s when an earlier response finishes late', + newerAcked => { + const earlier = '2026-10-12 09:45:12.945+00'; + const newer = '2026-10-12T09:46:00.000Z'; + ackSessionAttention('s1', 'question', earlier); + reconcileSessionAttention('s1', 'question', newer); + if (newerAcked) { + ackSessionAttention('s1', 'question', newer); + } + + ackSessionAttention('s1', 'question', earlier); + + expect(__peekSessionAttentionForTests('s1')).toEqual({ raiseId: newer }); + expect(isAttentionAcked('s1', { raiseId: newer, status: 'question' })).toBe(newerAcked); + expect(isAttentionAcked('s1', { raiseId: earlier, status: 'question' })).toBe(false); + } + ); + it('bumps revision on mutating reconciles and stays stable on no-ops', () => { const listener = vi.fn<() => void>(); const unsubscribe = subscribe(listener); @@ -599,12 +731,12 @@ describe('revision snapshot and listener notification', () => { let mutations = 0; // ack → mutation - ackSessionAttention('s1'); + ackSessionAttention('s1', 'question', 'R1'); if (getRevisionSnapshot() !== initial) { mutations += 1; } - // resolve → mutation + // the same raise observed → no-op reconcileSessionAttention('s1', 'question', 'R1'); if (getRevisionSnapshot() !== initial + mutations) { mutations += 1; @@ -637,8 +769,8 @@ describe('revision snapshot and listener notification', () => { const unsubscribe = subscribe(listener); const start = getRevisionSnapshot(); - ackSessionAttention('a'); - ackSessionAttention('b'); + ackSessionAttention('a', 'question'); + ackSessionAttention('b', 'question'); reconcileSessionAttention('a', 'question', 'R1'); reconcileSessionAttention('a', 'busy', null); // no-ops: @@ -663,7 +795,7 @@ describe('revision snapshot and listener notification', () => { const before = getRevisionSnapshot(); expect(() => { - ackSessionAttention('s1'); + ackSessionAttention('s1', 'question', 'R1'); }).not.toThrow(); expect(throwing).toHaveBeenCalledTimes(1); expect(good).toHaveBeenCalledTimes(1); @@ -677,13 +809,13 @@ describe('revision snapshot and listener notification', () => { const listener = vi.fn<() => void>(); const unsubscribe = subscribe(listener); - ackSessionAttention('s1'); + ackSessionAttention('s1', 'question', 'R1'); expect(listener).toHaveBeenCalledTimes(1); unsubscribe(); // A real mutation on a different session would notify a live listener; // after unsubscribe it must not. - ackSessionAttention('s2'); + ackSessionAttention('s2', 'question', 'R1'); expect(getRevisionSnapshot()).toBeGreaterThan(0); expect(listener).toHaveBeenCalledTimes(1); }); @@ -691,10 +823,9 @@ describe('revision snapshot and listener notification', () => { describe('sign-out cleanup', () => { it('drops the previous account acks so a reused session id is not suppressed', async () => { - ackSessionAttention('session-shared'); - reconcileSessionAttention('session-shared', 'question', '2026-08-24T10:00:00Z'); + ackSessionAttention('session-shared', 'question', '2026-08-24T10:00:00Z'); await __flushSessionAttentionWritesForTests(); - expect(isAttentionAcked('session-shared', '2026-08-24T10:00:00Z')).toBe(true); + expect(isAttentionAcked('session-shared', { raiseId: '2026-08-24T10:00:00Z' })).toBe(true); await clearSessionAttentionForSignOut(); @@ -702,14 +833,14 @@ describe('sign-out cleanup', () => { // prior ack cannot suppress the badge for the same session id. await __hydrateSessionAttentionForTests(); expect(__peekSessionAttentionForTests('session-shared')).toBeUndefined(); - expect(isAttentionAcked('session-shared', '2026-08-24T10:00:00Z')).toBe(false); + expect(isAttentionAcked('session-shared', { raiseId: '2026-08-24T10:00:00Z' })).toBe(false); expect( shouldShowNeedsInput({ status: 'question', raiseId: '2026-08-24T10:00:00Z', isAcked: false }) ).toBe(true); }); it('deletes the persisted blob and notifies subscribers', async () => { - ackSessionAttention('session-a'); + ackSessionAttention('session-a', 'question'); await __flushSessionAttentionWritesForTests(); expect(kvStore.get(storageKey(SESSION_ATTENTION_KEY, ATTENTION_ENTRY_KEY))).toBeDefined(); @@ -724,7 +855,7 @@ describe('sign-out cleanup', () => { }); it('is not defeated by a bump queued right before sign-out', async () => { - ackSessionAttention('session-a'); + ackSessionAttention('session-a', 'question'); // No flush: the write is still queued when the clear is chained behind it. await clearSessionAttentionForSignOut(); @@ -732,7 +863,7 @@ describe('sign-out cleanup', () => { }); it('never throws when the encrypted KV delete fails', async () => { - ackSessionAttention('session-a'); + ackSessionAttention('session-a', 'question'); await __flushSessionAttentionWritesForTests(); kvMock.clearScope.mockRejectedValueOnce(new Error('storage unavailable')); diff --git a/apps/mobile/src/lib/session-attention.ts b/apps/mobile/src/lib/session-attention.ts index dd75b7b92da..ed350362a05 100644 --- a/apps/mobile/src/lib/session-attention.ts +++ b/apps/mobile/src/lib/session-attention.ts @@ -1,3 +1,5 @@ +/* eslint-disable max-lines -- one cohesive ack store: identity, persistence, hydration, reconcile, and the test seams read together */ +import { parseTimestamp } from '@kilocode/app-shared/utils'; import { useSyncExternalStore } from 'react'; import { z } from 'zod'; @@ -26,8 +28,9 @@ import { SESSION_ATTENTION_KEY } from '@/lib/storage-keys'; * hydration completes the store is empty, so badges render from server status * alone; a restored ack then suppresses its raise. * - * No backend, tRPC, or shared-package imports: this is a mobile-local - * module so the web client can keep its own copy. + * No backend or tRPC imports: this is a mobile-local module so the web client + * can keep its own copy. The one shared import is the pure timestamp reader, + * because Hermes cannot parse the PostgreSQL text a stored row carries. */ const ATTENTION_STATUSES = new Set(['question', 'permission']); @@ -52,11 +55,36 @@ export function sessionNeedsInput(status: string | null | undefined): boolean { return status != null && ATTENTION_STATUSES.has(status); } +/** + * True when two raise ids name the same raise. + * + * Raise identity is `statusUpdatedAt ?? status`, and the time arrives in two + * spellings: ISO 8601 from the live rows, raw PostgreSQL text + * (`2026-10-12 09:45:12.945+00`) from stored rows. Same instant, different + * string, so a comparison that only used string equality would clear an ack + * the moment the other surface reconciled the same raise. `parseTimestamp` is + * the app's reader for both spellings: Hermes' `Date` cannot parse the + * PostgreSQL form on its own. + */ +function sameRaise(a: string | null, b: string | null): boolean { + if (a === b) { + return true; + } + if (a === null || b === null) { + return false; + } + const at = parseTimestamp(a).getTime(); + const bt = parseTimestamp(b).getTime(); + return !Number.isNaN(at) && !Number.isNaN(bt) && at === bt; +} + /** * One durable ack entry. `sessionId` is the map key and is added to the * persisted form at serialization time. * - * `raiseId === null` marks a pending ack (acked, raise not yet observed). + * `raiseId` names the raise the user answered (`statusUpdatedAt ?? status`), so + * the ack can only ever suppress that raise. A null `raiseId` only appears in a + * blob written before acks carried a raise; hydration drops those. * `ackedAt === null` marks a cleared ack (a same-session re-raise replaced * the raise and cleared the ack so the badge returns). */ @@ -166,7 +194,11 @@ function applyHydratedEntries(raw: string): boolean { return false; } const now = Date.now(); - const fresh = entries.filter(entry => entry.expiresAt === null || entry.expiresAt > now); + // An ack that names no raise (written before acks carried one) suppresses + // nothing: dropping it keeps a raise the user has not answered visible. + const fresh = entries.filter( + entry => entry.raiseId !== null && (entry.expiresAt === null || entry.expiresAt > now) + ); let applied = false; for (const entry of fresh) { // A mutation that landed after hydration began must win over the stale @@ -259,18 +291,40 @@ function getServerSnapshot(): number { * permission response, before the server status round-trip lands. * `reconcileSessionAttention` clears the entry once status leaves attention. * - * If the entry is already pending, nothing changes — skip the bump so we + * The ack names the raise the user answered, so it can only ever suppress that + * raise: `statusUpdatedAt` when the answered row carried it, the status alone + * otherwise. A caller that has no timestamp keeps the raise an existing entry + * already tracks for the same status — the raise that was on screen when the + * answer happened. + * + * A late response cannot replace a newer dated raise already tracked here. + * + * Re-acking the raise already acked changes nothing — skip the bump so we * don't fire a redundant global re-render. */ -export function ackSessionAttention(sessionId: string): void { +export function ackSessionAttention( + sessionId: string, + status: string, + statusUpdatedAt?: string | null +): void { const entry = store.entries.get(sessionId); - if (entry && entry.ackedAt !== null && entry.raiseId === null) { + // Only a cleared entry tracks the raise still waiting for an answer; an acked + // entry's raise was answered already and must not be reused for a new one. + const trackedRaise = entry?.ackedAt === null && entry.status === status ? entry.raiseId : null; + const raiseId = statusUpdatedAt ?? trackedRaise ?? status; + if ( + entry?.raiseId != null && + parseTimestamp(entry.raiseId).getTime() > parseTimestamp(raiseId).getTime() + ) { + return; + } + if (entry && entry.ackedAt !== null && sameRaise(entry.raiseId, raiseId)) { return; } const now = Date.now(); store.entries.set(sessionId, { - raiseId: null, - status: null, + raiseId, + status, ackedAt: now, expiresAt: now + SESSION_ATTENTION_EXPIRY_MS, }); @@ -284,9 +338,14 @@ export function ackSessionAttention(sessionId: string): void { * * - expired entry: delete it and notify * - non-attention status: delete the entry (if any) and notify - * - attention + existing pending entry: resolve it to the current raise - * - attention + resolved entry with a different raise: replace the raise and - * clear the ack (same-session re-raise) so the badge returns + * - observed raise is the acked raise: no-op + * - acked with its status alone and that same status observed: learn the raise + * (the ack keeps suppressing it; a later raise is then told apart) + * - observation without a timestamp of the acked status: keep the ack — the + * observer cannot name the raise, and the badge it draws belongs to the + * raise the user answered + * - any other raise: replace the raise and clear the ack (same-session + * re-raise) so the badge returns * - otherwise: no-op (does NOT bump the revision) */ export function reconcileSessionAttention( @@ -316,7 +375,7 @@ export function reconcileSessionAttention( if (entry.ackedAt === null) { // Cleared ack (re-raise): keep tracking the current raise, still unacked. - if (entry.raiseId !== raiseId) { + if (!sameRaise(entry.raiseId, raiseId)) { store.entries.set(sessionId, { raiseId, status: status ?? null, @@ -328,11 +387,16 @@ export function reconcileSessionAttention( return; } - if (entry.raiseId === null) { - // Pending ack resolves to the current raise. + if (sameRaise(entry.raiseId, raiseId)) { + return; + } + + if (entry.raiseId !== null && entry.raiseId === entry.status && entry.status === status) { + // Acked with the status alone: the first observation of that status is the + // raise the user answered. Learn it, so the next raise is told apart. store.entries.set(sessionId, { raiseId, - status: status ?? null, + status, ackedAt: entry.ackedAt, expiresAt: entry.expiresAt, }); @@ -340,17 +404,21 @@ export function reconcileSessionAttention( return; } - if (entry.raiseId !== raiseId) { - // Same-session re-raise: replace the raise and clear the ack so the badge - // returns. - store.entries.set(sessionId, { - raiseId, - status: status ?? null, - ackedAt: null, - expiresAt: null, - }); - commit(); + if (statusUpdatedAt == null && entry.status === status) { + // The observer has no timestamp for the acked status, so it cannot tell + // this raise from the answered one. Keeping the ack is what hides the + // badge the user already dismissed. + return; } + + // Different raise: replace the raise and clear the ack so the badge returns. + store.entries.set(sessionId, { + raiseId, + status: status ?? null, + ackedAt: null, + expiresAt: null, + }); + commit(); } /** @@ -386,12 +454,43 @@ export async function clearSessionAttentionForSignOut(): Promise { hydrationPromise = null; } -export function isAttentionAcked(sessionId: string, raiseId: string | null): boolean { +/** + * One raise as a reader sees it: the identity the row gives (`statusUpdatedAt ?? + * status`) plus, when the caller knows it, the row's status. + */ +export type AttentionRaise = { + raiseId: string | null; + status?: string | null; +}; + +/** + * True when the ack names this raise. + * + * Raise ids are compared as identities, so a stored row's PostgreSQL text and a + * live row's ISO form of the same instant are the same raise. A row the surface + * cannot date (`raiseId === status`), or an ack written from a status-only + * answer, matches by status: a miss there would draw the badge for a raise the + * user already answered. A caller that knows the row's status time asks about + * exactly one raise, so a re-raise under the same status shows again. + */ +export function isAttentionAcked(sessionId: string, raise: AttentionRaise): boolean { const entry = store.entries.get(sessionId); if (!entry || entry.ackedAt === null) { return false; } - return entry.raiseId === null || entry.raiseId === raiseId; + if (sameRaise(entry.raiseId, raise.raiseId)) { + return true; + } + if (raise.raiseId === null || entry.status === null) { + return false; + } + // The row could not be dated, so it names the acked status rather than a raise. + if (raise.raiseId === entry.status) { + return true; + } + // The ack itself was written from a status-only answer: nothing it holds can + // tell two raises of that status apart. + return entry.raiseId === entry.status && raise.status === entry.status; } export function shouldShowNeedsInput({ diff --git a/apps/web/src/lib/active-sessions-list.heartbeat.test.ts b/apps/web/src/lib/active-sessions-list.heartbeat.test.ts index d66f415a874..631aee127ba 100644 --- a/apps/web/src/lib/active-sessions-list.heartbeat.test.ts +++ b/apps/web/src/lib/active-sessions-list.heartbeat.test.ts @@ -32,7 +32,7 @@ const enrichmentRows = [ title: `Title ${row.session_id}`, git_url: null, last_activity_at: null, - status_updated_at: null, + status_updated_at: null as string | null, total_cost_microdollars: null, session_pr_platform: null, session_pr_url: null, @@ -49,6 +49,13 @@ const enrichmentRows = [ ...row, })); +const liveSessions = ['former-org', 'personal', 'active-org', 'unattributed'].map(id => ({ + id, + status: 'busy', + title: `Live ${id}`, + connectionId: 'cli', +})); + const where = jest.fn(async () => enrichmentRows); const chain = { from: jest.fn((): unknown => chain), @@ -76,12 +83,7 @@ jest.mock('@/lib/bounded-service-fetch', () => ({ fetchWithinBudget: jest.fn(async () => ({ ok: true, json: async () => ({ - sessions: ['former-org', 'personal', 'active-org', 'unattributed'].map(id => ({ - id, - status: 'busy', - title: `Live ${id}`, - connectionId: 'cli', - })), + sessions: liveSessions, }), })), })); @@ -102,6 +104,9 @@ beforeEach(() => { chain.from.mockClear(); chain.leftJoin.mockClear(); where.mockClear(); + for (const session of liveSessions) { + session.status = 'busy'; + } }); describe('listActiveSessions heartbeat enrichment', () => { @@ -123,4 +128,62 @@ describe('listActiveSessions heartbeat enrichment', () => { expect(byId.get('unattributed')?.organizationId).toBeNull(); expect(byId.has('former-org')).toBe(false); }); + + it.each([ + ['question', 'idle'], + ['question', 'busy'], + ['permission', 'idle'], + ['permission', 'busy'], + ])( + 'omits an earlier timestamp for live %s ahead of stored %s', + async (liveStatus, storedStatus) => { + liveSessions.forEach(session => { + session.status = liveStatus; + }); + where.mockResolvedValueOnce( + enrichmentRows.map(row => ({ + ...row, + status: storedStatus, + status_updated_at: '2026-09-22 09:15:00+00', + })) + ); + + const result = await listActiveSessions({ + userId: 'user-1', + organizationId: undefined, + includeCloudAgentSessions: false, + }); + + expect(result.sessions.find(session => session.id === 'personal')).toMatchObject({ + status: liveStatus, + }); + expect(result.sessions.find(session => session.id === 'personal')).not.toHaveProperty( + 'statusUpdatedAt' + ); + } + ); + + it.each(['question', 'permission', 'busy'])( + 'keeps the dated stored %s status when it remains the resolved status', + async storedStatus => { + where.mockResolvedValueOnce( + enrichmentRows.map(row => ({ + ...row, + status: storedStatus, + status_updated_at: '2026-09-22 09:15:00+00', + })) + ); + + const result = await listActiveSessions({ + userId: 'user-1', + organizationId: undefined, + includeCloudAgentSessions: false, + }); + + expect(result.sessions.find(session => session.id === 'personal')).toMatchObject({ + status: storedStatus, + statusUpdatedAt: '2026-09-22T09:15:00.000Z', + }); + } + ); }); diff --git a/apps/web/src/lib/active-sessions-list.test.ts b/apps/web/src/lib/active-sessions-list.test.ts index 857e777c982..9a7ff669737 100644 --- a/apps/web/src/lib/active-sessions-list.test.ts +++ b/apps/web/src/lib/active-sessions-list.test.ts @@ -203,4 +203,60 @@ describe('listActiveSessions cloud candidates', () => { expect(verified?.associatedPr).toMatchObject({ number: 7, verified: true }); expect(unverified?.associatedPr).toBeUndefined(); }); + + it.each([ + ['idle', true, 'busy', undefined], + ['busy', true, 'busy', '2026-09-22T09:15:00.000Z'], + ['question', true, 'question', '2026-09-22T09:15:00.000Z'], + ['permission', true, 'permission', '2026-09-22T09:15:00.000Z'], + ['idle', false, 'idle', '2026-09-22T09:15:00.000Z'], + ])( + 'dates stored %s with run_open=%s only when the resolved %s status matches', + async (storedStatus, runOpen, resolvedStatus, statusUpdatedAt) => { + limitMock.mockResolvedValue([ + { + session_id: 'candidate', + created_on_platform: null, + created_at: '2026-09-22T09:00:00.000Z', + updated_at: '2026-09-22T09:30:00.000Z', + status: storedStatus, + title: 'candidate', + organization_id: null, + git_url: null, + git_branch: null, + last_activity_at: null, + status_updated_at: '2026-09-22 09:15:00+00', + total_cost_microdollars: null, + cloud_agent_session_id: 'cas-1', + run_open: runOpen, + session_pr_platform: null, + session_pr_url: null, + session_pr_number: null, + session_pr_verified_at: null, + pr_url: null, + pr_number: null, + pr_state: null, + pr_title: null, + pr_head_sha: null, + pr_last_synced_at: null, + pr_review_decision: null, + review_decision_pending: null, + }, + ]); + + const result = await listActiveSessions({ + userId: 'user-1', + organizationId: undefined, + includeCloudAgentSessions: true, + }); + + expect(result.sessions).toHaveLength(1); + expect(result.sessions[0].status).toBe(resolvedStatus); + if (statusUpdatedAt === undefined) { + expect(result.sessions[0]).not.toHaveProperty('statusUpdatedAt'); + } else { + expect(result.sessions[0].statusUpdatedAt).toBe(statusUpdatedAt); + } + } + ); }); diff --git a/apps/web/src/lib/active-sessions-list.ts b/apps/web/src/lib/active-sessions-list.ts index 2b00f625d5d..11814aca536 100644 --- a/apps/web/src/lib/active-sessions-list.ts +++ b/apps/web/src/lib/active-sessions-list.ts @@ -38,9 +38,10 @@ export const activeSessionSchema = z.object({ /** * When this session's status last changed, from * `cli_sessions_v2.status_updated_at`, normalized to ISO 8601. Omitted when - * the column is NULL, unparseable, or the row was never enriched. The - * glanceable snapshot reads it to report how long the longest-waiting agent - * has needed input, and Hermes only parses the ISO form. + * the column is NULL, unparseable, dates a different status than the resolved + * status, or the row was never enriched. The glanceable snapshot reads it to + * report how long the longest-waiting agent has needed input, and Hermes only + * parses the ISO form. */ statusUpdatedAt: z.string().optional(), /** @@ -291,8 +292,11 @@ function mapEnrichedHeartbeatSession( if (row.last_activity_at != null) { mapped.lastActivityAt = row.last_activity_at; } + delete mapped.statusUpdatedAt; const statusUpdatedAt = - row.status_updated_at == null ? null : toIsoTimestamp(row.status_updated_at); + mapped.status !== row.status || row.status_updated_at == null + ? null + : toIsoTimestamp(row.status_updated_at); if (statusUpdatedAt !== null) { mapped.statusUpdatedAt = statusUpdatedAt; } @@ -329,7 +333,9 @@ function mapCloudCandidateRow(row: CloudCandidateRow): ActiveSession { mapped.lastActivityAt = row.last_activity_at; } const statusUpdatedAt = - row.status_updated_at == null ? null : toIsoTimestamp(row.status_updated_at); + mapped.status !== row.status || row.status_updated_at == null + ? null + : toIsoTimestamp(row.status_updated_at); if (statusUpdatedAt !== null) { mapped.statusUpdatedAt = statusUpdatedAt; } diff --git a/services/session-ingest/src/dos/UserConnectionDO.test.ts b/services/session-ingest/src/dos/UserConnectionDO.test.ts index 68537862026..af59d369ae4 100644 --- a/services/session-ingest/src/dos/UserConnectionDO.test.ts +++ b/services/session-ingest/src/dos/UserConnectionDO.test.ts @@ -574,6 +574,356 @@ describe('UserConnectionDO', () => { expect(cliWs.send).not.toHaveBeenCalled(); }); + it('persists authoritative idle over a question heartbeat before broadcasting and after hibernation', async () => { + const { doInstance, mockCtx, ctx } = setup(); + const cliWs = addCliSocket(mockCtx, 'cli-1'); + const webWs = addWebSocket(mockCtx); + sendHeartbeat(doInstance, cliWs, [makeSession('s1', 'question')]); + const changedAt = '2026-10-22T10:00:02.000Z'; + const snapshotsAtDelivery: Array> = []; + webWs.send.mockImplementation(() => { + snapshotsAtDelivery.push(doInstance.getActiveSessions()); + const restored = new UserConnectionDO(ctx as never, {} as Env); + snapshotsAtDelivery.push(restored.getActiveSessions()); + }); + + await doInstance.notifySessionEvent({ + type: 'session.status.updated', + data: { + source: 'v2', + sessionId: 's1', + previousStatus: 'question', + status: 'idle', + statusUpdatedAt: changedAt, + changedAt, + }, + }); + + expect(snapshotsAtDelivery).toMatchObject([ + [{ id: 's1', status: 'idle' }], + [{ id: 's1', status: 'idle' }], + ]); + expect(doInstance.getActiveSessions()).toMatchObject([{ id: 's1', status: 'idle' }]); + const restored = new UserConnectionDO(ctx as never, {} as Env); + expect(restored.getActiveSessions()).toMatchObject([{ id: 's1', status: 'idle' }]); + expect(allSent(webWs).at(-1)).toMatchObject({ + event: 'session.status.updated', + data: { sessionId: 's1', status: 'idle' }, + }); + }); + + it('allows a later undated heartbeat to raise a genuine new question after authoritative idle', async () => { + const { doInstance, mockCtx, ctx } = setup(); + const cliWs = addCliSocket(mockCtx, 'cli-1'); + const webWs = addWebSocket(mockCtx); + sendHeartbeat(doInstance, cliWs, [makeSession('s1', 'question')]); + const changedAt = '2026-10-22T10:00:02.000Z'; + await doInstance.notifySessionEvent({ + type: 'session.status.updated', + data: { + source: 'v2', + sessionId: 's1', + previousStatus: 'question', + status: 'idle', + statusUpdatedAt: changedAt, + changedAt, + }, + }); + const restored = new UserConnectionDO(ctx as never, {} as Env); + + sendHeartbeat(restored, cliWs, [makeSession('s1', 'question')]); + + expect(restored.getActiveSessions()).toMatchObject([{ id: 's1', status: 'question' }]); + expect(allSent(webWs).at(-1)).toMatchObject({ + event: 'sessions.heartbeat', + data: { sessions: [{ id: 's1', status: 'question' }] }, + }); + expect(new UserConnectionDO(ctx as never, {} as Env).getActiveSessions()).toMatchObject([ + { id: 's1', status: 'question' }, + ]); + }); + + it('preserves distinct same-millisecond explicit transitions in arrival order after hibernation', async () => { + const { doInstance, mockCtx, ctx } = setup(); + const cliWs = addCliSocket(mockCtx, 'cli-1'); + const webWs = addWebSocket(mockCtx); + sendHeartbeat(doInstance, cliWs, [makeSession('s1', 'question')]); + await doInstance.notifySessionEvent({ + type: 'session.status.updated', + data: { + source: 'v2', + sessionId: 's1', + previousStatus: 'question', + status: 'idle', + statusUpdatedAt: '2026-10-22T10:00:02.000Z', + changedAt: '2026-10-22T10:00:03.000Z', + }, + }); + webWs.send.mockClear(); + const restored = new UserConnectionDO(ctx as never, {} as Env); + + const result = await restored.notifySessionEvent({ + type: 'session.status.updated', + data: { + source: 'v2', + sessionId: 's1', + previousStatus: 'busy', + status: 'question', + statusUpdatedAt: '2026-10-22T10:00:02.000Z', + changedAt: '2026-10-22T10:00:04.000Z', + }, + }); + + expect(result).toEqual({ delivered: 1 }); + expect(allSent(webWs)).toMatchObject([ + { event: 'session.status.updated', data: { sessionId: 's1', status: 'question' } }, + ]); + expect(restored.getActiveSessions()).toMatchObject([{ id: 's1', status: 'question' }]); + }); + + it('updates full-row child statuses without erasing another live child raise', async () => { + const { doInstance, mockCtx, ctx } = setup(); + const cliWs = addCliSocket(mockCtx, 'cli-1'); + const webWs = addWebSocket(mockCtx); + sendHeartbeat(doInstance, cliWs, [ + { ...makeSession('root', 'scheduled'), scheduledAt: '2026-10-22T11:00:00.000Z' }, + makeSession('child-1', 'question', 'First child', 'root'), + makeSession('child-2', 'permission', 'Second child', 'root'), + ]); + for (const [sessionId, changedAt] of [ + ['child-2', '2026-10-22T10:00:02.000Z'], + ['child-1', '2026-10-22T10:00:03.000Z'], + ]) { + await doInstance.notifySessionEvent({ + type: 'session.status.updated', + data: { + source: 'v2', + session: { + source: 'v2', + sessionId, + createdAt: '2026-10-22T10:00:00.000Z', + updatedAt: changedAt, + title: 'Child', + createdOnPlatform: 'linux', + organizationId: null, + gitUrl: null, + gitBranch: null, + parentSessionId: 'root', + status: 'idle', + statusUpdatedAt: changedAt, + }, + previousStatus: sessionId === 'child-2' ? 'permission' : 'question', + status: 'idle', + statusUpdatedAt: changedAt, + changedAt, + }, + }); + const restored = new UserConnectionDO(ctx as never, {} as Env); + expect(restored.getActiveSessions()).toMatchObject([ + { id: 'root', status: sessionId === 'child-2' ? 'question' : 'scheduled' }, + ]); + expect(allSent(webWs).at(-1)).toMatchObject({ + event: 'session.status.updated', + data: { + sessionId: 'root', + status: sessionId === 'child-2' ? 'question' : 'scheduled', + statusUpdatedAt: null, + ...(sessionId === 'child-1' ? { scheduledAt: '2026-10-22T11:00:00.000Z' } : {}), + }, + }); + } + }); + + it('keeps independent child attention when the root itself becomes idle', async () => { + const { doInstance, mockCtx, ctx } = setup(); + const cliWs = addCliSocket(mockCtx, 'cli-1'); + const webWs = addWebSocket(mockCtx); + sendHeartbeat(doInstance, cliWs, [ + makeSession('root', 'question'), + makeSession('child', 'permission', 'Child', 'root'), + ]); + webWs.send.mockClear(); + + await doInstance.notifySessionEvent({ + type: 'session.status.updated', + data: { + source: 'v2', + sessionId: 'root', + previousStatus: 'question', + status: 'idle', + statusUpdatedAt: '2026-10-22T10:00:02.000Z', + changedAt: '2026-10-22T10:00:03.000Z', + }, + }); + + expect(doInstance.getActiveSessions()).toMatchObject([{ id: 'root', status: 'permission' }]); + expect(new UserConnectionDO(ctx as never, {} as Env).getActiveSessions()).toMatchObject([ + { id: 'root', status: 'permission' }, + ]); + expect(allSent(webWs)).toMatchObject([ + { + event: 'session.status.updated', + data: { sessionId: 'root', status: 'permission', statusUpdatedAt: null }, + }, + ]); + }); + + it('does not change the live snapshot or broadcast when attachment persistence fails', async () => { + const { doInstance, mockCtx, ctx } = setup(); + const cliWs = addCliSocket(mockCtx, 'cli-1'); + const webWs = addWebSocket(mockCtx); + sendHeartbeat(doInstance, cliWs, [makeSession('s1', 'question')]); + webWs.send.mockClear(); + vi.spyOn(cliWs, 'serializeAttachment').mockImplementation(() => { + throw new Error('Attachment too large'); + }); + + await expect( + doInstance.notifySessionEvent({ + type: 'session.status.updated', + data: { + source: 'v2', + sessionId: 's1', + previousStatus: 'question', + status: 'idle', + statusUpdatedAt: null, + changedAt: '2026-10-22T10:00:03.000Z', + }, + }) + ).rejects.toThrow('Attachment too large'); + + expect(webWs.send).not.toHaveBeenCalled(); + expect(doInstance.getActiveSessions()).toMatchObject([{ id: 's1', status: 'question' }]); + expect(new UserConnectionDO(ctx as never, {} as Env).getActiveSessions()).toMatchObject([ + { id: 's1', status: 'question' }, + ]); + }); + + it('updates only the current owner when another connected CLI retains an old session row', async () => { + const { doInstance, mockCtx, ctx } = setup(); + const firstCli = addCliSocket(mockCtx, 'cli-1'); + const currentCli = addCliSocket(mockCtx, 'cli-2'); + sendHeartbeat(doInstance, firstCli, [makeSession('s1', 'question')]); + sendHeartbeat(doInstance, currentCli, [makeSession('s1', 'permission')]); + + await doInstance.notifySessionEvent({ + type: 'session.status.updated', + data: { + source: 'v2', + sessionId: 's1', + previousStatus: 'permission', + status: 'idle', + statusUpdatedAt: null, + changedAt: '2026-10-22T10:00:03.000Z', + }, + }); + + expect(doInstance.getActiveSessions()).toMatchObject([ + { id: 's1', status: 'idle', connectionId: 'cli-2' }, + ]); + expect(new UserConnectionDO(ctx as never, {} as Env).getActiveSessions()).toMatchObject([ + { id: 's1', status: 'idle', connectionId: 'cli-2' }, + ]); + expect(firstCli.deserializeAttachment()).toMatchObject({ + sessions: [{ id: 's1', status: 'question' }], + }); + }); + + it('persists into the open replacement when a closing socket retains the same connection id', async () => { + const { doInstance, mockCtx, ctx } = setup(); + const oldCli = addCliSocket(mockCtx, 'cli-1'); + sendHeartbeat(doInstance, oldCli, [makeSession('s1', 'question')]); + oldCli.readyState = WebSocket.CLOSING; + const replacement = addCliSocket(mockCtx, 'cli-1'); + sendHeartbeat(doInstance, replacement, [makeSession('s1', 'question')]); + + await doInstance.notifySessionEvent({ + type: 'session.status.updated', + data: { + source: 'v2', + sessionId: 's1', + previousStatus: 'question', + status: 'idle', + statusUpdatedAt: null, + changedAt: '2026-10-22T10:00:03.000Z', + }, + }); + mockCtx.removeSocket(oldCli); + mockCtx.addSocket(oldCli); + expect(new UserConnectionDO(ctx as never, {} as Env).getActiveSessions()).toMatchObject([ + { id: 's1', status: 'idle' }, + ]); + mockCtx.removeSocket(oldCli); + + expect(doInstance.getActiveSessions()).toMatchObject([{ id: 's1', status: 'idle' }]); + expect(new UserConnectionDO(ctx as never, {} as Env).getActiveSessions()).toMatchObject([ + { id: 's1', status: 'idle' }, + ]); + }); + + it('does not clear the current root owner attention when a former owner child resolves', async () => { + const { doInstance, mockCtx } = setup(); + const firstCli = addCliSocket(mockCtx, 'cli-1'); + const currentCli = addCliSocket(mockCtx, 'cli-2'); + const webWs = addWebSocket(mockCtx); + sendHeartbeat(doInstance, firstCli, [ + makeSession('root', 'idle'), + makeSession('old-child', 'permission', 'Old child', 'root'), + ]); + sendHeartbeat(doInstance, currentCli, [ + makeSession('root', 'busy'), + makeSession('current-child', 'question', 'Current child', 'root'), + ]); + webWs.send.mockClear(); + + await doInstance.notifySessionEvent({ + type: 'session.status.updated', + data: { + source: 'v2', + sessionId: 'old-child', + previousStatus: 'permission', + status: 'idle', + statusUpdatedAt: null, + changedAt: '2026-10-22T10:00:03.000Z', + }, + }); + + expect(doInstance.getActiveSessions()).toMatchObject([ + { id: 'root', status: 'question', connectionId: 'cli-2' }, + ]); + expect(allSent(webWs)).toMatchObject([ + { event: 'session.status.updated', data: { sessionId: 'old-child', status: 'idle' } }, + ]); + }); + + it('clears unknown authoritative status and obsolete scheduled wake times without inventing idle', async () => { + const { doInstance, mockCtx, ctx } = setup(); + const cliWs = addCliSocket(mockCtx, 'cli-1'); + sendHeartbeat(doInstance, cliWs, [ + { + ...makeSession('s1', 'scheduled'), + scheduledAt: '2026-10-22T11:00:00.000Z', + }, + ]); + + await doInstance.notifySessionEvent({ + type: 'session.status.updated', + data: { + source: 'v2', + sessionId: 's1', + previousStatus: 'scheduled', + status: null, + statusUpdatedAt: null, + changedAt: '2026-10-22T10:00:03.000Z', + }, + }); + + expect(doInstance.getActiveSessions()).toMatchObject([{ id: 's1', status: '' }]); + const restored = new UserConnectionDO(ctx as never, {} as Env); + expect(restored.getActiveSessions()).toMatchObject([{ id: 's1', status: '' }]); + expect(restored.getActiveSessions()[0]).not.toHaveProperty('scheduledAt'); + }); + it('rejects invalid session event payloads without broadcasting', async () => { const { doInstance, mockCtx } = setup(); const webWs = addWebSocket(mockCtx); @@ -2014,6 +2364,37 @@ describe('UserConnectionDO', () => { ); }); + it('reconstructs a lone closing owner after hibernation before scheduling disconnect cleanup', async () => { + const { doInstance, mockCtx, ctx } = setup(); + const cliWs = addCliSocket(mockCtx, 'cli-1', [], undefined, 'usr_1'); + const otherCli = addCliSocket(mockCtx, 'cli-other'); + const webWs = addWebSocket(mockCtx); + sendHeartbeat(doInstance, cliWs, [makeSession('s-question', 'question')]); + sendHeartbeat(doInstance, otherCli, [makeSession('s-other', 'busy')]); + webWs.send.mockClear(); + cliWs.readyState = WebSocket.CLOSING; + const now = 1_700_000_000_000; + vi.spyOn(Date, 'now').mockReturnValue(now); + const restored = new UserConnectionDO(ctx as never, {} as Env); + + await disconnectCli(restored, cliWs); + + expect(ctx.storage.store.get('attentionReset:s-question')).toEqual({ + kiloUserId: 'usr_1', + dueAt: now + CLI_ABSENCE_ATTENTION_RESET_MS, + connectionId: 'cli-1', + }); + expect(ctx.storage.store.has('attentionReset:s-other')).toBe(false); + expect(restored.getActiveSessions()).toMatchObject([ + { id: 's-other', status: 'busy', connectionId: 'cli-other' }, + ]); + expect(allSent(webWs)).toContainEqual({ + type: 'system', + event: 'cli.disconnected', + data: { connectionId: 'cli-1' }, + }); + }); + it('does not hold attention when kiloUserId is missing on the attachment', async () => { const { doInstance, mockCtx, ctx } = setup(); const cliWs = addCliSocket(mockCtx, 'cli-1'); diff --git a/services/session-ingest/src/dos/UserConnectionDO.ts b/services/session-ingest/src/dos/UserConnectionDO.ts index 3080f6d2eb0..6a8deae5c10 100644 --- a/services/session-ingest/src/dos/UserConnectionDO.ts +++ b/services/session-ingest/src/dos/UserConnectionDO.ts @@ -335,7 +335,7 @@ export class UserConnectionDO extends DurableObject { private sessionOwners = new Map(); // Which web sockets want events for a session private webSubscriptions = new Map>(); - // Sessions per CLI connection (from heartbeat) + // Live session snapshots from heartbeats and authoritative status notifications. private connectionSessions = new Map(); // Protocol version per CLI connection (from heartbeat); absent = legacy CLI private connectionProtocolVersion = new Map(); @@ -388,11 +388,21 @@ export class UserConnectionDO extends DurableObject { let webCount = 0; let sessionCount = 0; - for (const ws of this.ctx.getWebSockets()) { + const sockets = this.ctx.getWebSockets(); + const openCliConnectionIds = new Set(); + for (const ws of sockets) { + if (ws.readyState !== WebSocket.OPEN) continue; + const attachment = ws.deserializeAttachment() as WSAttachment | null; + if (attachment?.role === 'cli') openCliConnectionIds.add(attachment.connectionId); + } + for (const ws of sockets) { const attachment = ws.deserializeAttachment() as WSAttachment | null; if (!attachment) continue; if (attachment.role === 'cli') { + if (ws.readyState !== WebSocket.OPEN && openCliConnectionIds.has(attachment.connectionId)) { + continue; + } cliCount++; const { connectionId, sessions, protocolVersion, capabilities } = attachment; this.connectionSessions.set(connectionId, sessions); @@ -2295,6 +2305,12 @@ export class UserConnectionDO extends DurableObject { } } + /** + * Persist authoritative status into the owning socket's live snapshot before + * fanout, so a subsequent list read or hibernation sees the same transition. + * Heartbeats have no status version: later heartbeats remain admissible, and + * notifications retain their existing arrival-order semantics. + */ async notifySessionEvent(event: SessionEventPayload): Promise<{ delivered: number }> { this.ensureState(); const parsed = SessionEventPayloadSchema.parse(event); @@ -2303,12 +2319,76 @@ export class UserConnectionDO extends DurableObject { event: parsed.type, data: parsed.data, }; + const rootStatusEvents: WebInboundMessage[] = []; + if (parsed.type === 'session.status.updated') { + const data = parsed.data; + const sessionId = 'session' in data ? data.session.sessionId : data.sessionId; + const owner = this.sessionOwners.get(sessionId); + const socket = this.ctx.getWebSockets('cli').find(ws => { + if (ws.readyState !== WebSocket.OPEN) return false; + const attachment = ws.deserializeAttachment() as WSAttachment | null; + return attachment?.role === 'cli' && attachment.connectionId === owner; + }); + const attachment = socket?.deserializeAttachment() as WSAttachment | null | undefined; + if (socket && attachment?.role === 'cli') { + const previousRootStatuses = new Map( + this.aggregateSessions().map(session => [session.id, session.status]) + ); + const previous = this.connectionSessions.get(attachment.connectionId) ?? []; + const status = data.status ?? ('session' in data ? data.session.status : null) ?? ''; + const sessions = previous.map(session => { + if (session.id !== sessionId) return session; + const { scheduledAt: _previousScheduledAt, ...rest } = session; + const scheduledAt = + data.scheduledAt ?? ('session' in data ? data.session.scheduledAt : null); + return { + ...rest, + status, + ...(status === 'scheduled' && scheduledAt ? { scheduledAt } : {}), + }; + }); + socket.serializeAttachment({ ...attachment, sessions }); + this.connectionSessions.set(attachment.connectionId, sessions); + + const currentRoots = this.aggregateSessions(); + const effectiveRootStatus = currentRoots.find(session => session.id === sessionId)?.status; + if (effectiveRootStatus !== undefined && effectiveRootStatus !== status) { + msg.data = { + ...data, + status: effectiveRootStatus, + statusUpdatedAt: null, + ...('session' in data + ? { session: { ...data.session, status: effectiveRootStatus, statusUpdatedAt: null } } + : {}), + }; + } + for (const root of currentRoots) { + const previousStatus = previousRootStatuses.get(root.id); + if (root.id === sessionId || previousStatus === root.status) continue; + rootStatusEvents.push({ + type: 'system', + event: 'session.status.updated', + data: { + source: 'v2', + sessionId: root.id, + previousStatus: previousStatus ?? null, + status: root.status, + ...(root.status === 'scheduled' && root.scheduledAt + ? { scheduledAt: root.scheduledAt } + : {}), + statusUpdatedAt: null, + changedAt: data.changedAt, + }, + }); + } + } + } let delivered = 0; - const json = JSON.stringify(msg); + const messages = [msg, ...rootStatusEvents].map(message => JSON.stringify(message)); for (const ws of this.activeWebSockets()) { try { - ws.send(json); + for (const json of messages) ws.send(json); delivered++; } catch (err) { console.warn('notifySessionEvent: skipping failed web socket:', err);