From 2362a7a3924f341ec958331438068fd21782ce3c Mon Sep 17 00:00:00 2001 From: Evgeny Shurakov Date: Tue, 7 Jul 2026 22:24:57 +0200 Subject: [PATCH] feat(session-ingest): push notifications for remote session attention signals Detect completed assistant turns and needs-input status transitions during ingest and dispatch a mobile push to the session owner for remote (vscode, agent-manager) root sessions. Pushes are suppressed when the user already has the session open in the web app (UserConnectionDO.hasSessionSubscribers). - SessionIngestDO emits AttentionSignal(s) on idle/question/permission transitions, pairing completed turns with the just-finished assistant message excerpt. First-ever status writes are skipped to avoid pushing about an old turn on full-history backfill. - queue-consumer collects signals across chunks and dispatches them via a NOTIFICATIONS service binding; partial flush on failure preserves signals from already-committed chunks. - Adds @kilocode/notifications dependency and wrangler NOTIFICATIONS binding. --- .../cloud-agent-next/CloudChatPage.tsx | 3 + apps/web/src/hooks/useCliSessionPresence.ts | 14 ++ packages/event-service/package.json | 1 + .../src/__tests__/presence.test.ts | 9 + packages/event-service/src/presence.ts | 3 + packages/notifications/src/rpc-schemas.ts | 1 + pnpm-lock.yaml | 3 + services/notifications/src/index.ts | 11 +- .../src/lib/cloud-agent-session-push.ts | 24 +- .../notifications-service-cloud-agent.test.ts | 50 +++- services/session-ingest/package.json | 1 + .../session-ingest/src/dos/SessionIngestDO.ts | 105 +++++++- .../src/dos/UserConnectionDO.test.ts | 66 ++++- .../src/dos/UserConnectionDO.ts | 33 +++ .../src/dos/session-ingest-attention.test.ts | 136 +++++++++++ .../src/dos/session-ingest-attention.ts | 79 ++++++ services/session-ingest/src/env.ts | 6 +- .../src/notifications-binding.ts | 18 ++ .../session-ingest/src/queue-consumer.test.ts | 175 +++++++++++++ services/session-ingest/src/queue-consumer.ts | 133 +++++++++- .../src/remote-session-notifications.test.ts | 124 ++++++++++ .../src/remote-session-notifications.ts | 80 ++++++ .../types/user-connection-protocol.test.ts | 23 ++ .../src/types/user-connection-protocol.ts | 5 + .../integration/session-ingest-do.test.ts | 230 ++++++++++++++++++ .../session-ingest/worker-configuration.d.ts | 1 + services/session-ingest/wrangler.jsonc | 5 + 27 files changed, 1320 insertions(+), 19 deletions(-) create mode 100644 apps/web/src/hooks/useCliSessionPresence.ts create mode 100644 packages/event-service/src/__tests__/presence.test.ts create mode 100644 services/session-ingest/src/dos/session-ingest-attention.test.ts create mode 100644 services/session-ingest/src/dos/session-ingest-attention.ts create mode 100644 services/session-ingest/src/notifications-binding.ts create mode 100644 services/session-ingest/src/remote-session-notifications.test.ts create mode 100644 services/session-ingest/src/remote-session-notifications.ts diff --git a/apps/web/src/components/cloud-agent-next/CloudChatPage.tsx b/apps/web/src/components/cloud-agent-next/CloudChatPage.tsx index 4c82dd6861..49b85efd9c 100644 --- a/apps/web/src/components/cloud-agent-next/CloudChatPage.tsx +++ b/apps/web/src/components/cloud-agent-next/CloudChatPage.tsx @@ -47,6 +47,7 @@ import { ContextUsageIndicator } from './ContextUsageIndicator'; import { resolveContextWindow } from './model-context-lengths'; import { useSlashCommandSets } from '@/hooks/useSlashCommandSets'; import { useCelebrationSound } from '@/hooks/useCelebrationSound'; +import { useCliSessionPresence } from '@/hooks/useCliSessionPresence'; import type { CloudAgentAttachments } from '@/lib/cloud-agent/constants'; import { SetPageTitle } from '@/components/SetPageTitle'; @@ -267,6 +268,8 @@ export default function CloudChatPage({ organizationId }: CloudChatPageProps) { const observedModel = useAtomValue(manager.atoms.observedModel); const remoteModelOverride = useAtomValue(manager.atoms.remoteModelOverride); + useCliSessionPresence(fetchedSessionData?.kiloSessionId ?? null); + const setSessionConfig = useSetAtom(manager.atoms.sessionConfig); const [attachmentMessageUuid] = useState(() => uuidv4()); diff --git a/apps/web/src/hooks/useCliSessionPresence.ts b/apps/web/src/hooks/useCliSessionPresence.ts new file mode 100644 index 0000000000..a83902dd17 --- /dev/null +++ b/apps/web/src/hooks/useCliSessionPresence.ts @@ -0,0 +1,14 @@ +'use client'; + +import { presenceContextForCliSession } from '@kilocode/event-service'; +import { usePresenceSubscription } from '@kilocode/kilo-chat-hooks'; + +import { useDocumentVisible } from './useDocumentVisible'; + +export function useCliSessionPresence(sessionId: string | null, enabled = true) { + const visible = useDocumentVisible(); + usePresenceSubscription( + sessionId ? presenceContextForCliSession(sessionId) : null, + Boolean(sessionId) && enabled && visible + ); +} diff --git a/packages/event-service/package.json b/packages/event-service/package.json index d14dc34908..cd0cbe3716 100644 --- a/packages/event-service/package.json +++ b/packages/event-service/package.json @@ -7,6 +7,7 @@ "types": "./src/index.ts", "exports": { ".": "./src/index.ts", + "./presence": "./src/presence.ts", "./types": "./src/types.ts" }, "scripts": { diff --git a/packages/event-service/src/__tests__/presence.test.ts b/packages/event-service/src/__tests__/presence.test.ts new file mode 100644 index 0000000000..fccbd869c4 --- /dev/null +++ b/packages/event-service/src/__tests__/presence.test.ts @@ -0,0 +1,9 @@ +import { describe, expect, it } from 'vitest'; + +import { presenceContextForCliSession } from '../presence'; + +describe('presence context builders', () => { + it('builds CLI session presence contexts', () => { + expect(presenceContextForCliSession('ses_1')).toBe('/presence/cli-session/ses_1'); + }); +}); diff --git a/packages/event-service/src/presence.ts b/packages/event-service/src/presence.ts index a267aa667c..efe042df9d 100644 --- a/packages/event-service/src/presence.ts +++ b/packages/event-service/src/presence.ts @@ -19,3 +19,6 @@ export const presenceContextForInstance = (sandboxId: string) => export const presenceContextForConversation = (sandboxId: string, conversationId: string) => `/presence${kiloclawConversationContext(sandboxId, conversationId)}` as const; + +export const presenceContextForCliSession = (sessionId: string) => + `/presence/cli-session/${sessionId}` as const; diff --git a/packages/notifications/src/rpc-schemas.ts b/packages/notifications/src/rpc-schemas.ts index ef99f9e2e4..ec804a7c84 100644 --- a/packages/notifications/src/rpc-schemas.ts +++ b/packages/notifications/src/rpc-schemas.ts @@ -143,6 +143,7 @@ export const sendCloudAgentSessionNotificationInputSchema = z.object({ executionId: z.string().min(1), status: cloudAgentSessionPushStatusSchema, body: z.string(), + suppressIfViewingSession: z.boolean().optional(), }); export type SendCloudAgentSessionNotificationParams = z.infer< typeof sendCloudAgentSessionNotificationInputSchema diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 6c5969d7a2..4f2259d06f 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -2784,6 +2784,9 @@ importers: '@kilocode/db': specifier: workspace:* version: link:../../packages/db + '@kilocode/notifications': + specifier: workspace:* + version: link:../../packages/notifications '@kilocode/session-ingest-contracts': specifier: workspace:* version: link:../../packages/session-ingest-contracts diff --git a/services/notifications/src/index.ts b/services/notifications/src/index.ts index a99b9bec44..63f38436e7 100644 --- a/services/notifications/src/index.ts +++ b/services/notifications/src/index.ts @@ -225,7 +225,7 @@ export class NotificationsService extends WorkerEntrypoint { let db: ReturnType | undefined; const getDbForCall = () => (db ??= getWorkerDb(this.env.HYPERDRIVE.connectionString)); - return dispatchCloudAgentSessionPush(params, { + const result = await dispatchCloudAgentSessionPush(params, { getSession: async (userId, cliSessionId) => { const [session] = await getDbForCall() .select({ @@ -262,6 +262,15 @@ export class NotificationsService extends WorkerEntrypoint { return stub.dispatchPush(input); }, }); + + console.log('sendCloudAgentSessionNotification', { + userId: params.userId, + cliSessionId: params.cliSessionId, + executionId: params.executionId, + dispatched: result.dispatched, + reason: result.reason, + }); + return result; } async sendScheduledActionNotice( diff --git a/services/notifications/src/lib/cloud-agent-session-push.ts b/services/notifications/src/lib/cloud-agent-session-push.ts index d6c9b06e16..562794556a 100644 --- a/services/notifications/src/lib/cloud-agent-session-push.ts +++ b/services/notifications/src/lib/cloud-agent-session-push.ts @@ -1,3 +1,4 @@ +import { presenceContextForCliSession } from '@kilocode/event-service/presence'; import { sendCloudAgentSessionNotificationInputSchema, type DispatchPushInput, @@ -28,6 +29,10 @@ export async function dispatchCloudAgentSessionPush( const session = await deps.getSession(parsed.userId, parsed.cliSessionId); if (!session) { + console.log('Cloud agent session push skipped: session not found', { + userId: parsed.userId, + cliSessionId: parsed.cliSessionId, + }); return { dispatched: false, reason: 'missing_session' }; } @@ -35,12 +40,19 @@ export async function dispatchCloudAgentSessionPush( session.organizationId && !(await deps.hasOrganizationAccess(parsed.userId, session.organizationId)) ) { + console.log('Cloud agent session push skipped: missing org access', { + userId: parsed.userId, + cliSessionId: parsed.cliSessionId, + organizationId: session.organizationId, + }); return { dispatched: false, reason: 'missing_session' }; } const outcome = await deps.dispatchPush({ userId: parsed.userId, - presenceContext: null, + presenceContext: parsed.suppressIfViewingSession + ? presenceContextForCliSession(parsed.cliSessionId) + : null, idempotencyKey: `cloud-agent:${parsed.cliSessionId}:${parsed.executionId}`, badge: null, push: { @@ -53,8 +65,18 @@ export async function dispatchCloudAgentSessionPush( } satisfies DispatchPushInput); if (outcome.kind === 'failed') { + console.warn('Cloud agent session push dispatch failed', { + userId: parsed.userId, + cliSessionId: parsed.cliSessionId, + error: outcome.error, + }); return { dispatched: false, reason: 'dispatch_failed' }; } + console.log('Cloud agent session push dispatched', { + userId: parsed.userId, + cliSessionId: parsed.cliSessionId, + outcomeKind: outcome.kind, + }); return { dispatched: true }; } 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 894873c831..75aacbf093 100644 --- a/services/notifications/src/lib/notifications-service-cloud-agent.test.ts +++ b/services/notifications/src/lib/notifications-service-cloud-agent.test.ts @@ -1,6 +1,10 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'; -import { type DispatchPushInput, type DispatchPushOutcome } from '@kilocode/notifications'; +import { + sendCloudAgentSessionNotificationInputSchema, + type DispatchPushInput, + type DispatchPushOutcome, +} from '@kilocode/notifications'; import { dispatchCloudAgentSessionPush, @@ -74,6 +78,27 @@ describe('dispatchCloudAgentSessionPush', () => { expect(deps.hasOrganizationAccess).not.toHaveBeenCalled(); }); + it('passes the CLI session presence context when requested', async () => { + const deps = createDeps(); + + const result = await dispatchCloudAgentSessionPush( + { + userId: 'user-1', + cliSessionId: 'ses_1', + executionId: 'exec_1', + status: 'completed', + body: 'Finished', + suppressIfViewingSession: true, + }, + deps + ); + + expect(result).toEqual({ dispatched: true }); + expect(mockDispatchPush).toHaveBeenCalledWith( + expect.objectContaining({ presenceContext: '/presence/cli-session/ses_1' }) + ); + }); + it('keeps follow-up executions in one session idempotent independently', async () => { const deps = createDeps(); @@ -206,3 +231,26 @@ describe('dispatchCloudAgentSessionPush', () => { expect(deps.getSession).not.toHaveBeenCalled(); }); }); + +describe('sendCloudAgentSessionNotificationInputSchema', () => { + it('accepts suppressIfViewingSession and strips unrelated fields', () => { + const parsed = sendCloudAgentSessionNotificationInputSchema.parse({ + userId: 'user-1', + cliSessionId: 'ses_1', + executionId: 'exec_1', + status: 'completed', + body: 'Finished', + suppressIfViewingSession: true, + extra: 'stripped', + }); + + expect(parsed).toEqual({ + userId: 'user-1', + cliSessionId: 'ses_1', + executionId: 'exec_1', + status: 'completed', + body: 'Finished', + suppressIfViewingSession: true, + }); + }); +}); diff --git a/services/session-ingest/package.json b/services/session-ingest/package.json index 85a27ed894..e106cda8a3 100644 --- a/services/session-ingest/package.json +++ b/services/session-ingest/package.json @@ -14,6 +14,7 @@ }, "dependencies": { "@kilocode/db": "workspace:*", + "@kilocode/notifications": "workspace:*", "@kilocode/session-ingest-contracts": "workspace:*", "@kilocode/worker-utils": "workspace:*", "@streamparser/json": "0.0.22", diff --git a/services/session-ingest/src/dos/SessionIngestDO.ts b/services/session-ingest/src/dos/SessionIngestDO.ts index 841cbf34e2..4a595c02e9 100644 --- a/services/session-ingest/src/dos/SessionIngestDO.ts +++ b/services/session-ingest/src/dos/SessionIngestDO.ts @@ -1,5 +1,5 @@ import { DurableObject } from 'cloudflare:workers'; -import { eq, ne, gt, gte, lt, and, or, inArray, isNull, isNotNull } from 'drizzle-orm'; +import { desc, eq, ne, gt, gte, lt, and, or, inArray, isNull, isNotNull } from 'drizzle-orm'; import { drizzle, type DrizzleSqliteDODatabase } from 'drizzle-orm/durable-sqlite'; import { migrate } from 'drizzle-orm/durable-sqlite/migrator'; @@ -17,6 +17,13 @@ import { extractNormalizedTitleFromItem, extractStatusFromItem, } from './session-ingest-extractors'; +import { + buildAssistantExcerpt, + completedAssistantMessageIdFromItemData, + isCompletedStatus, + isNeedsInputStatus, + type AttentionSignal, +} from './session-ingest-attention'; import { computeSessionMetrics, INACTIVITY_TIMEOUT_MS, @@ -71,6 +78,13 @@ function writeIngestMetaIfChanged( return { changed: true, value: params.incomingValue }; } +function hasIngestMeta(db: DrizzleSqliteDODatabase, key: IngestMetaKey): boolean { + return ( + db.select({ value: ingestMeta.value }).from(ingestMeta).where(eq(ingestMeta.key, key)).get() !== + undefined + ); +} + const INGEST_META_EXTRACTORS: Array<{ key: ExtractableMetaKey; extract: (item: IngestBatch[number]) => string | null | undefined; @@ -86,6 +100,9 @@ const INGEST_META_EXTRACTORS: Array<{ type Changes = Array<{ name: ExtractableMetaKey; value: string | null }>; +/** How many of the newest message rows to inspect when pairing an idle transition with the assistant turn that just finished. */ +const COMPLETED_MESSAGE_SCAN_LIMIT = 50; + type IngestLifecycleEvent = | { type: 'session_open' } | { @@ -137,6 +154,7 @@ export class SessionIngestDO extends DurableObject { r2References?: Record ): Promise<{ changes: Changes; + attentionSignals: AttentionSignal[]; }> { const deletedRow = this.db .select({ value: ingestMeta.value }) @@ -151,7 +169,7 @@ export class SessionIngestDO extends DurableObject { await this.env.SESSION_INGEST_R2.delete(keys); } } - return { changes: [] }; + return { changes: [], attentionSignals: [] }; } writeIngestMetaIfChanged(this.db, { key: 'kiloUserId', incomingValue: kiloUserId }); @@ -270,6 +288,11 @@ export class SessionIngestDO extends DurableObject { await this.ctx.storage.setAlarm(Date.now() + INACTIVITY_TIMEOUT_MS); } + // Read before the write loop below persists incoming values: whether this session had ever + // reported a status. The first-ever status write also registers as a change, and a + // full-history backfill of an already-idle session must not push about an old turn. + const hadPriorStatus = hasIngestMeta(this.db, 'status'); + const changes: Changes = []; for (const key of Object.keys(incomingByKey) as ExtractableMetaKey[]) { const incoming = incomingByKey[key]; @@ -299,11 +322,89 @@ export class SessionIngestDO extends DurableObject { ); } + const attentionSignals: AttentionSignal[] = []; + + const statusChange = changes.find(change => change.name === 'status'); + if (statusChange && isCompletedStatus(statusChange.value) && hadPriorStatus) { + // An idle transition means the assistant finished its turn. Pair it with the most recent + // completed assistant message so the signal carries that turn's excerpt; if none exists yet + // (e.g. a fresh session reporting idle before any turn), emit nothing rather than a spurious + // "Task completed" push. + const completedMessageId = this.findLastCompletedAssistantMessageId(); + if (completedMessageId) { + attentionSignals.push({ + signalId: completedMessageId, + kind: 'completed', + messageExcerpt: this.buildAssistantExcerptForMessage(completedMessageId), + }); + } + } else if (statusChange && isNeedsInputStatus(statusChange.value)) { + attentionSignals.push({ + signalId: `status:${statusChange.value}:${ingestedAt ?? Date.now()}`, + kind: 'needs_input', + messageExcerpt: '', + }); + } + return { changes, + attentionSignals, }; } + /** Builds a text excerpt for a completed assistant message from its already-ingested text parts. */ + private buildAssistantExcerptForMessage(messageId: string): string { + const range = getPartItemIdentityRange(messageId); + const rows = this.db + .select({ + item_data: ingestItems.item_data, + item_data_r2_key: ingestItems.item_data_r2_key, + }) + .from(ingestItems) + .where( + and( + eq(ingestItems.item_type, 'part'), + gte(ingestItems.item_id, range.start), + lt(ingestItems.item_id, range.end) + ) + ) + .orderBy(ingestItems.ingested_at, ingestItems.id) + .all(); + + // Parts offloaded to R2 (oversized items) store '{}' inline; skip them rather + // than fetching from R2 — this is a best-effort excerpt, not a full transcript. + return buildAssistantExcerpt( + rows.filter(row => !row.item_data_r2_key).map(row => row.item_data) + ); + } + + /** + * Finds the most recent completed assistant message id, scanning messages newest-first. Used to + * pair an idle transition with the assistant turn that just finished so the `completed` attention + * signal carries that turn's excerpt. R2-offloaded message rows (inline '{}') are skipped. + * + * The scan is bounded: the turn that just finished is effectively always among the newest + * messages, and an unbounded scan would load every message row on every turn end. + */ + private findLastCompletedAssistantMessageId(): string | null { + const rows = this.db + .select({ + item_data: ingestItems.item_data, + item_data_r2_key: ingestItems.item_data_r2_key, + }) + .from(ingestItems) + .where(eq(ingestItems.item_type, 'message')) + .orderBy(desc(ingestItems.ingested_at), desc(ingestItems.id)) + .limit(COMPLETED_MESSAGE_SCAN_LIMIT) + .all(); + for (const row of rows) { + if (row.item_data_r2_key) continue; + const messageId = completedAssistantMessageIdFromItemData(row.item_data); + if (messageId) return messageId; + } + return null; + } + async readKiloSdkSessionSnapshot(): Promise { return readKiloSdkSessionSnapshot(this.db, this.env.SESSION_INGEST_R2); } diff --git a/services/session-ingest/src/dos/UserConnectionDO.test.ts b/services/session-ingest/src/dos/UserConnectionDO.test.ts index 0ee029ef36..b3de4a0917 100644 --- a/services/session-ingest/src/dos/UserConnectionDO.test.ts +++ b/services/session-ingest/src/dos/UserConnectionDO.test.ts @@ -90,6 +90,13 @@ function makeSession(id: string, status = 'busy', title = 'Test', parentSessionI return parentSessionId ? { id, status, title, parentSessionId } : { id, status, title }; } +function makeSessionWithActivity( + id: string, + activity: { lastUserActivityAt?: number; lastMessageSource?: 'local' | 'remote' } +) { + return { ...makeSession(id), ...activity }; +} + function parseSent(ws: MockWS, callIndex = 0): unknown { const call = ws.send.mock.calls[callIndex]; if (!call) throw new Error(`No send call at index ${callIndex}`); @@ -181,7 +188,13 @@ function connectCliSocket(doInstance: UserConnectionDO, connectionId: string): M function addCliSocket( mockCtx: ReturnType, connectionId: string, - sessions: Array<{ id: string; status: string; title: string }> = [] + sessions: Array<{ + id: string; + status: string; + title: string; + lastUserActivityAt?: number; + lastMessageSource?: 'local' | 'remote'; + }> = [] ): MockWS { const attachment = { role: 'cli' as const, connectionId, sessions }; const ws = createMockWs(['cli'], attachment); @@ -205,7 +218,13 @@ function addWebSocket( function sendHeartbeat( doInstance: UserConnectionDO, cliWs: MockWS, - sessions: Array<{ id: string; status: string; title: string }>, + sessions: Array<{ + id: string; + status: string; + title: string; + lastUserActivityAt?: number; + lastMessageSource?: 'local' | 'remote'; + }>, protocolVersion?: string ) { const msg = JSON.stringify({ @@ -345,6 +364,49 @@ describe('UserConnectionDO', () => { }); }); + describe('getCliSessionPresence', () => { + it('tracks sessions and local activity reported by a connected CLI heartbeat', () => { + const { doInstance, mockCtx } = setup(); + const cliWs = addCliSocket(mockCtx, 'cli-1'); + + expect(doInstance.getCliSessionPresence('ses_1')).toEqual({ cliConnected: false }); + + sendHeartbeat(doInstance, cliWs, [ + makeSessionWithActivity('ses_1', { + lastUserActivityAt: 123, + lastMessageSource: 'local', + }), + ]); + + expect(doInstance.getCliSessionPresence('ses_1')).toEqual({ + cliConnected: true, + lastUserActivityAt: 123, + lastMessageSource: 'local', + }); + + mockCtx.removeSocket(cliWs); + disconnectCli(doInstance, cliWs); + + expect(doInstance.getCliSessionPresence('ses_1')).toEqual({ cliConnected: false }); + }); + + it('reconstructs heartbeat activity fields from hibernated attachments', () => { + const { doInstance, mockCtx } = setup(); + addCliSocket(mockCtx, 'cli-1', [ + makeSessionWithActivity('ses_1', { + lastUserActivityAt: 456, + lastMessageSource: 'remote', + }), + ]); + + expect(doInstance.getCliSessionPresence('ses_1')).toEqual({ + cliConnected: true, + lastUserActivityAt: 456, + lastMessageSource: 'remote', + }); + }); + }); + // ------------------------------------------------------------------------- // Heartbeat processing // ------------------------------------------------------------------------- diff --git a/services/session-ingest/src/dos/UserConnectionDO.ts b/services/session-ingest/src/dos/UserConnectionDO.ts index 2af8aea75e..39f873c137 100644 --- a/services/session-ingest/src/dos/UserConnectionDO.ts +++ b/services/session-ingest/src/dos/UserConnectionDO.ts @@ -4,6 +4,7 @@ import type { Env } from '../env'; import { CLIOutboundMessageSchema, type CLIInboundMessage, + type LastMessageSource, type SessionEventPayload, SessionEventPayloadSchema, type WebInboundMessage, @@ -17,6 +18,14 @@ type HeartbeatSession = { gitUrl?: string; gitBranch?: string; parentSessionId?: string; + lastUserActivityAt?: number; + lastMessageSource?: LastMessageSource; +}; + +export type CliSessionPresence = { + cliConnected: boolean; + lastUserActivityAt?: number; + lastMessageSource?: LastMessageSource; }; type WSAttachment = @@ -777,6 +786,30 @@ export class UserConnectionDO extends DurableObject { return { delivered }; } + /** Returns heartbeat-derived CLI presence for a session, if a live CLI owns it. */ + getCliSessionPresence(sessionId: string): CliSessionPresence { + this.ensureState(); + const ownerConnectionId = this.sessionOwners.get(sessionId); + if (!ownerConnectionId || !this.findCliByConnectionId(ownerConnectionId)) { + return { cliConnected: false }; + } + + const session = this.connectionSessions + .get(ownerConnectionId) + ?.find(candidate => candidate.id === sessionId); + if (!session) return { cliConnected: false }; + + return { + cliConnected: true, + ...(session.lastUserActivityAt !== undefined + ? { lastUserActivityAt: session.lastUserActivityAt } + : {}), + ...(session.lastMessageSource !== undefined + ? { lastMessageSource: session.lastMessageSource } + : {}), + }; + } + // --------------------------------------------------------------------------- // Helpers // --------------------------------------------------------------------------- diff --git a/services/session-ingest/src/dos/session-ingest-attention.test.ts b/services/session-ingest/src/dos/session-ingest-attention.test.ts new file mode 100644 index 0000000000..5ca0dda844 --- /dev/null +++ b/services/session-ingest/src/dos/session-ingest-attention.test.ts @@ -0,0 +1,136 @@ +import { describe, it, expect } from 'vitest'; +import { + buildAssistantExcerpt, + completedAssistantMessageIdFromItemData, + extractTextFromPartItemData, + isCompletedStatus, + isNeedsInputStatus, +} from './session-ingest-attention'; + +function completedAssistantMessageJson(data: Record): string { + return JSON.stringify({ id: 'msg-1', ...data }); +} + +describe('completedAssistantMessageIdFromItemData', () => { + it('returns the message id for a completed assistant message', () => { + expect( + completedAssistantMessageIdFromItemData( + completedAssistantMessageJson({ role: 'assistant', time: { created: 1, completed: 2 } }) + ) + ).toBe('msg-1'); + }); + + it('returns undefined for an assistant message without time.completed', () => { + expect( + completedAssistantMessageIdFromItemData( + completedAssistantMessageJson({ role: 'assistant', time: { created: 1 } }) + ) + ).toBeUndefined(); + }); + + it('returns undefined for a user message', () => { + expect( + completedAssistantMessageIdFromItemData( + completedAssistantMessageJson({ role: 'user', time: { created: 1, completed: 2 } }) + ) + ).toBeUndefined(); + }); + + it('returns undefined for malformed JSON', () => { + expect(completedAssistantMessageIdFromItemData('{not json')).toBeUndefined(); + }); +}); + +describe('extractTextFromPartItemData', () => { + it('extracts text from a text part', () => { + expect(extractTextFromPartItemData(JSON.stringify({ type: 'text', text: 'hello' }))).toBe( + 'hello' + ); + }); + + it('returns undefined for a non-text part', () => { + expect( + extractTextFromPartItemData(JSON.stringify({ type: 'tool', tool: 'bash' })) + ).toBeUndefined(); + }); + + it('returns undefined for malformed JSON', () => { + expect(extractTextFromPartItemData('{not json')).toBeUndefined(); + }); +}); + +describe('buildAssistantExcerpt', () => { + it('joins multiple text parts in order and trims the result', () => { + const rows = [ + JSON.stringify({ type: 'text', text: 'Hello ' }), + JSON.stringify({ type: 'text', text: 'world' }), + JSON.stringify({ type: 'text', text: ' ' }), + ]; + expect(buildAssistantExcerpt(rows)).toBe('Hello world'); + }); + + it('skips non-text parts when building the excerpt', () => { + const rows = [ + JSON.stringify({ type: 'tool', tool: 'bash' }), + JSON.stringify({ type: 'text', text: 'done' }), + ]; + expect(buildAssistantExcerpt(rows)).toBe('done'); + }); + + it('returns an empty string when there are no text parts', () => { + expect(buildAssistantExcerpt([JSON.stringify({ type: 'tool', tool: 'bash' })])).toBe(''); + }); + + it('collapses newlines and repeated whitespace into single spaces', () => { + const rows = [JSON.stringify({ type: 'text', text: 'First line.\n\nSecond line.' })]; + expect(buildAssistantExcerpt(rows)).toBe('First line. Second line.'); + }); + + it('truncates a long excerpt to 100 characters with an ellipsis', () => { + const rows = [JSON.stringify({ type: 'text', text: 'a'.repeat(250) })]; + const excerpt = buildAssistantExcerpt(rows); + expect(excerpt).toHaveLength(100); + expect(excerpt).toBe('a'.repeat(97) + '...'); + }); + + it('keeps an excerpt at exactly 100 characters untouched', () => { + const rows = [JSON.stringify({ type: 'text', text: 'b'.repeat(100) })]; + expect(buildAssistantExcerpt(rows)).toBe('b'.repeat(100)); + }); +}); + +describe('isCompletedStatus', () => { + it('returns true for idle', () => { + expect(isCompletedStatus('idle')).toBe(true); + }); + + it('returns false for busy, question, permission, and retry', () => { + expect(isCompletedStatus('busy')).toBe(false); + expect(isCompletedStatus('question')).toBe(false); + expect(isCompletedStatus('permission')).toBe(false); + expect(isCompletedStatus('retry')).toBe(false); + }); + + it('returns false for null and undefined', () => { + expect(isCompletedStatus(null)).toBe(false); + expect(isCompletedStatus(undefined)).toBe(false); + }); +}); + +describe('isNeedsInputStatus', () => { + it('returns true for question and permission', () => { + expect(isNeedsInputStatus('question')).toBe(true); + expect(isNeedsInputStatus('permission')).toBe(true); + }); + + it('returns false for idle, busy, and retry', () => { + expect(isNeedsInputStatus('idle')).toBe(false); + expect(isNeedsInputStatus('busy')).toBe(false); + expect(isNeedsInputStatus('retry')).toBe(false); + }); + + it('returns false for null and undefined', () => { + expect(isNeedsInputStatus(null)).toBe(false); + expect(isNeedsInputStatus(undefined)).toBe(false); + }); +}); diff --git a/services/session-ingest/src/dos/session-ingest-attention.ts b/services/session-ingest/src/dos/session-ingest-attention.ts new file mode 100644 index 0000000000..b5208b58bb --- /dev/null +++ b/services/session-ingest/src/dos/session-ingest-attention.ts @@ -0,0 +1,79 @@ +import { z } from 'zod'; + +export type AttentionSignalKind = 'completed' | 'needs_input'; + +export type AttentionSignal = { + signalId: string; + kind: AttentionSignalKind; + /** Push-ready excerpt: whitespace-collapsed and capped to fit a notification body. */ + messageExcerpt: string; +}; + +const NEEDS_INPUT_STATUSES = new Set(['question', 'permission']); + +/** Statuses that mean the session is waiting for the user to answer a question or approve a permission. */ +export function isNeedsInputStatus(status: string | null | undefined): boolean { + return status !== null && status !== undefined && NEEDS_INPUT_STATUSES.has(status); +} + +/** + * The root session went idle — the assistant finished its turn and is awaiting the next user message. + * Child-session statuses never reach this point: the kilo global feed drops any event whose + * `sessionID` is not the root session's, so only root idle transitions are ingested. + */ +export function isCompletedStatus(status: string | null | undefined): boolean { + return status === 'idle'; +} + +const CompletedAssistantMessageSchema = z.object({ + id: z.string().min(1), + role: z.literal('assistant'), + time: z.object({ completed: z.number() }), +}); + +/** + * Returns the message id when a message item's stored `item_data` JSON is a completed assistant + * message (role `assistant` with `time.completed`). Used to pair an idle transition with the + * assistant turn that just finished, so the `completed` signal can carry that turn's excerpt. + */ +export function completedAssistantMessageIdFromItemData(itemDataJson: string): string | undefined { + try { + const parsed = CompletedAssistantMessageSchema.safeParse(JSON.parse(itemDataJson)); + return parsed.success ? parsed.data.id : undefined; + } catch { + return undefined; + } +} + +const TextPartSchema = z.object({ type: z.literal('text'), text: z.string() }); + +/** Extracts the text of a `part` item's stored `item_data` JSON, or undefined if it isn't a text part. */ +export function extractTextFromPartItemData(itemDataJson: string): string | undefined { + try { + const parsed = TextPartSchema.safeParse(JSON.parse(itemDataJson)); + return parsed.success ? parsed.data.text : undefined; + } catch { + return undefined; + } +} + +// Expo/APNs reject oversized push payloads outright, so the excerpt must stay far below +// that limit. Matches the snippet length cloud-agent-next uses for the same notification. +const EXCERPT_MAX_LENGTH = 100; +const ELLIPSIS = '...'; + +function truncateExcerpt(text: string): string { + const singleLine = text.trim().replace(/\s+/g, ' '); + if (singleLine.length <= EXCERPT_MAX_LENGTH) return singleLine; + return singleLine.slice(0, EXCERPT_MAX_LENGTH - ELLIPSIS.length) + ELLIPSIS; +} + +/** Joins text parts (already in emission order) into a single push-ready excerpt. */ +export function buildAssistantExcerpt(partItemDataJsonRows: string[]): string { + const pieces: string[] = []; + for (const itemDataJson of partItemDataJsonRows) { + const text = extractTextFromPartItemData(itemDataJson); + if (text) pieces.push(text); + } + return truncateExcerpt(pieces.join('')); +} diff --git a/services/session-ingest/src/env.ts b/services/session-ingest/src/env.ts index 0ebae078dc..33b7a033e8 100644 --- a/services/session-ingest/src/env.ts +++ b/services/session-ingest/src/env.ts @@ -1,3 +1,7 @@ import type { O11YBinding } from './o11y-binding.js'; +import type { NotificationsBinding } from './notifications-binding.js'; -export type Env = Omit & { O11Y: O11YBinding }; +export type Env = Omit & { + O11Y: O11YBinding; + NOTIFICATIONS: NotificationsBinding; +}; diff --git a/services/session-ingest/src/notifications-binding.ts b/services/session-ingest/src/notifications-binding.ts new file mode 100644 index 0000000000..05ecac43d9 --- /dev/null +++ b/services/session-ingest/src/notifications-binding.ts @@ -0,0 +1,18 @@ +/** + * RPC method types for the NOTIFICATIONS service binding. + * + * `wrangler types` only sees `Fetcher` for service bindings; the actual RPC + * shape comes from the notifications worker's WorkerEntrypoint and is declared + * here from shared package types so the generated file can be freely regenerated. + */ + +import type { + SendCloudAgentSessionNotificationParams, + SendCloudAgentSessionNotificationResult, +} from '@kilocode/notifications'; + +export type NotificationsBinding = Fetcher & { + sendCloudAgentSessionNotification( + params: SendCloudAgentSessionNotificationParams + ): Promise; +}; diff --git a/services/session-ingest/src/queue-consumer.test.ts b/services/session-ingest/src/queue-consumer.test.ts index 7fc80c8fcb..6febf8fcef 100644 --- a/services/session-ingest/src/queue-consumer.test.ts +++ b/services/session-ingest/src/queue-consumer.test.ts @@ -27,6 +27,10 @@ vi.mock('./dos/SessionIngestDO', () => ({ getSessionIngestDO: vi.fn(), })); +vi.mock('./dos/UserConnectionDO', () => ({ + getUserConnectionDO: vi.fn(), +})); + vi.mock('./session-events', async importOriginal => { const actual = await importOriginal(); return { @@ -43,6 +47,7 @@ vi.mock('./util/ingest-limits', () => ({ import { getWorkerDb } from '@kilocode/db/client'; import { getSessionIngestDO } from './dos/SessionIngestDO'; +import { getUserConnectionDO } from './dos/UserConnectionDO'; import { notifyUserSessionEvent } from './session-events'; import { QUEUE_RETRY_DELAY_SECONDS, @@ -836,3 +841,173 @@ describe('queue status notifications', () => { ); }); }); + +describe('remote session attention notifications', () => { + const attentionSignal = { signalId: 'msg_1', kind: 'completed', messageExcerpt: 'All done' }; + + function setUpAttentionTest(params: { + sessionRow: { parent_session_id: string | null }; + cliPresence?: { + cliConnected: boolean; + lastUserActivityAt?: number; + lastMessageSource?: 'local' | 'remote'; + }; + ingest?: ReturnType; + stagedItems?: unknown[]; + }) { + const ingest = + params.ingest ?? vi.fn(async () => ({ changes: [], attentionSignals: [attentionSignal] })); + vi.mocked(getSessionIngestDO).mockReturnValue({ ingest } as never); + + // A single session lookup serves both the deleted-session guard and push eligibility. + const limit = vi.fn(async () => [{ session_id: 'ses_remote', ...params.sessionRow }]); + const where = vi.fn(() => ({ limit })); + const from = vi.fn(() => ({ where })); + const select = vi.fn(() => ({ from })); + vi.mocked(getWorkerDb).mockReturnValue({ select } as never); + + const getCliSessionPresence = vi.fn(async () => params.cliPresence ?? { cliConnected: true }); + vi.mocked(getUserConnectionDO).mockReturnValue({ + getCliSessionPresence, + } as never); + + const sendCloudAgentSessionNotification = vi.fn(async () => ({ dispatched: true })); + const body = JSON.stringify({ + data: params.stagedItems ?? [{ type: 'message', data: { id: 'msg_1' } }], + }); + const env = { + HYPERDRIVE: { connectionString: 'postgres://unused' }, + SESSION_INGEST_R2: { + get: vi.fn(async () => new Response(body)), + put: vi.fn(async () => undefined), + delete: vi.fn(async () => undefined), + }, + NOTIFICATIONS: { sendCloudAgentSessionNotification }, + } as never; + + return { + env, + ingest, + getCliSessionPresence, + sendCloudAgentSessionNotification, + }; + } + + async function runAttentionQueue(env: unknown) { + const ack = vi.fn(); + const retry = vi.fn(); + const waitUntilPromises: Promise[] = []; + const ctx = { + waitUntil: vi.fn((p: Promise) => waitUntilPromises.push(p)), + } as unknown as ExecutionContext; + + await queue( + { + messages: [ + { + body: { + r2Key: 'staging/attention', + kiloUserId: 'usr_remote', + sessionId: 'ses_remote', + ingestVersion: 1, + ingestedAt: 1, + }, + ack, + retry, + }, + ], + } as never, + env as never, + ctx + ); + await Promise.all(waitUntilPromises); + + return { ack, retry }; + } + + it('dispatches a push notification for an active root session', async () => { + const { env, getCliSessionPresence, sendCloudAgentSessionNotification } = setUpAttentionTest({ + sessionRow: { parent_session_id: null }, + cliPresence: { cliConnected: true }, + }); + + const { ack } = await runAttentionQueue(env); + + expect(ack).toHaveBeenCalledTimes(1); + expect(getCliSessionPresence).toHaveBeenCalledWith('ses_remote'); + expect(sendCloudAgentSessionNotification).toHaveBeenCalledWith({ + userId: 'usr_remote', + cliSessionId: 'ses_remote', + executionId: 'remote:msg_1', + status: 'completed', + body: 'All done', + suppressIfViewingSession: true, + }); + }); + + it('suppresses the push when recent local activity owns the current turn', async () => { + vi.spyOn(Date, 'now').mockReturnValue(10_000); + const { env, sendCloudAgentSessionNotification } = setUpAttentionTest({ + sessionRow: { parent_session_id: null }, + cliPresence: { + cliConnected: true, + lastMessageSource: 'local', + lastUserActivityAt: 9_000, + }, + }); + + await runAttentionQueue(env); + + expect(sendCloudAgentSessionNotification).not.toHaveBeenCalled(); + }); + + it('suppresses the push when no active CLI owns the root session', async () => { + const { env, getCliSessionPresence, sendCloudAgentSessionNotification } = setUpAttentionTest({ + sessionRow: { parent_session_id: null }, + cliPresence: { cliConnected: false }, + }); + + await runAttentionQueue(env); + + expect(getCliSessionPresence).toHaveBeenCalledWith('ses_remote'); + expect(sendCloudAgentSessionNotification).not.toHaveBeenCalled(); + }); + + it('suppresses the push for a child session', async () => { + const { env, sendCloudAgentSessionNotification } = setUpAttentionTest({ + sessionRow: { parent_session_id: 'ses_parent' }, + }); + + await runAttentionQueue(env); + + expect(sendCloudAgentSessionNotification).not.toHaveBeenCalled(); + }); + + it('dispatches signals collected from earlier chunks when a later chunk fails', async () => { + // 129 items force a mid-stream flush at the 128-item chunk cap; the leftover item + // flushes at the end as a second DO call. The first call commits and reports a + // signal, the second fails — the signal must still be dispatched because the retry + // will not re-emit the already-persisted status transition. + let ingestCalls = 0; + const ingest = vi.fn(async () => { + ingestCalls += 1; + if (ingestCalls > 1) throw new Error('DO unavailable'); + return { changes: [], attentionSignals: [attentionSignal] }; + }); + const stagedItems = Array.from({ length: 129 }, (_, i) => ({ + type: 'message', + data: { id: `msg_${i}` }, + })); + const { env, sendCloudAgentSessionNotification } = setUpAttentionTest({ + sessionRow: { parent_session_id: null }, + ingest, + stagedItems, + }); + + const { ack, retry } = await runAttentionQueue(env); + + expect(ack).not.toHaveBeenCalled(); + expect(retry).toHaveBeenCalledTimes(1); + expect(sendCloudAgentSessionNotification).toHaveBeenCalledTimes(1); + }); +}); diff --git a/services/session-ingest/src/queue-consumer.ts b/services/session-ingest/src/queue-consumer.ts index a56e8d0273..8b7bd39621 100644 --- a/services/session-ingest/src/queue-consumer.ts +++ b/services/session-ingest/src/queue-consumer.ts @@ -8,9 +8,15 @@ import { SessionItemSchema, type SessionDataItem } from './types/session-sync'; import { getItemIdentity } from './util/compaction'; import { MAX_INGEST_ITEM_BYTES, MAX_SINGLE_ITEM_BYTES } from './util/ingest-limits'; import { getSessionIngestDO } from './dos/SessionIngestDO'; +import { getUserConnectionDO } from './dos/UserConnectionDO'; +import type { AttentionSignal } from './dos/session-ingest-attention'; import { withDORetry, normalizeGitUrl } from '@kilocode/worker-utils'; import { mapSessionEventRow, notifyUserSessionEvent } from './session-events'; import { SessionStatusSchema } from './types/user-connection-protocol'; +import { + dispatchRemoteSessionAttentionSignal, + isEligibleForRemoteSessionAttention, +} from './remote-session-notifications'; export interface IngestQueueMessage { r2Key: string; @@ -165,13 +171,15 @@ async function processMessage( msg: IngestQueueMessage, ctx: ExecutionContext ): Promise { - if (await deleteStagingObjectIfSessionMissing(env, msg)) return; + const sessionRow = await loadSessionOrCleanupStaging(env, msg); + if (!sessionRow) return; const body = await getStagingObjectBody(env, msg.r2Key); const mergedChanges = new Map(); + const mergedAttentionSignals: AttentionSignal[] = []; try { - await ingestStagedSessionItems(env, msg, body, mergedChanges); + await ingestStagedSessionItems(env, msg, body, mergedChanges, mergedAttentionSignals); } catch (err) { // An earlier chunk may have committed to the DO before a later chunk (or the // JSON parse) failed. The DO reports a metadata change only when its stored @@ -179,10 +187,14 @@ async function processMessage( // re-emitted — Postgres would never catch up. Flush what we have now so the // two stores stay in sync. Best-effort: never mask the original error. await flushPartialMetadataChanges(env, msg, mergedChanges, ctx); + // Same reasoning for attention signals: a committed status transition won't + // re-emit on retry, so dispatch what was collected before the failure. + scheduleAttentionSignalDispatch(env, msg, sessionRow, mergedAttentionSignals, ctx); throw err; } await applyMetadataChanges(env, msg.kiloUserId, msg.sessionId, mergedChanges, ctx); + scheduleAttentionSignalDispatch(env, msg, sessionRow, mergedAttentionSignals, ctx); await env.SESSION_INGEST_R2.delete(msg.r2Key); } @@ -204,27 +216,39 @@ async function flushPartialMetadataChanges( } } -async function deleteStagingObjectIfSessionMissing( +type IngestSessionRow = { + session_id: string; + parent_session_id: string | null; +}; + +/** + * Loads the session row for the queued message, or cleans up the staging object and returns + * null if the session has been deleted since the message was queued. The parent column feeds + * the attention-push eligibility check later in processing. + */ +async function loadSessionOrCleanupStaging( env: Env, msg: IngestQueueMessage -): Promise { +): Promise { const { r2Key, kiloUserId, sessionId } = msg; - // Guard: skip processing if the session has been deleted since this message was queued const db = getWorkerDb(env.HYPERDRIVE.connectionString); const sessionRows = await db - .select({ session_id: cli_sessions_v2.session_id }) + .select({ + session_id: cli_sessions_v2.session_id, + parent_session_id: cli_sessions_v2.parent_session_id, + }) .from(cli_sessions_v2) .where( and(eq(cli_sessions_v2.session_id, sessionId), eq(cli_sessions_v2.kilo_user_id, kiloUserId)) ) .limit(1); - if (sessionRows[0]) return false; + if (sessionRows[0]) return sessionRows[0]; console.warn('Session no longer exists, cleaning up staging object', { r2Key, sessionId }); await env.SESSION_INGEST_R2.delete(r2Key); - return true; + return null; } async function getStagingObjectBody(env: Env, r2Key: string): Promise> { @@ -240,9 +264,10 @@ async function ingestStagedSessionItems( env: Env, msg: IngestQueueMessage, body: ReadableStream, - mergedChanges: Map + mergedChanges: Map, + mergedAttentionSignals: AttentionSignal[] ): Promise { - const chunker = createIngestChunker(env, msg, mergedChanges); + const chunker = createIngestChunker(env, msg, mergedChanges, mergedAttentionSignals); const parseError = await streamSessionItems(msg.r2Key, body, rawItem => chunker.stage(rawItem)); if (parseError) { @@ -322,7 +347,8 @@ function slimItemForR2Reference(item: SessionDataItem): SessionDataItem { function createIngestChunker( env: Env, msg: IngestQueueMessage, - mergedChanges: Map + mergedChanges: Map, + mergedAttentionSignals: AttentionSignal[] ) { const { r2Key, kiloUserId, sessionId, ingestVersion, ingestedAt } = msg; const encoder = new TextEncoder(); @@ -347,6 +373,7 @@ function createIngestChunker( for (const change of ingestResult.changes) { mergedChanges.set(change.name, change.value); } + mergedAttentionSignals.push(...(ingestResult.attentionSignals ?? [])); }; const stage = async (rawItem: Record): Promise => { @@ -612,6 +639,90 @@ async function applyMetadataChanges( } } +/** + * Best-effort mobile push dispatch for remote-session attention signals detected during + * ingest (completed assistant turn, or waiting for input). Suppressed for non-remote or + * child sessions, and for sessions the user is already watching in the web app. + * Never throws — a dispatch failure must not block ack'ing (or retrying) the message. + */ +function scheduleAttentionSignalDispatch( + env: Env, + msg: IngestQueueMessage, + sessionRow: IngestSessionRow, + signals: AttentionSignal[], + ctx: ExecutionContext +): void { + if (signals.length === 0) return; + if ( + !isEligibleForRemoteSessionAttention({ + parentSessionId: sessionRow.parent_session_id, + }) + ) { + console.log('Skipping attention signal dispatch (ineligible session)', { + sessionId: msg.sessionId, + kiloUserId: msg.kiloUserId, + parentSessionId: sessionRow.parent_session_id, + signalCount: signals.length, + }); + return; + } + + console.log('Scheduling attention signal dispatch', { + sessionId: msg.sessionId, + kiloUserId: msg.kiloUserId, + signalCount: signals.length, + kinds: signals.map(s => s.kind), + }); + ctx.waitUntil( + dispatchRemoteSessionAttentionSignals(env, msg.kiloUserId, msg.sessionId, signals).catch( + error => { + console.error('Failed to dispatch remote session attention signals (non-fatal)', { + sessionId: msg.sessionId, + kiloUserId: msg.kiloUserId, + error: error instanceof Error ? error.message : String(error), + }); + } + ) + ); +} + +async function dispatchRemoteSessionAttentionSignals( + env: Env, + kiloUserId: string, + sessionId: string, + signals: AttentionSignal[] +): Promise { + for (const signal of signals) { + try { + const outcome = await dispatchRemoteSessionAttentionSignal( + { kiloUserId, sessionId, signal }, + { + getCliPresence: async () => { + const stub = getUserConnectionDO(env, { kiloUserId }); + return stub.getCliSessionPresence(sessionId); + }, + sendPush: pushParams => env.NOTIFICATIONS.sendCloudAgentSessionNotification(pushParams), + } + ); + console.log('Remote session attention signal dispatch outcome', { + sessionId, + kiloUserId, + signalId: signal.signalId, + kind: signal.kind, + outcome, + }); + } catch (error) { + console.error('Failed to dispatch remote session attention notification (non-fatal)', { + sessionId, + kiloUserId, + signalId: signal.signalId, + kind: signal.kind, + error: error instanceof Error ? error.message : String(error), + }); + } + } +} + export async function queue( batch: MessageBatch, env: Env, diff --git a/services/session-ingest/src/remote-session-notifications.test.ts b/services/session-ingest/src/remote-session-notifications.test.ts new file mode 100644 index 0000000000..58d855a189 --- /dev/null +++ b/services/session-ingest/src/remote-session-notifications.test.ts @@ -0,0 +1,124 @@ +import { describe, it, expect, vi } from 'vitest'; +import type { AttentionSignal } from './dos/session-ingest-attention'; +import { + buildRemoteSessionAttentionPushBody, + dispatchRemoteSessionAttentionSignal, + isEligibleForRemoteSessionAttention, + LOCAL_ACTIVITY_WINDOW_MS, + shouldSuppressForLocalActivity, +} from './remote-session-notifications'; + +function completedSignal(messageExcerpt: string): AttentionSignal { + return { signalId: 'msg-1', kind: 'completed', messageExcerpt }; +} + +function needsInputSignal(): AttentionSignal { + return { signalId: 'status:question:123', kind: 'needs_input', messageExcerpt: '' }; +} + +describe('isEligibleForRemoteSessionAttention', () => { + it('is eligible for a root session', () => { + expect(isEligibleForRemoteSessionAttention({ parentSessionId: null })).toBe(true); + }); + + it('is not eligible for a child session', () => { + expect(isEligibleForRemoteSessionAttention({ parentSessionId: 'parent-1' })).toBe(false); + }); +}); + +describe('buildRemoteSessionAttentionPushBody', () => { + it('uses the message excerpt for a completed signal', () => { + expect(buildRemoteSessionAttentionPushBody(completedSignal('All done!'))).toBe('All done!'); + }); + + it('falls back to a default body when the excerpt is empty', () => { + expect(buildRemoteSessionAttentionPushBody(completedSignal(''))).toBe('Task completed'); + }); + + it('uses a fixed body for needs-input signals', () => { + expect(buildRemoteSessionAttentionPushBody(needsInputSignal())).toBe('Kilo needs your input.'); + }); +}); + +describe('dispatchRemoteSessionAttentionSignal', () => { + it('sends a push with a stable executionId and web-viewing suppression flag', async () => { + const sendPush = vi.fn(async () => ({ dispatched: true })); + const outcome = await dispatchRemoteSessionAttentionSignal( + { kiloUserId: 'usr_1', sessionId: 'ses_1', signal: completedSignal('Done') }, + { getCliPresence: async () => ({ cliConnected: true }), sendPush } + ); + + expect(outcome).toBe('sent'); + expect(sendPush).toHaveBeenCalledWith({ + userId: 'usr_1', + cliSessionId: 'ses_1', + executionId: 'remote:msg-1', + status: 'completed', + body: 'Done', + suppressIfViewingSession: true, + }); + }); + + it('suppresses the push when recent local activity owns the current turn', async () => { + const sendPush = vi.fn(async () => ({ dispatched: true })); + const now = 1_000_000; + vi.spyOn(Date, 'now').mockReturnValue(now); + const outcome = await dispatchRemoteSessionAttentionSignal( + { kiloUserId: 'usr_1', sessionId: 'ses_1', signal: completedSignal('Done') }, + { + getCliPresence: async () => ({ + cliConnected: true, + lastMessageSource: 'local', + lastUserActivityAt: now - 1_000, + }), + sendPush, + } + ); + + expect(outcome).toBe('suppressed'); + expect(sendPush).not.toHaveBeenCalled(); + }); + + it('suppresses the push when no connected CLI reports the session', async () => { + const getCliPresence = vi.fn(async () => ({ cliConnected: false })); + const sendPush = vi.fn(async () => ({ dispatched: true })); + const outcome = await dispatchRemoteSessionAttentionSignal( + { kiloUserId: 'usr_1', sessionId: 'ses_1', signal: completedSignal('Done') }, + { getCliPresence, sendPush } + ); + + expect(outcome).toBe('suppressed'); + expect(getCliPresence).toHaveBeenCalledOnce(); + expect(sendPush).not.toHaveBeenCalled(); + }); +}); + +describe('shouldSuppressForLocalActivity', () => { + it('suppresses local-sourced turns while activity is inside the window', () => { + expect( + shouldSuppressForLocalActivity( + { cliConnected: true, lastMessageSource: 'local', lastUserActivityAt: 10_000 }, + 10_000 + LOCAL_ACTIVITY_WINDOW_MS - 1 + ) + ).toBe(true); + }); + + it('does not suppress local-sourced turns after the activity window expires', () => { + expect( + shouldSuppressForLocalActivity( + { cliConnected: true, lastMessageSource: 'local', lastUserActivityAt: 10_000 }, + 10_000 + LOCAL_ACTIVITY_WINDOW_MS + ) + ).toBe(false); + }); + + it('does not suppress remote-sourced or legacy heartbeat turns', () => { + expect( + shouldSuppressForLocalActivity( + { cliConnected: true, lastMessageSource: 'remote', lastUserActivityAt: 10_000 }, + 11_000 + ) + ).toBe(false); + expect(shouldSuppressForLocalActivity({ cliConnected: true }, 11_000)).toBe(false); + }); +}); diff --git a/services/session-ingest/src/remote-session-notifications.ts b/services/session-ingest/src/remote-session-notifications.ts new file mode 100644 index 0000000000..8da7278a0f --- /dev/null +++ b/services/session-ingest/src/remote-session-notifications.ts @@ -0,0 +1,80 @@ +import type { + SendCloudAgentSessionNotificationParams, + SendCloudAgentSessionNotificationResult, +} from '@kilocode/notifications'; +import type { AttentionSignal } from './dos/session-ingest-attention'; +import type { CliSessionPresence } from './dos/UserConnectionDO'; + +export const LOCAL_ACTIVITY_WINDOW_MS = 2 * 60_000; + +export type RemoteSessionInfo = { + parentSessionId: string | null; +}; + +/** Only root sessions (no parent) can be eligible for attention pushes. */ +export function isEligibleForRemoteSessionAttention(session: RemoteSessionInfo): boolean { + return session.parentSessionId === null; +} + +const NEEDS_INPUT_BODY = 'Kilo needs your input.'; +const DEFAULT_COMPLETED_BODY = 'Task completed'; + +export function buildRemoteSessionAttentionPushBody( + signal: Pick +): string { + if (signal.kind === 'needs_input') return NEEDS_INPUT_BODY; + return signal.messageExcerpt.length > 0 ? signal.messageExcerpt : DEFAULT_COMPLETED_BODY; +} + +export type DispatchRemoteSessionAttentionDeps = { + getCliPresence: () => Promise; + sendPush: ( + params: SendCloudAgentSessionNotificationParams + ) => Promise; +}; + +export type DispatchRemoteSessionAttentionOutcome = 'sent' | 'suppressed'; + +export function shouldSuppressForLocalActivity( + presence: CliSessionPresence, + now: number = Date.now() +): boolean { + if (presence.lastMessageSource !== 'local') return false; + if (presence.lastUserActivityAt === undefined) return false; + return now - presence.lastUserActivityAt < LOCAL_ACTIVITY_WINDOW_MS; +} + +/** + * Sends a best-effort mobile push for a remote session attention signal, unless recent + * heartbeat data shows the user is actively driving the session locally. Web-viewing + * suppression is handled by the notifications service presence check. A connected CLI must + * currently report the session in its heartbeat. Callers are expected to have already + * confirmed `isEligibleForRemoteSessionAttention` for the owning session. + */ +export async function dispatchRemoteSessionAttentionSignal( + params: { kiloUserId: string; sessionId: string; signal: AttentionSignal }, + deps: DispatchRemoteSessionAttentionDeps +): Promise { + const presence = await deps.getCliPresence(); + if (!presence.cliConnected) { + return 'suppressed'; + } + + if (shouldSuppressForLocalActivity(presence)) { + return 'suppressed'; + } + + await deps.sendPush({ + userId: params.kiloUserId, + cliSessionId: params.sessionId, + executionId: `remote:${params.signal.signalId}`, + // The push status enum has no needs_input value and the notifications service ignores + // status when building the push — the body carries the real semantics. Extending the + // enum would fail validation on a notifications worker deployed with the old schema. + status: 'completed', + body: buildRemoteSessionAttentionPushBody(params.signal), + suppressIfViewingSession: true, + }); + + return 'sent'; +} diff --git a/services/session-ingest/src/types/user-connection-protocol.test.ts b/services/session-ingest/src/types/user-connection-protocol.test.ts index b5145be65f..1737c9dbc5 100644 --- a/services/session-ingest/src/types/user-connection-protocol.test.ts +++ b/services/session-ingest/src/types/user-connection-protocol.test.ts @@ -40,6 +40,29 @@ describe('CLIOutboundMessageSchema', () => { } }); + it('parses heartbeat with optional local activity fields', () => { + const msg = { + type: 'heartbeat', + sessions: [ + { + id: validSessionId, + status: 'busy', + title: 'Fix bug', + lastUserActivityAt: 123, + lastMessageSource: 'local', + }, + ], + }; + const result = CLIOutboundMessageSchema.safeParse(msg); + expect(result.success).toBe(true); + if (result.success && result.data.type === 'heartbeat') { + expect(result.data.sessions[0]).toMatchObject({ + lastUserActivityAt: 123, + lastMessageSource: 'local', + }); + } + }); + it('parses heartbeat without parentSessionId (backward compat)', () => { const msg = { type: 'heartbeat', diff --git a/services/session-ingest/src/types/user-connection-protocol.ts b/services/session-ingest/src/types/user-connection-protocol.ts index da3e79a550..240e2bdbd2 100644 --- a/services/session-ingest/src/types/user-connection-protocol.ts +++ b/services/session-ingest/src/types/user-connection-protocol.ts @@ -20,6 +20,8 @@ export const CLIOutboundMessageSchema = z.discriminatedUnion('type', [ gitUrl: z.string().optional(), gitBranch: z.string().optional(), parentSessionId: z.string().optional(), + lastUserActivityAt: z.number().optional(), + lastMessageSource: z.enum(['local', 'remote']).optional(), }) ), }), @@ -185,6 +187,9 @@ export const WebInboundMessageSchema = z.discriminatedUnion('type', [ // -- Inferred types ----------------------------------------------------------- export type CLIOutboundMessage = z.infer; +export type LastMessageSource = NonNullable< + Extract['sessions'][number]['lastMessageSource'] +>; export type CLIInboundMessage = z.infer; export type WebOutboundMessage = z.infer; export type WebInboundMessage = z.infer; diff --git a/services/session-ingest/test/integration/session-ingest-do.test.ts b/services/session-ingest/test/integration/session-ingest-do.test.ts index d770813243..e78b902231 100644 --- a/services/session-ingest/test/integration/session-ingest-do.test.ts +++ b/services/session-ingest/test/integration/session-ingest-do.test.ts @@ -2037,6 +2037,236 @@ describe('SessionIngestDO integration', () => { }); }); + describe('attention signals', () => { + /** A completed signal requires a previously stored status, so tests start sessions as busy. */ + async function ingestBusyStatus(stub: ReturnType, sessionId: string) { + await stub.ingest( + [{ type: 'session_status', data: { status: 'busy' } }], + kiloUserId, + sessionId, + 1 + ); + } + + it('emits a completed signal with a text excerpt when the status transitions to idle', async () => { + const sessionId = 'ses_attention_completed_0001'; + const stub = getStub(kiloUserId, sessionId); + await ingestBusyStatus(stub, sessionId); + + const result = await stub.ingest( + [ + { + type: 'part', + data: { id: 'part_1', messageID: 'msg_1', type: 'text', text: 'Hello ' }, + }, + { type: 'part', data: { id: 'part_2', messageID: 'msg_1', type: 'text', text: 'world' } }, + { + type: 'message', + data: { id: 'msg_1', role: 'assistant', time: { created: 1, completed: 2 } }, + }, + { type: 'session_status', data: { status: 'idle' } }, + ], + kiloUserId, + sessionId, + 1 + ); + + expect(result.attentionSignals).toEqual([ + { signalId: 'msg_1', kind: 'completed', messageExcerpt: 'Hello world' }, + ]); + }); + + it('finds an excerpt from parts ingested in an earlier call', async () => { + const sessionId = 'ses_attention_completed_0002'; + const stub = getStub(kiloUserId, sessionId); + + await stub.ingest( + [ + { type: 'part', data: { id: 'part_1', messageID: 'msg_1', type: 'text', text: 'Done!' } }, + { type: 'session_status', data: { status: 'busy' } }, + ], + kiloUserId, + sessionId, + 1 + ); + const result = await stub.ingest( + [ + { + type: 'message', + data: { id: 'msg_1', role: 'assistant', time: { created: 1, completed: 2 } }, + }, + { type: 'session_status', data: { status: 'idle' } }, + ], + kiloUserId, + sessionId, + 1 + ); + + expect(result.attentionSignals).toEqual([ + { signalId: 'msg_1', kind: 'completed', messageExcerpt: 'Done!' }, + ]); + }); + + it('truncates a long excerpt to a push-sized snippet', async () => { + const sessionId = 'ses_attention_truncated_0001'; + const stub = getStub(kiloUserId, sessionId); + await ingestBusyStatus(stub, sessionId); + + const longText = `First line.\n\n${'x'.repeat(200)}`; + const result = await stub.ingest( + [ + { + type: 'part', + data: { id: 'part_1', messageID: 'msg_1', type: 'text', text: longText }, + }, + { + type: 'message', + data: { id: 'msg_1', role: 'assistant', time: { created: 1, completed: 2 } }, + }, + { type: 'session_status', data: { status: 'idle' } }, + ], + kiloUserId, + sessionId, + 1 + ); + + const excerpt = result.attentionSignals[0]?.messageExcerpt; + expect(excerpt).toHaveLength(100); + expect(excerpt).toMatch(/^First line\. x+\.\.\.$/); + }); + + it('does not emit a completed signal when the assistant message has not finished', async () => { + const sessionId = 'ses_attention_incomplete_001'; + const stub = getStub(kiloUserId, sessionId); + await ingestBusyStatus(stub, sessionId); + + const result = await stub.ingest( + [ + { type: 'message', data: { id: 'msg_1', role: 'assistant', time: { created: 1 } } }, + { type: 'session_status', data: { status: 'idle' } }, + ], + kiloUserId, + sessionId, + 1 + ); + + expect(result.attentionSignals).toEqual([]); + }); + + it('does not emit a completed signal when only a user message has completed', async () => { + const sessionId = 'ses_attention_user_msg_0001'; + const stub = getStub(kiloUserId, sessionId); + await ingestBusyStatus(stub, sessionId); + + const result = await stub.ingest( + [ + { + type: 'message', + data: { id: 'msg_1', role: 'user', time: { created: 1, completed: 2 } }, + }, + { type: 'session_status', data: { status: 'idle' } }, + ], + kiloUserId, + sessionId, + 1 + ); + + expect(result.attentionSignals).toEqual([]); + }); + + it("does not emit a completed signal on the session's first reported status", async () => { + const sessionId = 'ses_attention_first_status_1'; + const stub = getStub(kiloUserId, sessionId); + + // A full-history backfill of an already-idle session must not push about an old turn. + const result = await stub.ingest( + [ + { type: 'part', data: { id: 'part_1', messageID: 'msg_1', type: 'text', text: 'Old' } }, + { + type: 'message', + data: { id: 'msg_1', role: 'assistant', time: { created: 1, completed: 2 } }, + }, + { type: 'session_status', data: { status: 'idle' } }, + ], + kiloUserId, + sessionId, + 1 + ); + + expect(result.attentionSignals).toEqual([]); + }); + + it('emits a needs-input signal when status becomes question', async () => { + const sessionId = 'ses_attention_question_0001'; + const stub = getStub(kiloUserId, sessionId); + + const result = await stub.ingest( + [{ type: 'session_status', data: { status: 'question' } }], + kiloUserId, + sessionId, + 1 + ); + + expect(result.attentionSignals).toHaveLength(1); + expect(result.attentionSignals[0]).toMatchObject({ kind: 'needs_input' }); + }); + + it('emits a needs-input signal when status becomes permission', async () => { + const sessionId = 'ses_attention_permission_001'; + const stub = getStub(kiloUserId, sessionId); + + const result = await stub.ingest( + [{ type: 'session_status', data: { status: 'permission' } }], + kiloUserId, + sessionId, + 1 + ); + + expect(result.attentionSignals).toHaveLength(1); + expect(result.attentionSignals[0]).toMatchObject({ kind: 'needs_input' }); + }); + + it('does not emit a completed signal when the session has no messages at all', async () => { + const sessionId = 'ses_attention_idle_0000001'; + const stub = getStub(kiloUserId, sessionId); + + await stub.ingest( + [{ type: 'session_status', data: { status: 'busy' } }], + kiloUserId, + sessionId, + 1 + ); + const result = await stub.ingest( + [{ type: 'session_status', data: { status: 'idle' } }], + kiloUserId, + sessionId, + 1 + ); + + expect(result.attentionSignals).toEqual([]); + }); + + it('does not re-emit a needs-input signal when the status is unchanged', async () => { + const sessionId = 'ses_attention_repeat_000001'; + const stub = getStub(kiloUserId, sessionId); + + await stub.ingest( + [{ type: 'session_status', data: { status: 'question' } }], + kiloUserId, + sessionId, + 1 + ); + const result = await stub.ingest( + [{ type: 'session_status', data: { status: 'question' } }], + kiloUserId, + sessionId, + 1 + ); + + expect(result.attentionSignals).toEqual([]); + }); + }); + describe('export produces valid JSON', () => { it('returns valid JSON from getAllStream even with no items', async () => { const sessionId = 'ses_export_empty_0000000007'; diff --git a/services/session-ingest/worker-configuration.d.ts b/services/session-ingest/worker-configuration.d.ts index fac45700de..12d49ca671 100644 --- a/services/session-ingest/worker-configuration.d.ts +++ b/services/session-ingest/worker-configuration.d.ts @@ -17,6 +17,7 @@ declare namespace Cloudflare { SESSION_ACCESS_CACHE_DO: DurableObjectNamespace; USER_CONNECTION_DO: DurableObjectNamespace; O11Y: Fetcher /* o11y */; + NOTIFICATIONS: Service /* entrypoint NotificationsService from notifications */; } } interface Env extends Cloudflare.Env {} diff --git a/services/session-ingest/wrangler.jsonc b/services/session-ingest/wrangler.jsonc index 84d519add0..fddfe386b6 100644 --- a/services/session-ingest/wrangler.jsonc +++ b/services/session-ingest/wrangler.jsonc @@ -96,6 +96,11 @@ "binding": "O11Y", "service": "o11y", }, + { + "binding": "NOTIFICATIONS", + "service": "notifications", + "entrypoint": "NotificationsService", + }, ], "secrets_store_secrets": [ {