diff --git a/apps/mobile/src/app/(app)/agent-chat/[session-id].tsx b/apps/mobile/src/app/(app)/agent-chat/[session-id].tsx index c6fc84d4be..bad0eb3bd2 100644 --- a/apps/mobile/src/app/(app)/agent-chat/[session-id].tsx +++ b/apps/mobile/src/app/(app)/agent-chat/[session-id].tsx @@ -13,6 +13,7 @@ import { ScreenHeader } from '@/components/screen-header'; import { Button } from '@/components/ui/button'; import { Text } from '@/components/ui/text'; import { useTRPC } from '@/lib/trpc'; +import { useAckSessionAttentionOnOpen } from '@/lib/session-attention'; export default function SessionDetailScreen() { const { @@ -26,6 +27,7 @@ export default function SessionDetailScreen() { }>(); const trpc = useTRPC(); const router = useRouter(); + useAckSessionAttentionOnOpen(sessionId); const sessionQuery = useQuery({ ...trpc.cliSessionsV2.get.queryOptions( { session_id: sessionId }, diff --git a/apps/mobile/src/components/agents/session-list-content.tsx b/apps/mobile/src/components/agents/session-list-content.tsx index 076cd6ea61..a127b3dc01 100644 --- a/apps/mobile/src/components/agents/session-list-content.tsx +++ b/apps/mobile/src/components/agents/session-list-content.tsx @@ -1,3 +1,4 @@ +import { useFocusEffect } from 'expo-router'; import { Bot, Plus, Search, SearchX } from 'lucide-react-native'; import { useCallback, useMemo, useRef, useState } from 'react'; import { @@ -24,6 +25,7 @@ import { type AgentSessionSortBy } from '@/lib/agent-session-sort'; import { type StoredSession } from '@/lib/hooks/use-agent-sessions'; import { useSessionMutations } from '@/lib/hooks/use-session-mutations'; import { useThemeColors } from '@/lib/hooks/use-theme-colors'; +import { getRevisionSnapshot } from '@/lib/session-attention'; import { getTabBarOverlayHeight } from '@/lib/tab-bar-layout'; // Height of the hidden-by-default search bar (mt-3 12 + border 1 + py-1.5 12 + line-20 + border 1 + mb-14 14 = 60). @@ -189,6 +191,25 @@ export function AgentSessionListContent({ [storedSessions] ); + // The tabs navigator uses `freezeOnBlur`, so while the session detail screen + // is pushed the Agents list is frozen. react-freeze reveals the previously + // rendered (cached) cells on return WITHOUT re-running them, so the attention + // store's `useSyncExternalStore` subscription does not re-render the list and + // the detail-screen mount ack is not reflected. Snapshot the attention + // revision only when the tab (re)gains focus, via `useFocusEffect`, which + // fires reliably after unfreeze. Keying the list on that focus snapshot + // remounts it exactly when an ack/reconcile happened while the list was away + // (e.g. returning from a session that was just opened) so frozen cells re-read + // the ack store — while a revision bump for some unrelated session that occurs + // *during* browsing does not touch the snapshot, so scroll is preserved. + const [attentionFocusRevision, setAttentionFocusRevision] = useState(getRevisionSnapshot); + useFocusEffect( + useCallback(() => { + setAttentionFocusRevision(getRevisionSnapshot()); + }, []) + ); + const attentionListKey = `${sortBy}:${attentionFocusRevision}`; + const handleRefresh = useCallback(() => { void (async () => { setRefreshing(true); @@ -304,10 +325,12 @@ export function AgentSessionListContent({ return ( + key={attentionListKey} sections={sections} renderItem={renderItem} renderSectionHeader={renderSectionHeader} keyExtractor={keyExtractor} + extraData={attentionListKey} ListHeaderComponent={listHeader} ListEmptyComponent={emptyComponent} ListFooterComponent={ diff --git a/apps/mobile/src/components/agents/session-row.tsx b/apps/mobile/src/components/agents/session-row.tsx index 9938e5e4e0..aba8b31640 100644 --- a/apps/mobile/src/components/agents/session-row.tsx +++ b/apps/mobile/src/components/agents/session-row.tsx @@ -1,11 +1,17 @@ import * as Haptics from 'expo-haptics'; -import { useRef, useState } from 'react'; +import { useEffect, useRef, useState } from 'react'; import { ActionSheetIOS, Alert, Modal, Platform, Pressable, TextInput, View } from 'react-native'; import { SessionRow } from '@/components/ui/session-row'; import { Text } from '@/components/ui/text'; import { type AgentSessionSortBy, getAgentSessionTimestamp } from '@/lib/agent-session-sort'; import { useThemeColors } from '@/lib/hooks/use-theme-colors'; +import { + isAttentionAcked, + reconcileSessionAttention, + shouldShowNeedsInput, + useSessionAttentionRevision, +} from '@/lib/session-attention'; import { parseTimestamp, timeAgo } from '@/lib/utils'; type StoredSessionRowProps = { @@ -19,6 +25,7 @@ type StoredSessionRowProps = { updated_at: string; git_branch: string | null; status: string | null; + status_updated_at: string | null; }; isLive: boolean; /** @@ -115,6 +122,17 @@ export function StoredSessionRow({ const renameTextRef = useRef(title); const agentLabel = platformLabel(session.created_on_platform); + const revision = useSessionAttentionRevision(); + const raiseId = session.status_updated_at ?? session.status ?? null; + const needsInput = shouldShowNeedsInput({ + status: session.status, + raiseId, + isAcked: isAttentionAcked(session.session_id, raiseId), + }); + useEffect(() => { + reconcileSessionAttention(session.session_id, session.status, session.status_updated_at); + }, [session.session_id, session.status, session.status_updated_at, revision]); + const handleRenameConfirm = () => { const newName = renameTextRef.current.trim(); setRenameVisible(false); @@ -167,7 +185,7 @@ export function StoredSessionRow({ @@ -236,14 +255,30 @@ export function StoredSessionRow({ export function RemoteSessionRow({ session, onPress }: Readonly) { const title = session.title.length > 0 ? session.title : 'Untitled session'; + const revision = useSessionAttentionRevision(); + const raiseId = session.status; + const needsInput = shouldShowNeedsInput({ + status: session.status, + raiseId, + isAcked: isAttentionAcked(session.id, raiseId), + }); + useEffect(() => { + reconcileSessionAttention(session.id, session.status, null); + }, [session.id, session.status, revision]); + return ( - + diff --git a/apps/mobile/src/components/ui/session-row.tsx b/apps/mobile/src/components/ui/session-row.tsx index 2a0cd0d56d..fccd6ebf0e 100644 --- a/apps/mobile/src/components/ui/session-row.tsx +++ b/apps/mobile/src/components/ui/session-row.tsx @@ -1,3 +1,4 @@ +import * as React from 'react'; import { ChevronRight } from 'lucide-react-native'; import { Pressable, View } from 'react-native'; @@ -18,6 +19,11 @@ type SessionRowProps = { meta?: string; /** When true, renders a pulsing good-tone StatusDot before the meta. */ live?: boolean; + /** + * When true, replaces the live dot / meta with a pulsing warn-tone dot + * and a `NEEDS INPUT` label. Highest priority in the eyebrow row. + */ + needsInput?: boolean; onPress?: () => void; /** Suppress bottom divider on the last row of a group. */ last?: boolean; @@ -43,6 +49,7 @@ export function SessionRow({ subtitle, meta, live, + needsInput = false, onPress, last, stripMode = 'edge', @@ -50,7 +57,28 @@ export function SessionRow({ }: Readonly) { const colors = useThemeColors(); const color = agentColor(agentLabel); - const dimStrip = !live; + const dimStrip = !live && !needsInput; + + let eyebrowRight: React.ReactNode = null; + if (needsInput) { + eyebrowRight = ( + + + + NEEDS INPUT + + + ); + } else if (live) { + eyebrowRight = ; + } else if (meta) { + eyebrowRight = ( + + {meta} + + ); + } + const row = ( {agentLabel} - {live && } - {!live && meta && ( - - {meta} - - )} + {eyebrowRight} {title} diff --git a/apps/mobile/src/components/ui/status-dot.tsx b/apps/mobile/src/components/ui/status-dot.tsx index 87b2cab109..b95ad90887 100644 --- a/apps/mobile/src/components/ui/status-dot.tsx +++ b/apps/mobile/src/components/ui/status-dot.tsx @@ -1,4 +1,14 @@ +import { useEffect } from 'react'; import { View } from 'react-native'; +import Animated, { + cancelAnimation, + Easing, + useAnimatedStyle, + useReducedMotion, + useSharedValue, + withRepeat, + withTiming, +} from 'react-native-reanimated'; import { cn } from '@/lib/utils'; @@ -7,6 +17,8 @@ export type StatusDotTone = 'good' | 'warn' | 'danger' | 'muted'; type StatusDotProps = { tone?: StatusDotTone; className?: string; + /** Soft opacity breathe on the dot+halo. Static when reduced motion is on. */ + pulse?: boolean; }; // Solid inner-dot and outer-halo classes per tone. The halo uses the @@ -19,12 +31,66 @@ const TONE: Record = { muted: { dot: 'bg-muted-soft', halo: 'bg-neutral-500/20' }, }; +// Soft breathe range and cadence. Mirrors the provisioning-step pulse +// pattern: 1.0 (fully visible) down to ~0.45 (faded), reversed, looping. +const PULSE_LOW = 0.45; +const PULSE_DURATION_MS = 1100; + /** * Status indicator dot with a halo (replaces CSS box-shadow). * 7px inner dot centered inside a 13px halo. + * + * When `pulse` is true, the entire dot+halo softly breathes via opacity — + * never a hard on/off blink. Respects `useReducedMotion()` and renders + * statically (fully visible) when motion is reduced. */ -export function StatusDot({ tone = 'good', className }: Readonly) { +export function StatusDot({ tone = 'good', className, pulse = false }: Readonly) { const styles = TONE[tone]; + + // Animated branch: opacity breathe on the wrapper so the inner dot and + // halo fade together. Static when reduced motion is on. + const reducedMotion = useReducedMotion(); + const opacity = useSharedValue(1); + useEffect(() => { + // Only the pulsing branch animates. Non-pulsing dots (every existing + // caller) must not start a perpetual invisible animation, and reduced + // motion keeps the dot fully visible. + if (!pulse || reducedMotion) { + cancelAnimation(opacity); + opacity.value = 1; + return undefined; + } + opacity.value = withRepeat( + withTiming(PULSE_LOW, { + duration: PULSE_DURATION_MS, + easing: Easing.inOut(Easing.ease), + }), + -1, + true + ); + return () => { + cancelAnimation(opacity); + }; + }, [opacity, pulse, reducedMotion]); + const pulseStyle = useAnimatedStyle(() => ({ + opacity: opacity.value, + })); + + if (pulse) { + return ( + + + + ); + } + return ( { + __resetSessionAttentionForTests(); +}); + +describe('sessionNeedsInput', () => { + it('returns true for question', () => { + expect(sessionNeedsInput('question')).toBe(true); + }); + + it('returns true for permission', () => { + expect(sessionNeedsInput('permission')).toBe(true); + }); + + it('returns false for idle', () => { + expect(sessionNeedsInput('idle')).toBe(false); + }); + + it('returns false for busy', () => { + expect(sessionNeedsInput('busy')).toBe(false); + }); + + it('returns false for retry', () => { + expect(sessionNeedsInput('retry')).toBe(false); + }); + + it('returns false for null', () => { + expect(sessionNeedsInput(null)).toBe(false); + }); + + it('returns false for undefined', () => { + expect(sessionNeedsInput(undefined)).toBe(false); + }); + + it('returns false for an unknown status string', () => { + expect(sessionNeedsInput('mystery')).toBe(false); + }); +}); + +describe('shouldShowNeedsInput', () => { + it('shows when status is attention and not acked', () => { + expect(shouldShowNeedsInput({ status: 'question', raiseId: 'R1', isAcked: false })).toBe(true); + }); + + it('hides when status is attention and acked', () => { + expect(shouldShowNeedsInput({ status: 'question', raiseId: 'R1', isAcked: true })).toBe(false); + }); + + it('hides when status is non-attention even if not acked', () => { + expect(shouldShowNeedsInput({ status: 'busy', raiseId: null, isAcked: false })).toBe(false); + }); + + it('hides when status is non-attention even if acked', () => { + expect(shouldShowNeedsInput({ status: 'idle', raiseId: null, isAcked: true })).toBe(false); + }); + + it('hides when status is null', () => { + expect(shouldShowNeedsInput({ status: null, raiseId: null, isAcked: false })).toBe(false); + }); +}); + +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('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('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('re-opening a session with a stale resolved ack re-pends 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 opens the session again → ack overwrites with pending + ackSessionAttention('s1'); + expect(isAttentionAcked('s1', 'R2')).toBe(true); + }); + + it('reconcile deletes the entry on a non-attention status, so the next raise shows', () => { + ackSessionAttention('s1'); + reconcileSessionAttention('s1', 'question', 'R1'); + expect(__peekSessionAttentionForTests('s1')).toEqual({ raiseId: 'R1' }); + + // status drops to busy — entry should be cleared + reconcileSessionAttention('s1', 'busy', null); + expect(__peekSessionAttentionForTests('s1')).toBeUndefined(); + + // next question raise (no timestamp — remote active-only row) + // is NOT acked and therefore visible + expect(isAttentionAcked('s1', 'question')).toBe(false); + }); + + it('timestamp-less remote question → busy → question cycle re-raises visibly', () => { + ackSessionAttention('s1'); + // resolve to status string (no statusUpdatedAt) + reconcileSessionAttention('s1', 'question', null); + expect(__peekSessionAttentionForTests('s1')).toEqual({ raiseId: 'question' }); + + // busy clears the entry + reconcileSessionAttention('s1', 'busy', null); + expect(__peekSessionAttentionForTests('s1')).toBeUndefined(); + + // fresh question raise — not acked, visible + expect(isAttentionAcked('s1', 'question')).toBe(false); + }); + + it('frozen-return sequence: pending entry resolves via reconcile, then next raise is not absorbed', () => { + // raise R1 observed, then user opens the session (ack → pending) + ackSessionAttention('s1'); + // reconcile resolves the pending entry to R1 + reconcileSessionAttention('s1', 'question', 'R1'); + expect(isAttentionAcked('s1', 'R1')).toBe(true); + + // new raise R2 — pending is already resolved to R1, so R2 is NOT acked + expect(isAttentionAcked('s1', 'R2')).toBe(false); + }); + + it('reconcile with attention status and no entry is a no-op', () => { + const before = getRevisionSnapshot(); + const listener = vi.fn<() => void>(); + const unsubscribe = subscribe(listener); + + reconcileSessionAttention('s1', 'question', 'R1'); + + expect(getRevisionSnapshot()).toBe(before); + expect(listener).not.toHaveBeenCalled(); + expect(isAttentionAcked('s1', 'R1')).toBe(false); + + unsubscribe(); + }); + + it('reconcile with non-attention status and no entry is a no-op (does not bump revision)', () => { + const before = getRevisionSnapshot(); + const listener = vi.fn<() => void>(); + const unsubscribe = subscribe(listener); + + reconcileSessionAttention('s1', 'busy', null); + + expect(getRevisionSnapshot()).toBe(before); + expect(listener).not.toHaveBeenCalled(); + + unsubscribe(); + }); + + it('reconcile with attention status and a resolved entry is a no-op (does not bump revision)', () => { + ackSessionAttention('s1'); + reconcileSessionAttention('s1', 'question', 'R1'); + // now entry.raiseId === 'R1' + + const before = getRevisionSnapshot(); + const listener = vi.fn<() => void>(); + const unsubscribe = subscribe(listener); + + // same raiseId → still resolved, no change + reconcileSessionAttention('s1', 'question', 'R1'); + expect(getRevisionSnapshot()).toBe(before); + expect(listener).not.toHaveBeenCalled(); + + // different raiseId → resolved entry blocks absorb, no change + reconcileSessionAttention('s1', 'question', 'R2'); + expect(getRevisionSnapshot()).toBe(before); + expect(listener).not.toHaveBeenCalled(); + expect(isAttentionAcked('s1', 'R1')).toBe(true); + expect(isAttentionAcked('s1', 'R2')).toBe(false); + + unsubscribe(); + }); +}); + +describe('revision snapshot and listener notification', () => { + it('bumps revision and notifies listeners on ackSessionAttention', () => { + const before = getRevisionSnapshot(); + const listener = vi.fn<() => void>(); + const unsubscribe = subscribe(listener); + + ackSessionAttention('s1'); + + expect(getRevisionSnapshot()).toBe(before + 1); + expect(listener).toHaveBeenCalledTimes(1); + + unsubscribe(); + }); + + it('does not bump revision or notify when re-acking an already-pending entry', () => { + ackSessionAttention('s1'); + + const after = getRevisionSnapshot(); + const listener = vi.fn<() => void>(); + const unsubscribe = subscribe(listener); + + // Repeated open of the same still-pending session is a no-op. + ackSessionAttention('s1'); + ackSessionAttention('s1'); + + expect(getRevisionSnapshot()).toBe(after); + expect(listener).not.toHaveBeenCalled(); + expect(isAttentionAcked('s1', '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 + + const after = getRevisionSnapshot(); + const listener = vi.fn<() => void>(); + const unsubscribe = subscribe(listener); + + ackSessionAttention('s1'); + + expect(getRevisionSnapshot()).toBe(after + 1); + expect(listener).toHaveBeenCalledTimes(1); + // re-pended: a new raise is once again absorbed + expect(isAttentionAcked('s1', 'R2')).toBe(true); + + unsubscribe(); + }); + + it('bumps revision on mutating reconciles (resolve, delete) and stays stable on no-ops', () => { + const listener = vi.fn<() => void>(); + const unsubscribe = subscribe(listener); + + const initial = getRevisionSnapshot(); + let mutations = 0; + + // ack → mutation + ackSessionAttention('s1'); + if (getRevisionSnapshot() !== initial) { + mutations += 1; + } + + // resolve → mutation + reconcileSessionAttention('s1', 'question', 'R1'); + if (getRevisionSnapshot() !== initial + mutations) { + mutations += 1; + } + + const afterMutations = getRevisionSnapshot(); + + // no-op reconciles: no entry → no change; resolved entry → no change + reconcileSessionAttention('s2', 'busy', null); + reconcileSessionAttention('s1', 'question', 'R1'); + reconcileSessionAttention('s1', 'question', 'R2'); + reconcileSessionAttention('s1', 'question', null); + + expect(getRevisionSnapshot()).toBe(afterMutations); + expect(listener).toHaveBeenCalledTimes(mutations); + + // delete → mutation + reconcileSessionAttention('s1', 'busy', null); + expect(getRevisionSnapshot()).toBe(afterMutations + 1); + expect(listener).toHaveBeenCalledTimes(mutations + 1); + + unsubscribe(); + }); + + it('listener notification count equals revision delta', () => { + const listener = vi.fn<() => void>(); + const unsubscribe = subscribe(listener); + + const start = getRevisionSnapshot(); + ackSessionAttention('a'); + ackSessionAttention('b'); + reconcileSessionAttention('a', 'question', 'R1'); + reconcileSessionAttention('a', 'busy', null); + // no-ops: + reconcileSessionAttention('a', 'busy', null); + reconcileSessionAttention('c', 'question', 'R9'); + reconcileSessionAttention('a', 'question', 'R1'); + const end = getRevisionSnapshot(); + + expect(end - start).toBe(listener.mock.calls.length); + expect(listener).toHaveBeenCalledTimes(4); + + unsubscribe(); + }); + + it('isolates a throwing listener so later subscribers are still notified once', () => { + const throwing = vi.fn(() => { + throw new Error('listener boom'); + }); + const good = vi.fn<() => void>(); + const unsubThrowing = subscribe(throwing); + const unsubGood = subscribe(good); + + const before = getRevisionSnapshot(); + expect(() => { + ackSessionAttention('s1'); + }).not.toThrow(); + expect(throwing).toHaveBeenCalledTimes(1); + expect(good).toHaveBeenCalledTimes(1); + expect(getRevisionSnapshot()).toBe(before + 1); + + unsubThrowing(); + unsubGood(); + }); + + it('unsubscribe stops further notifications', () => { + const listener = vi.fn<() => void>(); + const unsubscribe = subscribe(listener); + + ackSessionAttention('s1'); + expect(listener).toHaveBeenCalledTimes(1); + + unsubscribe(); + // A real mutation on a different session would notify a live listener; + // after unsubscribe it must not. + ackSessionAttention('s2'); + expect(getRevisionSnapshot()).toBeGreaterThan(0); + expect(listener).toHaveBeenCalledTimes(1); + }); +}); diff --git a/apps/mobile/src/lib/session-attention.ts b/apps/mobile/src/lib/session-attention.ts new file mode 100644 index 0000000000..c686aa1555 --- /dev/null +++ b/apps/mobile/src/lib/session-attention.ts @@ -0,0 +1,164 @@ +import { useEffect, useSyncExternalStore } from 'react'; + +/** + * Pure session-attention derivation + in-memory ack store for the mobile + * Agents session list "needs input" indicator. + * + * The detail screen is the only ack writer. Acks are intentionally NOT + * persisted across app restarts. Raise identity is `statusUpdatedAt ?? status` + * (stored rows carry server `status_updated_at`; remote active-only rows + * carry none so identity degrades to the status string). + * + * No backend, tRPC, or shared-package imports: this is a mobile-local + * module so the web client can keep its own copy. + */ + +const ATTENTION_STATUSES = new Set(['question', 'permission']); + +export function sessionNeedsInput(status: string | null | undefined): boolean { + return status != null && ATTENTION_STATUSES.has(status); +} + +type AckEntry = { raiseId: string | null }; + +type AttentionStore = { + listeners: Set<() => void>; + entries: Map; + revision: number; +}; + +const STORE_KEY = '__kiloSessionAttentionStore__'; +const globalScope = globalThis as typeof globalThis & { [STORE_KEY]?: AttentionStore }; +const store: AttentionStore = (globalScope[STORE_KEY] ??= { + listeners: new Set<() => void>(), + entries: new Map(), + revision: 0, +}); + +function bumpRevision(): void { + store.revision += 1; + // Isolate subscribers: one throwing listener must not prevent the rest from + // being notified of the revision change. + for (const listener of store.listeners) { + try { + listener(); + } catch { + // A subscriber's own error must not break store notification. + } + } +} + +export function subscribe(listener: () => void): () => void { + store.listeners.add(listener); + return () => { + store.listeners.delete(listener); + }; +} + +export function getRevisionSnapshot(): number { + return store.revision; +} + +/** + * Server snapshot for `useSyncExternalStore`. Stable across calls so the + * hook is SSR / RN-safe; the real revision is read on the client. + */ +function getServerSnapshot(): number { + return 0; +} + +export function ackSessionAttention(sessionId: string): void { + // Opening a session always leaves the entry pending. If it is already + // pending, nothing changes — skip the bump so we don't fire a redundant + // global re-render (e.g. React Strict Mode's double effect invocation). + if (store.entries.get(sessionId)?.raiseId === null) { + return; + } + store.entries.set(sessionId, { raiseId: null }); + bumpRevision(); +} + +/** + * Reconcile the ack store against the latest observed status. + * + * `raiseId = statusUpdatedAt ?? status`. + * + * - non-attention status: delete the entry (if any) and notify + * - attention + existing pending entry: resolve it to the current raise + * - otherwise: no-op (does NOT bump the revision) + */ +export function reconcileSessionAttention( + sessionId: string, + status: string | null | undefined, + statusUpdatedAt: string | null | undefined +): void { + if (!sessionNeedsInput(status)) { + if (store.entries.delete(sessionId)) { + bumpRevision(); + } + return; + } + + const raiseId = statusUpdatedAt ?? status ?? null; + if (store.entries.get(sessionId)?.raiseId === null) { + store.entries.set(sessionId, { raiseId }); + bumpRevision(); + } +} + +export function isAttentionAcked(sessionId: string, raiseId: string | null): boolean { + const entry = store.entries.get(sessionId); + if (!entry) { + return false; + } + return entry.raiseId === null || entry.raiseId === raiseId; +} + +export function shouldShowNeedsInput({ + status, + raiseId: _raiseId, + isAcked, +}: { + status: string | null | undefined; + raiseId: string | null; + isAcked: boolean; +}): boolean { + return sessionNeedsInput(status) && !isAcked; +} + +/** + * Subscribe a component to the ack store's revision counter. When the + * revision changes, the component re-renders and re-evaluates + * `isAttentionAcked` for its session. + */ +export function useSessionAttentionRevision(): number { + return useSyncExternalStore(subscribe, getRevisionSnapshot, getServerSnapshot); +} + +/** + * Ack a session's attention indicator when the detail screen opens. + * Re-runs if `sessionId` changes (e.g. switching sessions). + */ +export function useAckSessionAttentionOnOpen(sessionId: string): void { + useEffect(() => { + ackSessionAttention(sessionId); + }, [sessionId]); +} + +/** + * Test-only: clear all acks and reset the revision counter so each + * test starts from a known state. Not for production use. + */ +export function __resetSessionAttentionForTests(): void { + store.entries.clear(); + store.revision = 0; +} + +/** + * Test-only: peek at the current entry for a session (or undefined if + * no entry exists). Lets tests assert on the raw store shape without + * exposing it on the production API. + */ +export function __peekSessionAttentionForTests(sessionId: string): AckEntry | undefined { + return store.entries.get(sessionId); +} diff --git a/services/cloud-agent-next/src/persistence/CloudAgentSession.ts b/services/cloud-agent-next/src/persistence/CloudAgentSession.ts index f8feb88043..da5b3f8477 100644 --- a/services/cloud-agent-next/src/persistence/CloudAgentSession.ts +++ b/services/cloud-agent-next/src/persistence/CloudAgentSession.ts @@ -63,6 +63,8 @@ import { type IngestHandler, type IngestDOContext, } from '../websocket/ingest.js'; +import type { AttentionEvent } from '../websocket/ingest-attention-classifier.js'; +import { dispatchCloudAgentAttentionPush } from '../websocket/ingest-attention-classifier.js'; import type { StoredEvent } from '../websocket/types.js'; import type { WrapperCommand, CloudStatusData } from '../shared/protocol.js'; import { @@ -659,6 +661,43 @@ export class CloudAgentSession extends DurableObject { return this.messageSettlementOutbox; } + /** + * Best-effort push for a root-session `question.asked`/`permission.asked` + * event. Mirrors the terminal push gates in + * `MessageSettlementOutbox.settlePushNotificationEffect`: + * - metadata present, + * - source event is from the root kilocode session (not a child), + * - the run was created on the cloud-agent-web platform, + * - no /stream clients are currently connected. + * + * Runs under `ctx.waitUntil`; any failure is logged and swallowed so the + * ingest path is never broken. Replay dedup is owned by the notifications + * service via the stable `executionId: attention:{requestId}` key. + * + * Gate + dispatch logic lives in `dispatchCloudAgentAttentionPush` so it + * is unit-testable without constructing this DO. This method is the + * thin DO-specific adapter: fetch metadata, inject stream-client + push + * dependencies, and log on dispatch failure. + */ + private async handleAttentionEvent(event: AttentionEvent): Promise { + try { + const metadata = await this.getMetadata(); + await dispatchCloudAgentAttentionPush(event, metadata, { + hasConnectedStreamClients: () => getConnectedStreamClientCount(this.ctx) > 0, + sendPush: params => this.env.NOTIFICATIONS.sendCloudAgentSessionNotification(params), + }); + } catch (error) { + logger + .withFields({ + sessionId: this.sessionId, + requestId: event.requestId, + kind: event.kind, + error: error instanceof Error ? error.message : String(error), + }) + .warn('Cloud agent attention push notification dispatch failed'); + } + } + private getAgentRuntime(): AgentRuntime { if (!this.agentRuntime) { this.agentRuntime = createAgentRuntime({ @@ -839,6 +878,13 @@ export class CloudAgentSession extends DurableObject { : params ); }, + // Best-effort attention push: dispatch under ctx.waitUntil so ingest + // remains non-blocking. Duplicate replays invoke this each time; + // dedup is the notifications service's job (executionId is stable + // per raise: `attention:{requestId}`). + onAttentionEvent: event => { + this.ctx.waitUntil(this.handleAttentionEvent(event)); + }, }; this.ingestHandler = createIngestHandler( diff --git a/services/cloud-agent-next/src/websocket/ingest-attention-classifier.test.ts b/services/cloud-agent-next/src/websocket/ingest-attention-classifier.test.ts new file mode 100644 index 0000000000..245f62ce1b --- /dev/null +++ b/services/cloud-agent-next/src/websocket/ingest-attention-classifier.test.ts @@ -0,0 +1,340 @@ +import { describe, it, expect, vi } from 'vitest'; +import type { Mock } from 'vitest'; +import { + classifyAttentionKilocodeEvent, + dispatchCloudAgentAttentionPush, + type AttentionPushDeps, + type CloudAgentAttentionMetadata, +} from './ingest-attention-classifier.js'; +import type { AttentionEvent } from './ingest-attention-classifier.js'; + +const SOURCE_SESSION_ID = 'kilo_session_source'; + +describe('classifyAttentionKilocodeEvent', () => { + describe('raise mappings', () => { + it.each([ + ['question.asked', 'question'], + ['permission.asked', 'permission'], + ] as const)('%s with nested properties.id raises %s', (eventName, kind) => { + const result = classifyAttentionKilocodeEvent({ + event: eventName, + properties: { id: 'req_nested', sessionID: SOURCE_SESSION_ID }, + }); + expect(result).toEqual({ + requestId: 'req_nested', + kind, + sourceKiloSessionId: SOURCE_SESSION_ID, + }); + }); + + it.each([ + ['question.asked', 'question'], + ['permission.asked', 'permission'], + ] as const)('%s with direct top-level data.id fallback raises %s', (eventName, kind) => { + const result = classifyAttentionKilocodeEvent({ + event: eventName, + id: 'req_direct', + sessionID: SOURCE_SESSION_ID, + }); + expect(result).toEqual({ + requestId: 'req_direct', + kind, + sourceKiloSessionId: SOURCE_SESSION_ID, + }); + }); + + it('prefers properties.id over data.id for raise', () => { + const result = classifyAttentionKilocodeEvent({ + event: 'question.asked', + id: 'req_direct', + sessionID: 'top_session', + properties: { id: 'req_nested', sessionID: SOURCE_SESSION_ID }, + }); + expect(result).toEqual({ + requestId: 'req_nested', + kind: 'question', + sourceKiloSessionId: SOURCE_SESSION_ID, + }); + }); + + it('prefers properties.sessionID over data.sessionID', () => { + const result = classifyAttentionKilocodeEvent({ + event: 'permission.asked', + id: 'req_1', + sessionID: 'top_session', + properties: { id: 'req_1', sessionID: SOURCE_SESSION_ID }, + }); + expect(result).toEqual({ + requestId: 'req_1', + kind: 'permission', + sourceKiloSessionId: SOURCE_SESSION_ID, + }); + }); + }); + + describe('resolve events are explicitly ignored', () => { + it.each(['question.replied', 'question.rejected', 'permission.replied'])( + '%s returns null even with a valid id and source sessionID', + eventName => { + expect( + classifyAttentionKilocodeEvent({ + event: eventName, + properties: { id: 'req_x', requestID: 'req_x', sessionID: SOURCE_SESSION_ID }, + }) + ).toBeNull(); + } + ); + }); + + describe('ignored event types', () => { + it.each([ + 'session.status', + 'session.idle', + 'session.diff', + 'session.completed', + 'session.error', + 'session.network.asked', + 'session.network.restored', + 'message.part.delta', + 'message.part.updated', + 'message.updated', + 'message.part.removed', + 'session.created', + 'session.updated', + 'session.turn.close', + 'permission.ask', // partial + 'question.ask', // partial + 'retry.foo', + 'error.bar', + 'unknown', + ])('ignores %s', eventName => { + const result = classifyAttentionKilocodeEvent({ + event: eventName, + id: 'req_1', + properties: { id: 'req_1', sessionID: SOURCE_SESSION_ID }, + }); + expect(result).toBeNull(); + }); + }); + + describe('missing or invalid id or sessionID', () => { + it('ignores qualifying event with no id anywhere', () => { + const result = classifyAttentionKilocodeEvent({ + event: 'question.asked', + properties: { sessionID: SOURCE_SESSION_ID }, + }); + expect(result).toBeNull(); + }); + + it('ignores qualifying event with empty string id', () => { + const result = classifyAttentionKilocodeEvent({ + event: 'question.asked', + id: '', + sessionID: SOURCE_SESSION_ID, + properties: { id: '' }, + }); + expect(result).toBeNull(); + }); + + it('ignores qualifying event when source sessionID is missing', () => { + const result = classifyAttentionKilocodeEvent({ + event: 'question.asked', + properties: { id: 'req_present' }, + }); + expect(result).toBeNull(); + }); + + it('ignores qualifying event when source sessionID is empty', () => { + const result = classifyAttentionKilocodeEvent({ + event: 'question.asked', + properties: { id: 'req_present', sessionID: '' }, + }); + expect(result).toBeNull(); + }); + + it('ignores qualifying event with non-string id', () => { + const result = classifyAttentionKilocodeEvent({ + event: 'question.asked', + id: 123, + sessionID: SOURCE_SESSION_ID, + properties: { id: { foo: 'bar' }, sessionID: SOURCE_SESSION_ID }, + }); + expect(result).toBeNull(); + }); + + it('falls back to top-level id when properties is not an object', () => { + const result = classifyAttentionKilocodeEvent({ + event: 'question.asked', + properties: 'not-an-object', + id: 'req_direct', + sessionID: SOURCE_SESSION_ID, + }); + expect(result).toEqual({ + requestId: 'req_direct', + kind: 'question', + sourceKiloSessionId: SOURCE_SESSION_ID, + }); + }); + + it('returns null when properties is null and no top-level id', () => { + const result = classifyAttentionKilocodeEvent({ + event: 'question.asked', + properties: null, + }); + expect(result).toBeNull(); + }); + }); + + describe('non-object input', () => { + it.each([null, undefined, 'string', 42, true, []])('returns null for %s', value => { + expect(classifyAttentionKilocodeEvent(value)).toBeNull(); + }); + }); + + describe('missing event name', () => { + it('returns null when event is missing', () => { + expect(classifyAttentionKilocodeEvent({ id: 'req_1' })).toBeNull(); + }); + + it('returns null when event is not a string', () => { + expect(classifyAttentionKilocodeEvent({ event: 42, id: 'req_1' })).toBeNull(); + }); + }); +}); + +describe('dispatchCloudAgentAttentionPush', () => { + const baseMetadata: CloudAgentAttentionMetadata = { + auth: { kiloSessionId: 'kilo_root' }, + identity: { + sessionId: 'sess_1', + userId: 'user_1', + createdOnPlatform: 'cloud-agent-web', + }, + }; + + const baseEvent: AttentionEvent = { + requestId: 'req_1', + kind: 'question', + sourceKiloSessionId: 'kilo_root', + }; + + function createHarness(overrides?: { hasConnectedStreamClients?: Mock }): { + deps: AttentionPushDeps; + sendPush: Mock; + hasConnectedStreamClients: Mock; + } { + const hasConnectedStreamClients = overrides?.hasConnectedStreamClients ?? vi.fn(() => false); + const sendPush = vi.fn(() => Promise.resolve(undefined)); + return { + deps: { hasConnectedStreamClients, sendPush } satisfies AttentionPushDeps, + sendPush, + hasConnectedStreamClients, + }; + } + + it('dispatches with exact payload for question raise when all gates pass', async () => { + const { deps, sendPush } = createHarness(); + const result = await dispatchCloudAgentAttentionPush(baseEvent, baseMetadata, deps); + + expect(result).toBe('sent'); + expect(sendPush).toHaveBeenCalledTimes(1); + expect(sendPush).toHaveBeenCalledWith({ + userId: 'user_1', + cliSessionId: 'kilo_root', + executionId: 'attention:req_1', + status: 'completed', + body: 'Kilo needs your input.', + suppressIfViewingSession: true, + }); + }); + + it('dispatches with exact payload for permission raise when all gates pass', async () => { + const { deps, sendPush } = createHarness(); + const permissionEvent: AttentionEvent = { + requestId: 'req_2', + kind: 'permission', + sourceKiloSessionId: 'kilo_root', + }; + + const result = await dispatchCloudAgentAttentionPush(permissionEvent, baseMetadata, deps); + + expect(result).toBe('sent'); + expect(sendPush).toHaveBeenCalledTimes(1); + expect(sendPush).toHaveBeenCalledWith({ + userId: 'user_1', + cliSessionId: 'kilo_root', + executionId: 'attention:req_2', + status: 'completed', + body: 'Kilo needs your input.', + suppressIfViewingSession: true, + }); + }); + + it('suppresses when sourceKiloSessionId does not match metadata (non-root session)', async () => { + const { deps, sendPush } = createHarness(); + const childEvent: AttentionEvent = { + requestId: 'req_3', + kind: 'question', + sourceKiloSessionId: 'kilo_child', + }; + + const result = await dispatchCloudAgentAttentionPush(childEvent, baseMetadata, deps); + + expect(result).toBe('suppressed'); + expect(sendPush).not.toHaveBeenCalled(); + }); + + it('suppresses when metadata is null', async () => { + const { deps, sendPush } = createHarness(); + const result = await dispatchCloudAgentAttentionPush(baseEvent, null, deps); + + expect(result).toBe('suppressed'); + expect(sendPush).not.toHaveBeenCalled(); + }); + + it('suppresses when createdOnPlatform is not cloud-agent-web', async () => { + const { deps, sendPush } = createHarness(); + const cliMetadata: CloudAgentAttentionMetadata = { + auth: { kiloSessionId: 'kilo_root' }, + identity: { + sessionId: 'sess_1', + userId: 'user_1', + createdOnPlatform: 'cli', + }, + }; + + const result = await dispatchCloudAgentAttentionPush(baseEvent, cliMetadata, deps); + + expect(result).toBe('suppressed'); + expect(sendPush).not.toHaveBeenCalled(); + }); + + it('suppresses when metadata.auth.kiloSessionId is missing', async () => { + const { deps, sendPush } = createHarness(); + const noKiloSessionIdMetadata: CloudAgentAttentionMetadata = { + auth: {}, + identity: { + sessionId: 'sess_1', + userId: 'user_1', + createdOnPlatform: 'cloud-agent-web', + }, + }; + + const result = await dispatchCloudAgentAttentionPush(baseEvent, noKiloSessionIdMetadata, deps); + + expect(result).toBe('suppressed'); + expect(sendPush).not.toHaveBeenCalled(); + }); + + it('suppresses when hasConnectedStreamClients returns true', async () => { + const { deps, sendPush, hasConnectedStreamClients } = createHarness({ + hasConnectedStreamClients: vi.fn(() => true), + }); + + const result = await dispatchCloudAgentAttentionPush(baseEvent, baseMetadata, deps); + + expect(result).toBe('suppressed'); + expect(sendPush).not.toHaveBeenCalled(); + expect(hasConnectedStreamClients).toHaveBeenCalledTimes(1); + }); +}); diff --git a/services/cloud-agent-next/src/websocket/ingest-attention-classifier.ts b/services/cloud-agent-next/src/websocket/ingest-attention-classifier.ts new file mode 100644 index 0000000000..9c5df5a6d9 --- /dev/null +++ b/services/cloud-agent-next/src/websocket/ingest-attention-classifier.ts @@ -0,0 +1,153 @@ +/** + * Attention event classifier for kilocode ingest events. + * + * Cloud agent session attention is raise-only: `question.asked` and + * `permission.asked` indicate the wrapper is waiting for human input and + * should trigger a best-effort push notification. Resolves are not + * consumed (no outbox or scheduler) so they intentionally return `null`. + * + * The classifier is pure and synchronous, returning a stable requestId, + * a raise kind, and the source `kiloSessionId` so the caller can filter + * out events from child/sub-agent kilocode sessions of this Cloud Agent + * run, or null for non-attention events. + * + * Authoritative id source for raises: `properties.id`, with top-level + * `data.id` as the fallback (the wrapper's real-time shape spreads + * properties at the top level of `data`). + * + * The source `kiloSessionId` is extracted from `properties.sessionID` + * (authoritative) or top-level `data.sessionID` (fallback). A qualifying + * event without a non-empty source `sessionID` is ignored because the + * caller cannot verify it belongs to the root session. + */ + +import type { SessionMetadata } from '../persistence/session-metadata.js'; +import type { SendCloudAgentSessionNotificationParams } from '../notifications-binding.js'; + +export type AttentionEvent = { + requestId: string; + kind: 'question' | 'permission'; + sourceKiloSessionId: string; +}; + +/** + * The subset of `SessionMetadata` needed by the attention push gate. + * Kept narrow so the dispatch function can be called from the Durable + * Object (full `SessionMetadata`) and from unit tests (synthetic fixtures) + * without coupling tests to unrelated metadata fields. + */ +export type CloudAgentAttentionMetadata = Pick; + +/** + * Dependencies the dispatch function needs from its host. The Durable + * Object supplies these; unit tests inject spies. + */ +export type AttentionPushDeps = { + hasConnectedStreamClients: () => boolean; + sendPush: (params: SendCloudAgentSessionNotificationParams) => Promise; +}; + +const RAISE_KILO_EVENTS: ReadonlyMap = new Map([ + ['question.asked', 'question'], + ['permission.asked', 'permission'], +]); + +function readNonEmptyString( + record: Record | null, + key: string +): string | undefined { + if (!record) return undefined; + const value = record[key]; + return typeof value === 'string' && value.length > 0 ? value : undefined; +} + +function readSessionIdFromRecord(record: Record | null): string | undefined { + if (!record) return undefined; + return readNonEmptyString(record, 'sessionID'); +} + +/** + * Classify a kilocode ingest event for the attention system. + * + * Returns a stable requestId and raise kind, or null when the event is + * not an attention raise. The wrapper's real-time shape spreads + * properties at the top level of `data`, so the id is read from the + * nested `properties` first and falls back to the top-level field. + * + * @param data - The kilocode event data (already validated as an object) + * @returns AttentionEvent with requestId, kind, and sourceKiloSessionId, or null + */ +export function classifyAttentionKilocodeEvent(data: unknown): AttentionEvent | null { + if (typeof data !== 'object' || data === null) return null; + const dataRecord = data as Record; + const eventName = typeof dataRecord.event === 'string' ? dataRecord.event : undefined; + if (!eventName) return null; + + const kind = RAISE_KILO_EVENTS.get(eventName); + if (kind === undefined) return null; + + const properties = + typeof dataRecord.properties === 'object' && dataRecord.properties !== null + ? (dataRecord.properties as Record) + : null; + + // Authoritative nested id; top-level fallback for the wrapper's + // real-time spread shape. + const requestId = readNonEmptyString(properties, 'id') ?? readNonEmptyString(dataRecord, 'id'); + if (!requestId) return null; + + const sourceKiloSessionId = + readSessionIdFromRecord(properties) ?? readSessionIdFromRecord(dataRecord); + if (!sourceKiloSessionId) return null; + + return { requestId, kind, sourceKiloSessionId }; +} + +/** + * Gate + dispatch a best-effort push notification for a root-session + * `question.asked`/`permission.asked` event. Mirrors the terminal push + * gates used by the message settlement outbox. + * + * Gate order (each step short-circuits with `'suppressed'`): + * 1. metadata must be present + * 2. event's source `kiloSessionId` must equal the run's root + * `auth.kiloSessionId` (filters child/sub-agent sessions) + * 3. `identity.createdOnPlatform` must be `'cloud-agent-web'` + * 4. `auth.kiloSessionId` must be present (defensive; subsumed by + * step 2 today but explicit for future callers) + * 5. no `/stream` clients may currently be connected + * + * On the happy path, calls `deps.sendPush` with the stable + * `executionId: 'attention:' + event.requestId` and returns `'sent'`. + * The executionId is the dedup key the notifications service uses to + * drop replays for the same raise. + * + * This function is intentionally free of try/catch: any error from + * `deps.sendPush` propagates to the caller (the Durable Object's + * `handleAttentionEvent`), which logs with DO-specific context and + * swallows so the ingest path stays non-blocking. + */ +export async function dispatchCloudAgentAttentionPush( + event: AttentionEvent, + metadata: CloudAgentAttentionMetadata | null, + deps: AttentionPushDeps +): Promise<'sent' | 'suppressed'> { + if (!metadata) return 'suppressed'; + + const cliSessionId = metadata.auth.kiloSessionId; + if (event.sourceKiloSessionId !== cliSessionId) return 'suppressed'; + if (metadata.identity.createdOnPlatform !== 'cloud-agent-web') return 'suppressed'; + if (!cliSessionId) return 'suppressed'; + if (deps.hasConnectedStreamClients()) return 'suppressed'; + + await deps.sendPush({ + userId: metadata.identity.userId, + cliSessionId, + executionId: `attention:${event.requestId}`, + status: 'completed', + body: 'Kilo needs your input.', + suppressIfViewingSession: true, + }); + + return 'sent'; +} diff --git a/services/cloud-agent-next/src/websocket/ingest.test.ts b/services/cloud-agent-next/src/websocket/ingest.test.ts index 03f64f8e55..a4a185e88c 100644 --- a/services/cloud-agent-next/src/websocket/ingest.test.ts +++ b/services/cloud-agent-next/src/websocket/ingest.test.ts @@ -1653,4 +1653,218 @@ describe('createIngestHandler', () => { expect(await response.text()).toBe('Missing wrapperRunId parameter'); }); }); + + describe('handleIngestMessage — onAttentionEvent hook', () => { + const SOURCE_SESSION_ID = 'kilo_session_source'; + + function makeKilocodeData( + eventName: string, + props: Record, + options: { spreadId?: boolean } = {} + ): Record { + return options.spreadId + ? { ...props, event: eventName, type: eventName, properties: props } + : { event: eventName, type: eventName, properties: props }; + } + + function makeKilocodeMessageFromData(data: Record): string { + return JSON.stringify({ + streamEventType: 'kilocode', + data, + timestamp: new Date().toISOString(), + }); + } + + function createAttentionContext(): { + doContext: IngestDOContext; + onAttentionEvent: ReturnType; + } { + const doContext = createFakeDOContext(); + const onAttentionEvent = vi.fn(); + return { doContext: { ...doContext, onAttentionEvent }, onAttentionEvent }; + } + + it.each([ + { eventName: 'question.asked', kind: 'question' as const }, + { eventName: 'permission.asked', kind: 'permission' as const }, + ])('invokes onAttentionEvent for qualifying $eventName', async ({ eventName, kind }) => { + const { doContext, onAttentionEvent } = createAttentionContext(); + const broadcastFn = vi.fn(); + const handler = createIngestHandler( + createFakeState(), + createFakeEventQueries(), + SESSION_ID, + broadcastFn, + doContext + ); + const ws = createFakeWebSocket(makeAttachment()); + await handler.handleIngestMessage( + ws, + makeKilocodeMessageFromData( + makeKilocodeData(eventName, { id: 'req_nested', sessionID: SOURCE_SESSION_ID }) + ) + ); + expect(onAttentionEvent).toHaveBeenCalledTimes(1); + expect(onAttentionEvent).toHaveBeenCalledWith({ + requestId: 'req_nested', + kind, + sourceKiloSessionId: SOURCE_SESSION_ID, + }); + expect(broadcastFn).toHaveBeenCalledWith( + expect.objectContaining({ id: 0, stream_event_type: 'kilocode' }) + ); + }); + + it('invokes onAttentionEvent for top-level-spread data shape', async () => { + const { doContext, onAttentionEvent } = createAttentionContext(); + const broadcastFn = vi.fn(); + const handler = createIngestHandler( + createFakeState(), + createFakeEventQueries(), + SESSION_ID, + broadcastFn, + doContext + ); + const ws = createFakeWebSocket(makeAttachment()); + await handler.handleIngestMessage( + ws, + makeKilocodeMessageFromData( + makeKilocodeData( + 'question.asked', + { id: 'req_direct', sessionID: SOURCE_SESSION_ID }, + { spreadId: true } + ) + ) + ); + expect(onAttentionEvent).toHaveBeenCalledTimes(1); + expect(onAttentionEvent).toHaveBeenCalledWith({ + requestId: 'req_direct', + kind: 'question', + sourceKiloSessionId: SOURCE_SESSION_ID, + }); + expect(broadcastFn).toHaveBeenCalledWith( + expect.objectContaining({ id: 0, stream_event_type: 'kilocode' }) + ); + }); + + it.each(['session.diff', 'message.part.delta', 'session.completed'])( + 'does not invoke onAttentionEvent for non-attention kilocode event %s', + async eventName => { + const { doContext, onAttentionEvent } = createAttentionContext(); + const broadcastFn = vi.fn(); + const handler = createIngestHandler( + createFakeState(), + createFakeEventQueries(), + SESSION_ID, + broadcastFn, + doContext + ); + const ws = createFakeWebSocket(makeAttachment()); + await handler.handleIngestMessage( + ws, + makeKilocodeMessageFromData( + makeKilocodeData(eventName, { id: 'req_1', sessionID: SOURCE_SESSION_ID }) + ) + ); + expect(onAttentionEvent).not.toHaveBeenCalled(); + expect(broadcastFn).toHaveBeenCalledWith( + expect.objectContaining({ id: 0, stream_event_type: 'kilocode' }) + ); + } + ); + + it('does not invoke onAttentionEvent for resolve events', async () => { + const { doContext, onAttentionEvent } = createAttentionContext(); + const broadcastFn = vi.fn(); + const handler = createIngestHandler( + createFakeState(), + createFakeEventQueries(), + SESSION_ID, + broadcastFn, + doContext + ); + const ws = createFakeWebSocket(makeAttachment()); + await handler.handleIngestMessage( + ws, + makeKilocodeMessageFromData( + makeKilocodeData('question.replied', { + id: 'req_x', + requestID: 'req_x', + sessionID: SOURCE_SESSION_ID, + }) + ) + ); + expect(onAttentionEvent).not.toHaveBeenCalled(); + }); + + it('does not invoke onAttentionEvent when id is missing', async () => { + const { doContext, onAttentionEvent } = createAttentionContext(); + const broadcastFn = vi.fn(); + const handler = createIngestHandler( + createFakeState(), + createFakeEventQueries(), + SESSION_ID, + broadcastFn, + doContext + ); + const ws = createFakeWebSocket(makeAttachment()); + await handler.handleIngestMessage( + ws, + makeKilocodeMessageFromData({ + event: 'question.asked', + type: 'question.asked', + properties: { sessionID: SOURCE_SESSION_ID }, + }) + ); + expect(onAttentionEvent).not.toHaveBeenCalled(); + expect(broadcastFn).toHaveBeenCalledWith( + expect.objectContaining({ id: 0, stream_event_type: 'kilocode' }) + ); + }); + + it('does not invoke onAttentionEvent when source sessionID is missing', async () => { + const { doContext, onAttentionEvent } = createAttentionContext(); + const broadcastFn = vi.fn(); + const handler = createIngestHandler( + createFakeState(), + createFakeEventQueries(), + SESSION_ID, + broadcastFn, + doContext + ); + const ws = createFakeWebSocket(makeAttachment()); + await handler.handleIngestMessage( + ws, + makeKilocodeMessageFromData(makeKilocodeData('question.asked', { id: 'req_no_session' })) + ); + expect(onAttentionEvent).not.toHaveBeenCalled(); + expect(broadcastFn).toHaveBeenCalledWith( + expect.objectContaining({ id: 0, stream_event_type: 'kilocode' }) + ); + }); + + it('does not throw when onAttentionEvent is not registered', async () => { + const doContext = createFakeDOContext(); // no onAttentionEvent + const broadcastFn = vi.fn(); + const handler = createIngestHandler( + createFakeState(), + createFakeEventQueries(), + SESSION_ID, + broadcastFn, + doContext + ); + const ws = createFakeWebSocket(makeAttachment()); + await expect( + handler.handleIngestMessage( + ws, + makeKilocodeMessageFromData( + makeKilocodeData('question.asked', { id: 'req_1', sessionID: SOURCE_SESSION_ID }) + ) + ) + ).resolves.toBeUndefined(); + expect(broadcastFn).toHaveBeenCalledWith( + expect.objectContaining({ id: 0, stream_event_type: 'kilocode' }) + ); + }); + }); }); diff --git a/services/cloud-agent-next/src/websocket/ingest.ts b/services/cloud-agent-next/src/websocket/ingest.ts index 65d953ba36..63da454c36 100644 --- a/services/cloud-agent-next/src/websocket/ingest.ts +++ b/services/cloud-agent-next/src/websocket/ingest.ts @@ -40,6 +40,10 @@ import { cloudStatusForPreparingEvent, materializePreparationEvent, } from '../session/preparation-history.js'; +import { + classifyAttentionKilocodeEvent, + type AttentionEvent, +} from './ingest-attention-classifier.js'; // --------------------------------------------------------------------------- // Ingest Attachment @@ -257,6 +261,14 @@ export type IngestDOContext = { ) => Promise; /** Persist the slash-command catalog so connecting clients can be hydrated. */ setAvailableCommands: (commands: SlashCommandInfo[]) => Promise; + /** + * Optional callback invoked for qualifying question/permission kilocode + * events. Synchronous/fire-and-forget; the DO owns any `waitUntil` for + * external notification IO. Duplicate snapshot replays call it again + * and dedup is the notification dispatch's job (idempotency key on + * `executionId`), not this hook's. + */ + onAttentionEvent?: (event: AttentionEvent) => void; }; // --------------------------------------------------------------------------- @@ -743,6 +755,10 @@ export function createIngestHandler( } ); ws.serializeAttachment(attachment); + const attention = classifyAttentionKilocodeEvent(ingestEvent.data); + if (attention) { + doContext.onAttentionEvent?.(attention); + } } else { console.warn('Invalid kilocode event payload'); } diff --git a/services/session-ingest/src/remote-session-notifications.test.ts b/services/session-ingest/src/remote-session-notifications.test.ts index d06e6a7434..6fff2938dd 100644 --- a/services/session-ingest/src/remote-session-notifications.test.ts +++ b/services/session-ingest/src/remote-session-notifications.test.ts @@ -39,7 +39,7 @@ describe('buildRemoteSessionAttentionPushBody', () => { }); describe('dispatchRemoteSessionAttentionSignal', () => { - it('suppresses pushes while remote-session presence reporting is unavailable', async () => { + it('dispatches the push when enabled and the CLI reports an active session', async () => { const hasActiveCliSession = vi.fn(async () => true); const sendPush = vi.fn(async () => ({ dispatched: true })); const outcome = await dispatchRemoteSessionAttentionSignal( @@ -47,8 +47,29 @@ describe('dispatchRemoteSessionAttentionSignal', () => { { hasActiveCliSession, sendPush } ); + expect(outcome).toBe('sent'); + expect(hasActiveCliSession).toHaveBeenCalledTimes(1); + expect(sendPush).toHaveBeenCalledTimes(1); + expect(sendPush).toHaveBeenCalledWith({ + userId: 'usr_1', + cliSessionId: 'ses_1', + executionId: 'remote:msg-1', + status: 'completed', + body: 'Done', + suppressIfViewingSession: true, + }); + }); + + it('suppresses the push when the CLI has no active heartbeat, even with the flag enabled', async () => { + const hasActiveCliSession = vi.fn(async () => false); + const sendPush = vi.fn(async () => ({ dispatched: true })); + const outcome = await dispatchRemoteSessionAttentionSignal( + { kiloUserId: 'usr_1', sessionId: 'ses_1', signal: completedSignal('Done') }, + { hasActiveCliSession, sendPush } + ); + expect(outcome).toBe('suppressed'); - expect(hasActiveCliSession).not.toHaveBeenCalled(); + expect(hasActiveCliSession).toHaveBeenCalledTimes(1); expect(sendPush).not.toHaveBeenCalled(); }); }); diff --git a/services/session-ingest/src/remote-session-notifications.ts b/services/session-ingest/src/remote-session-notifications.ts index 8e611b3f21..d3355ebb82 100644 --- a/services/session-ingest/src/remote-session-notifications.ts +++ b/services/session-ingest/src/remote-session-notifications.ts @@ -16,8 +16,7 @@ export function isEligibleForRemoteSessionAttention(session: RemoteSessionInfo): const NEEDS_INPUT_BODY = 'Kilo needs your input.'; const DEFAULT_COMPLETED_BODY = 'Task completed'; -// Temporary until the CLI's session-presence reporting is released. -const REMOTE_SESSION_ATTENTION_PUSH_ENABLED = false; +const REMOTE_SESSION_ATTENTION_PUSH_ENABLED = true; export function buildRemoteSessionAttentionPushBody( signal: Pick