|
|
@@ -12,6 +12,7 @@ import { transportError } from '@deepseek-ai/dsh-host-apiproxy/api'
|
|
|
import { mergeOrderedBaseline } from '../ordered-baseline.ts'
|
|
|
import type { SessionListEntry, TitledSessionSummary } from './lineage.ts'
|
|
|
import { flattenLineage } from './lineage.ts'
|
|
|
+import type { PendingInteractionStatus } from './pending.ts'
|
|
|
// Type-only merge edge: the title domain's client-namespace outlet declares
|
|
|
// the 'title' projection key this manager projects into list rows (and any
|
|
|
// useProjection('title') consumer reads). Zero value imports by construction.
|
|
|
@@ -70,23 +71,30 @@ type SessionListMutation =
|
|
|
/** Local first-send flip: the sender clears blank without waiting for a host frame. */
|
|
|
| { kind: 'engaged'; sessionId: SessionId }
|
|
|
|
|
|
-/** Per-session cap for pre-instantiation approval/question buffering (low-frequency frames; a few dozen covers any real backlog). */
|
|
|
-const PENDING_BUFFER_CAP = 32
|
|
|
+/** Stable identity of a frame retained until an uninstantiated Session can consume it. */
|
|
|
+function bufferedRequestKey(envelope: RpcRequest<MuxFrame>): string | undefined {
|
|
|
+ const frame = envelope.payload
|
|
|
+ switch (frame.type) {
|
|
|
+ case 'approval/requested': return `a:${frame.approvalId}`
|
|
|
+ case 'question/requested': return `q:${envelope.rpcId}`
|
|
|
+ case 'session/queue': return 'queue'
|
|
|
+ default: return undefined
|
|
|
+ }
|
|
|
+}
|
|
|
|
|
|
|
|
|
/** Instance cluster + frame entry + the session list (see the web client architecture RFC). */
|
|
|
export class SessionManager {
|
|
|
private readonly sessions = new Map<SessionId, Session>()
|
|
|
- /** Approval/question frame buffer for uninstantiated sessions: pending interactions never hit
|
|
|
- * history (cannot be backfilled on open), the one frame class that must not take the
|
|
|
- * drop-and-backfill path; replayed and cleared on instantiation. Bounded per session (these
|
|
|
- * frames are low-frequency; overflow drops oldest) and dropped on session-removed (audit S7). */
|
|
|
+ /** Pre-instantiation buffer for answerable requests and the queued-turn snapshot, which history
|
|
|
+ * cannot reconstruct on open. Live requests remain until resolution; queue and replay duplicates
|
|
|
+ * compact by identity. Instantiation replays and clears it, while removal drops it (audit S7). */
|
|
|
private readonly pendingBuffers = new Map<SessionId, RpcRequest<MuxFrame>[]>()
|
|
|
- /** Outstanding approval questions per session, keyed by approvalId (idempotent under mux-open
|
|
|
- * replays of the same requested frame). Manager-owned rather than read off Session instances
|
|
|
- * because the sidebar must light up for sessions never instantiated. Cleared per connection
|
|
|
- * generation — the reopen replay re-adds still-pending questions — and on session-removed. */
|
|
|
- private readonly waitingApprovals = new Map<SessionId, Set<string>>()
|
|
|
+ /** Outstanding answerable interactions per session, keyed by their stable request identity.
|
|
|
+ * Manager-owned rather than read off Session instances because the sidebar must light up for
|
|
|
+ * sessions never instantiated. Cleared per connection generation — the reopen replay re-adds
|
|
|
+ * still-pending requests — and on session-removed. */
|
|
|
+ private readonly pendingInteractions = new Map<SessionId, Map<string, PendingInteractionStatus>>()
|
|
|
/** Per-session projection value stores, retained independently of instance arrival (the
|
|
|
* title-snapshot precedent, generalized): push frames land here whether or not the Session
|
|
|
* is instantiated (list rows read the 'title' key), and an instantiated Session adopts the
|
|
|
@@ -567,6 +575,26 @@ export class SessionManager {
|
|
|
return this.listSnapshotCache
|
|
|
}
|
|
|
|
|
|
+ /** Add or refresh one stable pending-interaction identity. */
|
|
|
+ private trackPending(sessionId: SessionId, key: string, status: PendingInteractionStatus): void {
|
|
|
+ let interactions = this.pendingInteractions.get(sessionId)
|
|
|
+ if (interactions === undefined) {
|
|
|
+ interactions = new Map()
|
|
|
+ this.pendingInteractions.set(sessionId, interactions)
|
|
|
+ }
|
|
|
+ if (interactions.get(key) === status) return
|
|
|
+ interactions.set(key, status)
|
|
|
+ this.notifier.markDirty()
|
|
|
+ }
|
|
|
+
|
|
|
+ /** Settle one pending-interaction identity without disturbing sibling waits. */
|
|
|
+ private resolvePending(sessionId: SessionId, key: string): void {
|
|
|
+ const interactions = this.pendingInteractions.get(sessionId)
|
|
|
+ if (interactions === undefined || !interactions.delete(key)) return
|
|
|
+ if (interactions.size === 0) this.pendingInteractions.delete(sessionId)
|
|
|
+ this.notifier.markDirty()
|
|
|
+ }
|
|
|
+
|
|
|
// ---- ConnectionController sinks (wired by boot) ----
|
|
|
|
|
|
/**
|
|
|
@@ -606,43 +634,56 @@ export class SessionManager {
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
- // List-level waiting-approval bit (the sidebar amber dot): tracked here for
|
|
|
- // every session, instantiated or not; approvalId keys make replays idempotent.
|
|
|
+ // List-level pending-interaction status (the sidebar amber dot): tracked
|
|
|
+ // for every session, instantiated or not; stable keys make replays idempotent.
|
|
|
if (frame.type === 'approval/requested') {
|
|
|
- let ids = this.waitingApprovals.get(frame.sessionId)
|
|
|
- if (ids === undefined) this.waitingApprovals.set(frame.sessionId, ids = new Set())
|
|
|
- if (!ids.has(frame.approvalId)) {
|
|
|
- ids.add(frame.approvalId)
|
|
|
- this.notifier.markDirty()
|
|
|
- }
|
|
|
+ this.trackPending(frame.sessionId, `a:${frame.approvalId}`, 'approval')
|
|
|
} else if (frame.type === 'approval/resolved') {
|
|
|
- const ids = this.waitingApprovals.get(frame.sessionId)
|
|
|
- if (ids !== undefined && ids.delete(frame.approvalId)) {
|
|
|
- if (ids.size === 0) this.waitingApprovals.delete(frame.sessionId)
|
|
|
- this.notifier.markDirty()
|
|
|
- }
|
|
|
+ this.resolvePending(frame.sessionId, `a:${frame.approvalId}`)
|
|
|
+ } else if (frame.type === 'question/requested') {
|
|
|
+ this.trackPending(
|
|
|
+ frame.sessionId,
|
|
|
+ `q:${envelope.rpcId}`,
|
|
|
+ frame.questions.length === 1 && frame.questions[0]?.intent?.kind === 'plan-review'
|
|
|
+ ? 'plan-review'
|
|
|
+ : 'question',
|
|
|
+ )
|
|
|
+ } else if (frame.type === 'question/resolved') {
|
|
|
+ this.resolvePending(frame.sessionId, `q:${frame.questionRpcId}`)
|
|
|
}
|
|
|
const session = this.sessions.get(frame.sessionId)
|
|
|
if (session === undefined) {
|
|
|
- // Approval/question/queue frames never hit history: buffer for replay on
|
|
|
- // instantiation; everything else drops (not instantiated — history fully
|
|
|
- // backfills on open).
|
|
|
+ // Answerable requests never hit history: retain each live identity until
|
|
|
+ // instantiation, compacting replay duplicates and resolutions so list
|
|
|
+ // status cannot outlive the PendingWait the user would need to answer.
|
|
|
+ // Queue is a latest-value snapshot; everything else drops because open
|
|
|
+ // backfills it from history.
|
|
|
switch (frame.type) {
|
|
|
case 'approval/requested':
|
|
|
- case 'approval/resolved':
|
|
|
case 'question/requested':
|
|
|
- case 'question/resolved':
|
|
|
case 'session/queue': {
|
|
|
const buffer = this.pendingBuffers.get(frame.sessionId) ?? []
|
|
|
- const prior = frame.type === 'session/queue'
|
|
|
- ? buffer.findIndex(item => item.payload.type === 'session/queue')
|
|
|
- : -1
|
|
|
- if (prior !== -1) buffer.splice(prior, 1)
|
|
|
- buffer.push(envelope)
|
|
|
- if (buffer.length > PENDING_BUFFER_CAP) buffer.splice(0, buffer.length - PENDING_BUFFER_CAP)
|
|
|
+ const key = frame.type === 'approval/requested'
|
|
|
+ ? `a:${frame.approvalId}`
|
|
|
+ : frame.type === 'question/requested' ? `q:${envelope.rpcId}` : 'queue'
|
|
|
+ const prior = buffer.findIndex(item => bufferedRequestKey(item) === key)
|
|
|
+ if (prior === -1) buffer.push(envelope)
|
|
|
+ else buffer[prior] = envelope
|
|
|
this.pendingBuffers.set(frame.sessionId, buffer)
|
|
|
return
|
|
|
}
|
|
|
+ case 'approval/resolved':
|
|
|
+ case 'question/resolved': {
|
|
|
+ const buffer = this.pendingBuffers.get(frame.sessionId)
|
|
|
+ if (buffer === undefined) return
|
|
|
+ const key = frame.type === 'approval/resolved'
|
|
|
+ ? `a:${frame.approvalId}`
|
|
|
+ : `q:${frame.questionRpcId}`
|
|
|
+ const prior = buffer.findIndex(item => bufferedRequestKey(item) === key)
|
|
|
+ if (prior !== -1) buffer.splice(prior, 1)
|
|
|
+ if (buffer.length === 0) this.pendingBuffers.delete(frame.sessionId)
|
|
|
+ return
|
|
|
+ }
|
|
|
default:
|
|
|
return
|
|
|
}
|
|
|
@@ -689,7 +730,7 @@ export class SessionManager {
|
|
|
this.sessions.get(frame.sessionId)?.handleRemoved()
|
|
|
}
|
|
|
this.pendingBuffers.delete(frame.sessionId) // a removed session's buffered frames must not replay on a future instantiation
|
|
|
- this.waitingApprovals.delete(frame.sessionId) // a removed session cannot wait on anyone
|
|
|
+ this.pendingInteractions.delete(frame.sessionId) // a removed session cannot wait on anyone
|
|
|
if (!durableSubagent) this.projectionStores.delete(frame.sessionId)
|
|
|
// A pull already in flight was requested before this removal and can
|
|
|
// carry the pre-removal parentAvailable:true, which would resurrect
|
|
|
@@ -735,14 +776,14 @@ export class SessionManager {
|
|
|
* The moment a connection generation dies (before any next-generation frame
|
|
|
* can arrive — onConnected waits for the readiness handshake while replayed
|
|
|
* frames flow from stream open, so clearing there would race the replay):
|
|
|
- * drop generation-scoped live state. Approvals resolved while disconnected
|
|
|
- * send no frame, so the stale bits and the buffered answerable frames must
|
|
|
- * not survive into the next generation — the mux-open replay re-adds every
|
|
|
- * still-pending question with its live rpcId.
|
|
|
+ * drop generation-scoped live state. Interactions resolved while disconnected
|
|
|
+ * send no frame, so stale statuses and buffered answerable frames must not
|
|
|
+ * survive into the next generation — mux-open replay re-adds every still-pending
|
|
|
+ * request with its live rpcId.
|
|
|
*/
|
|
|
handleDisconnected(): void {
|
|
|
- if (this.waitingApprovals.size > 0) {
|
|
|
- this.waitingApprovals.clear()
|
|
|
+ if (this.pendingInteractions.size > 0) {
|
|
|
+ this.pendingInteractions.clear()
|
|
|
this.notifier.markDirty()
|
|
|
}
|
|
|
for (const [sessionId, buffer] of [...this.pendingBuffers]) {
|
|
|
@@ -855,7 +896,15 @@ export class SessionManager {
|
|
|
...(projectionValues === undefined ? {} : { projectionValues }),
|
|
|
}
|
|
|
})
|
|
|
- const fresh = flattenLineage(merged, new Set(this.waitingApprovals.keys()))
|
|
|
+ const pendingInteractions = new Map<SessionId, PendingInteractionStatus>()
|
|
|
+ for (const [sessionId, interactions] of this.pendingInteractions) {
|
|
|
+ const statuses = [...interactions.values()]
|
|
|
+ // The composer selects the first question ahead of approval. Mirror that
|
|
|
+ // answer order so the sidebar names the interaction the user can act on.
|
|
|
+ const status = statuses.find(candidate => candidate !== 'approval') ?? statuses[0]
|
|
|
+ if (status !== undefined) pendingInteractions.set(sessionId, status)
|
|
|
+ }
|
|
|
+ const fresh = flattenLineage(merged, pendingInteractions)
|
|
|
const items = fresh.map((entry) => {
|
|
|
const prev = this.entryCache.get(entry.sessionId)
|
|
|
if (
|
|
|
@@ -863,7 +912,7 @@ export class SessionManager {
|
|
|
&& prev.blank === entry.blank
|
|
|
&& prev.parentSessionId === entry.parentSessionId && prev.cwd === entry.cwd
|
|
|
&& prev.origin === entry.origin && prev.title === entry.title && prev.depth === entry.depth
|
|
|
- && prev.waitingApproval === entry.waitingApproval
|
|
|
+ && prev.pendingInteraction === entry.pendingInteraction
|
|
|
&& prev.projectionValues === entry.projectionValues
|
|
|
) return prev
|
|
|
this.entryCache.set(entry.sessionId, entry)
|