DeepSeek/deepseek-harness
1
0
Fork 0
mirror of https://github.com/deepseek-ai/deepseek-harness synced 2026-08-15 21:04:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into web2-todo

Browse Source 
# Conflicts:
#	packages/client/connection/src/client/fixture.ts
#	packages/client/runtime/README.i18n.yaml
#	packages/client/runtime/README.md
#	packages/client/runtime/README.zh.md
#	packages/client/runtime/src/client/index.ts
#	packages/client/runtime/src/client/sessions/session.ts
#	packages/client/ui-conversation/src/client/apply.ts
#	packages/client/ui-conversation/src/client/skeleton/ConversationRoot.tsx
#	packages/client/ui-conversation/tests/chat-stats-bash-sample.spec.tsx
#	packages/client/ui-conversation/tests/chat-toolview-slot.spec.tsx
#	packages/client/ui-conversation/tests/chat-view.spec.tsx
#	packages/client/ui-conversation/tests/gate-branch-tails.spec.tsx
#	packages/client/ui-conversation/tests/skeleton.spec.tsx
#	packages/client/ui-trajectory/tests/views.spec.tsx
This commit is contained in:
imccyu
2026-07-27 10:07:51 +08:00
parent 68d6ed2c9c f12a014481
commit 9d63d75a6f
887 changed files with 28254 additions and 9617 deletions
Download Patch File Download Diff File Expand all files Collapse all files

70
packages/client/runtime/src/client/agents/scope.ts Normal file
View File

@@ -0,0 +1,70 @@
/**
* Client Agent-scope primitive: mint a Cordis context tagged with the owning
* Agent's identity. The mechanism mirrors the host `dsh-scope` architecture
* (no-op plugin fiber + context tag + `Context.filter` routing predicate);
* the shape deliberately diverges: the filter lives on the actx itself
* instead of a separate carrier object, so scoped dispatch is plain cordis —
* `actx.bail(actx, event, payload)` / `actx.emit(actx, ...)` — with no
* wrapper. The host needs a detached carrier because its dispatch subject is
* the business Agent object; client scope events carry only ids, so the
* actx is the natural subject. The second divergence stands: the scope key
* is the branded `SessionId` (value compared), not an object identity — the
* agent and its session share one id (1:1, same axis; no separate AgentId
* brand), and a client scope's identity IS that wire id. Third divergence,
* deliberate: the client scopes the Agent IDENTITY, not a live Agent object
* — a cold session's host Agent is already disposed while its client actx
* stays alive for history viewing.
*/
import { Context as CordisContext } from 'cordis'
import type { Context, Fiber } from 'cordis'
import type { SessionId } from '@deepseek-ai/dsh-client-connection/client'
/** Context tag written by {@link createScope}. */
const kScope = Symbol('dsh.client.scope')
/** A minted Agent scope and its disposal boundary. */
export interface AgentScopeHandle {
/**
* Tagged context: scope-owned registrations and scoped dispatch both go
* through it (passing it as the dispatch subject routes to this agent's
* tagged listeners plus every untagged one).
*/
ctx: Context
/** Backing fiber (dispose tears down every scope-owned registration). */
fiber: Fiber
}
/** Shared no-op plugin backing each Agent scope fiber. */
function agentScope(): void {}
/**
* Mint an Agent scope under `ctx`: a no-op plugin fiber whose context
* carries the agent tag and the dispatch filter — untagged listeners are
* admitted globally, tagged listeners only for a matching agent.
* Registrations through the returned ctx dispose with the fiber.
* @param ctx - client root context the scope fiber mounts under.
* @param key - owning agent identity (the routing tag; agent id === session id).
* @returns the tagged context and its backing fiber.
*/
export function createScope(ctx: Context, key: SessionId): AgentScopeHandle {
const fiber = ctx.plugin(agentScope)
return {
fiber,
ctx: fiber.ctx.extend({
[kScope]: key,
[CordisContext.filter](listenerCtx: Context): boolean {
const tag = scopeOf(listenerCtx)
return tag === undefined || tag === key
},
}),
}
}
/**
* Read the nearest agent tag inherited by a context.
* @param ctx - any client context.
* @returns its agent identity (the session id), or undefined for root contexts.
*/
export function scopeOf(ctx: Context): SessionId | undefined {
return (ctx as Context & { [kScope]?: SessionId })[kScope]
}

42
packages/client/runtime/src/client/index.ts
View File

@@ -1,7 +1,7 @@
/** Browser runtime services for slots, sessions, workspaces, and connection-stream delivery. */
import type { Context } from 'cordis'
import type { ConnectionHandle, SessionId } from '@deepseek-ai/dsh-client-connection/client'
import type { SnapshotSelectorHook } from '@deepseek-ai/dsh-client-ui-slots'
import type { MaybeSnapshotSelectorHook, SnapshotSelectorHook } from '@deepseek-ai/dsh-client-ui-slots'
import { SlotsService } from './slots.ts'
import { SessionsService } from './sessions/service.ts'
import type { SessionListState } from './sessions/service.ts'
@@ -11,10 +11,14 @@ import type { ConversationSnapshot, RunningToolCall, ToolResultNode } from './se
export { SlotsService } from './slots.ts'
export type { RootOwnerProps } from './slots.ts'
export { SessionCreateError, SessionsService, scopeOf, workspaceTitleOf } from './sessions/service.ts'
export { createScope } from './agents/scope.ts'
export type { AgentScopeHandle } from './agents/scope.ts'
export { WorkspacesService } from './workspaces/service.ts'
export type { Session } from './sessions/session.ts'
export type { SessionBinding, SessionListState, SessionSummary } from './sessions/service.ts'
export type { SessionIntentListSnapshot, SessionListPhase } from './sessions/manager.ts'
export type {
SessionBinding, SessionListState, SessionProvideContribution, SessionProvideDescriptor, SessionSummary,
} from './sessions/service.ts'
export type { SessionListPhase } from './sessions/manager.ts'
export type { WorkspaceListPhase } from './workspaces/manager.ts'
export type { WorkspaceListState } from './workspaces/service.ts'
export type { WorkspaceId, WorkspaceView } from '@deepseek-ai/dsh-client-connection/client'
@@ -24,8 +28,8 @@ export type {
EngineStoreHandle, EngineStoreInstance, ObservableSnapshot, SnapshotStore,
} from './contract/store.ts'
export type {
AssistantBlock, AssistantMessageNode, ComposerPhase, ContextMessageNode, ConversationNode,
ConversationSnapshot, PendingPrompt, RunningToolCall, SessionIntentSnapshot, SessionIntentTarget,
AssistantBlock, AssistantMessageNode, CodeSubCall, ComposerPhase, ContextMessageNode, ConversationNode,
ConversationSnapshot, QueuedMessage, RunningToolCall,
SteeringMessageNode, TodoItem, ToolResultNode, UnknownSurfaceNode, UserMessageNode,
} from './sessions/conversation.ts'
export { PendingWait } from './sessions/pending.ts'
@@ -56,6 +60,12 @@ declare module '@deepseek-ai/dsh-client-ui-slots' {
/** The framework-resolved session id (owners never pass it). */
sessionId: SessionId
}
/** Standard kit for slots that remain mounted while current session changes. */
interface SessionMaybeStandardProps {
useSession: MaybeSnapshotSelectorHook<ConversationSnapshot>
/** Current session id; absent in the no-session state. */
sessionId: SessionId | undefined
}
/** Props injected into every global slot component. */
interface GlobalStandardProps {
useSessions: SnapshotSelectorHook<SessionListState>
@@ -72,6 +82,20 @@ declare module 'cordis' {
* @param key - the mutated SlotMap key.
*/
'slots/changed'(key: string): void
/**
* The host command registry changed (host/commands-changed passthrough).
* Pure invalidation signal: subscribers refetch `command.list` in the
* background rather than diffing.
* @mode emit
*/
'commands/changed'(): void
/**
* A connection generation was (re-)established. Wire-derived caches must
* treat their state as stale and repull (commands directory; the queue
* mirrors reset themselves through the session resync path).
* @mode emit
*/
'connection/reset'(): void
}
interface Context {
slots: import('./slots.ts').SlotsService
@@ -91,15 +115,23 @@ export function apply(ctx: Context): void {
const connection = ctx.get('connection') as ConnectionHandle
const sessions = new SessionsService(ctx, connection.api)
const workspaces = new WorkspacesService(ctx, connection.api, sessions)
ctx.effect(
() => workspaces.startInitialSelection(),
'runtime: initial Workspace selection',
)
const loop = connection.start({
onMuxEnvelope: (envelope) => { sessions.handleMuxEnvelope(envelope) },
onHostEnvelope: (envelope) => {
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')
},
onConnected: () => {
sessions.handleConnected()
workspaces.handleConnected()
ctx.emit('connection/reset')
},
})
ctx.effect(() => () => { loop.stop() }, 'runtime: connection stream loop')

71
packages/client/runtime/src/client/sessions/conversation.ts
View File

@@ -6,7 +6,7 @@
import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
import type { TodoItem } from '@deepseek-ai/dsh-session/types'
import type {
RpcError, SessionId, ToolCallView, ToolResultView, WorkspaceId,
RpcError, SessionId, ToolCallView, ToolResultView,
} from '@deepseek-ai/dsh-client-connection/client'
import type { PendingInteraction } from './pending.ts'
@@ -130,6 +130,21 @@ export type ConversationNode =
| ToolResultNode
| UnknownSurfaceNode
/**
* One `run_code` sub-dispatch materialized in the native call-block shapes so
* every consumer (tool rows, details panel) renders it through the exact
* components that render a native call: a started-but-unsettled sub-call is a
* {@link RunningToolCall} (rows derive the running state from the shape,
* exactly as for native calls) and its `tool/code-dispatch` settlement
* replaces it in place with the {@link ToolResultNode} form. Never part of
* the surface `nodes` flow — sub-calls live under their parent via
* {@link ConversationSnapshot.codeDispatches}. `callId` is the deterministic
* sub-call id (`<parent>:code:<n>`); the call side carries the sub-tool name
* and its JSON-stringified logged arguments; `content`/`isError` are the
* settled sub-call's complete logged outcome.
*/
export type CodeSubCall = RunningToolCall | ToolResultNode
/** In-flight tool card material: tool/call seen, tool/result not yet. */
export interface RunningToolCall {
callId: string
@@ -144,6 +159,12 @@ export interface RunningToolCall {
}
/** One queued-message row mirrored from `session/queued` frames (key: the enqueueing prompt's rpcId when wire-sourced). */
export interface QueuedMessage {
readonly key: string
readonly preview: string
}
/** In-progress assistant output (chunk accumulator product). */
export interface PartialAssistant {
turn: number
@@ -182,30 +203,6 @@ export interface PromptError {
error: RpcError
}
/** Workspace target of a frontend-only Session. */
export type SessionIntentTarget =
| { kind: 'workspace'; workspaceId: WorkspaceId }
| { kind: 'workspace-intent' }
/** Publication state owned by a frontend Session before it joins the Host. */
export interface SessionIntentSnapshot {
target: SessionIntentTarget
phase: 'ready' | 'connecting'
error?: { step: 'session'; message: string }
}
/** One editable prompt retained by its Session until the Host accepts it. */
export interface PendingPrompt {
text: string
phase: 'editing' | 'sending' | 'failed'
/** Failed prerequisite retried before sending, or the send itself. */
retry: 'connect' | 'send'
/** Workspace needed when retrying Session attachment. */
workspaceId?: WorkspaceId
/** Last failure diagnostic, absent while editing or sending. */
error?: string
}
/** The immutable snapshot contract Session hands to uSES (see the web client architecture RFC). */
export interface ConversationSnapshot {
sessionId: SessionId
@@ -215,7 +212,16 @@ export interface ConversationSnapshot {
foldDegraded: boolean
partial: PartialAssistant | null
runningCalls: readonly RunningToolCall[]
/**
* `run_code` sub-dispatches grouped under their parent callId, in dispatch
* order. Populated from in-window `tool/code-dispatch` events (live and
* replay identically); the per-parent array reference is stable across
* unrelated snapshot swaps (memo premise, same regime as `nodes`).
*/
codeDispatches: ReadonlyMap<string, readonly CodeSubCall[]>
pending: readonly PendingInteraction[]
/** Read-only inbox mirror (session/queued frames + mux-open baseline; cleared by the leave-running flip). */
queue: readonly QueuedMessage[]
running: boolean
/** Input-area shape (see {@link ComposerPhase}); derived here, switched on by consumers. */
composerPhase: ComposerPhase
@@ -226,10 +232,17 @@ export interface ConversationSnapshot {
hasMore: boolean
loadingOlder: boolean
promptError: PromptError | null
/** Frontend-only publication state; null for a Host-connected Session. */
intent: SessionIntentSnapshot | null
/** Session-owned editable prompt waiting for connection, attachment, or send. */
pendingPrompt: PendingPrompt | null
/**
* Whether this session still has an empty log (no user message yet).
* Mirrors the host summary's derived blank bit: seeded from `session.list`
* / the `host/session-added` frame, flipped false by the first ACCEPTED
* prompt locally (on the RPC success response — acceptance proves the
* user message is in the host log; a rejected first prompt keeps the
* session blank and reusable) and by any `running: true` status remotely,
* and re-aligned by every list re-pull (the summary stays authoritative).
* Blank sessions are hidden from session lists and reused by New Session.
*/
blank: boolean
lastAgentError: string | null
/** Latest `todo/write` whole-list snapshot in the window (last write wins); empty = no plan. */
todos: readonly TodoItem[]

2
packages/client/runtime/src/client/sessions/lineage.ts
View File

@@ -15,6 +15,8 @@ export interface SessionListEntry {
title?: string
updatedAt: number
running: boolean
/** Empty-log bit mirrored from the summary; lists hide blank sessions (filtering stays with the consumer). */
blank: boolean
parentSessionId?: SessionId
cwd?: string
/** Lineage indent depth: root = 0; the UI just multiplies by the indent width. */

207
packages/client/runtime/src/client/sessions/manager.ts
View File

@@ -11,7 +11,6 @@ import type { SessionListEntry, TitledSessionSummary } from './lineage.ts'
import { flattenLineage } from './lineage.ts'
import { Notifier } from './notifier.ts'
import { Session } from './session.ts'
import type { SessionIntentSnapshot, SessionIntentTarget } from './conversation.ts'
/**
* List arrival lifecycle, orthogonal to the pull-activity `state` axis:
@@ -23,19 +22,11 @@ import type { SessionIntentSnapshot, SessionIntentTarget } from './conversation.
*/
export type SessionListPhase = 'pending' | 'ready'
/** Session-owned frontend Intent projected into the global list snapshot. */
export interface SessionIntentListSnapshot extends SessionIntentSnapshot {
sessionId: SessionId
prompt: string
}
/** Immutable session-list snapshot for useSessionList. */
export interface SessionListSnapshot {
items: readonly SessionListEntry[]
/** Selected real or frontend-only Session id. */
/** Selected Session id (validated against items; masked to undefined while its session is off the list). */
current: SessionId | undefined
/** Sole page-local frontend Session projection; its state remains owned by Session. */
intent: SessionIntentListSnapshot | undefined
state: 'idle' | 'loading' | 'error'
/** Arrival lifecycle (see {@link SessionListPhase}); `state` stays the pull-activity axis. */
phase: SessionListPhase
@@ -46,6 +37,8 @@ type SessionListMutation =
| { kind: 'upsert'; summary: SessionSummary }
| { kind: 'remove'; sessionId: SessionId }
| { kind: 'status'; sessionId: SessionId; running: boolean }
/** 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
@@ -76,8 +69,6 @@ export class SessionManager {
private listMutations: SessionListMutation[] | null = null
private selected: SessionId | undefined
private intentSessionId: SessionId | undefined
private stopIntentWatch: (() => void) | undefined
private listSnapshotCache: SessionListSnapshot
/** Entry-identity cache (§C.2 reference stability): list rebuilds reuse the previous entry
@@ -101,87 +92,38 @@ export class SessionManager {
this.listSnapshotCache = this.buildListSnapshot()
}
// ---- Selection and client-local intents ----
// ---- Selection ----
/**
* Select a real Session and discard the unmaterialized intent.
* @param sessionId - listed real Session id.
* Select a listed Session.
* @param sessionId - listed Session id.
*/
select(sessionId: SessionId): void {
if (!this.summaries.some(summary => summary.sessionId === sessionId)) {
throw new Error(`sessions.select: unknown session ${sessionId}`)
}
this.discardIntent()
this.selected = sessionId
this.notifier.notifyNow()
}
/** Clear selection and abandon any frontend-only Session. */
/** Clear the selection (the layout falls to the no-session view state). */
clearSelection(): void {
this.discardIntent()
this.selected = undefined
this.notifier.notifyNow()
}
/**
* Start a frontend Session against a real or still-local Workspace target.
* @param target - real Workspace or the WorkspacesService-owned local target.
* @param prompt - optional prompt retained when retargeting from a picker.
* @returns the frontend Session object that owns the Intent.
*/
startIntent(target: SessionIntentTarget, prompt = ''): Session {
this.discardIntent()
const sessionId = `client-session-${crypto.randomUUID()}` as SessionId
const session = this.createSession(sessionId, { target, prompt })
this.sessions.set(sessionId, session)
this.intentSessionId = sessionId
this.selected = sessionId
this.stopIntentWatch = session.subscribe(() => {
if (this.intentSessionId !== sessionId) return
if (session.getSnapshot().intent === null) {
this.intentSessionId = undefined
this.stopIntentWatch?.()
this.stopIntentWatch = undefined
}
this.notifier.markDirty()
})
this.notifier.notifyNow()
return session
}
/**
* Resolve the active frontend Session Intent.
* @returns the active frontend Session, if one remains selected.
*/
getIntent(): Session | undefined {
return this.intentSessionId === undefined ? undefined : this.sessions.get(this.intentSessionId)
}
/**
* Update the retained prompt of the active frontend Session.
* @param text - exact controlled-input value for the active frontend Session.
*/
updateIntent(text: string): void {
const session = this.getIntent()
if (session === undefined) return
session.updatePendingPrompt(text)
// The intent watch defers via markDirty, but the hero composer reads this
// prompt from the LIST snapshot as a controlled value: it must flush in
// the same tick as onChange (see Notifier.notifyNow) or React rolls the
// textarea back and IME composition breaks.
this.notifier.notifyNow()
}
private discardIntent(): void {
const session = this.getIntent()
this.intentSessionId = undefined
this.stopIntentWatch?.()
this.stopIntentWatch = undefined
session?.abandonIntent()
}
// ---- Instance management ----
/**
* Drop a session instance (scope-prune companion, decision 12: instance
* and scope share one lifecycle). The host session log is the durable
* truth — a later get() lazily rebuilds and open() backfills history.
* @param sessionId - the session to drop.
*/
drop(sessionId: SessionId): void {
this.sessions.delete(sessionId)
}
/**
* Lazy build: return the existing instance or construct one (no auto-open —
* open is triggered by the container's select callback).
@@ -193,31 +135,33 @@ export class SessionManager {
if (session === undefined) {
session = this.createSession(sessionId)
this.sessions.set(sessionId, session)
// Sync the running bit from the list snapshot into the new instance (consistency when the list precedes open).
const summary = this.summaries.find(s => s.sessionId === sessionId)
if (summary !== undefined) session.handleRunning(summary.running)
// Replay approval/question frames buffered before instantiation (rpcId verbatim, same semantics as the subscribed baseline replay).
// Replay approval/question/queued frames buffered before instantiation (rpcId
// verbatim, same semantics as the subscribed baseline replay). Replay happens
// 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).
const buffered = this.pendingBuffers.get(sessionId)
if (buffered !== undefined) {
this.pendingBuffers.delete(sessionId)
for (const envelope of buffered) session.handleMuxEnvelope(envelope.rpcId, envelope.payload)
}
// 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)
if (summary !== undefined) {
session.handleBlank(summary.blank)
session.handleRunning(summary.running)
}
}
return session
}
private createSession(
sessionId: SessionId,
intent?: { target: SessionIntentTarget; prompt: string },
): Session {
private createSession(sessionId: SessionId): Session {
return new Session(sessionId, this.api, {
...(intent === undefined ? {} : { intent }),
onPublished: (published) => {
this.sessions.set(published.sessionId, published)
this.recordMutation({
kind: 'upsert',
summary: { sessionId: published.sessionId, updatedAt: Date.now(), running: false },
})
// The sender's local first-send flip mirrors into the list row so the
// session surfaces (lists filter on blank) before any host frame lands.
onEngaged: (engaged) => {
this.recordMutation({ kind: 'engaged', sessionId: engaged.sessionId })
},
})
}
@@ -244,8 +188,13 @@ export class SessionManager {
this.summaries = summaries
this.listState = 'idle'
this.listPhase = 'ready'
// Push running bits down to instantiated Sessions (the list is the authoritative summary source).
for (const s of this.summaries) this.sessions.get(s.sessionId)?.handleRunning(s.running)
// Push running/blank bits down to instantiated Sessions (the list is the authoritative summary source).
for (const s of this.summaries) {
const session = this.sessions.get(s.sessionId)
if (session === undefined) continue
session.handleBlank(s.blank)
session.handleRunning(s.running)
}
} else {
this.listState = 'error'
this.listError = result.error
@@ -266,7 +215,8 @@ export class SessionManager {
/**
* Contract session.create; on success merge into summaries immediately (no
* wait for the next refresh).
* wait for the next refresh). A created session is blank by definition
* (entity birth precedes the first message).
* @param opts - target workspace or working directory, plus an optional caller-owned id.
* @returns the create result.
*/
@@ -274,16 +224,14 @@ export class SessionManager {
opts: { workspaceId?: WorkspaceId; cwd?: string; sessionId?: SessionId } = {},
): Promise<RpcResult<{ sessionId: SessionId }>> {
try {
const shared = opts.sessionId === undefined ? {} : { sessionId: opts.sessionId }
const payload = opts.workspaceId !== undefined
? { workspaceId: opts.workspaceId, ...(opts.sessionId === undefined ? {} : { sessionId: opts.sessionId }) }
: {
...(opts.cwd === undefined ? {} : { cwd: opts.cwd }),
...(opts.sessionId === undefined ? {} : { sessionId: opts.sessionId }),
}
? { workspaceId: opts.workspaceId, ...shared }
: { ...(opts.cwd === undefined ? {} : { cwd: opts.cwd }), ...shared }
const { result } = await this.api.sessions.create(payload)
if (result.ok) {
this.recordMutation({ kind: 'upsert', summary: {
sessionId: result.value.sessionId, updatedAt: Date.now(), running: false,
sessionId: result.value.sessionId, updatedAt: Date.now(), running: false, blank: true,
...(opts.cwd !== undefined ? { cwd: opts.cwd } : {}),
} })
} else {
@@ -296,6 +244,7 @@ export class SessionManager {
sessionId: publishedSessionId,
updatedAt: Date.now(),
running: false,
blank: true,
} })
}
}
@@ -370,16 +319,31 @@ export class SessionManager {
this.titleSnapshots.delete(frame.sessionId)
this.notifier.markDirty()
}
// New mux-generation baseline: buffered session/queued frames belong to
// the previous generation and the host is about to resend the live
// snapshot — drop them, or every reconnect appends a duplicate batch
// (and enough reconnects push real approval/question frames past the
// cap). Same re-baseline signal Session uses for its own mirror.
const buffered = this.pendingBuffers.get(frame.sessionId)
if (buffered !== undefined) {
const kept = buffered.filter(item => item.payload.type !== 'session/queued')
if (kept.length !== buffered.length) {
if (kept.length === 0) this.pendingBuffers.delete(frame.sessionId)
else this.pendingBuffers.set(frame.sessionId, kept)
}
}
}
const session = this.sessions.get(frame.sessionId)
if (session === undefined) {
// Approval/question frames never hit history: buffer for replay on instantiation;
// everything else drops (not instantiated — history fully backfills on open).
// Approval/question/queued frames never hit history: buffer for replay on
// instantiation; everything else drops (not instantiated — history fully
// backfills on open).
switch (frame.type) {
case 'approval/requested':
case 'approval/resolved':
case 'question/requested':
case 'question/resolved': {
case 'question/resolved':
case 'session/queued': {
const buffer = this.pendingBuffers.get(frame.sessionId) ?? []
buffer.push(envelope)
if (buffer.length > PENDING_BUFFER_CAP) buffer.splice(0, buffer.length - PENDING_BUFFER_CAP)
@@ -402,11 +366,11 @@ export class SessionManager {
switch (frame.type) {
case 'host/session-added': {
this.mergeSummary({
sessionId: frame.sessionId, updatedAt: Date.now(), running: false,
sessionId: frame.sessionId, updatedAt: Date.now(), running: false, blank: frame.blank,
...(frame.parentSessionId !== undefined ? { parentSessionId: frame.parentSessionId } : {}),
...(frame.cwd !== undefined ? { cwd: frame.cwd } : {}),
})
this.sessions.get(frame.sessionId)?.handlePublished()
this.sessions.get(frame.sessionId)?.handleBlank(frame.blank)
return
}
case 'host/session-removed': {
@@ -448,6 +412,7 @@ export class SessionManager {
const prev = this.entryCache.get(entry.sessionId)
if (
prev !== undefined && prev.updatedAt === entry.updatedAt && prev.running === entry.running
&& prev.blank === entry.blank
&& prev.parentSessionId === entry.parentSessionId && prev.cwd === entry.cwd
&& prev.title === entry.title && prev.depth === entry.depth
) return prev
@@ -459,24 +424,13 @@ export class SessionManager {
}
const sameOrder = items.length === this.itemsCache.length && items.every((e, i) => e === this.itemsCache[i])
if (!sameOrder) this.itemsCache = items
const intentSession = this.getIntent()
const intentState = intentSession?.getSnapshot()
const intent = intentSession !== undefined
&& intentState !== undefined && intentState.intent !== null && intentState.pendingPrompt !== null
? {
sessionId: intentSession.sessionId,
...intentState.intent,
prompt: intentState.pendingPrompt.text,
}
: undefined
const selected = this.selected
const current = selected !== undefined && (
intent?.sessionId === selected || items.some(item => item.sessionId === selected)
) ? selected : undefined
const current = selected !== undefined && items.some(item => item.sessionId === selected)
? selected
: undefined
return {
items: this.itemsCache,
current,
intent,
state: this.listState,
phase: this.listPhase,
error: this.listError,
@@ -492,18 +446,29 @@ function applyMutation(summaries: readonly SessionSummary[], mutation: SessionLi
if (existing === undefined) return [mutation.summary, ...summaries]
const filled: SessionSummary = {
...existing,
// Blank only lowers: a stale true (session-added racing the local
// first send) never re-hides an already-surfaced session.
blank: existing.blank && mutation.summary.blank,
...(existing.cwd === undefined && mutation.summary.cwd !== undefined ? { cwd: mutation.summary.cwd } : {}),
...(existing.parentSessionId === undefined && mutation.summary.parentSessionId !== undefined
? { parentSessionId: mutation.summary.parentSessionId } : {}),
}
if (filled.cwd === existing.cwd && filled.parentSessionId === existing.parentSessionId) return [...summaries]
if (filled.cwd === existing.cwd && filled.parentSessionId === existing.parentSessionId
&& filled.blank === existing.blank) return [...summaries]
return summaries.map(summary => summary.sessionId === mutation.summary.sessionId ? filled : summary)
}
case 'remove':
return summaries.filter(summary => summary.sessionId !== mutation.sessionId)
case 'status':
return summaries.map(summary => summary.sessionId === mutation.sessionId && summary.running !== mutation.running
? { ...summary, running: mutation.running }
// running:true doubles as the cross-端 blank flip (a blank session
// never runs, so the first running frame proves a message landed).
return summaries.map(summary => summary.sessionId === mutation.sessionId
&& (summary.running !== mutation.running || (mutation.running && summary.blank))
? { ...summary, running: mutation.running, blank: summary.blank && !mutation.running }
: summary)
case 'engaged':
return summaries.map(summary => summary.sessionId === mutation.sessionId && summary.blank
? { ...summary, blank: false }
: summary)
}
}

25
packages/client/runtime/src/client/sessions/notifier.ts
View File

@@ -3,11 +3,17 @@
// the flush rebuilds the snapshot cache BEFORE notifying (useSyncExternalStore requires a stable
// getSnapshot reference). With no listeners the rebuild is skipped and only the dirty bit is set
// (keeps frame storms cheap); the next getSnapshot rebuilds lazily.
//
// Freshness and notification are SEPARATE bits: a pull (ensureFresh) between
// markDirty and the scheduled flush rebuilds the snapshot but must not
// swallow the notification — push subscribers (object-layer watchers) would
// otherwise starve whenever any reader pulls first.
/** Subscription + microtask-batched notification primitive (shared by Session and SessionManager). */
export class Notifier {
private listeners = new Set<() => void>()
private dirty = false
private notifyPending = false
private scheduled = false
/** @param rebuild - snapshot rebuild function injected by the owner (writes the owner's snapshotCache). */
@@ -28,14 +34,18 @@ export class Notifier {
/** State-change entry: mark dirty and schedule the batched flush. */
markDirty(): void {
this.dirty = true
this.notifyPending = true
if (this.scheduled) return
this.scheduled = true
queueMicrotask(() => {
this.scheduled = false
if (!this.dirty) return
if (this.listeners.size === 0) return // lazy: no subscribers, keep dirty for the next getSnapshot
this.dirty = false
this.rebuild()
if (!this.notifyPending) return
if (this.listeners.size === 0) return // lazy: no subscribers; dirty (if still set) rebuilds on next getSnapshot
this.notifyPending = false
if (this.dirty) {
this.dirty = false
this.rebuild()
}
for (const listener of this.listeners) listener()
})
}
@@ -46,13 +56,18 @@ export class Notifier {
*/
notifyNow(): void {
this.dirty = true
this.notifyPending = true
if (this.listeners.size === 0) return // lazy: same as markDirty, next getSnapshot rebuilds
this.notifyPending = false
this.dirty = false
this.rebuild()
for (const listener of this.listeners) listener()
}
/** Pre-getSnapshot check: rebuild synchronously when dirty (read path before first subscribe / while unobserved). */
/**
* Pre-getSnapshot check: rebuild synchronously when dirty (read path
* before first subscribe / while unobserved). Notification stays pending.
*/
ensureFresh(): void {
if (!this.dirty) return
this.dirty = false

316
packages/client/runtime/src/client/sessions/service.ts
View File

@@ -2,8 +2,9 @@
* SessionsService: root sessions service — list snapshot store (manager
* projection; carries `current`, the persisted selection every
* session-scoped surface keys off — migrated here from ui-layout per the
* slot-parity design), session scope tree (mintScope pattern: no-op plugin
* Fiber + ctx.extend scope tag), stable SessionBinding cache, ancestry walk.
* slot-parity design), Agent scope tree (mintScope pattern: no-op plugin
* Fiber + ctx.extend scope tag; one scope per session, agent id === session
* id), stable SessionBinding cache, ancestry walk.
*
* Scope lifecycle is stage-driven: a scope is minted lazily on first
* resolution (pure — resolution has no side effects and is render-safe);
@@ -16,15 +17,15 @@
*/
import type { Context, Fiber } from 'cordis'
import type { IApiClient, RpcError, SessionId, WorkspaceId } from '@deepseek-ai/dsh-client-connection/client'
import type { SessionCell } from '@deepseek-ai/dsh-client-ui-slots'
import type {
HostObservable, SessionMaybeProvideInfo, SessionProvideInfo,
} from '@deepseek-ai/dsh-client-ui-slots'
import type { SnapshotStore } from '../contract/store.ts'
import { createSnapshotStore } from '../contract/store.ts'
import { createScope, scopeOf as scopeTagOf } from '../agents/scope.ts'
import { SessionManager } from './manager.ts'
import type {
SessionIntentListSnapshot, SessionListPhase,
} from './manager.ts'
import type { SessionListPhase } from './manager.ts'
import type { Session } from './session.ts'
import type { SessionIntentTarget } from './conversation.ts'
/** Session list row projected from the host list RPC plus live stream increments. */
export interface SessionSummary {
@@ -36,6 +37,13 @@ export interface SessionSummary {
cwd?: string
parentId?: SessionId
running: boolean
/**
* Empty-log bit (host summary derivation mirror). New Session reuses a blank
* one targeting the same workspace. Filtering stays with the consumer: the
* store carries every row, while the Workspace browser shows only the
* selected blank entry.
*/
blank: boolean
updatedAt: number
}
@@ -48,17 +56,13 @@ export interface SessionListState {
ids: SessionId[]
byId: Record<SessionId, SessionSummary>
current: SessionId | undefined
/** Frontend Session Intent projected from its owning Session object. */
intent: SessionIntentListSnapshot | undefined
/** Arrival lifecycle projected 1:1 from the manager snapshot (see SessionListPhase): empty-with-ready means "truly no sessions". */
phase: SessionListPhase
}
/** Structured session-create failure preserving partial publication identity. */
/** Structured session-create failure. */
export class SessionCreateError extends Error {
override readonly name = 'SessionCreateError'
/** Definitely published by Host before Workspace attachment failed. */
readonly publishedSessionId: SessionId | undefined
/**
* @param rpcError - Host business or folded transport error.
@@ -69,9 +73,6 @@ export class SessionCreateError extends Error {
readonly requestedSessionId: SessionId | undefined,
) {
super(`session create failed: ${rpcError.code}: ${rpcError.message}`)
this.publishedSessionId = rpcError.code === 'workspace-attach-failed'
? rpcError.details.sessionId
: undefined
}
}
@@ -82,20 +83,10 @@ export interface SessionBinding {
readonly ctx: Context
}
/** Scope tag key (client counterpart of the host dsh-scope pattern). */
const kScope = Symbol('dsh.client.scope')
/**
* Read the session scope tag off a context.
* @param ctx - any client context.
* @returns the session id, or undefined on root contexts.
*/
export function scopeOf(ctx: Context): SessionId | undefined {
return (ctx as Context & { [kScope]?: SessionId })[kScope]
}
/** Shared no-op plugin backing each session scope fiber. */
function sessionScope(): void {}
// Scope primitives live in ../agents/scope.ts (the client mirror of host
// dsh-scope, keyed by Agent identity); re-exported here so existing
// consumers keep their import site.
export { scopeOf } from '../agents/scope.ts'
/**
* Workspace display title of a session cwd: the path's last non-empty
@@ -128,8 +119,30 @@ interface ScopeRecord {
fiber: Fiber
ctx: Context
binding: SessionBinding
/** Render-layer standard kit (identity-stable per scope; the renderer's per-cell caches key off it). */
cell: SessionCell
/** Render-layer standard-props bundle (identity-stable per scope; the renderer's per-info caches key off it). */
provideInfo: SessionProvideInfo
}
/** One plugin's per-session standard-props contribution (see {@link SessionsService.provide}). */
export interface SessionProvideContribution {
/** Bare observable sources, keyed by hook base name ('input' → useInput). */
hooks?: Record<string, HostObservable<unknown>>
/** Stable plain members (action callbacks etc.), spread into standard props verbatim. */
props?: Record<string, unknown>
}
/**
* Static declaration plus per-session resolver for one standard-kit
* contribution. The declared names let the renderer construct the same hook
* and prop surface while no session is current.
*/
export interface SessionProvideDescriptor {
/** Hook base names (`input` becomes `useInput`). */
hooks?: readonly string[]
/** Plain standard-prop names. */
props?: readonly string[]
/** Resolve every declared member for one definite session. */
resolve(binding: SessionBinding): SessionProvideContribution
}
/** Root sessions service: list store, current selection, object-layer manager, scope tree, bindings, ancestry. */
@@ -150,6 +163,10 @@ export class SessionsService {
private readonly selection: SnapshotStore<{ sessionId?: SessionId }>
private readonly scopes = new Map<SessionId, ScopeRecord>()
/** Registered per-session standard-props providers, in registration order. */
private readonly providers: SessionProvideDescriptor[] = []
/** Static no-session projection, rebuilt only when the provider roster changes. */
private maybeInfo: SessionMaybeProvideInfo
/**
* The staged session id — follows `list.current` exactly, holding its last
* defined value across masked gaps (a transiently absent selection blanks
@@ -170,7 +187,7 @@ export class SessionsService {
{ persist: { name: 'dsh.sessions.current' } })
this.manager = new SessionManager(api, this.selection.getSnapshot().sessionId)
this.list = createSnapshotStore<SessionListState>({
ids: [], byId: {}, current: undefined, intent: undefined, phase: 'pending',
ids: [], byId: {}, current: undefined, phase: 'pending',
})
// The manager owns wire truth; the store is its projection. Manager
// notifications are already microtask-batched.
@@ -182,9 +199,97 @@ export class SessionsService {
// the follower writes no list state — session.open()'s synchronous prefix
// touches only session-side state and its own microtask-batched notifier.
this.list.subscribe(() => { this.followCurrent() })
// The runtime's own contribution comes first: useSession rides the same
// provide channel every plugin uses (no renderer special case).
this.providers.push({
hooks: ['session'],
resolve: binding => ({ hooks: { session: binding.session } }),
})
this.maybeInfo = this.materializeMaybeProvideInfo()
rootCtx.reflect.provide('sessions', this, undefined)
}
/**
* Register a per-session standard-props provider: every session-scope slot
* component receives the contributed members as standard props (`hooks`
* sources become `use<Name>` selector hooks on the render side; `props`
* spread verbatim). Contributions materialize lazily with the session's
* scope record and die with it. Registration order is resolution order;
* duplicate member names fail loud at materialization.
* @param descriptor - static member roster plus per-session resolver.
* @returns disposer removing the provider (already-materialized bundles keep their members until their scope drops).
*/
provide(descriptor: SessionProvideDescriptor): () => void {
this.providers.push(descriptor)
// Scopes may already exist (boot order: the list lands and resolves
// scopes before later plugins register) — their bundles must include
// every provider by first render, so re-materialize on roster change.
this.rematerializeProvideBundles()
return () => {
const at = this.providers.indexOf(descriptor)
if (at >= 0) this.providers.splice(at, 1)
this.rematerializeProvideBundles()
}
}
/** Rebuild every live scope's standard-props bundle after a provider roster change. */
private rematerializeProvideBundles(): void {
this.maybeInfo = this.materializeMaybeProvideInfo()
for (const record of this.scopes.values()) {
record.provideInfo = this.materializeProvideInfo(record.binding)
}
}
/** Build the static no-session kit and reject duplicate declared names. */
private materializeMaybeProvideInfo(): SessionMaybeProvideInfo {
const hooks: Record<string, undefined> = {}
const props: Record<string, undefined> = {}
for (const descriptor of this.providers) {
for (const name of descriptor.hooks ?? []) {
if (Object.hasOwn(hooks, name)) throw new Error(`sessions.provide: duplicate hook "${name}"`)
hooks[name] = undefined
}
for (const name of descriptor.props ?? []) {
if (Object.hasOwn(props, name)) throw new Error(`sessions.provide: duplicate prop "${name}"`)
props[name] = undefined
}
}
return { sessionId: undefined, hooks, props }
}
/** Materialize the standard-props bundle for one session (fails loud on duplicate member names). */
private materializeProvideInfo(binding: SessionBinding): SessionProvideInfo {
const hooks: Record<string, HostObservable<unknown>> = {}
const props: Record<string, unknown> = {}
for (const descriptor of this.providers) {
const contribution = descriptor.resolve(binding)
const contributedHooks = contribution.hooks ?? {}
const contributedProps = contribution.props ?? {}
for (const name of Object.keys(contributedHooks)) {
if (!(descriptor.hooks ?? []).includes(name)) {
throw new Error(`sessions.provide: undeclared hook "${name}"`)
}
}
for (const name of Object.keys(contributedProps)) {
if (!(descriptor.props ?? []).includes(name)) {
throw new Error(`sessions.provide: undeclared prop "${name}"`)
}
}
for (const name of descriptor.hooks ?? []) {
const source = contributedHooks[name]
if (source === undefined) throw new Error(`sessions.provide: missing hook "${name}"`)
if (Object.hasOwn(hooks, name)) throw new Error(`sessions.provide: duplicate hook "${name}"`)
hooks[name] = source
}
for (const name of descriptor.props ?? []) {
if (!Object.hasOwn(contributedProps, name)) throw new Error(`sessions.provide: missing prop "${name}"`)
if (Object.hasOwn(props, name)) throw new Error(`sessions.provide: duplicate prop "${name}"`)
props[name] = contributedProps[name]
}
}
return { sessionId: binding.sessionId, hooks, props }
}
/**
* Select a session as current. Unknown ids fail loud instead of navigating
* nowhere.
@@ -205,32 +310,6 @@ export class SessionsService {
this.manager.clearSelection()
}
/**
* Start or retarget the sole client-local Session intent.
* @param target - resolved real or frontend-only Workspace target.
* @param prompt - optional prompt retained across retargeting.
* @returns the frontend Session object that owns the Intent.
*/
startIntent(target: SessionIntentTarget, prompt = ''): Session {
return this.manager.startIntent(target, prompt)
}
/**
* Resolve the active frontend Session Intent.
* @returns the active frontend Session object, if one exists.
*/
intent(): Session | undefined {
return this.manager.getIntent()
}
/**
* Update the retained prompt of the active frontend Session.
* @param text - exact controlled-input value for the current Session Intent.
*/
updateIntent(text: string): void {
this.manager.updateIntent(text)
}
/**
* Refresh the real Session baseline, reusing an in-flight pull.
* @returns completion of the current or newly started baseline pull.
@@ -261,21 +340,26 @@ export class SessionsService {
}
/**
* Create a session on the host.
* Create a session on the host. Resolution guarantee: by the time the
* promise resolves, the created session is in the list store and
* {@link SessionsService.binding} resolves it — callers (New Session
* draft hand-off) may address the scope synchronously, without waiting a
* notifier flush. The synchronous projection below makes this structural
* rather than an accident of microtask ordering.
* @param opts - target workspace or directory and an optional preallocated id.
* @returns the new session id.
* @throws {SessionCreateError} with the requested id and, after an attach
* failure, the definitely published id.
* @throws {SessionCreateError} with the requested id.
*/
async create(opts: { workspaceId?: WorkspaceId; cwd?: string; sessionId?: SessionId } = {}): Promise<SessionId> {
const result = await this.manager.create(opts)
if (!result.ok) throw new SessionCreateError(result.error, opts.sessionId)
this.projectList()
return result.value.sessionId
}
/**
* Resolve a session-scoped context view (use-and-discard).
* @param id - session id.
* Resolve an Agent-scoped context view (use-and-discard).
* @param id - session id (the agent identity — 1:1 same axis).
* @returns scoped ctx, or undefined for a session neither listed nor already scoped.
*/
scope(id: SessionId): Context | undefined {
@@ -283,7 +367,7 @@ export class SessionsService {
}
/**
* Read the session scope tag off a context. Service-method seam: fetch
* Read the Agent scope tag off a context. Service-method seam: fetch
* bundles must reach scope resolution through ctx.sessions — a cross-bundle
* value import of the standalone helper would inline a second module
* instance whose private tag Symbol never matches.
@@ -291,7 +375,22 @@ export class SessionsService {
* @returns the session id, or undefined on root contexts.
*/
scopeOf(ctx: Context): SessionId | undefined {
return scopeOf(ctx)
return scopeTagOf(ctx)
}
/**
* Resolve the business Session behind an Agent-scoped context — the one
* hop every scoped consumer (event listeners, per-session controllers)
* takes from ctx-space into object-space (the client mirror of host
* `agent.session`). Same service-method seam as
* {@link SessionsService.scopeOf}.
* @param ctx - an Agent-scoped context.
* @returns the Session, or undefined when the ctx is untagged or its scope was pruned.
*/
sessionOf(ctx: Context): Session | undefined {
const id = scopeTagOf(ctx)
if (id === undefined) return undefined
return this.scopes.get(id)?.binding.session
}
/**
@@ -305,16 +404,26 @@ export class SessionsService {
}
/**
* Resolve the render-layer session cell (SessionProvider's feed through
* the renderer host; ctx never enters the render layer). Pure resolution —
* render-safe: SessionProvider calls this during render, so no staging, no
* window side effects (StrictMode double-invokes and concurrent discarded
* passes must stay free).
* Resolve the render-layer standard-props bundle (SessionProvider's feed
* through the renderer host; ctx never enters the render layer). Pure
* resolution — render-safe: SessionProvider calls this during render, so no
* staging, no window side effects (StrictMode double-invokes and concurrent
* discarded passes must stay free).
* @param id - session id.
* @returns cell, or undefined for a session neither listed nor already scoped.
* @returns the provide info, or undefined for a session neither listed nor already scoped.
*/
cell(id: string): SessionCell | undefined {
return this.resolve(id as SessionId)?.cell
provideInfo(id: string): SessionProvideInfo | undefined {
return this.resolve(id as SessionId)?.provideInfo
}
/**
* Resolve the current-session-optional standard kit. Unknown or absent ids
* return the static no-session projection rather than removing hook props.
* @param id - current session id, when selected.
* @returns a definite or no-session provide bundle.
*/
maybeProvideInfo(id: string | undefined): SessionMaybeProvideInfo {
return (id === undefined ? undefined : this.provideInfo(id)) ?? this.maybeInfo
}
/**
@@ -360,29 +469,42 @@ export class SessionsService {
return chain
}
/** Lazily mint the scope + binding for a listed (or already-scoped) session. */
/**
* Lazily mint the scope + binding for an eligible session. Eligibility and
* prune share one predicate (decision 12): listed on the host — a scope is
* born when its session enters the client's view (list mirror row from the
* baseline pull, a create() echo, or the session-added frame) and dies with
* the prune when the row leaves.
*/
private resolve(id: SessionId): ScopeRecord | undefined {
const existing = this.scopes.get(id)
if (existing !== undefined) return existing
// Frozen scopes outlive the list; new scopes are only minted for listed sessions.
if (this.list.getSnapshot().byId[id] === undefined) return undefined
const fiber = this.rootCtx.plugin(sessionScope)
const ctx = fiber.ctx.extend({ [kScope]: id })
if (!this.eligible(id)) return undefined
const { fiber, ctx } = createScope(this.rootCtx, id)
const session = this.manager.get(id)
// The Session owns its scoped dispatch point (host Agent.loopCtx mirror);
// mint and bind are one step so a live scope record implies a bound actx.
session.bindScope(ctx)
const binding: SessionBinding = { sessionId: id, session, ctx }
const record: ScopeRecord = {
fiber,
ctx,
binding: { sessionId: id, session, ctx },
// Session is the observable; React binds a selector hook at its own seam.
cell: { sessionId: id, session },
binding,
// Sources are bare observables; React binds selector hooks at its own seam.
provideInfo: this.materializeProvideInfo(binding),
}
this.scopes.set(id, record)
return record
}
/** The one aliveness predicate shared by scope mint and prune: host-listed. */
private eligible(id: SessionId): boolean {
return this.list.getSnapshot().byId[id] !== undefined
}
/** Project the manager's list snapshot into the store (title derivation is display-only). */
private projectList(): void {
const { items, current, intent, phase } = this.manager.getListSnapshot()
const { items, current, phase } = this.manager.getListSnapshot()
const ids: SessionId[] = []
const byId: Record<SessionId, SessionSummary> = {}
for (const entry of items) {
@@ -391,6 +513,7 @@ export class SessionsService {
id: entry.sessionId,
displayTitle: displayTitleOf(entry.title, entry.cwd, entry.sessionId),
running: entry.running,
blank: entry.blank,
updatedAt: entry.updatedAt,
...(entry.title !== undefined ? { title: entry.title } : {}),
...(entry.cwd !== undefined ? { cwd: entry.cwd } : {}),
@@ -398,19 +521,22 @@ export class SessionsService {
}
}
const persisted = this.selection.getSnapshot().sessionId
if (intent?.sessionId === current) {
// No current (cleared, or masked gap) wipes the persisted cell — a reload
// stays on empty; the in-memory selection still resurfaces a masked id.
if (current === undefined) {
if (persisted !== undefined) this.selection.set({})
} else if (current !== undefined && byId[current] !== undefined && persisted !== current) {
} else if (byId[current] !== undefined && persisted !== current) {
this.selection.set({ sessionId: current })
}
this.list.set({ ids, byId, current, intent, phase })
this.list.set({ ids, byId, current, phase })
this.pruneScopes(byId)
}
/** Tear down scopes for removed sessions off stage; the staged one defers until the stage moves. */
/** Tear down scope + instance for no-longer-eligible sessions off stage; the staged one defers until the stage moves. */
private pruneScopes(byId: Record<SessionId, SessionSummary>): void {
void byId
for (const [id, record] of this.scopes) {
if (byId[id] !== undefined) continue
if (this.eligible(id)) continue
if (id === this.watched) {
this.deferredRemovals.add(id)
continue
@@ -421,12 +547,22 @@ export class SessionsService {
}
}
/** Dispose a scope fiber and its session-keyed slot-store instances together (single lifecycle axis). */
/**
* One teardown for the whole per-session axis (decision 12): the scope
* fiber (cascading every actx-registered effect: input shell, slash
* controller, popup, plugin stores, listeners), the session-keyed slot
* stores, and the Session instance itself — the host session log is the
* durable truth, a reopen lazily rebuilds and backfills via open().
*/
private dropScope(id: SessionId, record: ScopeRecord): void {
void record.fiber.dispose()
// Release the Session's dispatch point with the scope it belongs to (a
// surviving instance — the live Intent — rebinds when resolve re-mints).
record.binding.session.unbindScope()
// Optional lookup: slots and sessions are sibling services with no
// declared dependency; a slots-less boot (object-layer tests) skips.
this.rootCtx.get('slots')?.pruneStoreScope(id)
this.manager.drop(id)
}
/** Run deferred teardowns whose session is no longer staged (called when the stage moves). */
@@ -436,8 +572,8 @@ export class SessionsService {
* stage move sweeps first, so the set cannot contain the id the stage just
* moved to; kept as a guard against future extra sweep call sites. */
if (id === this.watched) continue
// Still absent from the list? (A re-added id cancels the deferred teardown.)
if (this.list.getSnapshot().byId[id] !== undefined) {
// Eligible again? (A re-added id cancels the deferred teardown.)
if (this.eligible(id)) {
this.deferredRemovals.delete(id)
continue
}

590
packages/client/runtime/src/client/sessions/service.ts.orig Normal file
View File

@@ -0,0 +1,590 @@
/**
* SessionsService: root sessions service — list snapshot store (manager
* projection; carries `current`, the persisted selection every
* session-scoped surface keys off — migrated here from ui-layout per the
* slot-parity design), Agent scope tree (mintScope pattern: no-op plugin
* Fiber + ctx.extend scope tag; one scope per session, agent id === session
* id), stable SessionBinding cache, ancestry walk.
*
* Scope lifecycle is stage-driven: a scope is minted lazily on first
* resolution (pure — resolution has no side effects and is render-safe);
* the event window and deferred teardown key off the STAGED session, which
* follows `list.current` exactly. Staging is the open signal: the window
* opens ⟺ the session is on stage (today the stage is `current`; the staged
* state can widen to a multi-pane list later). A session leaving the list
* tears its scope down immediately unless it is the staged one, whose scope
* survives frozen (read-only view) until the stage moves on.
*/
import type { Context, Fiber } from 'cordis'
import type { IApiClient, RpcError, SessionId, WorkspaceId } from '@deepseek-ai/dsh-client-connection/client'
import type {
HostObservable, SessionMaybeProvideInfo, SessionProvideInfo,
} from '@deepseek-ai/dsh-client-ui-slots'
import type { SnapshotStore } from '../contract/store.ts'
import { createSnapshotStore } from '../contract/store.ts'
import { createScope, scopeOf as scopeTagOf } from '../agents/scope.ts'
import { SessionManager } from './manager.ts'
import type { SessionListPhase } from './manager.ts'
import type { Session } from './session.ts'
/** Session list row projected from the host list RPC plus live stream increments. */
export interface SessionSummary {
id: SessionId
/** Latest durable log-backed title, absent until the host projects one. */
title?: string
/** Human-facing label: durable title, project basename, then session id. */
displayTitle: string
cwd?: string
parentId?: SessionId
running: boolean
/**
* Empty-log bit (host summary derivation mirror). List surfaces hide blank
* sessions; New Session reuses a blank one targeting the same workspace.
* Filtering stays with the consumer — the store carries every row.
*/
blank: boolean
updatedAt: number
}
/**
* Session list store shape. `current` rides the same snapshot (arbitrated:
* the single useSessions standard hook reads list and selection together —
* sidebar highlighting and SessionProvider share one fact source).
*/
export interface SessionListState {
ids: SessionId[]
byId: Record<SessionId, SessionSummary>
current: SessionId | undefined
/** Arrival lifecycle projected 1:1 from the manager snapshot (see SessionListPhase): empty-with-ready means "truly no sessions". */
phase: SessionListPhase
}
/** Structured session-create failure. */
export class SessionCreateError extends Error {
override readonly name = 'SessionCreateError'
/**
* @param rpcError - Host business or folded transport error.
* @param requestedSessionId - caller-preallocated id used for later stream/list reconciliation.
*/
constructor(
readonly rpcError: RpcError,
readonly requestedSessionId: SessionId | undefined,
) {
super(`session create failed: ${rpcError.code}: ${rpcError.message}`)
}
}
/** Session assembly handle for SessionProvider/inject factories (identity-stable per session). */
export interface SessionBinding {
readonly sessionId: SessionId
readonly session: Session
readonly ctx: Context
}
// Scope primitives live in ../agents/scope.ts (the client mirror of host
// dsh-scope, keyed by Agent identity); re-exported here so existing
// consumers keep their import site.
export { scopeOf } from '../agents/scope.ts'
/**
* Workspace display title of a session cwd: the path's last non-empty
* segment (both separators accepted; trailing separators ignored), or ''
* for separator-only paths — callers own their fallback (session id, raw
* cwd, default-directory copy). The repo-wide single basename derivation —
* every surface naming a workspace (picker rows, toggle labels, list titles)
* calls this instead of re-splitting paths.
* @param cwd - workspace directory path.
* @returns basename title, or '' when no non-empty segment exists.
*/
export function workspaceTitleOf(cwd: string): string {
return cwd.replace(/[/\\]+$/, '').split(/[/\\]/).pop() ?? ''
}
/**
* Display title projection: durable title, project directory basename, then
* the raw id.
*/
function displayTitleOf(title: string | undefined, cwd: string | undefined, id: SessionId): string {
if (title !== undefined) return title
if (cwd !== undefined && cwd !== '') {
const base = workspaceTitleOf(cwd)
if (base !== '') return base
}
return id
}
interface ScopeRecord {
fiber: Fiber
ctx: Context
binding: SessionBinding
/** Render-layer standard-props bundle (identity-stable per scope; the renderer's per-info caches key off it). */
provideInfo: SessionProvideInfo
}
/** One plugin's per-session standard-props contribution (see {@link SessionsService.provide}). */
export interface SessionProvideContribution {
/** Bare observable sources, keyed by hook base name ('input' → useInput). */
hooks?: Record<string, HostObservable<unknown>>
/** Stable plain members (action callbacks etc.), spread into standard props verbatim. */
props?: Record<string, unknown>
}
/**
* Static declaration plus per-session resolver for one standard-kit
* contribution. The declared names let the renderer construct the same hook
* and prop surface while no session is current.
*/
export interface SessionProvideDescriptor {
/** Hook base names (`input` becomes `useInput`). */
hooks?: readonly string[]
/** Plain standard-prop names. */
props?: readonly string[]
/** Resolve every declared member for one definite session. */
resolve(binding: SessionBinding): SessionProvideContribution
}
/** Root sessions service: list store, current selection, object-layer manager, scope tree, bindings, ancestry. */
export class SessionsService {
/** List snapshot store (list RPC + host stream increments; re-pulled on reconnect) — the useSessions standard feed, current included. */
readonly list: SnapshotStore<SessionListState>
/** The object-layer instance cluster and frame dispatch entry. */
private readonly manager: SessionManager
/**
* Persisted selection cell (the durable half of `list.current`). Private on
* purpose: reads go through the list snapshot; writes through {@link
* SessionsService.open} / {@link SessionsService.clear}. Projection
* validates it against the live list instead of destructively pruning, so a
* selection survives transient list states (reconnect re-pull) and
* resurfaces when its session returns.
*/
private readonly selection: SnapshotStore<{ sessionId?: SessionId }>
private readonly scopes = new Map<SessionId, ScopeRecord>()
/** Registered per-session standard-props providers, in registration order. */
private readonly providers: SessionProvideDescriptor[] = []
/** Static no-session projection, rebuilt only when the provider roster changes. */
private maybeInfo: SessionMaybeProvideInfo
/**
* The staged session id — follows `list.current` exactly, holding its last
* defined value across masked gaps (a transiently absent selection blanks
* `current` without moving the stage, so reconnect re-pulls and removals
* keep the staged scope's frozen view alive until the stage moves on).
*/
private watched: SessionId | undefined
/** Removed-while-staged sessions whose teardown waits for the stage to move away. */
private readonly deferredRemovals = new Set<SessionId>()
/**
* @param ctx - client root context (scope fibers mount under it).
* @param api - wire client shared with every Session.
*/
constructor(private readonly rootCtx: Context, api: IApiClient) {
this.selection = createSnapshotStore<{ sessionId?: SessionId }>(
{},
{ persist: { name: 'dsh.sessions.current' } })
this.manager = new SessionManager(api, this.selection.getSnapshot().sessionId)
this.list = createSnapshotStore<SessionListState>({
ids: [], byId: {}, current: undefined, phase: 'pending',
})
// The manager owns wire truth; the store is its projection. Manager
// notifications are already microtask-batched.
this.manager.subscribe(() => { this.projectList() })
// Stage follower: every current write (open() and projection alike)
// re-evaluates staging, so startup restore (persisted selection validated
// by the projection) and reconnect resurfacing open their window with no
// dedicated code path. Safe to run synchronously inside the store notify:
// the follower writes no list state — session.open()'s synchronous prefix
// touches only session-side state and its own microtask-batched notifier.
this.list.subscribe(() => { this.followCurrent() })
// The runtime's own contribution comes first: useSession rides the same
// provide channel every plugin uses (no renderer special case).
this.providers.push({
hooks: ['session'],
resolve: binding => ({ hooks: { session: binding.session } }),
})
this.maybeInfo = this.materializeMaybeProvideInfo()
rootCtx.reflect.provide('sessions', this, undefined)
}
/**
* Register a per-session standard-props provider: every session-scope slot
* component receives the contributed members as standard props (`hooks`
* sources become `use<Name>` selector hooks on the render side; `props`
* spread verbatim). Contributions materialize lazily with the session's
* scope record and die with it. Registration order is resolution order;
* duplicate member names fail loud at materialization.
* @param descriptor - static member roster plus per-session resolver.
* @returns disposer removing the provider (already-materialized bundles keep their members until their scope drops).
*/
provide(descriptor: SessionProvideDescriptor): () => void {
this.providers.push(descriptor)
// Scopes may already exist (boot order: the list lands and resolves
// scopes before later plugins register) — their bundles must include
// every provider by first render, so re-materialize on roster change.
this.rematerializeProvideBundles()
return () => {
const at = this.providers.indexOf(descriptor)
if (at >= 0) this.providers.splice(at, 1)
this.rematerializeProvideBundles()
}
}
/** Rebuild every live scope's standard-props bundle after a provider roster change. */
private rematerializeProvideBundles(): void {
this.maybeInfo = this.materializeMaybeProvideInfo()
for (const record of this.scopes.values()) {
record.provideInfo = this.materializeProvideInfo(record.binding)
}
}
/** Build the static no-session kit and reject duplicate declared names. */
private materializeMaybeProvideInfo(): SessionMaybeProvideInfo {
const hooks: Record<string, undefined> = {}
const props: Record<string, undefined> = {}
for (const descriptor of this.providers) {
for (const name of descriptor.hooks ?? []) {
if (Object.hasOwn(hooks, name)) throw new Error(`sessions.provide: duplicate hook "${name}"`)
hooks[name] = undefined
}
for (const name of descriptor.props ?? []) {
if (Object.hasOwn(props, name)) throw new Error(`sessions.provide: duplicate prop "${name}"`)
props[name] = undefined
}
}
return { sessionId: undefined, hooks, props }
}
/** Materialize the standard-props bundle for one session (fails loud on duplicate member names). */
private materializeProvideInfo(binding: SessionBinding): SessionProvideInfo {
const hooks: Record<string, HostObservable<unknown>> = {}
const props: Record<string, unknown> = {}
for (const descriptor of this.providers) {
const contribution = descriptor.resolve(binding)
const contributedHooks = contribution.hooks ?? {}
const contributedProps = contribution.props ?? {}
for (const name of Object.keys(contributedHooks)) {
if (!(descriptor.hooks ?? []).includes(name)) {
throw new Error(`sessions.provide: undeclared hook "${name}"`)
}
}
for (const name of Object.keys(contributedProps)) {
if (!(descriptor.props ?? []).includes(name)) {
throw new Error(`sessions.provide: undeclared prop "${name}"`)
}
}
for (const name of descriptor.hooks ?? []) {
const source = contributedHooks[name]
if (source === undefined) throw new Error(`sessions.provide: missing hook "${name}"`)
if (Object.hasOwn(hooks, name)) throw new Error(`sessions.provide: duplicate hook "${name}"`)
hooks[name] = source
}
for (const name of descriptor.props ?? []) {
if (!Object.hasOwn(contributedProps, name)) throw new Error(`sessions.provide: missing prop "${name}"`)
if (Object.hasOwn(props, name)) throw new Error(`sessions.provide: duplicate prop "${name}"`)
props[name] = contributedProps[name]
}
}
return { sessionId: binding.sessionId, hooks, props }
}
/**
* Select a session as current. Unknown ids fail loud instead of navigating
* nowhere.
* @param id - session id (must exist in the list store).
*/
open(id: SessionId): void {
this.manager.select(id)
}
/**
* Clear the current selection so the layout shows the no-session empty
* state (new-session affordance and the workspace preselection flow).
* Wipes the persisted selection too — a reload stays on empty until the
* user opens or starts a session. The staged scope keeps its frozen view
* per the masked-gap contract until the next open() moves the stage.
*/
clear(): void {
this.manager.clearSelection()
}
/**
* Refresh the real Session baseline, reusing an in-flight pull.
* @returns completion of the current or newly started baseline pull.
*/
refresh(): Promise<void> {
return this.manager.refreshList()
}
/**
* Route a mux stream envelope into the Session object layer.
* @param envelope - validated mux stream envelope.
*/
handleMuxEnvelope(envelope: Parameters<SessionManager['handleMuxEnvelope']>[0]): void {
this.manager.handleMuxEnvelope(envelope)
}
/**
* Route a Host stream envelope into the Session object layer.
* @param envelope - validated Host stream envelope.
*/
handleHostEnvelope(envelope: Parameters<SessionManager['handleHostEnvelope']>[0]): void {
this.manager.handleHostEnvelope(envelope)
}
/** Rebuild the Session baseline and every opened window after connection. */
handleConnected(): void {
this.manager.handleConnected()
}
/**
* Create a session on the host. Resolution guarantee: by the time the
* promise resolves, the created session is in the list store and
* {@link SessionsService.binding} resolves it — callers (New Session
* draft hand-off) may address the scope synchronously, without waiting a
* notifier flush. The synchronous projection below makes this structural
* rather than an accident of microtask ordering.
* @param opts - target workspace or directory and an optional preallocated id.
* @returns the new session id.
* @throws {SessionCreateError} with the requested id.
*/
async create(opts: { workspaceId?: WorkspaceId; cwd?: string; sessionId?: SessionId } = {}): Promise<SessionId> {
const result = await this.manager.create(opts)
if (!result.ok) throw new SessionCreateError(result.error, opts.sessionId)
this.projectList()
return result.value.sessionId
}
/**
* Resolve an Agent-scoped context view (use-and-discard).
* @param id - session id (the agent identity — 1:1 same axis).
* @returns scoped ctx, or undefined for a session neither listed nor already scoped.
*/
scope(id: SessionId): Context | undefined {
return this.resolve(id)?.ctx
}
/**
* Read the Agent scope tag off a context. Service-method seam: fetch
* bundles must reach scope resolution through ctx.sessions — a cross-bundle
* value import of the standalone helper would inline a second module
* instance whose private tag Symbol never matches.
* @param ctx - any client context.
* @returns the session id, or undefined on root contexts.
*/
scopeOf(ctx: Context): SessionId | undefined {
return scopeTagOf(ctx)
}
/**
* Resolve the business Session behind an Agent-scoped context — the one
* hop every scoped consumer (event listeners, per-session controllers)
* takes from ctx-space into object-space (the client mirror of host
* `agent.session`). Same service-method seam as
* {@link SessionsService.scopeOf}.
* @param ctx - an Agent-scoped context.
* @returns the Session, or undefined when the ctx is untagged or its scope was pruned.
*/
sessionOf(ctx: Context): Session | undefined {
const id = scopeTagOf(ctx)
if (id === undefined) return undefined
return this.scopes.get(id)?.binding.session
}
/**
* Resolve the stable session binding (scope-addressed assembly feed). Pure
* resolution — no staging, no window side effects.
* @param id - session id.
* @returns binding, or undefined for a session neither listed nor already scoped.
*/
binding(id: SessionId): SessionBinding | undefined {
return this.resolve(id)?.binding
}
/**
* Resolve the render-layer standard-props bundle (SessionProvider's feed
* through the renderer host; ctx never enters the render layer). Pure
* resolution — render-safe: SessionProvider calls this during render, so no
* staging, no window side effects (StrictMode double-invokes and concurrent
* discarded passes must stay free).
* @param id - session id.
* @returns the provide info, or undefined for a session neither listed nor already scoped.
*/
provideInfo(id: string): SessionProvideInfo | undefined {
return this.resolve(id as SessionId)?.provideInfo
}
/**
* Resolve the current-session-optional standard kit. Unknown or absent ids
* return the static no-session projection rather than removing hook props.
* @param id - current session id, when selected.
* @returns a definite or no-session provide bundle.
*/
maybeProvideInfo(id: string | undefined): SessionMaybeProvideInfo {
return (id === undefined ? undefined : this.provideInfo(id)) ?? this.maybeInfo
}
/**
* Move the stage to the list's current session: sweep teardowns deferred
* behind the previous occupant and pull the new occupant's history window.
* Staging IS the open signal — the window opens ⟺ the session is on stage
* — and open() is idempotent (an in-flight or completed open no-ops; a
* failed one retries the next time current is touched).
*/
private followCurrent(): void {
const snapshot = this.list.getSnapshot()
const current = snapshot.current
// A masked gap (current blanked while the selection's session is
// transiently absent) holds the stage: tearing down on the gap would
// destroy exactly the frozen scope the mask exists to preserve.
if (current === undefined || snapshot.byId[current] === undefined || current === this.watched) return
this.watched = current
this.sweepDeferred()
const record = this.resolve(current)
/* v8 ignore next 3 -- defensive: current is always a listed id (open()
* validates and the projection masks absent selections), so resolve
* cannot miss; kept so a future current writer cannot crash the notify. */
if (record !== undefined) {
void record.binding.session.open()
}
}
/**
* Breadcrumb feed: walk parentId links inside the list store.
* @param id - session id.
* @returns summaries from root ancestor to the session itself (empty when unknown; a broken link stops the walk).
*/
ancestry(id: SessionId): SessionSummary[] {
const { byId } = this.list.getSnapshot()
const chain: SessionSummary[] = []
let cursor: SessionId | undefined = id
while (cursor !== undefined) {
const summary: SessionSummary | undefined = byId[cursor]
if (summary === undefined || chain.includes(summary)) break
chain.unshift(summary)
cursor = summary.parentId
}
return chain
}
/**
* Lazily mint the scope + binding for an eligible session. Eligibility and
* prune share one predicate (decision 12): listed on the host — a scope is
* born when its session enters the client's view (list mirror row from the
* baseline pull, a create() echo, or the session-added frame) and dies with
* the prune when the row leaves.
*/
private resolve(id: SessionId): ScopeRecord | undefined {
const existing = this.scopes.get(id)
if (existing !== undefined) return existing
if (!this.eligible(id)) return undefined
const { fiber, ctx } = createScope(this.rootCtx, id)
const session = this.manager.get(id)
// The Session owns its scoped dispatch point (host Agent.loopCtx mirror);
// mint and bind are one step so a live scope record implies a bound actx.
session.bindScope(ctx)
const binding: SessionBinding = { sessionId: id, session, ctx }
const record: ScopeRecord = {
fiber,
ctx,
binding,
// Sources are bare observables; React binds selector hooks at its own seam.
provideInfo: this.materializeProvideInfo(binding),
}
this.scopes.set(id, record)
return record
}
/** The one aliveness predicate shared by scope mint and prune: host-listed. */
private eligible(id: SessionId): boolean {
return this.list.getSnapshot().byId[id] !== undefined
}
/** Project the manager's list snapshot into the store (title derivation is display-only). */
private projectList(): void {
const { items, current, phase } = this.manager.getListSnapshot()
const ids: SessionId[] = []
const byId: Record<SessionId, SessionSummary> = {}
for (const entry of items) {
ids.push(entry.sessionId)
byId[entry.sessionId] = {
id: entry.sessionId,
displayTitle: displayTitleOf(entry.title, entry.cwd, entry.sessionId),
running: entry.running,
blank: entry.blank,
updatedAt: entry.updatedAt,
...(entry.title !== undefined ? { title: entry.title } : {}),
...(entry.cwd !== undefined ? { cwd: entry.cwd } : {}),
...(entry.parentSessionId !== undefined ? { parentId: entry.parentSessionId } : {}),
}
}
const persisted = this.selection.getSnapshot().sessionId
// No current (cleared, or masked gap) wipes the persisted cell — a reload
// stays on empty; the in-memory selection still resurfaces a masked id.
if (current === undefined) {
if (persisted !== undefined) this.selection.set({})
} else if (byId[current] !== undefined && persisted !== current) {
this.selection.set({ sessionId: current })
}
this.list.set({ ids, byId, current, phase })
this.pruneScopes(byId)
}
/** Tear down scope + instance for no-longer-eligible sessions off stage; the staged one defers until the stage moves. */
private pruneScopes(byId: Record<SessionId, SessionSummary>): void {
void byId
for (const [id, record] of this.scopes) {
if (this.eligible(id)) continue
if (id === this.watched) {
this.deferredRemovals.add(id)
continue
}
this.scopes.delete(id)
this.deferredRemovals.delete(id)
this.dropScope(id, record)
}
}
/**
* One teardown for the whole per-session axis (decision 12): the scope
* fiber (cascading every actx-registered effect: input shell, slash
* controller, popup, plugin stores, listeners), the session-keyed slot
* stores, and the Session instance itself — the host session log is the
* durable truth, a reopen lazily rebuilds and backfills via open().
*/
private dropScope(id: SessionId, record: ScopeRecord): void {
void record.fiber.dispose()
// Release the Session's dispatch point with the scope it belongs to (a
// surviving instance — the live Intent — rebinds when resolve re-mints).
record.binding.session.unbindScope()
// Optional lookup: slots and sessions are sibling services with no
// declared dependency; a slots-less boot (object-layer tests) skips.
this.rootCtx.get('slots')?.pruneStoreScope(id)
this.manager.drop(id)
}
/** Run deferred teardowns whose session is no longer staged (called when the stage moves). */
private sweepDeferred(): void {
for (const id of [...this.deferredRemovals]) {
/* v8 ignore next -- defensive: only the staged id ever defers, and every
* stage move sweeps first, so the set cannot contain the id the stage just
* moved to; kept as a guard against future extra sweep call sites. */
if (id === this.watched) continue
// Eligible again? (A re-added id cancels the deferred teardown.)
if (this.eligible(id)) {
this.deferredRemovals.delete(id)
continue
}
const record = this.scopes.get(id)
this.deferredRemovals.delete(id)
/* v8 ignore next -- defensive: prune deletes a scope and its deferral
* together, so a deferred id always still owns its record; kept so a
* future teardown path cannot double-dispose. */
if (record !== undefined) {
this.scopes.delete(id)
this.dropScope(id, record)
}
}
}
}

422
packages/client/runtime/src/client/sessions/session.ts
View File

@@ -1,18 +1,19 @@
// Sessions remain resident after creation so they continue consuming mux frames off-screen.
import type { Context } from 'cordis'
import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
import type { SessionEvent, TodoItem } from '@deepseek-ai/dsh-session/types'
import type {
HistoryEntry, IApiClient, MuxFrame, RpcError, RpcId, RpcResult,
SessionId, ToolEventView, WorkspaceId,
SessionId, ToolEventView,
} from '@deepseek-ai/dsh-client-connection/client'
// Value import from the inline-safe wire layer (not the connection plugin):
// plugin-to-plugin value imports are a bundle purity error.
import { transportError } from '@deepseek-ai/dsh-host-apiproxy/api'
import type { ObservableSnapshot } from '../contract/store.ts'
import type {
ComposerPhase, ConversationNode, ConversationSnapshot, OpenState, PendingPrompt,
PromptError, RunningToolCall, SessionIntentSnapshot, SessionIntentTarget,
CodeSubCall, ComposerPhase, ConversationNode, ConversationSnapshot, OpenState,
PromptError, QueuedMessage, RunningToolCall,
} from './conversation.ts'
import type { PendingInteraction } from './pending.ts'
import { PendingWait } from './pending.ts'
@@ -23,10 +24,37 @@ import { PartialAccumulator } from './partial.ts'
/** Messages requested per history page. */
export const PAGE_MESSAGES = 50
/** Optional frontend Intent and publication observer for a Session object. */
/** Manager-owned observers of a Session object's local state edges. */
export interface SessionOptions {
intent?: { target: SessionIntentTarget; prompt: string }
onPublished?(session: Session): void
/**
* First ACCEPTED prompt on a blank session (fires at most once, on the
* prompt RPC's success response): the manager mirrors the blank→false flip
* into its list row so the session surfaces without waiting for a host
* frame. Acceptance is the flip point because it proves the user message
* is in the host log; a rejected first prompt keeps the session blank
* (hidden, still reusable by connectWorkspace).
*/
onEngaged?(session: Session): void
}
/** Queue-row preview cap: the dock renders one line, the full content never leaves the host mirror. */
const QUEUE_PREVIEW_CHARS = 200
/** Internal inbox-mirror entry: the snapshot row plus the retirement-matching fields the frames carry. */
interface QueuedEntry {
row: QueuedMessage
steering: boolean
/** JSON-serialized MessageSource (steering retirement matches by source, the host-mirror precedent). */
sourceJson: string
}
/** Single-line queue-row preview: text blocks flattened, non-text as tags, capped by code point. */
function queuePreviewOf(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
}
/**
@@ -64,10 +92,20 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
private callsCache: { rev: number; value: RunningToolCall[] } | null = null
private pendingRev = 0
private pendingCache: { rev: number; value: PendingInteraction[] } | null = null
/** Inbox mirror (session/queued frames + mux-open baseline). Queue frames never hit history,
* so this is stream-only state: reconnect clears it and the fresh baseline re-populates. */
private queued: QueuedEntry[] = []
private queueRev = 0
private queueCache: { rev: number; value: QueuedMessage[] } | null = null
private frozenRev = 0
private nodesCache: { folded: readonly ConversationNode[]; frozenRev: number; value: readonly ConversationNode[] } | null = null
/** Latest todo/write whole-list snapshot in the window (last write wins on replay). */
private todos: readonly TodoItem[] = []
/** `run_code` sub-dispatches by parent callId (window-derived, like openCalls). Appends
* copy-on-write the per-parent array so published snapshot references never mutate. */
private codeDispatches = new Map<string, readonly CodeSubCall[]>()
private dispatchesRev = 0
private dispatchesCache: { rev: number; value: ReadonlyMap<string, readonly CodeSubCall[]> } | null = null
private running = false
/**
* Sticky send marker, private input of the composerPhase derivation: set
@@ -75,12 +113,10 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
* engaging edge of the phase machine (see ComposerPhase).
*/
private promptAttempted = false
/** Empty-log mirror (see ConversationSnapshot.blank); monotone false once flipped. */
private blankBit = false
private removed = false
private promptError: PromptError | null = null
private intent: SessionIntentSnapshot | null
private pendingPrompt: PendingPrompt | null
private intentGeneration = 0
private published: boolean
private lastAgentError: string | null = null
/** Live events buffered during open/resync and stitched by sequence once history lands. */
private liveBuffer: { event: SessionEvent; view: ToolEventView | undefined }[] = []
@@ -93,27 +129,46 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
private readonly notifier = new Notifier(() => {
this.snapshotCache = this.buildSnapshot()
})
/**
* Agent-scoped cordis context, bound once by SessionsService when it
* mints the scope (the client mirror of the host Agent's loopCtx). The
* Session dispatches its own scoped events through it; undefined means
* unbound (bare object-layer construction) or already pruned — both skip
* dispatch-dependent behavior rather than fail.
*/
private actx: Context | undefined
/**
* @param sessionId - stable identity shared by the frontend Intent and Host entity.
* @param sessionId - Host session identity (client sessions are always Host-born).
* @param api - shared wire client.
* @param options - optional frontend-only initial state and publication observer.
* @param options - optional manager-owned state observers.
*/
constructor(
readonly sessionId: SessionId,
private readonly api: IApiClient,
private readonly options: SessionOptions = {},
) {
this.intent = options.intent === undefined
? null
: { target: options.intent.target, phase: 'ready' }
this.pendingPrompt = options.intent === undefined
? null
: { text: options.intent.prompt, phase: 'editing', retry: 'send' }
this.published = options.intent === undefined
this.snapshotCache = this.buildSnapshot()
}
/**
* Bind the Agent-scoped context minted by SessionsService (single write;
* a second bind is a wiring error and throws). Direction stays one-way at
* the seam: consumers still reach the Session via `sessions.sessionOf`,
* while the Session holds its own dispatch point (host Agent.loopCtx
* mirror).
* @param actx - the agent's scoped context.
*/
bindScope(actx: Context): void {
if (this.actx !== undefined) throw new Error(`session ${this.sessionId} already has a bound scope`)
this.actx = actx
}
/** Release the bound scope at prune time (a later rebind accompanies a freshly minted scope). */
unbindScope(): void {
this.actx = undefined
}
// ---- Operations ----
/**
@@ -139,64 +194,22 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
if (!result.ok) {
this.promptError = { op: 'send', error: result.error }
this.notifier.markDirty()
return result
}
// Blank flips on ACCEPTANCE, not attempt: an accepted prompt has logged
// its user/message on the host (events.length > 0 is fact, not
// optimism), while a rejected first prompt must keep the session blank
// — the client-side blank mirror only ever lowers, so flipping early on
// a failure would surface the session forever and strip its
// connectWorkspace reuse eligibility against the host's authority.
if (this.blankBit) {
this.blankBit = false
this.options.onEngaged?.(this)
this.notifier.markDirty()
}
return result
}
/**
* Update this Session's retained prompt while it remains editable.
* @param text - exact controlled value of this Session's retained prompt.
*/
updatePendingPrompt(text: string): void {
const pending = this.pendingPrompt
if (pending === null || pending.phase === 'sending') return
this.pendingPrompt = { ...pending, text }
this.notifier.notifyNow()
}
/**
* Connect this frontend Session to a real Workspace and flush its retained prompt.
* @param workspaceId - real Workspace target.
*/
connect(workspaceId: WorkspaceId): void {
const intent = this.intent
const pending = this.pendingPrompt
if (intent === null || intent.phase === 'connecting' || pending === null || pending.text.trim() === '') return
const connecting: SessionIntentSnapshot = {
target: { kind: 'workspace', workspaceId },
phase: 'connecting',
}
const queued: PendingPrompt = {
...pending,
phase: 'sending',
retry: 'connect',
workspaceId,
}
delete queued.error
this.intent = connecting
this.pendingPrompt = queued
this.notifier.notifyNow()
void this.flushPendingPrompt()
}
/** Stop a superseded frontend Intent from automatically sending after publication. */
abandonIntent(): void {
if (this.intent === null) return
this.intentGeneration += 1
}
/** Retry this Session's retained prompt from its failed prerequisite. */
retryPendingPrompt(): void {
const pending = this.pendingPrompt
if (pending === null || pending.phase === 'sending' || pending.text.trim() === '') return
const sending: PendingPrompt = { ...pending, phase: 'sending' }
delete sending.error
this.pendingPrompt = sending
this.promptError = null
this.notifier.markDirty()
void this.flushPendingPrompt()
}
/**
* Stop: contract session.cancel 1:1; failures land in promptError (same error-strip display slot).
* @returns the cancel result.
@@ -269,6 +282,11 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
* in-flight open first — its history request rode the dead connection and must not settle
* the fresh generation into 'error' (audit S4). */
async resync(): Promise<void> {
// The queue mirror is NOT cleared here: onConnected (which drives resync)
// races the mux frames — the fresh generation's baseline may have landed
// already, and the host never resends it. The mirror re-baselines on the
// session/subscribed frame instead (same stream as the queue snapshot
// that follows it, so ordering is guaranteed).
if (this.openState === 'cold') return // never opened: no window to rebuild (doOpen flips to 'loading' synchronously, so cold implies no in-flight open)
this.openGeneration++
this.openPromise = null
@@ -317,12 +335,35 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
handleMuxEnvelope(rpcId: RpcId, frame: MuxFrame): void {
switch (frame.type) {
case 'session/event': {
this.retireQueued(frame.event)
this.acceptLiveEvent(frame.event, frame.view)
return
}
case 'session/queued': {
// Row key: the enqueueing prompt's rpcId when it rode this wire (the
// provisional-echo reconciliation key); otherwise the frame envelope id.
const key = 'rpcId' in frame.source ? String(frame.source.rpcId) : `f:${rpcId}`
this.queued.push({
row: { key, preview: queuePreviewOf(frame.content) },
steering: frame.steering,
sourceJson: JSON.stringify(frame.source),
})
this.queueRev++
this.notifier.markDirty()
return
}
case 'session/subscribed': {
this.subscribedLastSeq = frame.lastSeq
return // pure baseline bookkeeping, no visible change
// New mux-generation baseline: the host pushes this session's queue
// snapshot AFTER the subscribed frame on the same stream, so the
// stale mirror clears here — race-free against onConnected/resync
// timing (clearing there could wipe a baseline that already landed).
if (this.queued.length > 0) {
this.queued = []
this.queueRev++
this.notifier.markDirty()
}
return
}
case 'approval/requested': {
const { type: _type, sessionId: _sid, ...payload } = frame
@@ -359,14 +400,38 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
* @param running - the new running state.
*/
handleRunning(running: boolean): void {
// Leave-running sweep (host queuedMirror precedent): discard paths (cancel,
// terminal steering drop) have no per-entry frame, so ANY not-running signal
// with a nonempty mirror clears it — checked before the equality return so a
// stale replay on an already-idle session still sweeps.
if (!running && this.queued.length > 0) {
this.queued = []
this.queueRev++
this.notifier.markDirty()
}
// Turn-start conversion: a blank session never runs, so the first
// running:true proves another端's first message landed (设计稿 2.2).
if (running && this.blankBit) {
this.blankBit = false
this.notifier.markDirty()
}
if (this.running === running) return
this.running = running
this.notifier.markDirty()
}
/** Mark that Host publication is known without resolving an uncertain local create response. */
handlePublished(): void {
this.markPublished()
/**
* Blank-bit relay from the authoritative summary source (list baseline and
* the session-added frame). Monotone: once any signal (local first send,
* running flip, an earlier summary) cleared it, a stale true never
* re-blanks.
* @param blank - the summary's derived empty-log bit.
*/
handleBlank(blank: boolean): void {
if (blank === this.blankBit) return
if (blank && (this.promptAttempted || this.running)) return
this.blankBit = blank
this.notifier.markDirty()
}
/** host/session-removed relay: flag the snapshot (instance survives — resident-instance rule). */
@@ -402,112 +467,6 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
this.pendingRev++
}
/** Advance the retained prompt through Session attachment and submission. */
private async flushPendingPrompt(): Promise<void> {
const pending = this.pendingPrompt
if (pending?.phase === 'sending') {
const ready = pending.retry === 'connect'
? await this.attachPendingPrompt(pending)
: pending
if (ready !== null) await this.sendPendingPrompt(ready)
}
}
/** Complete the Host Session prerequisite and return the prompt's send step. */
private async attachPendingPrompt(pending: PendingPrompt): Promise<PendingPrompt | null> {
const workspaceId = pending.workspaceId
if (workspaceId === undefined) throw new Error('a Session attachment requires a Workspace id')
const originIntent = this.intent
const originGeneration = this.intentGeneration
let result: RpcResult<{ sessionId: SessionId }>
try {
result = (await this.api.sessions.create({ sessionId: this.sessionId, workspaceId })).result
} catch (error) {
result = transportError(error)
}
let ready: PendingPrompt | null = null
if (result.ok) {
ready = this.completePendingAttachment(pending, originIntent, originGeneration)
} else {
this.failPendingAttachment(pending, originIntent, originGeneration, result.error)
}
this.notifier.markDirty()
return ready
}
/** Move a published Session to the send step unless its page intent was superseded. */
private completePendingAttachment(
pending: PendingPrompt,
originIntent: SessionIntentSnapshot | null,
originGeneration: number,
): PendingPrompt | null {
this.markPublished()
this.intent = null
this.promptAttempted = true
const superseded = originIntent !== null && originGeneration !== this.intentGeneration
const next: PendingPrompt = {
...pending,
phase: superseded ? 'failed' : 'sending',
retry: 'send',
...(superseded ? { error: 'Message was not sent because you navigated away.' } : {}),
}
if (!superseded) delete next.error
this.pendingPrompt = next
return superseded ? null : next
}
/** Retain the prompt at the failed attachment step that owns the retry. */
private failPendingAttachment(
pending: PendingPrompt,
originIntent: SessionIntentSnapshot | null,
originGeneration: number,
error: RpcError,
): void {
const partiallyPublished = error.code === 'workspace-attach-failed'
if (partiallyPublished) {
this.markPublished()
this.intent = null
this.promptAttempted = true
}
const activeIntent = !partiallyPublished
&& originIntent !== null
&& originGeneration === this.intentGeneration
&& this.intent === originIntent
if (activeIntent) {
this.intent = {
target: originIntent.target,
phase: 'ready',
error: { step: 'session', message: rpcErrorMessage(error) },
}
this.pendingPrompt = { ...pending, phase: 'editing' }
}
if (!activeIntent && (partiallyPublished || originIntent === null) && this.pendingPrompt === pending) {
this.pendingPrompt = { ...pending, phase: 'failed', error: rpcErrorMessage(error) }
}
}
/** Submit the retained prompt and keep it only when Host rejects the send. */
private async sendPendingPrompt(pending: PendingPrompt): Promise<void> {
const result = await this.prompt([{ type: 'text', text: pending.text.trim() }], 'queue')
if (this.pendingPrompt === pending) {
this.pendingPrompt = result.ok
? null
: {
...pending,
retry: 'send',
phase: 'failed',
error: rpcErrorMessage(result.error),
}
this.notifier.markDirty()
}
}
private markPublished(): void {
if (this.published) return
this.published = true
this.options.onPublished?.(this)
}
/** @param generation - openGeneration at launch; every await re-checks it and a stale pass
* drops all writes (resync superseded this open — its outcome belongs to a dead connection). */
private async doOpen(generation: number): Promise<void> {
@@ -614,9 +573,89 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
}
}
/** Consumption-event retirement, mirroring the host queuedMirror rules: a message-triggered
* turn/start claims the oldest non-steering entry; a steering/message drains the oldest
* steering entry with the same source (loop-authored steering matches nothing and drops none). */
private retireQueued(event: SessionEvent): void {
if (this.queued.length === 0) return
let index = -1
if (event.type === 'turn/start') {
if (event.data.trigger.kind !== 'message') return
index = this.queued.findIndex(entry => !entry.steering)
} else if (event.type === 'steering/message') {
const source = JSON.stringify(event.data.source)
index = this.queued.findIndex(entry => entry.steering && entry.sourceJson === source)
} else {
return
}
if (index < 0) return
this.queued.splice(index, 1)
this.queueRev++
this.notifier.markDirty()
}
/** Per-event side effects (right column of the §A.9 dispatch table):
* chunk accumulation / partial clear on finalize / openCalls add-remove. */
private applyEventSideEffects(event: SessionEvent, view?: ToolEventView): void {
// The `tool/code-dispatch-start`/`tool/code-dispatch` pair is declared by
// the host-side dsh-tools plugin whose types cannot enter the client
// program (its host Context merges collide with the client's), so this
// wire consumer narrows them structurally — the same posture as every
// other cross-wire event payload.
if ((event.type as string) === 'tool/code-dispatch-start') {
// A started sub-dispatch enters the index as a RunningToolCall — the
// exact shape a native in-flight call renders from — under its parent
// run_code callId; it never joins the surface flow.
const data = event.data as unknown as {
parentCallId: string
subCallId: string
name: string
arguments: unknown
}
const running: CodeSubCall = {
callId: data.subCallId, name: data.name,
argsRaw: JSON.stringify(data.arguments),
turn: 0, step: 0, time: event.time, callView: null,
}
const siblings = this.codeDispatches.get(data.parentCallId) ?? []
this.codeDispatches.set(data.parentCallId, [...siblings, running])
this.dispatchesRev++
return
}
if ((event.type as string) === 'tool/code-dispatch') {
// Settlement replaces the running entry in place (same array position,
// so parallel sub-calls keep their start order) with the
// ToolResultNode form; a settle with no observed start (history window
// cut mid-pair, or a pre-start-event log) appends directly.
const data = event.data as unknown as {
parentCallId: string
subCallId: string
name: string
arguments: unknown
isError: boolean
content: ContentBlock[]
}
const siblings = this.codeDispatches.get(data.parentCallId) ?? []
const at = siblings.findIndex(sub => sub.callId === data.subCallId)
const started = at === -1 ? undefined : siblings[at]
const settled: CodeSubCall = {
kind: 'tool-result', seq: event.seq, time: event.time,
callId: data.subCallId,
call: { name: data.name, argsRaw: JSON.stringify(data.arguments) },
// Duration source: the paired start's time when observed; null =
// unknown (settle-only window), matching the native tool-result
// contract so views never present a fabricated zero duration.
callTime: started === undefined ? null : started.time,
content: data.content, isError: data.isError,
callView: null, resultView: null,
}
this.codeDispatches.set(
data.parentCallId,
at === -1 ? [...siblings, settled] : siblings.map((sub, index) => (index === at ? settled : sub)),
)
this.dispatchesRev++
return
}
switch (event.type) {
case 'assistant/chunk': {
const { turn, step, chunk } = event.data
@@ -703,6 +742,8 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
this.callsRev++
this.frozenNodes = []
this.frozenRev++
this.codeDispatches = new Map()
this.dispatchesRev++
for (let i = 0; i < this.events.length; i++) {
const event = this.events[i]
/* v8 ignore next -- dense-array guard: i stays within events.length, so the undefined arm needs a sparse array no caller builds. */
@@ -735,6 +776,12 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
if (this.pendingCache === null || this.pendingCache.rev !== this.pendingRev) {
this.pendingCache = { rev: this.pendingRev, value: [...this.pending.values()] }
}
if (this.dispatchesCache === null || this.dispatchesCache.rev !== this.dispatchesRev) {
this.dispatchesCache = { rev: this.dispatchesRev, value: new Map(this.codeDispatches) }
}
if (this.queueCache === null || this.queueCache.rev !== this.queueRev) {
this.queueCache = { rev: this.queueRev, value: this.queued.map(entry => entry.row) }
}
const partial = this.partial?.toPartial() ?? null
return {
sessionId: this.sessionId,
@@ -743,6 +790,8 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
partial,
runningCalls: this.callsCache.value,
pending: this.pendingCache.value,
codeDispatches: this.dispatchesCache.value,
queue: this.queueCache.value,
running: this.running,
composerPhase: derivePhase(
nodes.length > 0 || partial !== null || this.running || this.pendingCache.value.length > 0,
@@ -754,18 +803,13 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
hasMore: this.hasMore,
loadingOlder: this.loadingOlder,
promptError: this.promptError,
intent: this.intent,
pendingPrompt: this.pendingPrompt,
blank: this.blankBit,
lastAgentError: this.lastAgentError,
todos: this.todos,
}
}
}
function rpcErrorMessage(error: RpcError): string {
return `${error.code}: ${error.message}`
}
/**
* The composerPhase judgment — the single site that knows the predicate
* (consumers switch on the result, never re-derive). Monotone per session

11
packages/client/runtime/src/client/slots.ts
View File

@@ -235,7 +235,7 @@ export class SlotsService extends Service {
}
}
/** Build once after both object-layer services mount; session cells still resolve lazily. */
/** Build once after both object-layer services mount; per-session provide bundles still resolve lazily. */
private hostFace(): SlotRendererHost {
if (this._host !== undefined) return this._host
const sessions = this.ctx.get('sessions')
@@ -264,7 +264,8 @@ export class SlotsService extends Service {
sessions: {
list: sessions.list,
current,
cell: id => sessions.cell(id),
provideInfo: id => sessions.provideInfo(id),
maybeProvideInfo: id => sessions.maybeProvideInfo(id),
},
workspaces: { list: workspaces.list },
}
@@ -275,13 +276,13 @@ export class SlotsService extends Service {
private resolveStore(handle: EngineStoreHandle, sessionId: string | undefined): StoreInstanceLike {
const record = this._stores.get(handle)
if (record === undefined) throw new Error('store handle is not registered (entry unloaded, or the handle never went through register)')
const key = record.scope === 'session' ? sessionId : ROOT_INSTANCE_KEY
if (key === undefined) throw new Error('session-scoped store resolution requires a session id')
const key = record.scope === 'root' ? ROOT_INSTANCE_KEY : sessionId
if (key === undefined) throw new Error(`${record.scope} store resolution requires a session id`)
let instance = record.instances.get(key)
if (instance === undefined) {
// Session instances get the scope key (the engine suffixes the persist
// key per session); root instances stay keyless.
instance = record.scope === 'session' ? handle.create(key) : handle.create()
instance = record.scope === 'root' ? handle.create() : handle.create(key)
record.instances.set(key, instance)
}
return instance

48
packages/client/runtime/src/client/workspaces/manager.ts
View File

@@ -6,11 +6,7 @@ import type {
import { transportError } from '@deepseek-ai/dsh-host-apiproxy/api'
import { mergeOrderedBaseline } from '../ordered-baseline.ts'
import { Notifier } from '../sessions/notifier.ts'
import {
Workspace, type WorkspaceCreateInput, type WorkspaceIntentSnapshot,
} from './workspace.ts'
export type { WorkspaceIntentSnapshot } from './workspace.ts'
import { Workspace, type WorkspaceCreateInput } from './workspace.ts'
/** Monotone workspace-list arrival lifecycle. */
export type WorkspaceListPhase = 'pending' | 'ready'
@@ -18,8 +14,6 @@ export type WorkspaceListPhase = 'pending' | 'ready'
/** Immutable workspace-list snapshot. */
export interface WorkspaceListSnapshot {
items: readonly WorkspaceView[]
/** The sole page-local Workspace intent; never persisted or sent over the Host stream. */
intent: WorkspaceIntentSnapshot | undefined
state: 'idle' | 'loading' | 'error'
phase: WorkspaceListPhase
error: RpcError | null
@@ -28,7 +22,6 @@ export interface WorkspaceListSnapshot {
/** Workspace object cluster driven by one list baseline and changed-frame upserts. */
export class WorkspaceManager {
private items: Workspace[] = []
private intent: Workspace | undefined
private itemViewsSource: readonly Workspace[] | null = null
private itemViewsCache: readonly WorkspaceView[] = []
private state: WorkspaceListSnapshot['state'] = 'idle'
@@ -46,44 +39,6 @@ export class WorkspaceManager {
this.snapshotCache = this.buildSnapshot()
}
/**
* Replace the current client-local Workspace intent object.
* @param name - directory/display name used if the intent is materialized.
* @returns the new intent snapshot.
*/
startIntent(name = 'workspace'): WorkspaceIntentSnapshot {
this.intent = new Workspace(this.api, { name })
this.notifier.notifyNow()
return this.intent.getSnapshot().intent as WorkspaceIntentSnapshot
}
/** Discard the current client-local Workspace intent. */
discardIntent(): void {
if (this.intent === undefined) return
this.intent = undefined
this.notifier.notifyNow()
}
/**
* Materialize the current Workspace intent through the ordinary Host create seam.
* A superseded intent is never cleared by an older completion.
* @returns the Host create result, or undefined when no intent exists.
*/
async materializeIntent(): Promise<RpcResult<{ workspace: WorkspaceView; created: boolean }> | undefined> {
const intent = this.intent
if (intent?.getSnapshot().intent?.phase !== 'ready') return undefined
const completion = intent.materialize()
if (completion === undefined) return undefined
this.notifier.notifyNow()
const result = await completion
if (result.ok) {
this.upsert(result.value.workspace, intent)
if (this.intent === intent) this.intent = undefined
}
this.notifier.markDirty()
return result
}
/**
* Refresh from workspace.list. The first successful response establishes
* Host order; later responses update membership and values without moving
@@ -212,7 +167,6 @@ export class WorkspaceManager {
private buildSnapshot(): WorkspaceListSnapshot {
return {
items: this.itemViews(),
intent: this.intent?.getSnapshot().intent,
state: this.state,
phase: this.phase,
error: this.error,

155
packages/client/runtime/src/client/workspaces/service.ts
View File

@@ -7,13 +7,11 @@ import type {
import type { SnapshotStore } from '../contract/store.ts'
import { createSnapshotStore } from '../contract/store.ts'
import type { SessionsService } from '../sessions/service.ts'
import { WorkspaceManager, type WorkspaceIntentSnapshot, type WorkspaceListPhase } from './manager.ts'
import { WorkspaceManager, type WorkspaceListPhase } from './manager.ts'
/** Workspace list plus the two-baseline readiness and default-target projection. */
export interface WorkspaceListState {
items: readonly WorkspaceView[]
/** Sole client-local Workspace projection; its state remains owned by Workspace. */
intent: WorkspaceIntentSnapshot | undefined
state: 'idle' | 'loading' | 'error'
phase: WorkspaceListPhase
error: RpcError | null
@@ -29,64 +27,128 @@ export class WorkspacesService {
readonly list: SnapshotStore<WorkspaceListState>
/** Workspace baseline and frame owner. */
private readonly manager: WorkspaceManager
private initialSessionResolved = false
private composingIntent = false
/** In-flight blank-session creates keyed by workspace (connectWorkspace coalescing). */
private readonly connecting = new Map<WorkspaceId, Promise<SessionId>>()
/** Guards the runtime-owned one-shot initial-selection subscription. */
private initialSelectionStarted = false
/**
* @param ctx - client root context.
* @param api - shared wire client.
* @param sessions - lower-level Session service used for recency and cross-domain intent orchestration.
* @param sessions - lower-level Session service used for recency and blank-session reuse.
*/
constructor(ctx: Context, api: IApiClient, private readonly sessions: SessionsService) {
this.manager = new WorkspaceManager(api)
this.list = createSnapshotStore<WorkspaceListState>({
items: [], intent: undefined, state: 'idle', phase: 'pending', error: null,
items: [], state: 'idle', phase: 'pending', error: null,
baselinesReady: false, recentWorkspaceId: undefined,
})
this.manager.subscribe(() => { if (!this.composingIntent) this.project() })
this.sessions.list.subscribe(() => { if (!this.composingIntent) this.project() })
this.manager.subscribe(() => { this.project() })
this.sessions.list.subscribe(() => { this.project() })
ctx.reflect.provide('workspaces', this, undefined)
}
/**
* Start the sole Session intent, resolving the default Workspace here.
* @param workspaceId - optional explicit real Workspace target.
* @param prompt - optional prompt retained while retargeting.
* Resolve the session a New Session flow lands in once this Workspace is
* chosen: reuse the workspace's existing blank session when one is in the
* list mirror, else create a fresh one on the host (`session.create` births
* the full Session+Agent — the client holds no intermediate state). The
* caller owns navigation: take the returned id to `sessions.open`.
* Resolution guarantee (both arms): the returned id is already in the list
* store and `sessions.binding(id)` resolves synchronously — draft hand-off
* may write the new scope's machine before opening.
* @param workspaceId - chosen Workspace (must be in the workspace list).
* @returns the reused or newly created session id.
*/
startSession(workspaceId?: WorkspaceId, prompt = ''): void {
const snapshot = this.list.getSnapshot()
const resolved = workspaceId ?? snapshot.recentWorkspaceId ?? snapshot.items[0]?.workspaceId
this.composingIntent = true
try {
if (resolved === undefined) {
this.manager.startIntent()
this.sessions.startIntent({ kind: 'workspace-intent' }, prompt)
} else {
this.manager.discardIntent()
this.sessions.startIntent({ kind: 'workspace', workspaceId: resolved }, prompt)
async connectWorkspace(workspaceId: WorkspaceId): Promise<SessionId> {
const workspace = this.list.getSnapshot().items.find(item => item.workspaceId === workspaceId)
if (workspace === undefined) throw new Error(`workspaces.connectWorkspace: unknown workspace ${workspaceId}`)
// Coalesce concurrent connects: a create's summary lands without cwd
// until the host frame arrives, so a second call inside that window
// would miss the reuse scan and mint another hidden blank session.
const inflight = this.connecting.get(workspaceId)
if (inflight !== undefined) return inflight
// Reuse: blank && same canonical cwd (workspace.path is the host realpath
// canon; summary cwd is the session header passthrough of the same canon).
const sessions = this.sessions.list.getSnapshot()
for (const id of sessions.ids) {
const summary = sessions.byId[id]
if (summary !== undefined && summary.blank && summary.cwd === workspace.path) return summary.id
}
const attempt = this.sessions.create({ workspaceId })
.finally(() => { this.connecting.delete(workspaceId) })
this.connecting.set(workspaceId, attempt)
return attempt
}
/**
* Follow the first complete Workspace/Session baseline and select a default
* session exactly once. A restored current session wins; otherwise the most
* recent Workspace is connected (reusing or creating its blank session).
* Later explicit clears stay cleared instead of retriggering this startup
* policy. A failed connect may retry on the next baseline projection.
* @returns disposer for the baseline subscription; late work cannot navigate after disposal.
*/
startInitialSelection(): () => void {
if (this.initialSelectionStarted) {
throw new Error('workspaces.startInitialSelection: already started')
}
this.initialSelectionStarted = true
let state: 'waiting' | 'connecting' | 'done' = 'waiting'
let disposed = false
const reconcile = (): void => {
if (disposed || state !== 'waiting') return
const workspace = this.list.getSnapshot()
if (!workspace.baselinesReady) return
const current = this.sessions.list.getSnapshot().current
const target = workspace.recentWorkspaceId
if (current !== undefined || target === undefined) {
state = 'done'
return
}
} finally {
this.composingIntent = false
this.project()
state = 'connecting'
void this.connectWorkspace(target).then(
(sessionId) => {
if (disposed) return
if (this.sessions.list.getSnapshot().current === undefined) {
this.sessions.open(sessionId)
}
state = 'done'
},
(reason: unknown) => {
if (disposed) return
state = 'waiting'
console.warn('initial workspace selection failed:', reason)
},
)
}
const unsubscribe = this.list.subscribe(reconcile)
reconcile()
return () => {
disposed = true
unsubscribe()
}
}
/** Connect the current frontend Workspace and Session, then flush the Session-owned prompt. */
sendSession(): void {
const session = this.sessions.intent()
const target = session?.getSnapshot().intent?.target
if (session === undefined || target === undefined) return
if (target.kind === 'workspace') {
session.connect(target.workspaceId)
/**
* The shared New Session action behind the shell entry points (sidebar
* button, workspace browser): resolve the target Workspace — explicit wins,
* else the recent-Workspace projection — connect its blank session and
* navigate there; with no Workspace at all, clear the selection into the
* New Session view state. Connect failures are non-fatal (console
* diagnostics; the current view stays usable).
* @param workspaceId - explicit target Workspace for scoped actions.
*/
startSession(workspaceId?: WorkspaceId): void {
const target = workspaceId ?? this.list.getSnapshot().recentWorkspaceId
if (target === undefined) {
this.sessions.clear()
return
}
if (session.getSnapshot().pendingPrompt?.text.trim() === '') return
void this.manager.materializeIntent().then((result) => {
if (this.sessions.intent() !== session) return
if (result?.ok) {
session.connect(result.value.workspace.workspaceId)
}
})
void this.connectWorkspace(target).then(
(sessionId) => { this.sessions.open(sessionId) },
(reason: unknown) => { console.warn('new session failed:', reason) },
)
}
/**
@@ -153,20 +215,15 @@ export class WorkspacesService {
private project(): void {
const workspace = this.manager.getSnapshot()
const sessions = this.sessions.list.getSnapshot()
if (workspace.intent !== undefined && sessions.intent?.target.kind !== 'workspace-intent') {
this.manager.discardIntent()
return
}
const baselinesReady = workspace.phase === 'ready' && sessions.phase === 'ready'
this.list.set({
...workspace,
items: workspace.items,
state: workspace.state,
phase: workspace.phase,
error: workspace.error,
baselinesReady,
recentWorkspaceId: baselinesReady ? recentWorkspace(workspace.items, sessions.byId) : undefined,
})
if (!this.initialSessionResolved && baselinesReady) {
this.initialSessionResolved = true
if (sessions.current === undefined && sessions.intent === undefined) this.startSession()
}
}
}