feat(session-ingest): push notifications for remote session attention signals

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());

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
+  );
+}

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": {

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');
+  });
+});

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;

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

services/notifications/src/index.ts

@@ -225,7 +225,7 @@ export class NotificationsService extends WorkerEntrypoint<Env> {
     let db: ReturnType<typeof getWorkerDb> | 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<Env> {
         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(

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,19 +29,30 @@ 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' };
   }
 
   if (
     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 };
 }

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,
+    });
+  });
+});

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",

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<Env> {
     r2References?: Record<string, string>
   ): Promise<{
     changes: Changes;
+    attentionSignals: AttentionSignal[];
   }> {
     const deletedRow = this.db
       .select({ value: ingestMeta.value })
@@ -151,7 +169,7 @@ export class SessionIngestDO extends DurableObject<Env> {
           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<Env> {
       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<Env> {
       );
     }
 
+    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<KiloSdkSessionSnapshotRead> {
     return readKiloSdkSessionSnapshot(this.db, this.env.SESSION_INGEST_R2);
   }