Merge remote-tracking branch 'origin/master' into worktree/web-model-request-retry

# Conflicts: # apps/cli/README.i18n.yaml # packages/client/runtime/README.i18n.yaml # packages/client/runtime/README.md # packages/client/runtime/README.zh.md # packages/client/ui-conversation/README.i18n.yaml # packages/client/ui-conversation/src/client/chat/ChatView.tsx # packages/client/ui-conversation/src/client/chat/MessageItem.tsx # packages/client/ui-conversation/tests/chat-view.spec.tsx
2026-07-31 10:19:14 +08:00
parent 5b9c088fb7 1351565681
commit d3c6638b6d
933 changed files with 28910 additions and 4079 deletions
@@ -29,6 +29,17 @@ export interface ISessions {
  open(id: SessionId): void
  /** Clear the current selection into the no-session view state. */
  clear(): void
+  /**
+   * Fork a session from a completed-turn prefix of the source; on resolution
+   * the child is in the list store and `open()` can target it.
+   * @param opts - source session id, the optional event seq anchoring the
+   *   cut (the boundary is the first turn/end at or after it; an in-log
+   *   anchor in an open turn is unavailable rather than clipped backward),
+   *   and whether to increment an inherited durable title before resolving.
+   * @returns the child session id.
+   * @throws when the fork fails, or when a requested child-title rename fails after creation.
+   */
+  fork(opts: { sessionId: SessionId; atSeq?: number; increaseTitle?: boolean }): Promise<SessionId>
  /**
   * Register a per-session standard-props provider (hooks become `use<Name>`
   * selector hooks on the render side; props spread verbatim).
@@ -122,6 +122,28 @@ declare module 'cordis' {
     * @mode emit
     */
    'commands/changed'(): void
+    /**
+     * One settings namespace's resolved value changed on the host
+     * (host/settings-changed passthrough). Subscribers refetch
+     * `settings.describe`; the frame carries no values.
+     * @mode emit
+     * @param ns - the namespace whose resolved value changed.
+     */
+    'settings/changed'(ns: string): void
+    /**
+     * One credential reference's state changed on the host
+     * (host/credentials-changed passthrough). The ref is an
+     * environment-variable NAME — never a value.
+     * @mode emit
+     * @param ref - the reference whose configured state changed.
+     */
+    'credentials/changed'(ref: string): void
+    /**
+     * The host provider topology changed (host/models-changed passthrough).
+     * Subscribers refetch `llm.providers`/`llm.models`/`session.models`.
+     * @mode emit
+     */
+    'models/changed'(): void
    /**
     * A connection generation was (re-)established. Wire-derived caches must
     * treat their state as stale and repull (commands directory; the queue
@@ -170,8 +192,13 @@ export function apply(ctx: Context): void {
      sessions.handleHostEnvelope(envelope)
      workspaces.handleHostEnvelope(envelope)
      // Typed-event bridge: the session layer ignores registry frames (no
-      // session routing); consumers (command directory caches) subscribe on ctx.
-      if (envelope.payload.type === 'host/commands-changed') ctx.emit('commands/changed')
+      // session routing); consumers (command directory caches, the settings
+      // and model surfaces) subscribe on ctx.
+      const frame = envelope.payload
+      if (frame.type === 'host/commands-changed') ctx.emit('commands/changed')
+      else if (frame.type === 'host/settings-changed') ctx.emit('settings/changed', frame.ns)
+      else if (frame.type === 'host/credentials-changed') ctx.emit('credentials/changed', frame.ref)
+      else if (frame.type === 'host/models-changed') ctx.emit('models/changed')
      try {
        sessionHistory.handleHostEnvelope(envelope)
      } catch (error) {
@@ -1,12 +1,14 @@
 import type {
  HistoryEntry, IApiClient, MuxFrame, RpcError, SessionId,
 } from '@deepseek-ai/dsh-client-connection/client'
+import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
 import { transportError } from '@deepseek-ai/dsh-host-apiproxy/api'
 import type {
  SessionHistoryFace, SessionHistorySnapshot,
 } from '../contract/session-history.ts'
 import { createHistoryInspection } from '../sessions/history.ts'
 import { Notifier } from '../sessions/notifier.ts'
+import { PartialAccumulator } from '../sessions/partial.ts'

 const HISTORY_PAGE_MESSAGES = 50

@@ -33,6 +35,9 @@ export class SessionHistorySource implements SessionHistoryFace {
    entries: readonly HistoryEntry[]
    value: SessionHistorySnapshot['inspection']
  } | null = null
+  private streamPublishToken: object | null = null
+  private streamBaseInspection: SessionHistorySnapshot['inspection'] | null = null
+  private streamPartial: PartialAccumulator | null = null
  private snapshotCache: SessionHistorySnapshot
  private readonly notifier = new Notifier(() => {
    this.snapshotCache = this.buildSnapshot()
@@ -125,7 +130,7 @@ export class SessionHistorySource implements SessionHistoryFace {
    if (this.state !== 'cold') {
      this.state = 'cold'
      this.error = null
-      this.notifier.markDirty()
+      this.publishDirtyNow()
    }
  }

@@ -143,7 +148,7 @@ export class SessionHistorySource implements SessionHistoryFace {
    this.hasMore = false
    this.state = 'cold'
    this.error = null
-    this.notifier.markDirty()
+    this.publishDirtyNow()
    void this.loadForConsumers()
  }

@@ -155,6 +160,9 @@ export class SessionHistorySource implements SessionHistoryFace {
    this.openPromise = null
    this.olderPromise = null
    this.liveBuffer = []
+    this.streamPublishToken = null
+    this.streamBaseInspection = null
+    this.streamPartial = null
  }

  private open(): Promise<void> {
@@ -188,7 +196,7 @@ export class SessionHistorySource implements SessionHistoryFace {
  private async doOpen(generation: number): Promise<void> {
    this.state = 'loading'
    this.error = null
-    this.notifier.markDirty()
+    this.publishDirtyNow()
    try {
      let { result } = await this.api.sessions.history({
        sessionId: this.sessionId,
@@ -222,7 +230,7 @@ export class SessionHistorySource implements SessionHistoryFace {
      /* v8 ignore next -- transportError always returns the error branch. */
      this.error = folded.ok ? null : folded.error
    } finally {
-      if (generation === this.generation) this.notifier.markDirty()
+      if (generation === this.generation) this.publishDirtyNow()
    }
  }

@@ -261,7 +269,7 @@ export class SessionHistorySource implements SessionHistoryFace {
    const settled = operation.finally(() => {
      if (this.olderPromise !== settled) return
      this.olderPromise = null
-      this.notifier.markDirty()
+      this.publishDirtyNow()
    })
    this.olderPromise = settled
    return settled
@@ -286,7 +294,7 @@ export class SessionHistorySource implements SessionHistoryFace {
    const buffered = this.liveBuffer
    this.liveBuffer = []
    for (const entry of buffered) this.appendLive(entry)
-    this.notifier.markDirty()
+    this.publishDirtyNow()
  }

  private acceptLive(entry: HistoryEntry): void {
@@ -301,8 +309,16 @@ export class SessionHistorySource implements SessionHistoryFace {
      void this.repairGap()
      return
    }
+    if (
+      entry.event.type === 'assistant/chunk'
+      && entry.event.data.chunk.type !== 'usage'
+    ) {
+      if (!this.appendIncrementalChunk(entry, entry.event)) return
+      this.publishStreamDirty()
+      return
+    }
    this.appendLive(entry)
-    this.notifier.markDirty()
+    this.publishDirtyNow()
  }

  private appendLive(entry: HistoryEntry): void {
@@ -311,6 +327,66 @@ export class SessionHistorySource implements SessionHistoryFace {
    this.entries = [...this.entries, entry]
  }

+  /** Append a chunk against the cached finalized projection; false means no visible publish. */
+  private appendIncrementalChunk(
+    entry: HistoryEntry,
+    event: SessionEvent<'assistant/chunk'>,
+  ): boolean {
+    const { turn, step, chunk } = event.data
+    if (!isVisibleAssistantChunk(chunk.type)) {
+      const inspection = this.currentInspection()
+      this.appendLive(entry)
+      this.inspectionCache = { entries: this.entries, value: inspection }
+      return false
+    }
+    const base = this.streamBaseInspection ?? this.currentInspection()
+    this.streamBaseInspection = base
+    if (
+      this.streamPartial === null
+      || this.streamPartial.turn !== turn
+      || this.streamPartial.step !== step
+    ) {
+      const current = base.partial
+      this.streamPartial = new PartialAccumulator(
+        turn,
+        step,
+        current?.turn === turn && current.step === step ? current.blocks : [],
+      )
+    }
+    this.streamPartial.push(chunk)
+    this.appendLive(entry)
+    this.inspectionCache = {
+      entries: this.entries,
+      value: { ...base, partial: this.streamPartial.toPartial() },
+    }
+    return true
+  }
+
+  /** Coalesce token-stream projection and rendering work to one publish per browser frame. */
+  private publishStreamDirty(): void {
+    if (this.streamPublishToken !== null) return
+    const token = {}
+    this.streamPublishToken = token
+    const publish = () => {
+      if (this.streamPublishToken !== token) return
+      this.streamPublishToken = null
+      this.notifier.markDirty()
+    }
+    if (typeof globalThis.requestAnimationFrame === 'function') {
+      globalThis.requestAnimationFrame(publish)
+    } else {
+      queueMicrotask(publish)
+    }
+  }
+
+  /** Publish structural changes immediately and invalidate an older scheduled stream publish. */
+  private publishDirtyNow(): void {
+    this.streamPublishToken = null
+    this.streamBaseInspection = null
+    this.streamPartial = null
+    this.notifier.markDirty()
+  }
+
  private async repairGap(): Promise<void> {
    if (this.stitching) return
    this.stitching = true
@@ -335,6 +411,16 @@ export class SessionHistorySource implements SessionHistoryFace {
  }

  private buildSnapshot(): SessionHistorySnapshot {
+    return {
+      state: this.state,
+      error: this.error,
+      hasMore: this.hasMore,
+      inspection: this.currentInspection(),
+    }
+  }
+
+  /** Inspection pinned to the source's current immutable entry array. */
+  private currentInspection(): SessionHistorySnapshot['inspection'] {
    if (this.inspectionCache?.entries !== this.entries) {
      const entries = this.entries
      this.inspectionCache = {
@@ -342,11 +428,14 @@ export class SessionHistorySource implements SessionHistoryFace {
        value: createHistoryInspection(() => entries),
      }
    }
-    return {
-      state: this.state,
-      error: this.error,
-      hasMore: this.hasMore,
-      inspection: this.inspectionCache.value,
-    }
+    return this.inspectionCache.value
  }
 }
+
+function isVisibleAssistantChunk(type: string): boolean {
+  return type === 'block-start'
+    || type === 'text-delta'
+    || type === 'reasoning-delta'
+    || type === 'tool-call-delta'
+    || type === 'block-end'
+}
@@ -289,6 +289,40 @@ export class SessionManager {
    }
  }

+  /**
+   * Contract session.fork; on success merge the child into summaries
+   * immediately (same synchronous-addressability guarantee as create). The
+   * child carries the source's history, so it is never blank; lineage rides
+   * parentSessionId so the list nests it under its source. A child published
+   * before Workspace attachment fails is also reconciled into the list.
+   * @param opts - source session and the optional seq anchoring the cut.
+   * @returns the fork result (the child session id).
+   */
+  async fork(
+    opts: { sessionId: SessionId; atSeq?: number },
+  ): Promise<RpcResult<{ sessionId: SessionId }>> {
+    try {
+      const source = this.summaries.find(s => s.sessionId === opts.sessionId)
+      const { result } = await this.api.sessions.fork({
+        sessionId: opts.sessionId,
+        ...opts.atSeq === undefined ? {} : { atSeq: opts.atSeq },
+      })
+      const childId = result.ok
+        ? result.value.sessionId
+        : workspaceAttachSessionId(result.error)
+      if (childId !== undefined) {
+        this.recordMutation({ kind: 'upsert', summary: {
+          sessionId: childId, updatedAt: Date.now(), running: false, blank: false,
+          parentSessionId: opts.sessionId,
+          ...(source?.cwd !== undefined ? { cwd: source.cwd } : {}),
+        } })
+      }
+      return result
+    } catch (error) {
+      return transportError(error)
+    }
+  }
+
  /**
   * Insert-or-enrich a locally synthesized summary: a new id prepends; an
   * existing entry only gains fields it lacks (the session-added frame and the
@@ -13,8 +13,18 @@ export class PartialAccumulator {
  private changed = true
  private snapshot: PartialAssistant

-  constructor(readonly turn: number, readonly step: number) {
-    this.snapshot = { turn, step, blocks: [] }
+  /**
+   * @param turn - Owning agent turn.
+   * @param step - Owning model step.
+   * @param initialBlocks - Materialized prefix when accumulation begins after history replay.
+   */
+  constructor(
+    readonly turn: number,
+    readonly step: number,
+    initialBlocks: readonly AssistantBlock[] = [],
+  ) {
+    this.blocks = [...initialBlocks]
+    this.snapshot = { turn, step, blocks: initialBlocks }
  }

  /**
@@ -81,6 +81,22 @@ export class SessionCreateError extends Error {
  }
 }

+/** Structured session-fork failure. */
+export class SessionForkError extends Error {
+  override readonly name = 'SessionForkError'
+
+  /**
+   * @param rpcError - Host business or folded transport error.
+   * @param sourceSessionId - the session the fork was cut from.
+   */
+  constructor(
+    readonly rpcError: RpcError,
+    readonly sourceSessionId: SessionId,
+  ) {
+    super(`session fork failed: ${rpcError.code}: ${rpcError.message}`)
+  }
+}
+
 /** Session assembly handle for SessionProvider/inject factories (identity-stable per session). */
 export interface SessionBinding {
  readonly sessionId: SessionId
@@ -121,6 +137,24 @@ function displayTitleOf(title: string | undefined, cwd: string | undefined, id:
  return id
 }

+/**
+ * Increment a trailing fork number while preserving its half-width or
+ * full-width parentheses; an unnumbered title starts with ` (1)`.
+ * @param title - source session's durable title.
+ * @returns the title assigned to the fork child.
+ */
+function increasedForkTitle(title: string): string {
+  const ascii = /^(.*?)\((\d+)\)$/u.exec(title)
+  if (ascii?.[1] !== undefined && ascii[2] !== undefined) {
+    return `${ascii[1]}(${BigInt(ascii[2]) + 1n})`
+  }
+  const fullWidth = /^(.*?)（(\d+)）$/u.exec(title)
+  if (fullWidth?.[1] !== undefined && fullWidth[2] !== undefined) {
+    return `${fullWidth[1]}（${BigInt(fullWidth[2]) + 1n}）`
+  }
+  return `${title} (1)`
+}
+
 interface ScopeRecord {
  fiber: Fiber
  ctx: Context
@@ -317,6 +351,42 @@ export class SessionsService implements ISessions {
    return result.value.sessionId
  }

+  /**
+   * Fork a session from a completed-turn prefix of the source (same
+   * synchronous-addressability guarantee as {@link SessionsService.create}:
+   * on resolution the child is in the list store and open() can target it).
+   * @param opts - source session id, the optional event seq anchoring the
+   *   cut (the boundary is the first turn/end at or after it; an in-log
+   *   anchor in an open turn is unavailable rather than clipped backward),
+   *   and whether to increment an inherited durable title before resolving.
+   * @returns the child session id.
+   * @throws {SessionForkError} with the source id.
+   * @throws {Error} when a requested child-title rename fails after creation.
+   */
+  async fork(opts: {
+    sessionId: SessionId
+    atSeq?: number
+    increaseTitle?: boolean
+  }): Promise<SessionId> {
+    const sourceTitle = opts.increaseTitle
+      ? this.list.getSnapshot().byId[opts.sessionId]?.title
+      : undefined
+    const result = await this.manager.fork({
+      sessionId: opts.sessionId,
+      ...(opts.atSeq === undefined ? {} : { atSeq: opts.atSeq }),
+    })
+    if (!result.ok) throw new SessionForkError(result.error, opts.sessionId)
+    this.projectList()
+    const childId = result.value.sessionId
+    if (sourceTitle !== undefined) {
+      const child = this.binding(childId)?.session
+      if (child === undefined) throw new Error(`fork child "${childId}" is not locally addressable`)
+      const renamed = await child.rename(increasedForkTitle(sourceTitle))
+      if (!renamed.ok) throw new Error(`fork child rename failed: ${renamed.error.code}: ${renamed.error.message}`)
+    }
+    return childId
+  }
+
  /**
   * Resolve an Agent-scoped context view (use-and-discard).
   * @param id - session id (the agent identity — 1:1 same axis).