fix(web): restore pending Inbox after restart

This commit is contained in:
_Kerman committed 2026-08-27 12:37:00 +08:00
1 parent 6bfda7a0e5
commit 8c8c5b2680
89 files changed
+523 -979

No files matched your search

@@ -1,20 +1,8 @@
/** Session-owned observable state excluding Conversation target data. */
import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
import type { MessageId } from '@deepseek-ai/dsh-llm/brand'
import type { SessionId } from '@deepseek-ai/dsh-session/types'
import type { SubagentAddress } from '@deepseek-ai/dsh-subagent/client'
import type { ClientFailure } from './result.ts'
/** One transient inbox occurrence from the authoritative queue snapshot. */
export interface QueuedMessage {
readonly id: MessageId
readonly messageId: MessageId
readonly placement: 'queued' | 'steering' | 'context'
readonly content: readonly ContentBlock[]
readonly preview: string
readonly text: string | null
}
/** History-open lifecycle of a Session event window. */
export type OpenState = 'cold' | 'loading' | 'open' | 'error'
@@ -27,7 +15,6 @@ export interface PromptError {
/** Immutable Session lifecycle and control snapshot. */
export interface SessionSnapshot {
readonly sessionId: SessionId
readonly queue: readonly QueuedMessage[]
readonly running: boolean
readonly subagent: {
readonly address: SubagentAddress
@@ -54,7 +54,6 @@ export type {
export type {
OpenState,
PromptError,
QueuedMessage,
SessionSnapshot,
} from './contract/snapshot.ts'
export type { ClientFailure, ClientResult } from './contract/result.ts'
@@ -8,7 +8,6 @@ import type { WorkspaceId } from '@deepseek-ai/dsh-workspace/types'
import type {
SessionControlBaseline,
SessionControlFrame,
SessionQueuedItem,
SessionError,
SessionSummary,
SessionJob as JobView,
@@ -93,8 +92,6 @@ export class SessionManager {
private readonly sessions = new Map<SessionId, Session>()
/** In-flight Session disposals remain here after instances leave `sessions`, so manager disposal can await quiescence. */
private readonly sessionDisposals = new Set<Promise<void>>()
/** Latest transient queues, retained independently of Session object materialization. */
private readonly queues = new Map<SessionId, readonly SessionQueuedItem[]>()
/**
* Sessions that finished running while not selected — the sidebar's green
* "done" reminder (manager-owned, survives connection generations; cleared
@@ -287,11 +284,6 @@ export class SessionManager {
if (session === undefined) {
session = this.createSession(sessionId)
this.sessions.set(sessionId, session)
// Install the latest control baseline before the running-bit sync: a
// not-running summary must sweep replayed queue
// rows the same way a live status flip would (their retirement events dropped
// while the session was uninstantiated).
session.replaceControl(this.queues.get(sessionId) ?? [])
// Sync the running and blank bits from the list snapshot into the new
// instance (consistency when the list precedes open).
const summary = this.summaries.find(s => s.sessionId === sessionId)
@@ -679,22 +671,12 @@ export class SessionManager {
this.notifier.markDirty()
return
}
if (frame.type === 'jobs') {
if (frame.jobs.length === 0) this.jobsBySession.delete(frame.sessionId)
else this.jobsBySession.set(frame.sessionId, frame.jobs)
this.notifier.markDirty()
return
}
this.queues.set(frame.sessionId, frame.items)
this.sessions.get(frame.sessionId)?.handleControlFrame(frame)
if (frame.jobs.length === 0) this.jobsBySession.delete(frame.sessionId)
else this.jobsBySession.set(frame.sessionId, frame.jobs)
this.notifier.markDirty()
}
private replaceControlBaseline(baseline: SessionControlBaseline): void {
this.queues.clear()
for (const [sessionId, items] of Object.entries(baseline.queues)) {
this.queues.set(sessionId as SessionId, items)
}
this.jobsBySession.clear()
for (const [sessionId, jobs] of Object.entries(baseline.jobs)) {
if (jobs.length > 0) this.jobsBySession.set(sessionId as SessionId, jobs)
@@ -705,9 +687,6 @@ export class SessionManager {
store.truncate(block.asOfSeq)
store.seed(block)
}
for (const [sessionId, session] of this.sessions) {
session.replaceControl(this.queues.get(sessionId) ?? [])
}
this.notifier.markDirty()
}
@@ -747,7 +726,6 @@ export class SessionManager {
this.updateCatalogActivity(sessionId, false)
if (durableSubagent) this.sessions.get(sessionId)?.handleRunning(false)
else this.sessions.get(sessionId)?.handleRemoved()
this.queues.delete(sessionId)
this.jobsBySession.delete(sessionId)
if (!durableSubagent) this.projectionStores.delete(sessionId)
const inflightCatalog = this.catalogInflight.get(sessionId)
@@ -1,67 +0,0 @@
import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
import type { SessionQueuedItem } from '../../types.ts'
import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
import type { QueuedMessage } from '../contract/snapshot.ts'
const QUEUE_PREVIEW_CHARS = 200
function previewOf(content: readonly ContentBlock[]): string {
const flat = content
.map(block => (block.type === 'text' ? block.text : `[${block.type}]`))
.join(' ').replace(/\s+/g, ' ').trim()
const chars = Array.from(flat)
return chars.length > QUEUE_PREVIEW_CHARS ? `${chars.slice(0, QUEUE_PREVIEW_CHARS).join('')}…` : flat
}
function textOf(content: readonly ContentBlock[]): string | null {
if (!content.every(block => block.type === 'text')) return null
return content.map(block => block.text).join('')
}
type QueueItems = readonly SessionQueuedItem[]
/** Authoritative transient queue projection and durable steering handoff. */
export class SessionQueueMirror {
private current: readonly QueuedMessage[] = []
/**
* Return the current immutable queue projection.
* @returns current queue rows.
*/
snapshot(): readonly QueuedMessage[] {
return this.current
}
/**
* Replace from one authoritative stream queue frame.
* @param items - complete host queue snapshot.
*/
replace(items: QueueItems): void {
this.current = items.map((item) => {
const content = item.message.content as unknown as readonly ContentBlock[]
return {
id: item.id,
messageId: item.message.id,
placement: item.placement,
content,
preview: previewOf(content),
text: textOf(content),
}
})
}
/**
* Retire a transient steering row once its durable message enters the log.
* @param event - newly contiguous durable Session event.
* @returns whether the projection changed.
*/
acceptDurable(event: SessionEvent): boolean {
if (event.type !== 'user/message') return false
const messageId = event.data.id
const index = this.current.findIndex(item =>
item.placement === 'steering' && item.messageId === messageId)
if (index < 0) return false
this.current = this.current.filter((_item, candidate) => candidate !== index)
return true
}
}
@@ -15,8 +15,6 @@ import type {
PromptContentPart,
QueueAction,
SessionAddress,
SessionControlFrame,
SessionQueuedItem,
SessionRequestId,
SessionError,
} from '../../types.ts'
@@ -36,7 +34,6 @@ import type { SessionRemotes } from './remotes.ts'
import { ProjectionValueStore } from './projection-store.ts'
import type { ProjectionsBaseline } from './projection-store.ts'
import { resolvedClientTimeZone } from '../time-zone.ts'
import { SessionQueueMirror } from './queue-mirror.ts'
/** Messages requested per history page. */
export const PAGE_MESSAGES = 50
@@ -81,8 +78,6 @@ export class Session implements SessionFace {
* passes drop all writes once the generation moves on. */
private openGeneration = 0
private loadingOlder = false
/** Authoritative stream-only inbox snapshot; pending work never hits history. */
private readonly queueMirror = new SessionQueueMirror()
private running = false
private address: SubagentAddress | undefined
private parentAvailable: boolean | undefined
@@ -393,7 +388,7 @@ export class Session implements SessionFace {
}
/** Rebuild an opened history source after address replacement.
* Invalidates any in-flight open first; queue state belongs to the independently
* Invalidates any in-flight open first; projection state belongs to the independently
* reconnecting control stream and remains untouched. */
async resync(): Promise<void> {
if (this.openState === 'cold') return // never opened: no window to rebuild (doOpen flips to 'loading' synchronously, so cold implies no in-flight open)
@@ -431,24 +426,6 @@ export class Session implements SessionFace {
// ---- Manager-only entry points (@internal; never called by the UI) ----
/**
* Replace every transient control value for this Session from one stream baseline.
* @param queue - complete pending queue for this Session.
*/
replaceControl(queue: readonly SessionQueuedItem[]): void {
this.queueMirror.replace(queue)
this.notifier.markDirty()
}
/**
* Apply one Session-addressed live control update.
* @param frame - queue replacement addressed to this Session.
*/
handleControlFrame(frame: Extract<SessionControlFrame, { type: 'queue' }>): void {
this.queueMirror.replace(frame.items)
this.notifier.markDirty()
}
/**
* Running-bit relay from the host stream (list entry and snapshot stay consistent).
* @param running - the new running state.
@@ -599,9 +576,8 @@ export class Session implements SessionFace {
const event = entry.event
const awaitingFirstTurn = this.firstPromptPendingTurn
if (event.type === 'turn/start') this.firstPromptPendingTurn = false
const queueChanged = this.queueMirror.acceptDurable(event)
this.eventSource.append(entry)
return queueChanged || awaitingFirstTurn !== this.firstPromptPendingTurn
return awaitingFirstTurn !== this.firstPromptPendingTurn
}
/** Publish a terminal background failure only while this stream still owns the Session. */
@@ -619,7 +595,6 @@ export class Session implements SessionFace {
private buildSnapshot(): SessionSnapshot {
return {
sessionId: this.sessionId,
queue: this.queueMirror.snapshot(),
running: this.running,
subagent: this.address === undefined
? null