Merge session format migration into live assistant stream

# Conflicts:
#	docs/event-producer-consumer.i18n.yaml
#	docs/event-producer-consumer.md
#	docs/event-producer-consumer.zh.md

packages/api/session-controller/src/index.ts

@@ -17,7 +17,11 @@ import { SessionCommandController } from './commands.ts'
 import { SessionControlController } from './control.ts'
 import { SessionHistoryController } from './history.ts'
 import { SessionFileReferences } from './file-references.ts'
-import { ApiSessionList, DEFAULT_COLD_BLANK_PROBE_MAX_BYTES } from './list.ts'
+import {
+  ApiSessionList,
+  DEFAULT_COLD_BLANK_PROBE_MAX_BYTES,
+  DEFAULT_COLD_BLANK_PROBE_MAX_EVENTS,
+} from './list.ts'
 import { buildModelCatalog } from './catalog.ts'
 import { installModelSelectionProjection } from './model-selection-projection.ts'
 import { SessionSkillCatalog } from './skill-catalog.ts'
@@ -66,7 +70,9 @@ declare module '@deepseek-ai/cordis' {
 
 /** Session Controller deployment policy. */
 export interface Config {
-  /** Maximum cold Session artifact size eligible for one full projection observation. */
+  /** Maximum stat-reported event count eligible for one full cold projection observation; `0` disables the event-count gate. */
+  readonly coldBlankProbeMaxEvents?: number
+  /** Maximum stat-reported artifact byte size eligible for one full cold projection observation; `0` disables the byte-size gate. */
   readonly coldBlankProbeMaxBytes?: number
   /** Override platform desktop-opener detection. */
   readonly nativeOpen?: boolean
@@ -95,6 +101,7 @@ export class SessionController extends TypertRemoteService {
   ]
 
   static Config: z<Config> = z.object({
+    coldBlankProbeMaxEvents: z.natural().default(DEFAULT_COLD_BLANK_PROBE_MAX_EVENTS),
     coldBlankProbeMaxBytes: z.natural().default(DEFAULT_COLD_BLANK_PROBE_MAX_BYTES),
     nativeOpen: z.boolean(),
   })
@@ -110,7 +117,8 @@ export class SessionController extends TypertRemoteService {
 
   /**
    * @param ctx - Host context containing the Session capability assembly.
-   * @param config - cold-list observation policy.
+   * @param config - cold-list observation and native-opener deployment policy.
+   * @param internals - host integrations replaceable by direct unit tests.
    */
   constructor(ctx: Context, config: Config, internals: SessionControllerInternals = {}) {
     super(ctx, 'sessionController', { namespace: 'session' })
@@ -124,10 +132,10 @@ export class SessionController extends TypertRemoteService {
       await Promise.allSettled([...this.promotions])
     }, 'session-controller.promotions')
     this.history = new SessionHistoryController(ctx, (observation) => { this.promote(observation) })
-    this.listState = new ApiSessionList(
-      ctx,
-      config.coldBlankProbeMaxBytes ?? DEFAULT_COLD_BLANK_PROBE_MAX_BYTES,
-    )
+    this.listState = new ApiSessionList(ctx, {
+      coldBlankProbeMaxEvents: config.coldBlankProbeMaxEvents ?? DEFAULT_COLD_BLANK_PROBE_MAX_EVENTS,
+      coldBlankProbeMaxBytes: config.coldBlankProbeMaxBytes ?? DEFAULT_COLD_BLANK_PROBE_MAX_BYTES,
+    })
     this.openPath = internals.openPath ?? openNativePath
     this.canOpenPath = internals.canOpenPath
       ?? (() => config.nativeOpen ?? (internals.openPath !== undefined || canOpenNativePath()))

packages/api/session-controller/src/list.ts

@@ -1,6 +1,5 @@
 /** Cold-safe Session list and search projection. */
 
-import { stat } from 'node:fs/promises'
 import type { Context } from '@deepseek-ai/cordis'
 import type {} from '@deepseek-ai/dsh-agent-presets'
 import type { ImageAttachmentLimits } from '@deepseek-ai/dsh-attachment'
@@ -8,11 +7,7 @@ import { SessionLogOffset } from '@deepseek-ai/dsh-session'
 import type { Session, SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
 import type {} from '@deepseek-ai/dsh-session-projection'
 import type {} from '@deepseek-ai/dsh-session-projection-cache'
-import {
-  SessionQueryError,
-  type SessionRecord,
-  type SessionSearchCursor,
-} from '@deepseek-ai/dsh-session-query'
+import { SessionQueryError, type SessionSearchCursor } from '@deepseek-ai/dsh-session-query'
 import { RemoteError } from '@deepseek-ai/dsh-typert-protocol'
 import { z } from 'zod'
 import {
@@ -24,9 +19,20 @@ import type {
   SessionSearchValue, SessionSummary,
 } from './types.ts'
 
-/** Default maximum artifact size eligible for one cold projection observation. */
+/** Default maximum stat-reported event count eligible for one cold projection observation. */
+export const DEFAULT_COLD_BLANK_PROBE_MAX_EVENTS = 16
+
+/** Default maximum stat-reported artifact size eligible for one cold projection observation. */
 export const DEFAULT_COLD_BLANK_PROBE_MAX_BYTES = 1024
 
+/** Resolved cold-blank probe policy: each threshold gates its stat metric; `0` disables that gate. */
+export interface ColdBlankProbePolicy {
+  /** Maximum stat-reported `eventCount` eligible for a full observation. */
+  readonly coldBlankProbeMaxEvents: number
+  /** Maximum stat-reported `sizeBytes` eligible for a full observation. */
+  readonly coldBlankProbeMaxBytes: number
+}
+
 const COLD_SUMMARY_BATCH_SIZE = 16
 const SEARCH_PROVIDER_CALL_LIMIT = 100
 const SESSION_SEARCH_QUERY_MAX_CHARS = 500
@@ -86,11 +92,11 @@ export function truncateUnicodeCodePoints(value: string, maximum: number): strin
 export class ApiSessionList {
   /**
    * @param ctx - Host context carrying Session, query, persistence, and projection services.
-   * @param coldBlankProbeMaxBytes - maximum physical artifact size eligible for a full observation.
+   * @param probe - stat-metadata thresholds gating a full cold observation.
    */
   constructor(
     private readonly ctx: Context,
-    private readonly coldBlankProbeMaxBytes: number,
+    private readonly probe: ColdBlankProbePolicy,
   ) {
     ctx.sessionProjections.register<'sessionListMetadata', SessionListMetadata>({
       key: 'sessionListMetadata',
@@ -143,7 +149,7 @@ export class ApiSessionList {
     const records = await this.ctx.sessionQuery.listSessions(signal)
     signal?.throwIfAborted()
     const items: SessionSummary[] = []
-    const cold: SessionRecord[] = []
+    const cold: SessionHeader[] = []
     for (const record of records) {
       const live = this.ctx.sessions.get(record.header.id)
       if (live !== undefined) {
@@ -151,11 +157,11 @@ export class ApiSessionList {
         continue
       }
       if (record.header.cwd === undefined) continue
-      cold.push(record)
+      cold.push(record.header)
     }
     for (let offset = 0; offset < cold.length; offset += COLD_SUMMARY_BATCH_SIZE) {
       const settled = await Promise.allSettled(cold.slice(offset, offset + COLD_SUMMARY_BATCH_SIZE)
-        .map(record => this.summarizeCold(record, signal)))
+        .map(header => this.summarizeCold(header, signal)))
       for (const result of settled) {
         if (result.status === 'rejected') throw result.reason
         items.push(result.value)
@@ -166,14 +172,13 @@ export class ApiSessionList {
   }
 
   private async summarizeCold(
-    record: SessionRecord,
+    header: SessionHeader,
     signal: AbortSignal | undefined,
   ): Promise<SessionSummary> {
-    const { header } = record
     const cached = this.projectionsFor(header, undefined)
     const projections = cached?.values.sessionListMetadata?.blank === false
       ? cached
-      : await this.probeSmallCold(record, signal) ?? cached
+      : await this.probeSmallCold(header, signal) ?? cached
     const raced = this.ctx.sessions.get(header.id)
     if (raced !== undefined) return this.summaryFor(raced)
     const metadata = projections?.values.sessionListMetadata
@@ -181,7 +186,7 @@ export class ApiSessionList {
       sessionId: header.id,
       updatedAt: updatedAt(header, metadata),
       running: false,
-      // A large or inaccessible cache miss remains unknown and visible.
+      // A large, metadata-less, or inaccessible cache miss remains unknown and visible.
       blank: metadata?.blank ?? false,
       ...listFields(header),
       ...(projections === undefined ? {} : { projections }),
@@ -189,19 +194,33 @@ export class ApiSessionList {
   }
 
   private async probeSmallCold(
-    record: SessionRecord,
+    header: SessionHeader,
     signal: AbortSignal | undefined,
   ): Promise<SessionProjectionHints | undefined> {
-    if (this.coldBlankProbeMaxBytes === 0) return undefined
-    const { header } = record
+    const { coldBlankProbeMaxEvents, coldBlankProbeMaxBytes } = this.probe
+    if (coldBlankProbeMaxEvents === 0 && coldBlankProbeMaxBytes === 0) return undefined
     const persistence = this.ctx.get('sessionPersistence')
-    const location = record.location ?? persistence?.locate(header)
-    if (location === undefined) return undefined
+    if (persistence === undefined) return undefined
     signal?.throwIfAborted()
+    let snapshot: Awaited<ReturnType<typeof persistence.stat>>
     try {
-      if ((await stat(location.path)).size > this.coldBlankProbeMaxBytes) return undefined
-    } catch {
+      snapshot = await persistence.stat(header.id, signal === undefined ? {} : { signal })
+    } catch (error: unknown) {
+      // An unreadable single session degrades to unknown state instead of
+      // failing the whole list request.
       signal?.throwIfAborted()
+      this.ctx.logger.warn(
+        `api-session.list: cold stat for "${header.id}" failed; serving it as visible: ${String(error)}`,
+      )
+      return undefined
+    }
+    if (snapshot === undefined) return undefined
+    if (snapshot.eventCount !== undefined) {
+      if (coldBlankProbeMaxEvents === 0 || snapshot.eventCount > coldBlankProbeMaxEvents) return undefined
+    } else if (snapshot.sizeBytes !== undefined) {
+      if (coldBlankProbeMaxBytes === 0 || snapshot.sizeBytes > coldBlankProbeMaxBytes) return undefined
+    } else {
+      // The backend offers no cheap size hint, so a full observation is unbounded work.
       return undefined
     }
     try {