From 3ca893523f27464968c7393f36c62af62aac67a0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Fri, 17 Jul 2026 00:43:21 +0200 Subject: [PATCH 1/9] feat(session-ingest): enable remote session attention pushes --- .../src/remote-session-notifications.test.ts | 25 +++++++++++++++++-- .../src/remote-session-notifications.ts | 3 +-- 2 files changed, 24 insertions(+), 4 deletions(-) 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 From 32d0a34195acd57e7dd7b60703e2ff7a4e87dc6a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Fri, 17 Jul 2026 00:43:27 +0200 Subject: [PATCH 2/9] feat(cloud-agent): push notification when a session needs input --- .../src/persistence/CloudAgentSession.ts | 46 +++ .../ingest-attention-classifier.test.ts | 340 ++++++++++++++++++ .../websocket/ingest-attention-classifier.ts | 153 ++++++++ .../src/websocket/ingest.test.ts | 214 +++++++++++ .../cloud-agent-next/src/websocket/ingest.ts | 16 + 5 files changed, 769 insertions(+) create mode 100644 services/cloud-agent-next/src/websocket/ingest-attention-classifier.test.ts create mode 100644 services/cloud-agent-next/src/websocket/ingest-attention-classifier.ts diff --git a/services/cloud-agent-next/src/persistence/CloudAgentSession.ts b/services/cloud-agent-next/src/persistence/CloudAgentSession.ts index f8feb88043..097c6a2834 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 { + const metadata = await this.getMetadata(); + try { + 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'); } From 8f0fbf54f49f308fd92a20e5770679d9e861fb1b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Fri, 17 Jul 2026 23:07:00 +0200 Subject: [PATCH 3/9] feat(mobile): session attention derivation and ack store --- apps/mobile/src/lib/session-attention.test.ts | 323 ++++++++++++++++++ apps/mobile/src/lib/session-attention.ts | 148 ++++++++ 2 files changed, 471 insertions(+) create mode 100644 apps/mobile/src/lib/session-attention.test.ts create mode 100644 apps/mobile/src/lib/session-attention.ts diff --git a/apps/mobile/src/lib/session-attention.test.ts b/apps/mobile/src/lib/session-attention.test.ts new file mode 100644 index 0000000000..6601349097 --- /dev/null +++ b/apps/mobile/src/lib/session-attention.test.ts @@ -0,0 +1,323 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +import { + __peekSessionAttentionForTests, + __resetSessionAttentionForTests, + ackSessionAttention, + getRevisionSnapshot, + isAttentionAcked, + reconcileSessionAttention, + sessionNeedsInput, + shouldShowNeedsInput, + subscribe, +} from './session-attention'; + +beforeEach(() => { + __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('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..0ab1e17fc6 --- /dev/null +++ b/apps/mobile/src/lib/session-attention.ts @@ -0,0 +1,148 @@ +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 }; + +const listeners = new Set<() => void>(); +const entries = new Map(); +let revision = 0; + +function bumpRevision(): void { + revision += 1; + for (const listener of listeners) { + listener(); + } +} + +export function subscribe(listener: () => void): () => void { + listeners.add(listener); + return () => { + listeners.delete(listener); + }; +} + +export function getRevisionSnapshot(): number { + return 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 (entries.get(sessionId)?.raiseId === null) { + return; + } + 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 (entries.delete(sessionId)) { + bumpRevision(); + } + return; + } + + const raiseId = statusUpdatedAt ?? status ?? null; + if (entries.get(sessionId)?.raiseId === null) { + entries.set(sessionId, { raiseId }); + bumpRevision(); + } +} + +export function isAttentionAcked(sessionId: string, raiseId: string | null): boolean { + const entry = 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 { + entries.clear(); + 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 entries.get(sessionId); +} From b43be8111a60e20526bcaa93b44e865f550dfcc7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Fri, 17 Jul 2026 23:07:01 +0200 Subject: [PATCH 4/9] feat(mobile): soft pulse option for StatusDot --- apps/mobile/src/components/ui/status-dot.tsx | 68 +++++++++++++++++++- 1 file changed, 67 insertions(+), 1 deletion(-) 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 ( Date: Fri, 17 Jul 2026 23:07:02 +0200 Subject: [PATCH 5/9] feat(mobile): needs-input indicator in session list --- .../src/app/(app)/agent-chat/[session-id].tsx | 2 + .../src/components/agents/session-row.tsx | 41 +++++++++++++++++-- apps/mobile/src/components/ui/session-row.tsx | 37 +++++++++++++---- 3 files changed, 70 insertions(+), 10 deletions(-) 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-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} From 160271954e22a85f4027717ae0089a1d3d1dd4e4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Sat, 18 Jul 2026 01:49:11 +0200 Subject: [PATCH 6/9] fix(mobile): reflect needs-input clear-on-open across frozen tab Anchor the attention ack store on globalThis so the single-writer detail screen and the session-list readers share one instance across Expo Router route bundles, and remount the Agents list on attention-revision change after refocus so the freezeOnBlur'd list re-reads the ack store on return. --- .../agents/session-list-content.tsx | 25 +++++++++++ apps/mobile/src/lib/session-attention.ts | 44 ++++++++++++------- 2 files changed, 52 insertions(+), 17 deletions(-) diff --git a/apps/mobile/src/components/agents/session-list-content.tsx b/apps/mobile/src/components/agents/session-list-content.tsx index 076cd6ea61..025aaf65d9 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 { useSessionAttentionRevision } 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,27 @@ 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. `useFocusEffect` fires + // reliably on refocus (after unfreeze); bump a tick so the content re-renders + // and re-reads the current attention revision. + const attentionRevision = useSessionAttentionRevision(); + const [, bumpFocusTick] = useState(0); + useFocusEffect( + useCallback(() => { + bumpFocusTick(tick => tick + 1); + }, []) + ); + // Remount the list only when the attention revision changed since it was last + // rendered — e.g. returning from a session that was just opened (acked). A + // remount is the only reliable way to force frozen cells to re-read the ack + // store; keying on the revision (re-read above on focus) leaves scroll intact + // during ordinary focus changes where no ack/reconcile occurred. + const attentionListKey = `${sortBy}:${attentionRevision}`; + const handleRefresh = useCallback(() => { void (async () => { setRefreshing(true); @@ -304,10 +327,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/lib/session-attention.ts b/apps/mobile/src/lib/session-attention.ts index 0ab1e17fc6..b574f14751 100644 --- a/apps/mobile/src/lib/session-attention.ts +++ b/apps/mobile/src/lib/session-attention.ts @@ -21,26 +21,36 @@ export function sessionNeedsInput(status: string | null | undefined): boolean { type AckEntry = { raiseId: string | null }; -const listeners = new Set<() => void>(); -const entries = new Map(); -let revision = 0; +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 { - revision += 1; - for (const listener of listeners) { + store.revision += 1; + for (const listener of store.listeners) { listener(); } } export function subscribe(listener: () => void): () => void { - listeners.add(listener); + store.listeners.add(listener); return () => { - listeners.delete(listener); + store.listeners.delete(listener); }; } export function getRevisionSnapshot(): number { - return revision; + return store.revision; } /** @@ -55,10 +65,10 @@ 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 (entries.get(sessionId)?.raiseId === null) { + if (store.entries.get(sessionId)?.raiseId === null) { return; } - entries.set(sessionId, { raiseId: null }); + store.entries.set(sessionId, { raiseId: null }); bumpRevision(); } @@ -77,21 +87,21 @@ export function reconcileSessionAttention( statusUpdatedAt: string | null | undefined ): void { if (!sessionNeedsInput(status)) { - if (entries.delete(sessionId)) { + if (store.entries.delete(sessionId)) { bumpRevision(); } return; } const raiseId = statusUpdatedAt ?? status ?? null; - if (entries.get(sessionId)?.raiseId === null) { - entries.set(sessionId, { raiseId }); + if (store.entries.get(sessionId)?.raiseId === null) { + store.entries.set(sessionId, { raiseId }); bumpRevision(); } } export function isAttentionAcked(sessionId: string, raiseId: string | null): boolean { - const entry = entries.get(sessionId); + const entry = store.entries.get(sessionId); if (!entry) { return false; } @@ -134,8 +144,8 @@ export function useAckSessionAttentionOnOpen(sessionId: string): void { * test starts from a known state. Not for production use. */ export function __resetSessionAttentionForTests(): void { - entries.clear(); - revision = 0; + store.entries.clear(); + store.revision = 0; } /** @@ -144,5 +154,5 @@ export function __resetSessionAttentionForTests(): void { * exposing it on the production API. */ export function __peekSessionAttentionForTests(sessionId: string): AckEntry | undefined { - return entries.get(sessionId); + return store.entries.get(sessionId); } From dc86ab99474c50582e35dc627df99929d3de5958 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Sat, 18 Jul 2026 02:27:52 +0200 Subject: [PATCH 7/9] fix: address review findings for needs-input feature - cloud-agent-next: fetch session metadata inside the attention-push try/catch so a metadata failure is logged and swallowed as documented, never breaking the best-effort ingest path. - mobile: key the Agents list on an attention revision snapshotted at tab focus, so it remounts to re-read the ack store only on return from a just-opened session, not on any unrelated session's attention change during browsing (preserving scroll). - mobile: isolate attention-store subscribers so one throwing listener cannot prevent the rest from being notified. --- .../agents/session-list-content.tsx | 24 +++++++++---------- apps/mobile/src/lib/session-attention.test.ts | 18 ++++++++++++++ apps/mobile/src/lib/session-attention.ts | 8 ++++++- .../src/persistence/CloudAgentSession.ts | 2 +- 4 files changed, 37 insertions(+), 15 deletions(-) diff --git a/apps/mobile/src/components/agents/session-list-content.tsx b/apps/mobile/src/components/agents/session-list-content.tsx index 025aaf65d9..a127b3dc01 100644 --- a/apps/mobile/src/components/agents/session-list-content.tsx +++ b/apps/mobile/src/components/agents/session-list-content.tsx @@ -25,7 +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 { useSessionAttentionRevision } from '@/lib/session-attention'; +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). @@ -195,22 +195,20 @@ export function AgentSessionListContent({ // 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. `useFocusEffect` fires - // reliably on refocus (after unfreeze); bump a tick so the content re-renders - // and re-reads the current attention revision. - const attentionRevision = useSessionAttentionRevision(); - const [, bumpFocusTick] = useState(0); + // 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(() => { - bumpFocusTick(tick => tick + 1); + setAttentionFocusRevision(getRevisionSnapshot()); }, []) ); - // Remount the list only when the attention revision changed since it was last - // rendered — e.g. returning from a session that was just opened (acked). A - // remount is the only reliable way to force frozen cells to re-read the ack - // store; keying on the revision (re-read above on focus) leaves scroll intact - // during ordinary focus changes where no ack/reconcile occurred. - const attentionListKey = `${sortBy}:${attentionRevision}`; + const attentionListKey = `${sortBy}:${attentionFocusRevision}`; const handleRefresh = useCallback(() => { void (async () => { diff --git a/apps/mobile/src/lib/session-attention.test.ts b/apps/mobile/src/lib/session-attention.test.ts index 6601349097..6940dc95c7 100644 --- a/apps/mobile/src/lib/session-attention.test.ts +++ b/apps/mobile/src/lib/session-attention.test.ts @@ -306,6 +306,24 @@ describe('revision snapshot and listener notification', () => { 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); diff --git a/apps/mobile/src/lib/session-attention.ts b/apps/mobile/src/lib/session-attention.ts index b574f14751..c686aa1555 100644 --- a/apps/mobile/src/lib/session-attention.ts +++ b/apps/mobile/src/lib/session-attention.ts @@ -37,8 +37,14 @@ const store: AttentionStore = (globalScope[STORE_KEY] ??= { 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) { - listener(); + try { + listener(); + } catch { + // A subscriber's own error must not break store notification. + } } } diff --git a/services/cloud-agent-next/src/persistence/CloudAgentSession.ts b/services/cloud-agent-next/src/persistence/CloudAgentSession.ts index 097c6a2834..da5b3f8477 100644 --- a/services/cloud-agent-next/src/persistence/CloudAgentSession.ts +++ b/services/cloud-agent-next/src/persistence/CloudAgentSession.ts @@ -680,8 +680,8 @@ export class CloudAgentSession extends DurableObject { * dependencies, and log on dispatch failure. */ private async handleAttentionEvent(event: AttentionEvent): Promise { - const metadata = await this.getMetadata(); try { + const metadata = await this.getMetadata(); await dispatchCloudAgentAttentionPush(event, metadata, { hasConnectedStreamClients: () => getConnectedStreamClientCount(this.ctx) > 0, sendPush: params => this.env.NOTIFICATIONS.sendCloudAgentSessionNotification(params), From feeed497fef5bbe9324b99e793e2030f2d193e6c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Sat, 18 Jul 2026 02:40:04 +0200 Subject: [PATCH 8/9] test(mobile): use block body in listener-isolation test oxlint no-confusing-void-expression forbids returning the void expression from the arrow shorthand passed to expect(). --- apps/mobile/src/lib/session-attention.test.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/apps/mobile/src/lib/session-attention.test.ts b/apps/mobile/src/lib/session-attention.test.ts index 6940dc95c7..1147da8acb 100644 --- a/apps/mobile/src/lib/session-attention.test.ts +++ b/apps/mobile/src/lib/session-attention.test.ts @@ -315,7 +315,9 @@ describe('revision snapshot and listener notification', () => { const unsubGood = subscribe(good); const before = getRevisionSnapshot(); - expect(() => ackSessionAttention('s1')).not.toThrow(); + expect(() => { + ackSessionAttention('s1'); + }).not.toThrow(); expect(throwing).toHaveBeenCalledTimes(1); expect(good).toHaveBeenCalledTimes(1); expect(getRevisionSnapshot()).toBe(before + 1); From a4cb7e8b3453464bf7068dd097f77f5c154cfc98 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Sun, 19 Jul 2026 01:38:15 +0200 Subject: [PATCH 9/9] feat(notifications): lead session pushes with the session title Thread the heartbeat session title from UserConnectionDO through SessionIngestDO into the notifications service so attention and session-ready push notifications lead with the session title. - packages/notifications: optional title on sendSessionReadyNotificationInputSchema. - services/notifications: add a code-point-safe title sanitizer (collapse whitespace, <=80 code points incl. ellipsis, no split surrogate) used by dispatchCloudAgentSessionPush and the ready push. Ready push prefers the heartbeat title hint, then the DB title, then the fixed 'Kilo session ready'. - services/session-ingest: forward the heartbeat title from UserConnectionDO.claimSessionReadyPush into SessionIngestDO and on to sendSessionReadyNotification. - Tests cover title composition, long-title and surrogate-pair truncation, whitespace handling, hint-vs-DB precedence, and schema strip-on-parse. --- packages/notifications/src/rpc-schemas.ts | 1 + .../src/lib/cloud-agent-session-push.ts | 18 +- .../notifications-service-cloud-agent.test.ts | 226 ++++++++++++++++++ .../src/dos/SessionIngestDO.test.ts | 17 +- .../session-ingest/src/dos/SessionIngestDO.ts | 3 +- .../src/dos/UserConnectionDO.test.ts | 4 +- .../src/dos/UserConnectionDO.ts | 6 +- 7 files changed, 265 insertions(+), 10 deletions(-) diff --git a/packages/notifications/src/rpc-schemas.ts b/packages/notifications/src/rpc-schemas.ts index 5b5cdfc97a..78ec16f327 100644 --- a/packages/notifications/src/rpc-schemas.ts +++ b/packages/notifications/src/rpc-schemas.ts @@ -162,6 +162,7 @@ export type SendCloudAgentSessionNotificationResult = z.infer< export const sendSessionReadyNotificationInputSchema = z.object({ userId: z.string().min(1), cliSessionId: z.string().min(1), + title: z.string().optional(), }); export type SendSessionReadyNotificationParams = z.infer< typeof sendSessionReadyNotificationInputSchema diff --git a/services/notifications/src/lib/cloud-agent-session-push.ts b/services/notifications/src/lib/cloud-agent-session-push.ts index 20735c3d54..cb38c0aa53 100644 --- a/services/notifications/src/lib/cloud-agent-session-push.ts +++ b/services/notifications/src/lib/cloud-agent-session-push.ts @@ -15,6 +15,17 @@ type CloudAgentNotificationSession = { organizationId: string | null; }; +const TITLE_MAX_LENGTH = 80; + +function sanitizeTitle(title: string | null | undefined): string | null { + if (title == null) return null; + const collapsed = title.replace(/\s+/g, ' ').trim(); + if (collapsed === '') return null; + const codePoints = Array.from(collapsed); + if (codePoints.length <= TITLE_MAX_LENGTH) return collapsed; + return `${codePoints.slice(0, TITLE_MAX_LENGTH - 3).join('')}...`; +} + export type DispatchCloudAgentSessionPushDeps = { getSession: ( userId: string, @@ -85,7 +96,7 @@ export async function dispatchCloudAgentSessionPush( ? presenceContextForCliSession(parsed.cliSessionId) : null, idempotencyKey: `cloud-agent:${parsed.cliSessionId}:${parsed.executionId}`, - title: session.title ?? 'Agent session', + title: sanitizeTitle(session.title) ?? 'Agent session', body: parsed.body, }), deps @@ -105,10 +116,11 @@ export async function dispatchSessionReadyPush( return dispatchSessionPush( parsed.userId, parsed.cliSessionId, - () => ({ + session => ({ presenceContext: presenceContextForPlatform('app'), idempotencyKey: `cloud-agent:${parsed.cliSessionId}:session-ready`, - title: 'Kilo session ready', + title: + sanitizeTitle(parsed.title ?? null) ?? sanitizeTitle(session.title) ?? 'Kilo session ready', body: 'Your Kilo session is ready to control from your phone', }), deps diff --git a/services/notifications/src/lib/notifications-service-cloud-agent.test.ts b/services/notifications/src/lib/notifications-service-cloud-agent.test.ts index 482bb620ea..8ad427d561 100644 --- a/services/notifications/src/lib/notifications-service-cloud-agent.test.ts +++ b/services/notifications/src/lib/notifications-service-cloud-agent.test.ts @@ -2,6 +2,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'; import { sendCloudAgentSessionNotificationInputSchema, + sendSessionReadyNotificationInputSchema, type DispatchPushInput, type DispatchPushOutcome, } from '@kilocode/notifications'; @@ -252,6 +253,148 @@ describe('dispatchCloudAgentSessionPush', () => { ).rejects.toThrow(); expect(deps.getSession).not.toHaveBeenCalled(); }); + + it('composes the push title from the session title for an attention dispatch', async () => { + const deps = createDeps({ + session: { title: 'Refactor auth module', organizationId: null }, + }); + + const result = await dispatchCloudAgentSessionPush( + { + userId: 'user-1', + cliSessionId: 'ses_1', + executionId: 'attention:req_1', + status: 'completed', + body: 'Kilo needs your input.', + }, + deps + ); + + expect(result).toEqual({ dispatched: true }); + expect(mockDispatchPush).toHaveBeenCalledWith( + expect.objectContaining({ + push: expect.objectContaining({ + title: 'Refactor auth module', + body: 'Kilo needs your input.', + }), + }) + ); + }); + + it('falls back to the fixed title when the session title is null', async () => { + const deps = createDeps({ session: { title: null, organizationId: null } }); + + await dispatchCloudAgentSessionPush( + { + userId: 'user-1', + cliSessionId: 'ses_1', + executionId: 'exec_null', + status: 'completed', + body: 'Finished', + }, + deps + ); + + expect(mockDispatchPush).toHaveBeenCalledWith( + expect.objectContaining({ push: expect.objectContaining({ title: 'Agent session' }) }) + ); + }); + + it('falls back to the fixed title when the session title is only whitespace', async () => { + const deps = createDeps({ session: { title: ' \n \t', organizationId: null } }); + + await dispatchCloudAgentSessionPush( + { + userId: 'user-1', + cliSessionId: 'ses_1', + executionId: 'exec_ws', + status: 'completed', + body: 'Finished', + }, + deps + ); + + expect(mockDispatchPush).toHaveBeenCalledWith( + expect.objectContaining({ push: expect.objectContaining({ title: 'Agent session' }) }) + ); + }); + + it('truncates a long session title to at most 80 code points ending in ellipsis', async () => { + // Mix ASCII with a surrogate-pair emoji to exercise the codepoint-safe path. + const longTitle = 'x'.repeat(70) + ' 🦊 '.repeat(5) + ' tail'; + const deps = createDeps({ session: { title: longTitle, organizationId: null } }); + + await dispatchCloudAgentSessionPush( + { + userId: 'user-1', + cliSessionId: 'ses_1', + executionId: 'exec_long', + status: 'completed', + body: 'Finished', + }, + deps + ); + + const calls = mockDispatchPush.mock.calls[0]?.[0] as DispatchPushInput; + const pushedTitle = calls.push.title; + expect(pushedTitle.endsWith('...')).toBe(true); + expect(pushedTitle.includes('\n')).toBe(false); + expect(Array.from(pushedTitle).length).toBeLessThanOrEqual(80); + // No lone surrogate: if the truncation keeps the emoji, it must remain intact; + // otherwise the truncation point must be on an ASCII/space boundary. + if (pushedTitle.includes('🦊')) { + // Surviving emoji must be the full code point sequence, not a lone surrogate. + expect(pushedTitle.includes('\uD83E')).toBe(true); + } + }); + + it('truncates surrogate-pair emoji by code points, not UTF-16 code units', async () => { + const emojiTitle = '🦊'.repeat(85); + const expected = '🦊'.repeat(77) + '...'; + const deps = createDeps({ session: { title: emojiTitle, organizationId: null } }); + + await dispatchCloudAgentSessionPush( + { + userId: 'user-1', + cliSessionId: 'ses_1', + executionId: 'exec_emoji_surrogate', + status: 'completed', + body: 'Finished', + }, + deps + ); + + const calls = mockDispatchPush.mock.calls[0]?.[0] as DispatchPushInput; + const pushedTitle = calls.push.title; + expect(pushedTitle).toBe(expected); + expect(Array.from(pushedTitle).length).toBe(80); + expect( + /[\uD800-\uDBFF](?![\uDC00-\uDFFF])|(? { + const deps = createDeps({ + session: { title: 'Refactor\n\nthe\t auth module', organizationId: null }, + }); + + await dispatchCloudAgentSessionPush( + { + userId: 'user-1', + cliSessionId: 'ses_1', + executionId: 'exec_ws', + status: 'completed', + body: 'Finished', + }, + deps + ); + + expect(mockDispatchPush).toHaveBeenCalledWith( + expect.objectContaining({ + push: expect.objectContaining({ title: 'Refactor the auth module' }), + }) + ); + }); }); describe('dispatchSessionReadyPush', () => { @@ -322,6 +465,59 @@ describe('dispatchSessionReadyPush', () => { expect(result).toEqual({ dispatched: false, reason: 'dispatch_failed' }); }); + + it('prefers the heartbeat hint title over the DB title when both are present', async () => { + const deps = createDeps({ + session: { title: 'DB title', organizationId: null }, + }); + + const result = await dispatchSessionReadyPush( + { userId: 'user-1', cliSessionId: 'ses_1', title: 'Heartbeat title' }, + deps + ); + + expect(result).toEqual({ dispatched: true }); + expect(mockDispatchPush).toHaveBeenCalledWith( + expect.objectContaining({ + push: expect.objectContaining({ + title: 'Heartbeat title', + body: 'Your Kilo session is ready to control from your phone', + }), + }) + ); + }); + + it('uses the DB title when no heartbeat hint is provided', async () => { + const deps = createDeps({ + session: { title: 'DB title', organizationId: null }, + }); + + await dispatchSessionReadyPush({ userId: 'user-1', cliSessionId: 'ses_1' }, deps); + + expect(mockDispatchPush).toHaveBeenCalledWith( + expect.objectContaining({ + push: expect.objectContaining({ + title: 'DB title', + body: 'Your Kilo session is ready to control from your phone', + }), + }) + ); + }); + + it('falls back to the fixed copy when no hint and no DB title are present', async () => { + const deps = createDeps({ session: { title: null, organizationId: null } }); + + await dispatchSessionReadyPush({ userId: 'user-1', cliSessionId: 'ses_1' }, deps); + + expect(mockDispatchPush).toHaveBeenCalledWith( + expect.objectContaining({ + push: expect.objectContaining({ + title: 'Kilo session ready', + body: 'Your Kilo session is ready to control from your phone', + }), + }) + ); + }); }); describe('sendCloudAgentSessionNotificationInputSchema', () => { @@ -346,3 +542,33 @@ describe('sendCloudAgentSessionNotificationInputSchema', () => { }); }); }); + +describe('sendSessionReadyNotificationInputSchema', () => { + it('accepts input without title and strips unknown keys', () => { + const parsed = sendSessionReadyNotificationInputSchema.parse({ + userId: 'user-1', + cliSessionId: 'ses_1', + extra: 'stripped', + }); + + expect(parsed).toEqual({ + userId: 'user-1', + cliSessionId: 'ses_1', + }); + }); + + it('accepts input with a title', () => { + const parsed = sendSessionReadyNotificationInputSchema.parse({ + userId: 'user-1', + cliSessionId: 'ses_1', + title: 'Heartbeat title', + extra: 'stripped', + }); + + expect(parsed).toEqual({ + userId: 'user-1', + cliSessionId: 'ses_1', + title: 'Heartbeat title', + }); + }); +}); diff --git a/services/session-ingest/src/dos/SessionIngestDO.test.ts b/services/session-ingest/src/dos/SessionIngestDO.test.ts index 75588abda7..bf9e7dd918 100644 --- a/services/session-ingest/src/dos/SessionIngestDO.test.ts +++ b/services/session-ingest/src/dos/SessionIngestDO.test.ts @@ -291,20 +291,35 @@ describe('SessionIngestDO session-ready push', () => { it('pushes on first claim and never again', async () => { const { durableObject, sendSessionReadyNotification, settle } = makeHarness(); - durableObject.claimSessionReadyPush('usr_push', 'ses_push'); + durableObject.claimSessionReadyPush('usr_push', 'ses_push', 'My title'); await settle(); expect(sendSessionReadyNotification).toHaveBeenCalledTimes(1); expect(sendSessionReadyNotification).toHaveBeenCalledWith({ userId: 'usr_push', cliSessionId: 'ses_push', + title: 'My title', }); // Re-claims (CLI reconnect, UserConnectionDO eviction) must not re-push. + durableObject.claimSessionReadyPush('usr_push', 'ses_push', 'My title'); + await settle(); + + expect(sendSessionReadyNotification).toHaveBeenCalledTimes(1); + }); + + it('forwards an undefined title when none is supplied', async () => { + const { durableObject, sendSessionReadyNotification, settle } = makeHarness(); + durableObject.claimSessionReadyPush('usr_push', 'ses_push'); await settle(); expect(sendSessionReadyNotification).toHaveBeenCalledTimes(1); + expect(sendSessionReadyNotification).toHaveBeenCalledWith({ + userId: 'usr_push', + cliSessionId: 'ses_push', + title: undefined, + }); }); it('never pushes for a deleted session', async () => { diff --git a/services/session-ingest/src/dos/SessionIngestDO.ts b/services/session-ingest/src/dos/SessionIngestDO.ts index 163a8bd8ff..25ca4143fc 100644 --- a/services/session-ingest/src/dos/SessionIngestDO.ts +++ b/services/session-ingest/src/dos/SessionIngestDO.ts @@ -362,7 +362,7 @@ export class SessionIngestDO extends DurableObject { * CLI reconnects and UserConnectionDO evictions can't re-arm it. Push * failures are non-fatal: log and move on. */ - claimSessionReadyPush(kiloUserId: string, sessionId: string): void { + claimSessionReadyPush(kiloUserId: string, sessionId: string, title?: string): void { const deletedRow = this.db .select({ value: ingestMeta.value }) .from(ingestMeta) @@ -380,6 +380,7 @@ export class SessionIngestDO extends DurableObject { this.env.NOTIFICATIONS.sendSessionReadyNotification({ userId: kiloUserId, cliSessionId: sessionId, + title, }).catch((error: unknown) => { console.error('Failed to send session-ready push (non-fatal)', { sessionId, diff --git a/services/session-ingest/src/dos/UserConnectionDO.test.ts b/services/session-ingest/src/dos/UserConnectionDO.test.ts index e660d6555d..c0fc83950c 100644 --- a/services/session-ingest/src/dos/UserConnectionDO.test.ts +++ b/services/session-ingest/src/dos/UserConnectionDO.test.ts @@ -3094,7 +3094,7 @@ describe('UserConnectionDO', () => { sendHeartbeat(doInstance, cliWs, [makeSession('ses_main')]); expect(claimSessionReadyPush).toHaveBeenCalledTimes(1); - expect(claimSessionReadyPush).toHaveBeenCalledWith('usr_1', 'ses_main'); + expect(claimSessionReadyPush).toHaveBeenCalledWith('usr_1', 'ses_main', 'Test'); // Subsequent heartbeats for the same session must not re-claim. sendHeartbeat(doInstance, cliWs, [makeSession('ses_main')]); @@ -3111,7 +3111,7 @@ describe('UserConnectionDO', () => { ]); expect(claimSessionReadyPush).toHaveBeenCalledTimes(1); - expect(claimSessionReadyPush).toHaveBeenCalledWith('usr_1', 'ses_main'); + expect(claimSessionReadyPush).toHaveBeenCalledWith('usr_1', 'ses_main', 'Test'); }); it('does not claim on sockets without a kiloUserId (legacy attachment)', () => { diff --git a/services/session-ingest/src/dos/UserConnectionDO.ts b/services/session-ingest/src/dos/UserConnectionDO.ts index b132a3ff0e..984d1fd2ee 100644 --- a/services/session-ingest/src/dos/UserConnectionDO.ts +++ b/services/session-ingest/src/dos/UserConnectionDO.ts @@ -420,7 +420,7 @@ export class UserConnectionDO extends DurableObject { // remote-controllable — the only moment the session-ready push fires. // The durable claim in SessionIngestDO makes reconnect re-sights no-ops. if (!previousOwner && !session.parentSessionId && attachment.kiloUserId) { - this.claimSessionReadyPush(attachment.kiloUserId, session.id); + this.claimSessionReadyPush(attachment.kiloUserId, session.id, session.title); } this.sessionOwners.set(session.id, connectionId); } @@ -474,10 +474,10 @@ export class UserConnectionDO extends DurableObject { * Fire-and-forget "session ready to control from your phone" push via the * session's SessionIngestDO, which holds the durable once-ever claim. */ - private claimSessionReadyPush(kiloUserId: string, sessionId: string): void { + private claimSessionReadyPush(kiloUserId: string, sessionId: string, title: string): void { const stub = getSessionIngestDO(this.env, { kiloUserId, sessionId }); this.ctx.waitUntil( - stub.claimSessionReadyPush(kiloUserId, sessionId).catch((error: unknown) => { + stub.claimSessionReadyPush(kiloUserId, sessionId, title).catch((error: unknown) => { console.error('Failed to claim session-ready push (non-fatal)', { sessionId, error: error instanceof Error ? error.message : String(error),